diff --git a/apps/api/routers/leads.py b/apps/api/routers/leads.py index 9e5ae26..e97d6d9 100644 --- a/apps/api/routers/leads.py +++ b/apps/api/routers/leads.py @@ -1399,15 +1399,14 @@ async def tech_stack_endpoint( @router.post("/leads/domain-intel") async def domain_intelligence(body: dict, ctx: WorkspaceCtx = Depends(require_editor)): """Analyze a domain — RDAP registration, DNS, hosting, email provider, legitimacy score.""" - domain = body.get("domain", "").strip() - if not domain: - raise HTTPException(400, "domain is required") + from apps.api.services.leadgen.enrichment.domain_intel import clean_domain - # Strip protocol/path if provided as URL - if "://" in domain: - from urllib.parse import urlparse - domain = urlparse(domain).hostname or domain - domain = domain.replace("www.", "").strip("/") + raw_domain = body.get("domain") + if not isinstance(raw_domain, str) or not raw_domain.strip(): + raise HTTPException(400, "domain is required") + domain = clean_domain(raw_domain) + if not domain: + raise HTTPException(400, "domain must be a valid domain name or URL") from apps.api.services.leadgen.enrichment.domain_intel import analyze_domain result = await analyze_domain(domain) diff --git a/apps/api/services/leadgen/enrichment/domain_intel.py b/apps/api/services/leadgen/enrichment/domain_intel.py index 06ab4af..ec0deab 100644 --- a/apps/api/services/leadgen/enrichment/domain_intel.py +++ b/apps/api/services/leadgen/enrichment/domain_intel.py @@ -14,8 +14,10 @@ import asyncio import logging +import re from typing import Dict, Optional from datetime import datetime +from urllib.parse import urlparse logger = logging.getLogger("leadgen.domain_intel") @@ -56,6 +58,40 @@ } +_HOSTNAME = re.compile( + r"^(?=.{1,253}$)[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?" + r"(?:\.[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?)+$" +) + + +def clean_domain(value: object) -> str: + """The bare, lowercase hostname of a domain or URL, or "" when it is not one. + + The result goes into RDAP and crt.sh request URLs and is compared with + lowercase certificate names, so it must not carry a scheme, path, query, + port, credentials, case, a trailing dot or a leading ``www.``. Only that + prefix is removed, never a ``www.`` elsewhere in the name. + """ + if not isinstance(value, str): + return "" + text = value.strip() + if not text: + return "" + try: + url = text if "://" in text or text.startswith("//") else "//" + text + host = urlparse(url).hostname or "" + except ValueError: + return "" + host = host.strip(".").lower() + if host.startswith("www."): + host = host[4:] + try: + host = host.encode("idna").decode("ascii") # bücher.de -> xn--bcher-kva.de + except UnicodeError: + return "" + return host if _HOSTNAME.match(host) else "" + + async def get_rdap_info(domain: str) -> Dict: """Query RDAP for domain registration data (free, no auth).""" import aiohttp diff --git a/tests/test_domain_intel_route.py b/tests/test_domain_intel_route.py new file mode 100644 index 0000000..807a5d3 --- /dev/null +++ b/tests/test_domain_intel_route.py @@ -0,0 +1,112 @@ +"""POST /api/leads/domain-intel analyzes a clean, lowercase hostname.""" + +from types import SimpleNamespace + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from apps.api.routers import leads +from apps.api.services.leadgen.enrichment import domain_intel +from apps.api.services.leadgen.enrichment.domain_intel import clean_domain + + +_MALFORMED_URLS = [ + "https://[", + "[example.com", + "https://example.com]", + "//[not-an-ip]", + "https://example.com\uff0fpath", + "example.com\uff1a443", + "//example.com\uff20evil.com", +] + + +@pytest.fixture +def analyzed(monkeypatch): + seen = [] + + async def fake_analyze(domain): + seen.append(domain) + return {"domain": domain} + + monkeypatch.setattr(domain_intel, "analyze_domain", fake_analyze) + return seen + + +@pytest.fixture +def client(): + ctx = SimpleNamespace(workspace_id="ws-fixture") + app = FastAPI() + app.include_router(leads.router) + app.dependency_overrides[leads.require_editor] = lambda: ctx + with TestClient(app) as http: + yield http + + +@pytest.mark.parametrize("raw,expected", [ + ("example.com", "example.com"), + ("Example.COM", "example.com"), + ("//example.com", "example.com"), + ("//WWW.Example.com:8443/pricing?x=1#top", "example.com"), + (" www.Example.com. ", "example.com"), + ("https://WWW.Example.com:8443/pricing?x=1#top", "example.com"), + ("example.com/careers", "example.com"), + ("user:pw@example.com", "example.com"), + # Only a leading www. goes; a www. inside the name is part of it. + ("awww.example.com", "awww.example.com"), + ("blog.www.example.com", "blog.www.example.com"), + ("wwwexample.com", "wwwexample.com"), + ("bücher.de", "xn--bcher-kva.de"), +]) +def test_clean_domain_keeps_only_the_hostname(raw, expected): + assert clean_domain(raw) == expected + + +@pytest.mark.parametrize("raw", [ + "", " ", "localhost", "exa mple.com", "example.com&exclude=none", + "example..com", "-example.com", "https://", 123, None, ["example.com"], +]) +def test_clean_domain_rejects_what_is_not_a_hostname(raw): + assert clean_domain(raw) == "" + + +@pytest.mark.parametrize("raw", [ + "https://WWW.Example.com/about", + "//example.com", + "//WWW.Example.com:8443/pricing?x=1#top", +]) +def test_route_analyzes_the_cleaned_domain(client, analyzed, raw): + response = client.post("/api/leads/domain-intel", json={"domain": raw}) + + assert response.status_code == 200, response.text + assert analyzed == ["example.com"] + + +def test_route_keeps_a_www_inside_the_name(client, analyzed): + assert client.post("/api/leads/domain-intel", json={"domain": "awww.example.com"}).status_code == 200 + assert analyzed == ["awww.example.com"] + + +@pytest.mark.parametrize("body", [ + {}, {"domain": ""}, {"domain": 42}, {"domain": ["a.com"]}, + {"domain": "example.com&exclude=none"}, {"domain": "not a domain"}, +]) +def test_route_rejects_input_that_is_not_a_domain(client, analyzed, body): + response = client.post("/api/leads/domain-intel", json=body) + + assert response.status_code == 400, response.text + assert analyzed == [] + + +@pytest.mark.parametrize("raw", _MALFORMED_URLS) +def test_clean_domain_rejects_malformed_urls_without_raising(raw): + assert clean_domain(raw) == "" + + +@pytest.mark.parametrize("raw", _MALFORMED_URLS) +def test_route_rejects_malformed_urls_before_analysis(client, analyzed, raw): + response = client.post("/api/leads/domain-intel", json={"domain": raw}) + + assert response.status_code == 400, response.text + assert analyzed == []