diff --git a/src/adcp/signing/__init__.py b/src/adcp/signing/__init__.py index f3dd39f33..b0007c1c4 100644 --- a/src/adcp/signing/__init__.py +++ b/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,6 +464,7 @@ 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", @@ -465,6 +472,7 @@ def __init__(self, *args: object, **kwargs: object) -> None: "signing_operation", "unauthorized_response_headers", "validate_jwks_uri", + "validate_brand_domain", "verify_detached_jws", "verify_flask_request", "verify_from_agent_url", diff --git a/src/adcp/signing/brand_authz.py b/src/adcp/signing/brand_authz.py index 13ae6297a..ba27742a7 100644 --- a/src/adcp/signing/brand_authz.py +++ b/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, diff --git a/src/adcp/signing/etld.py b/src/adcp/signing/etld.py index 5c457b2c5..93335346a 100644 --- a/src/adcp/signing/etld.py +++ b/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", ] diff --git a/tests/test_brand_authz.py b/tests/test_brand_authz.py index 76b6b90fd..0396dc166 100644 --- a/tests/test_brand_authz.py +++ b/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 ----- diff --git a/tests/test_etld.py b/tests/test_etld.py index 533218536..05d06c297 100644 --- a/tests/test_etld.py +++ b/tests/test_etld.py @@ -14,7 +14,15 @@ import pytest -from adcp.signing.etld import host_from, registrable_domain, same_registrable_domain +from adcp.signing.etld import ( + BrandDomainValidationError, + host_from, + is_development_brand_domain, + registrable_domain, + same_development_brand_domain, + same_registrable_domain, + validate_brand_domain, +) # ----- host_from ----- @@ -165,3 +173,48 @@ def test_registrable_domain_reserved_tld_returns_none() -> None: assert registrable_domain("brand.example") is None assert registrable_domain("foo.test") is None assert registrable_domain("svc.invalid") is None + + +def test_validate_brand_domain_accepts_public_and_private_psl_names() -> None: + assert validate_brand_domain("Ads.Brand.COM") == "ads.brand.com" + assert validate_brand_domain("brand.co.uk") == "brand.co.uk" + assert validate_brand_domain("tenant.github.io") == "tenant.github.io" + + +@pytest.mark.parametrize( + "domain", + ["localhost", "unknown", "co.uk", "brand.unknown", "1.2.3.4", "https://brand.com"], +) +def test_validate_brand_domain_rejects_non_registrable_names(domain: str) -> None: + with pytest.raises(BrandDomainValidationError): + validate_brand_domain(domain) + + +@pytest.mark.parametrize( + "domain", + [ + "brand.localhost", + "brand.test", + "brand.example", + "brand.invalid", + "example.com", + "example.net", + "example.org", + ], +) +def test_validate_brand_domain_requires_explicit_development_option(domain: str) -> None: + with pytest.raises(BrandDomainValidationError, match="special-use"): + validate_brand_domain(domain) + assert validate_brand_domain(domain, allow_development_domains=True) == domain + assert is_development_brand_domain(domain) is True + + +def test_validate_brand_domain_never_allows_mdns_local() -> None: + assert is_development_brand_domain("brand.local") is False + with pytest.raises(BrandDomainValidationError, match="special-use"): + validate_brand_domain("brand.local", allow_development_domains=True) + + +def test_same_development_brand_domain_uses_test_namespace_boundary() -> None: + assert same_development_brand_domain("ads.brand.example", "brand.example") is True + assert same_development_brand_domain("ads.brand.example", "other.example") is False