fix(leads): complete domain-intel input validation

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)

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

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 == []