feat!: validate brand domains in signing authorization

src/adcp/signing/__init__.py

@@ -196,9 +196,13 @@
     SignatureVerificationError,
 )
 from adcp.signing.etld import (
+    BrandDomainValidationError,
     host_from,
+    is_development_brand_domain,
     registrable_domain,
+    same_development_brand_domain,
     same_registrable_domain,
+    validate_brand_domain,
 )
 from adcp.signing.ip_pinned_transport import (
     AsyncIpPinnedTransport,
@@ -431,6 +435,7 @@ def __init__(self, *args: object, **kwargs: object) -> None:
     "build_capability_cache_key",
     "build_ip_pinned_transport",
     "build_signature_base",
+    "BrandDomainValidationError",
     "canonicalize_authority",
     "canonicalize_target_uri",
     "check_key_origin_consistency",
@@ -445,6 +450,7 @@ def __init__(self, *args: object, **kwargs: object) -> None:
     "format_signature_header",
     "generate_signing_keypair",
     "host_from",
+    "is_development_brand_domain",
     "install_signing_event_hook",
     "load_private_key_pem",
     "operation_needs_signing",
@@ -458,13 +464,15 @@ def __init__(self, *args: object, **kwargs: object) -> None:
     "validate_resolved_ip",
     "validate_uri_static",
     "same_registrable_domain",
+    "same_development_brand_domain",
     "sign_request",
     "sign_signature_base",
     "signing_profile_for_adcp_version",
     "sign_standard_webhook",
     "signing_operation",
     "unauthorized_response_headers",
     "validate_jwks_uri",
+    "validate_brand_domain",
     "verify_detached_jws",
     "verify_flask_request",
     "verify_from_agent_url",

src/adcp/signing/brand_authz.py

@@ -53,7 +53,13 @@
     _BrandJsonSnapshot,
     _ClientFactory,
 )
-from adcp.signing.etld import host_from, registrable_domain, same_registrable_domain
+from adcp.signing.etld import (
+    BrandDomainValidationError,
+    registrable_domain,
+    same_development_brand_domain,
+    same_registrable_domain,
+    validate_brand_domain,
+)
 
 #: Reason a brand-authorization check resolved the way it did. Used
 #: for verifier error attribution and adopter logging. The framework
@@ -149,13 +155,14 @@ def __init__(
         max_redirects: int = DEFAULT_MAX_REDIRECTS,
         max_body_bytes: int = DEFAULT_MAX_BRAND_JSON_BYTES,
         allow_private_destinations: bool = False,
+        allow_development_domains: bool = False,
         timeout_seconds: float = DEFAULT_BRAND_JSON_TIMEOUT_SECONDS,
         clock: Callable[[], float] | None = None,
         _client_factory: _ClientFactory | None = None,
         _fetcher: _BrandJsonFetcher | None = None,
     ) -> None:
         self._clock = clock or time.time
-        self._allow_private = allow_private_destinations
+        self._allow_development_domains = allow_development_domains
         self._fetcher = _fetcher or _BrandJsonFetcher(
             brand_json_url,
             min_cooldown_seconds=min_cooldown_seconds,
@@ -207,10 +214,11 @@ async def check(
         # blank / IP-literal brand domain match anything via shared
         # binding semantics downstream.
         try:
-            brand_host = host_from(brand_domain)
-        except ValueError:
-            return BrandAuthorizationResult(False, reason="brand_domain_invalid")
-        if registrable_domain(brand_host) is None:
+            brand_host = validate_brand_domain(
+                brand_domain,
+                allow_development_domains=self._allow_development_domains,
+            )
+        except BrandDomainValidationError:
             return BrandAuthorizationResult(False, reason="brand_domain_invalid")
 
         snap = await self._snapshot()
@@ -255,7 +263,9 @@ async def check(
         matched = listing[0]
 
         # Step 2a: eTLD+1 binding.
-        if same_registrable_domain(agent_url, brand_host):
+        if same_registrable_domain(agent_url, brand_host) or (
+            self._allow_development_domains and same_development_brand_domain(agent_url, brand_host)
+        ):
             return BrandAuthorizationResult(
                 True,
                 reason="etld1_match",
@@ -323,6 +333,7 @@ def build_brand_json_resolvers(
     max_redirects: int = DEFAULT_MAX_REDIRECTS,
     max_body_bytes: int = DEFAULT_MAX_BRAND_JSON_BYTES,
     allow_private_destinations: bool = False,
+    allow_development_domains: bool = False,
     timeout_seconds: float = DEFAULT_BRAND_JSON_TIMEOUT_SECONDS,
     clock: Callable[[], float] | None = None,
 ) -> tuple[BrandJsonJwksResolver, BrandJsonAuthorizationResolver]:
@@ -371,6 +382,7 @@ def build_brand_json_resolvers(
         max_redirects=max_redirects,
         max_body_bytes=max_body_bytes,
         allow_private_destinations=allow_private_destinations,
+        allow_development_domains=allow_development_domains,
         timeout_seconds=timeout_seconds,
         clock=clock,
         _fetcher=fetcher,

src/adcp/signing/etld.py

@@ -37,6 +37,7 @@
 
 from __future__ import annotations
 
+import re
 from functools import lru_cache
 from urllib.parse import urlsplit
 
@@ -45,6 +46,120 @@
 
 from ._idna_canonicalize import canonicalize_host
 
+_DOTTED_WIRE_DOMAIN_RE = re.compile(
+    r"^[a-z0-9]([a-z0-9-]*[a-z0-9])?(\.[a-z0-9]([a-z0-9-]*[a-z0-9])?)+$"
+)
+_DEVELOPMENT_SUFFIXES = ("localhost", "test", "example", "invalid")
+_DEVELOPMENT_EXACT_NAMES = frozenset({"example.com", "example.net", "example.org"})
+_SPECIAL_USE_SUFFIXES = (
+    "alt",
+    "6tisch.arpa",
+    "eap.arpa",
+    "eap-noob.arpa",
+    "home.arpa",
+    "in-addr.arpa",
+    "ip6.arpa",
+    "ipv4only.arpa",
+    "resolver.arpa",
+    "service.arpa",
+    "example",
+    "example.com",
+    "example.net",
+    "example.org",
+    "invalid",
+    "local",
+    "localhost",
+    "onion",
+    "test",
+)
+
+
+class BrandDomainValidationError(ValueError):
+    """A BrandRef/BrandKey domain is not valid for the selected runtime."""
+
+    def __init__(self, code: str, domain: str) -> None:
+        self.code = code
+        self.domain = domain
+        messages = {
+            "invalid_syntax": "brand domain must be a bare, lowercase-compatible dotted hostname",
+            "not_registrable": (
+                "brand domain must have a registrable public or private-suffix domain"
+            ),
+            "special_use_not_allowed": "IANA special-use brand domain is not allowed in production",
+        }
+        super().__init__(messages[code])
+
+
+def _has_domain_suffix(domain: str, suffix: str) -> bool:
+    return domain == suffix or domain.endswith(f".{suffix}")
+
+
+def is_development_brand_domain(domain: str) -> bool:
+    """Return whether ``domain`` is in the protocol's narrow test-name set."""
+
+    return domain in _DEVELOPMENT_EXACT_NAMES or any(
+        domain.endswith(f".{suffix}") for suffix in _DEVELOPMENT_SUFFIXES
+    )
+
+
+def _is_special_use_domain(domain: str) -> bool:
+    return any(_has_domain_suffix(domain, suffix) for suffix in _SPECIAL_USE_SUFFIXES)
+
+
+def validate_brand_domain(domain: str, *, allow_development_domains: bool = False) -> str:
+    """Validate and canonicalize a BrandRef/BrandKey domain.
+
+    Production names must have an eTLD+1 in the SDK's pinned ICANN+PRIVATE
+    Public Suffix List and must not be IANA special-use names. Development
+    callers may explicitly admit subdomains of ``.localhost``, ``.test``,
+    ``.example``, or ``.invalid`` and the reserved example.com/net/org names.
+    Bare ``localhost`` remains invalid, and ``.local`` is never admitted.
+    """
+
+    if not isinstance(domain, str) or not domain or re.search(r"[\s/:@?#]", domain):
+        raise BrandDomainValidationError("invalid_syntax", domain)
+    try:
+        canonical = canonicalize_host(domain)
+    except (ValueError, UnicodeError) as exc:
+        raise BrandDomainValidationError("invalid_syntax", domain) from exc
+    if _DOTTED_WIRE_DOMAIN_RE.fullmatch(canonical) is None:
+        raise BrandDomainValidationError("invalid_syntax", domain)
+
+    development_name = is_development_brand_domain(canonical)
+    if development_name and allow_development_domains:
+        return canonical
+    if development_name or _is_special_use_domain(canonical):
+        raise BrandDomainValidationError("special_use_not_allowed", domain)
+    if registrable_domain(canonical) is None:
+        raise BrandDomainValidationError("not_registrable", domain)
+    return canonical
+
+
+def _development_registrable_domain(domain: str) -> str | None:
+    if domain in _DEVELOPMENT_EXACT_NAMES or any(
+        domain.endswith(f".{name}") for name in _DEVELOPMENT_EXACT_NAMES
+    ):
+        for name in _DEVELOPMENT_EXACT_NAMES:
+            if _has_domain_suffix(domain, name):
+                return name
+    for suffix in _DEVELOPMENT_SUFFIXES:
+        marker = f".{suffix}"
+        if domain.endswith(marker):
+            labels = domain.split(".")
+            return ".".join(labels[-2:]) if len(labels) >= 2 else None
+    return None
+
+
+def same_development_brand_domain(a: str, b: str) -> bool:
+    """Compare two hosts within an explicitly enabled development namespace."""
+
+    try:
+        da = _development_registrable_domain(host_from(a))
+        db = _development_registrable_domain(host_from(b))
+    except ValueError:
+        return False
+    return da is not None and da == db
+
 
 @lru_cache(maxsize=1)
 def _extractor() -> tldextract.TLDExtract:
@@ -164,7 +279,11 @@ def same_registrable_domain(a: str, b: str) -> bool:
 
 
 __all__ = [
+    "BrandDomainValidationError",
     "host_from",
+    "is_development_brand_domain",
     "registrable_domain",
+    "same_development_brand_domain",
     "same_registrable_domain",
+    "validate_brand_domain",
 ]

tests/test_brand_authz.py

@@ -148,7 +148,7 @@ async def test_authz_stale_on_error_is_bounded() -> None:
 
 
 @pytest.mark.asyncio
-async def test_authz_etld1_match_with_subdomain_brand_url() -> None:
+async def test_authz_etld1_match_with_subdomain_brand_domain() -> None:
     body = _brand_json(
         {"agents": [{"type": "signals", "id": "x", "url": "https://api.brand.com/x"}]}
     )
@@ -160,7 +160,7 @@ async def test_authz_etld1_match_with_subdomain_brand_url() -> None:
 
     assert await resolver.is_authorized(
         agent_url="https://api.brand.com/x",
-        brand_domain="https://www.brand.com/",
+        brand_domain="www.brand.com",
     )
 
 
@@ -537,6 +537,75 @@ async def test_authz_rejects_localhost_brand_domain() -> None:
     assert result.reason == "brand_domain_invalid"
 
 
+@pytest.mark.asyncio
+async def test_authz_development_domain_requires_separate_explicit_options() -> None:
+    body = _brand_json(
+        {
+            "agents": [
+                {
+                    "type": "signals",
+                    "url": "https://ads.brand.example/signals",
+                }
+            ]
+        }
+    )
+    url = "https://brand.example/.well-known/brand.json"
+    transport = _MockTransport({url: {"body": body}})
+
+    production_resolver = BrandJsonAuthorizationResolver(
+        url,
+        _client_factory=_factory(transport),
+    )
+    production_result = await production_resolver.check(
+        agent_url="https://ads.brand.example/signals",
+        brand_domain="brand.example",
+    )
+    assert production_result.authorized is False
+    assert production_result.reason == "brand_domain_invalid"
+
+    private_only_resolver = BrandJsonAuthorizationResolver(
+        url,
+        allow_private_destinations=True,
+        _client_factory=_factory(transport),
+    )
+    private_only_result = await private_only_resolver.check(
+        agent_url="https://ads.brand.example/signals",
+        brand_domain="brand.example",
+    )
+    assert private_only_result.authorized is False
+    assert private_only_result.reason == "brand_domain_invalid"
+
+    development_resolver = BrandJsonAuthorizationResolver(
+        url,
+        allow_private_destinations=True,
+        allow_development_domains=True,
+        _client_factory=_factory(transport),
+    )
+    development_result = await development_resolver.check(
+        agent_url="https://ads.brand.example/signals",
+        brand_domain="brand.example",
+    )
+    assert development_result.authorized is True
+    assert development_result.reason == "etld1_match"
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("brand_domain", ["https://brand.com/path", "brand.com/path", None])
+async def test_authz_rejects_non_bare_or_malformed_brand_domain(
+    brand_domain: str | None,
+) -> None:
+    resolver = BrandJsonAuthorizationResolver(
+        "https://brand.com/.well-known/brand.json",
+    )
+
+    result = await resolver.check(
+        agent_url="https://ads.brand.com/x",
+        brand_domain=brand_domain,  # type: ignore[arg-type]
+    )
+    assert result.authorized is False
+    assert result.reason == "brand_domain_invalid"
+
+
 # ----- fetch errors -----