From dc549281aaf7ccb44e7b10a119db903cb105b14a Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Thu, 29 Oct 2026 20:26:29 +0000 Subject: [PATCH 1/3] feat(adagents)!: attribute properties to a publisher domain, gate permissive resolution, verify cross-domain claims A sales agent syncing adagents.json for publisher domain D had three authorization decisions left to it, and each implementation made them differently (#1697): - properties_for_publisher_domain(properties, D) answers "which of these authorized properties are D's" for shared and network files. A declared publisher_domain wins; otherwise a type=domain identifier naming D or any subdomain (www., m., shop., *.D) attributes the property. Domainless properties (bundle-only apps, venues) are excluded unless include_domainless=True. www.D is treated as D; hosts compare in A-label form; D must be registrable (an IP, single label or public suffix such as co.uk or github.io raises). - resolve_properties_for_agent(mode="permissive") now requires publisher_domain=, the domain the document was fetched for. The bare-entry fallback keeps only properties whose identifiers are all type=domain under that domain, so a file on attacker.example can no longer bare-list an agent and claim another publisher's site, app bundle or DOOH venue. A property whose own publisher_domain names another domain, or that carries a bundle, store or venue identifier, is dropped. allow_any_domain=True keeps the SDK 9 result. Typed selectors and strict mode are unchanged. - verify_cross_domain_properties(properties, source_domain=A, agent_url=...) keeps a claim about another publisher's domain B only when B vouches: the document governing B, fetched with fetch_adagents (so B's authoritative_location, or its ads.txt MANAGERDOMAIN when B has no file, is followed), authorizes the agent under strict resolution for every identifier of the property. A domain identifier is vouched for by its registrable domain; a declared publisher_domain by that domain. It fails closed: every named publisher must vouch for the whole property, any fetch failure or bare entry vouches for nothing, unusable domain values are invalid_domain, and at most max_domains (50) distinct domains are fetched. The fetcher is injectable. Tests are written from the attacker's side: lookalike and suffix domains, mixed-domain and relabelled properties, bundles bolted onto a vouched site, a relay publisher vouching for a victim, private-suffix tenants, fetch failures, fan-out bounds, and delegation through authoritative_location and MANAGERDOMAIN with real fetch_adagents over a mock transport. BREAKING CHANGE: resolve_properties_for_agent(mode="permissive") raises ValueError unless publisher_domain= (or allow_any_domain=True) is passed, and with publisher_domain its bare-entry fallback returns only domain properties under that domain. See MIGRATION_v9_to_v10.md. Closes #1697 Co-Authored-By: Claude Opus 5.5 (1M context) --- MIGRATION_v9_to_v10.md | 1 + docs/adagents-discovery.md | 66 +++ docs/adagents-fetch.md | 8 +- src/adcp/__init__.py | 18 + src/adcp/adagents.py | 404 +++++++++++++++++- tests/test_adagents.py | 7 + tests/test_adagents_fetch_api.py | 7 +- tests/test_adagents_matching.py | 5 +- tests/test_adagents_publisher_domain.py | 532 ++++++++++++++++++++++++ 9 files changed, 1041 insertions(+), 7 deletions(-) create mode 100644 tests/test_adagents_publisher_domain.py diff --git a/MIGRATION_v9_to_v10.md b/MIGRATION_v9_to_v10.md index 2186a256d..35dd3bf8a 100644 --- a/MIGRATION_v9_to_v10.md +++ b/MIGRATION_v9_to_v10.md @@ -30,3 +30,4 @@ the SDK-relevant inventory is [docs/adcp-3.3-sdk-inventory.md](docs/adcp-3.3-sdk | A received `null` is re-emitted; a Python `None` keyword argument is "not supplied" | A property whose schema admits `null` and that arrived as `null` (`model_validate`, `model_validate_json`, nested validation) stays in `model_fields_set` and is written back as `null` by `model_dump()`. Passing `None` as a keyword argument to a constructor, assigning `None` to a field, or passing `None` in `model_copy(update=...)` now leaves the field *unset*: it is omitted from the dump, and `model_dump(exclude_unset=True)` no longer lists it. A `None` inside a dict you validate is a wire `null`. Properties whose schema does not admit `null` are unchanged (`NoExplicitNull` refuses a JSON `null` and never emits one). | | `model_dump()` leaves an unset schema default off | A property the schema lists outside `required` with a `default` (or a `const`) is emitted only when it was set: by the document that was parsed, by a keyword argument, or by assignment. `sandbox: false`, `timezone: "inventory_local"`, `assignment_limit: 50` and the other schema defaults no longer appear on a dump unless you set them; the attribute still reads the default. Required discriminators (`asset_type`, `pricing_model`) are emitted either way. Pass `exclude_unset=False` for the SDK 9 output. Golden-file tests that compare whole dumps need regenerating. If you relied on a default reaching the peer, set it explicitly. `model_dump_as_received()` is deprecated: it returns the same document as `model_dump(mode="json")` and is removed in SDK 11. Generated root models (a union root such as `LocalizedCreativeAsset`) now subclass `AdCPRootModel` and dump with `exclude_none=True` like every other AdCP model. | | Unknown keys on open objects are kept and written back | `AdCPBaseModel` defaults to `extra='allow'`: a property the pinned model does not declare (a field a newer spec added, a sibling arm's property the schema admits) survives `model_validate` → `model_dump` in `model_extra`, including a `null` value. Generated classes whose schema object is closed (`additionalProperties: false`) keep `extra='forbid'`. SDK 9 dropped unknown keys (`extra='ignore'`). If your code passed stray keyword arguments and relied on them being dropped, remove them; `ADCP_STRICT_VALIDATION=1` turns the default into `forbid` so tests find them. | +| Permissive adagents.json resolution requires the fetched domain | `resolve_properties_for_agent(doc, agent_url, mode="permissive")` raises `ValueError` without `publisher_domain=`. Pass the domain the document was fetched for: a bare entry's top-level properties are then kept only when every identifier is a `type: "domain"` under that domain, so a file on another host can no longer claim a publisher's site, app bundle or DOOH venue through a bare entry. Properties identified by bundle, store or venue IDs, and properties whose `publisher_domain` names another domain, are dropped from the bare fallback. `allow_any_domain=True` restores the SDK 9 result. Typed selectors and strict mode are unchanged. See [docs/adagents-discovery.md](docs/adagents-discovery.md) for `properties_for_publisher_domain` and `verify_cross_domain_properties`. | diff --git a/docs/adagents-discovery.md b/docs/adagents-discovery.md index 42dca3a66..8f35ab952 100644 --- a/docs/adagents-discovery.md +++ b/docs/adagents-discovery.md @@ -20,3 +20,69 @@ Malformed sibling entries and non-string or empty URL values are skipped during lookup. `validate_adagents_structure(document)` reports invalid URL types through its existing `missing_url` diagnostic. The listing helper returns matching entry objects in source order and does not modify the document. + +## Attributing properties to a publisher domain + +One adagents.json is often served at several publisher domains (a media +group's shared file) or names a network's publishers. When syncing a file for +publisher domain `D`, `properties_for_publisher_domain(properties, D)` returns +the properties that are `D`'s: + +- a property that declares `publisher_domain` belongs to `D` when that value + names `D` (its identifiers are not consulted); +- otherwise it belongs to `D` when a `type: "domain"` identifier names `D` or + any subdomain (`www.D`, `m.D`, `shop.D`, the pattern `*.D`); +- a property with no domain at all (an app known only by bundle ID, a DOOH + venue) is excluded unless `include_domainless=True`. + +`www.D` is treated as `D`, and hosts compare case-insensitively in A-label +form. `D` must be a registrable hostname: an IP literal, a single label or a +public suffix (`co.uk`, `github.io`) raises `AdagentsValidationError`. +Attribution does not authorize anything; apply it to the output of +`get_properties_by_agent` or `resolve_properties_for_agent`. + +```python +from adcp import get_properties_by_agent, properties_for_publisher_domain + +authorized = get_properties_by_agent(document, agent_url) +ours = properties_for_publisher_domain(authorized, "example.org") +apps = [ + p + for p in properties_for_publisher_domain(authorized, "example.org", include_domainless=True) + if p not in ours +] +``` + +## Permissive resolution is gated to the fetched domain + +A bare `authorized_agents` entry has no typed binding, so nothing in the file +stops `attacker.example` from bare-listing an agent and claiming another +publisher's site, app or venue. `resolve_properties_for_agent(..., +mode="permissive")` therefore requires `publisher_domain=`, the domain the +document was fetched for. The bare-entry fallback then keeps only properties +whose identifiers are all `type: "domain"` and name that domain or a +subdomain. A property whose own `publisher_domain` names another domain is +dropped, as is any property with a bundle, store or venue identifier. Typed +selectors are unaffected. `allow_any_domain=True` keeps the unfiltered SDK 9 +result for diagnostics. + +## Verifying cross-domain claims + +When a file served at `A` authorizes an agent for a property on another +publisher's domain `B`, `await verify_cross_domain_properties(properties, +source_domain=A, agent_url=...)` keeps the claim only when `B` vouches: the +document governing `B` (fetched with `fetch_adagents`, so `B`'s +`authoritative_location`, or its ads.txt `MANAGERDOMAIN` when `B` publishes no +file, is followed) authorizes the agent, under strict resolution and `B`'s +revocations, for every identifier of the property. A domain identifier is +vouched for by its registrable domain (`shop.b.com` by `b.com`); a declared +`publisher_domain` by that domain. Properties on `A` or its subdomains are +`A`'s own claims and need no fetch. + +The result has `verified`, `unverified` (each an `UnverifiedProperty` with the +`domain` and a `reason`: `not_vouched`, `fetch_failed`, `invalid_domain` or +`domain_limit`) and `domainless`. The check fails closed: a property naming +several publishers needs each to vouch for all of it, a failed fetch or a bare +entry in `B`'s file vouches for nothing, and at most `max_domains` (default 50) +distinct domains are fetched per call. Pass `fetcher=` to reuse a cache; it +receives a domain and returns the parsed document. diff --git a/docs/adagents-fetch.md b/docs/adagents-fetch.md index d505da4eb..721dbd24d 100644 --- a/docs/adagents-fetch.md +++ b/docs/adagents-fetch.md @@ -10,10 +10,16 @@ from adcp import fetch_adagents, validate_adagents_structure, resolve_properties async def inspect_publisher(domain: str, agent_url: str): document = await fetch_adagents(domain, validate_structure=False) report = validate_adagents_structure(document) - properties = resolve_properties_for_agent(document, agent_url, mode="permissive") + properties = resolve_properties_for_agent( + document, agent_url, mode="permissive", publisher_domain=domain + ) return report, properties ``` +Permissive mode requires `publisher_domain=`: a bare entry's top-level +properties are kept only when they are domains under the fetched publisher (see +[publisher agent discovery](adagents-discovery.md#permissive-resolution-is-gated-to-the-fetched-domain)). + The opt-out preserves JSON-object and endpoint checks, authorization-array shape, renderer catalog trust checks, URL/DNS SSRF gates, response size caps, and redirect constraints. `fetch_adagents_with_cache` and `validate_adagents_domain` accept the diff --git a/src/adcp/__init__.py b/src/adcp/__init__.py index 1a354434f..e33732527 100644 --- a/src/adcp/__init__.py +++ b/src/adcp/__init__.py @@ -166,7 +166,11 @@ def _resolve_version() -> str: "DiscoveryMethod", "DivergenceReport", "EntryErrorKind", + "AdagentsFetcher", + "CrossDomainFailureReason", + "CrossDomainVerification", "PublisherDivergence", + "UnverifiedProperty", "detect_publisher_properties_divergence", "domain_matches", "fetch_adagents", @@ -180,11 +184,13 @@ def _resolve_version() -> str: "get_properties_by_agent", "identifiers_match", "parse_managerdomains", + "properties_for_publisher_domain", "resolve_properties_for_agent", "validate_adagents_domain", "validate_adagents_structure", "verify_agent_authorization", "verify_agent_for_property", + "verify_cross_domain_properties", ), "adcp.canonical_formats": ( "format_is_supported", @@ -1604,6 +1610,12 @@ def get_adcp_version() -> str: "get_all_tags", "get_properties_by_agent", "resolve_properties_for_agent", + "properties_for_publisher_domain", + "verify_cross_domain_properties", + "AdagentsFetcher", + "CrossDomainFailureReason", + "CrossDomainVerification", + "UnverifiedProperty", # Test helpers "test_agent", "test_agent_a2a", @@ -1825,12 +1837,15 @@ def get_adcp_version() -> str: from adcp.adagents import ( AdagentsCacheEntry, AdagentsEntryError, + AdagentsFetcher, AdagentsFetchResult, AdagentsTransportFactory, AdagentsValidationReport, AdAgentsValidationResult, AgentAuthorizationsDirectoryResult, AuthorizationContext, + CrossDomainFailureReason, + CrossDomainVerification, DirectoryDiscoveryMethod, DirectoryEdgeStatus, DirectoryPublisherEntry, @@ -1838,6 +1853,7 @@ def get_adcp_version() -> str: DivergenceReport, EntryErrorKind, PublisherDivergence, + UnverifiedProperty, detect_publisher_properties_divergence, domain_matches, fetch_adagents, @@ -1851,11 +1867,13 @@ def get_adcp_version() -> str: get_properties_by_agent, identifiers_match, parse_managerdomains, + properties_for_publisher_domain, resolve_properties_for_agent, validate_adagents_domain, validate_adagents_structure, verify_agent_authorization, verify_agent_for_property, + verify_cross_domain_properties, ) from adcp.canonical_formats import ( format_is_supported, diff --git a/src/adcp/adagents.py b/src/adcp/adagents.py index 54a9101b8..b508dd01d 100644 --- a/src/adcp/adagents.py +++ b/src/adcp/adagents.py @@ -14,7 +14,7 @@ import logging import re import socket -from collections.abc import Callable +from collections.abc import Awaitable, Callable from contextlib import AbstractAsyncContextManager from dataclasses import dataclass, field from datetime import datetime @@ -2099,6 +2099,8 @@ def resolve_properties_for_agent( agent_url: str, *, mode: PropertyResolutionMode = "strict", + publisher_domain: str | None = None, + allow_any_domain: bool = False, ) -> list[dict[str, Any]]: """Resolve properties for an agent with an explicit strict/permissive mode. @@ -2115,30 +2117,60 @@ def resolve_properties_for_agent( explicit or unknown selector field, or has multiple same-URL entries, the resolver still returns the strict result. + A bare entry carries no typed binding, so nothing in the file stops a + document served at ``attacker.example`` from bare-listing an agent and + claiming another publisher's site, app or venue. Permissive mode + therefore requires ``publisher_domain``, the domain the document was + fetched for: the bare-entry fallback keeps only properties whose + identifiers are all ``type: "domain"`` and name that domain or one of + its subdomains (``www.D`` is treated as ``D``). A property whose own + ``publisher_domain`` names another domain is dropped; the claim is not + trusted, since the file is not attested. Properties identified by a + bundle ID, store ID or venue are dropped, because a domain cannot vouch + for them. ``allow_any_domain=True`` instead keeps the SDK 9 behavior of + returning every top-level property; use it only for diagnostics. + Args: adagents_data: Parsed adagents.json data agent_url: URL of the agent to filter by mode: ``"strict"`` for spec-conformant resolution, ``"permissive"`` to opt into bare-entry top-level property fallback. + publisher_domain: Permissive mode only: the publisher domain the + document was fetched for, gating the bare-entry fallback. + allow_any_domain: Permissive mode only: skip the domain gate. Returns: List of properties for the specified agent. Raises: - AdagentsValidationError: If adagents_data is malformed - ValueError: If mode is not ``"strict"`` or ``"permissive"`` + AdagentsValidationError: If adagents_data is malformed, or + ``publisher_domain`` is not a registrable hostname + ValueError: If mode is not ``"strict"`` or ``"permissive"``, or the + domain gate options do not fit the mode """ if mode == "strict": + if publisher_domain is not None or allow_any_domain: + raise ValueError("publisher_domain and allow_any_domain apply to mode='permissive'") return _resolve_properties_for_agent( adagents_data, agent_url, permissive_bare_top_level=False, ) if mode == "permissive": + if publisher_domain is None and not allow_any_domain: + raise ValueError( + "mode='permissive' requires publisher_domain= (the domain the document " + "was fetched for), or allow_any_domain=True to skip the domain gate" + ) + if publisher_domain is not None and allow_any_domain: + raise ValueError("pass publisher_domain or allow_any_domain=True, not both") return _resolve_properties_for_agent( adagents_data, agent_url, permissive_bare_top_level=True, + bare_entry_domain=( + None if publisher_domain is None else _publisher_domain_host(publisher_domain) + ), ) raise ValueError("mode must be 'strict' or 'permissive'") @@ -2148,6 +2180,7 @@ def _resolve_properties_for_agent( agent_url: str, *, permissive_bare_top_level: bool, + bare_entry_domain: str | None = None, ) -> list[dict[str, Any]]: """Implementation shared by strict and opt-in permissive property resolution.""" matches = find_authorized_agent_entries(adagents_data, agent_url) @@ -2185,11 +2218,374 @@ def _resolve_properties_for_agent( and len(matches) == 1 and _is_bare_authorized_agent_entry(matches[0]) ): - return [p for p in revoked_top_level if isinstance(p, dict)] + return [ + p + for p in revoked_top_level + if isinstance(p, dict) + and (bare_entry_domain is None or _only_names_domain(p, bare_entry_domain)) + ] return [] +# --------------------------------------------------------------------------- +# Publisher-domain attribution and cross-domain verification (#1697) +# --------------------------------------------------------------------------- + +_HOSTNAME_RE = re.compile( + r"[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?(?:\.[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?)+" +) + + +def _domain_host(value: object, *, wildcard: bool = False) -> str | None: + """Canonical lowercase A-label hostname of ``value``, or None if it is not one. + + ``wildcard=True`` accepts a leading ``*.`` (a property identifier + pattern) and returns the host it ranges over. IP literals, single-label + names, URLs and anything else that is not a hostname return None. + """ + if not isinstance(value, str): + return None + candidate = value.strip() + if wildcard and candidate.startswith("*."): + candidate = candidate[2:] + if candidate.endswith("."): + candidate = candidate[:-1] + try: + host = idna.encode(candidate, uts46=True).decode("ascii").lower() + except (idna.IDNAError, UnicodeError): + return None + if len(host) > 253 or _HOSTNAME_RE.fullmatch(host) is None: + return None + try: + ipaddress.ip_address(host) + except ValueError: + return host + return None + + +def _publisher_host(value: object, *, wildcard: bool = False) -> str | None: + """:func:`_domain_host` with a leading ``www.`` removed (``www.D`` names ``D``). + + ``www.com`` is a domain in its own right, so the label is removed only + when a dotted name remains. + """ + host = _domain_host(value, wildcard=wildcard) + if host is not None and host.startswith("www.") and "." in host[4:]: + return host[4:] + return host + + +def _publisher_domain_host(publisher_domain: str) -> str: + """Validate a caller-supplied publisher domain for attribution and gating. + + A public suffix (``co.uk``, ``github.io``) is refused: every tenant + under it would count as one of its subdomains. + """ + host = _publisher_host(publisher_domain) + if host is None or registrable_domain(host) is None: + raise AdagentsValidationError( + f"publisher_domain must be a registrable hostname, got {publisher_domain!r}" + ) + return host + + +def _names_domain(host: str, domain: str) -> bool: + """``host`` is ``domain`` or one of its subdomains.""" + return host == domain or host.endswith(f".{domain}") + + +def _domain_identifiers(prop: dict[str, Any]) -> list[Any]: + identifiers = prop.get("identifiers") + if not isinstance(identifiers, list): + return [] + return [ + i.get("value") for i in identifiers if isinstance(i, dict) and i.get("type") == "domain" + ] + + +def _only_names_domain(prop: dict[str, Any], domain: str) -> bool: + """Permissive-mode gate: every identifier is a domain under ``domain``.""" + if "publisher_domain" in prop and _publisher_host(prop["publisher_domain"]) != domain: + return False + identifiers = prop.get("identifiers") + if not isinstance(identifiers, list) or not identifiers: + return False + for identifier in identifiers: + if not isinstance(identifier, dict) or identifier.get("type") != "domain": + return False + host = _domain_host(identifier.get("value"), wildcard=True) + if host is None or not _names_domain(host, domain): + return False + return True + + +def properties_for_publisher_domain( + properties: list[dict[str, Any]], + publisher_domain: str, + *, + include_domainless: bool = False, +) -> list[dict[str, Any]]: + """Return the properties that belong to ``publisher_domain``. + + One adagents.json is often served at several publisher domains (a + media group's shared file) or names a network's publishers. When + syncing it for publisher domain ``D``, this answers "which of these + properties are ``D``'s", using the AdCP domain rules: + + - A property that declares ``publisher_domain`` belongs to ``D`` when + that value names ``D``. Its identifiers are not consulted. + - Otherwise it belongs to ``D`` when a ``type: "domain"`` identifier + names ``D`` or any subdomain of it (``www.D``, ``m.D``, ``shop.D``, + or the pattern ``*.D``). + - A property with neither (an app known only by bundle ID, a DOOH + venue) cannot be attributed. It is excluded unless + ``include_domainless=True``; call twice to report those separately. + + A ``publisher_domain`` given as ``www.D`` is treated as ``D``. Hosts + are compared case-insensitively in A-label form. This attributes + properties; it does not authorize them. Pass the output of + :func:`get_properties_by_agent` or :func:`resolve_properties_for_agent`, + and see :func:`verify_cross_domain_properties` for claims about + another publisher's domain. + + Raises: + AdagentsValidationError: ``publisher_domain`` is not a registrable + hostname (an IP literal, a single label, a public suffix). + """ + domain = _publisher_domain_host(publisher_domain) + attributed: list[dict[str, Any]] = [] + for prop in (p for p in properties if isinstance(p, dict)): + if "publisher_domain" in prop: + if _publisher_host(prop["publisher_domain"]) == domain: + attributed.append(prop) + continue + values = _domain_identifiers(prop) + if not values: + if include_domainless: + attributed.append(prop) + continue + if any( + (host := _domain_host(value, wildcard=True)) is not None and _names_domain(host, domain) + for value in values + ): + attributed.append(prop) + return attributed + + +CrossDomainFailureReason = Literal["not_vouched", "fetch_failed", "invalid_domain", "domain_limit"] +AdagentsFetcher: TypeAlias = Callable[[str], Awaitable[dict[str, Any]]] +"""Fetch the adagents.json document governing a publisher domain.""" + + +@dataclass(frozen=True) +class UnverifiedProperty: + """A cross-domain claim that could not be verified. + + ``domain`` is the publisher domain that did not vouch for the claim, or + the raw value that is not a usable hostname (``invalid_domain``). + """ + + prop: dict[str, Any] + domain: str + reason: CrossDomainFailureReason + + +@dataclass(frozen=True) +class CrossDomainVerification: + """Result of :func:`verify_cross_domain_properties`. + + ``verified``: properties on the source domain (or its subdomains) and + cross-domain claims every named publisher vouched for, in input order. + ``unverified``: cross-domain claims to drop, with the reason. + ``domainless``: properties naming no domain at all, which no publisher + domain can vouch for; the caller decides. + """ + + verified: list[dict[str, Any]] = field(default_factory=list) + unverified: list[UnverifiedProperty] = field(default_factory=list) + domainless: list[dict[str, Any]] = field(default_factory=list) + + +def _vouching_domains(prop: dict[str, Any], source: str) -> list[str] | str | None: + """Publisher domains that must vouch for ``prop``, read from its claims. + + Returns the list of other publishers' domains (empty when every claim is + on ``source``), None when the property names no domain at all, or the + offending raw value (as ``str``) when a claim is not a usable hostname. + A declared ``publisher_domain`` is where the spec says to check, so it + vouches as written. A domain identifier is vouched for by its + registrable domain (eTLD+1), which controls every host beneath it. + """ + domains: list[str] = [] + named = False + if "publisher_domain" in prop: + named = True + declared = _publisher_host(prop["publisher_domain"]) + if declared is None: + return str(prop["publisher_domain"]) + if not _names_domain(declared, source): + domains.append(declared) + for value in _domain_identifiers(prop): + named = True + host = _domain_host(value, wildcard=True) + if host is None: + return str(value) + if _names_domain(host, source): + continue + registrable = registrable_domain(host) + if registrable is None: + return str(value) + if registrable not in domains: + domains.append(registrable) + return domains if named else None + + +def _identifier_covered(claimed: dict[str, Any], authorized: list[dict[str, Any]]) -> bool: + """One claimed identifier is authorized by some identifier the voucher lists.""" + claimed_type = claimed.get("type") + claimed_value = claimed.get("value") + if not isinstance(claimed_type, str) or not isinstance(claimed_value, str): + return False + for prop in authorized: + identifiers = prop.get("identifiers") + if not isinstance(identifiers, list): + continue + for identifier in identifiers: + if not isinstance(identifier, dict) or identifier.get("type") != claimed_type: + continue + value = identifier.get("value") + if not isinstance(value, str): + continue + if claimed_type == "domain": + if domain_matches(claimed_value, value): + return True + elif value == claimed_value: + return True + return False + + +async def verify_cross_domain_properties( + properties: list[dict[str, Any]], + *, + source_domain: str, + agent_url: str, + fetcher: AdagentsFetcher | None = None, + timeout: float = 10.0, + client: httpx.AsyncClient | None = None, + max_domains: int = 50, + max_concurrency: int = 10, +) -> CrossDomainVerification: + """Keep only the cross-domain claims the named publishers vouch for. + + ``properties`` were resolved for ``agent_url`` from the adagents.json + fetched for ``source_domain`` (``A``). A property on ``A`` or one of + its subdomains is ``A``'s own claim and is verified. A property that + names another publisher's domain ``B`` (a declared ``publisher_domain``, + or a ``type: "domain"`` identifier outside ``A``) is verified only when + ``B`` vouches for it: the document governing ``B``, fetched with + :func:`fetch_adagents`, authorizes ``agent_url`` (strict resolution, + honoring ``B``'s revocations) for every identifier of the property. + That covers ``B`` authorizing the agent itself and ``B`` delegating to + ``A``, since :func:`fetch_adagents` follows ``B``'s + ``authoritative_location`` and, when ``B`` publishes no file, its + ads.txt ``MANAGERDOMAIN``. A domain identifier is vouched for by its + registrable domain (``shop.b.example`` by ``b.example``). + + The check fails closed. A property naming several other publishers + needs each of them to vouch for the whole property. A fetch that fails + for any reason, a value that is not a hostname, and a bare entry in + ``B``'s file vouch for nothing. At most ``max_domains`` distinct + publisher domains are fetched (a hostile file cannot fan the sync out + to arbitrary hosts); claims beyond that are ``domain_limit``. + Properties that name no domain are returned in ``domainless``. + + Args: + properties: Properties resolved for ``agent_url`` from ``A``'s document. + source_domain: The publisher domain the document was fetched for. + agent_url: The agent the properties were resolved for. + fetcher: Async callable returning the document governing a domain. + Defaults to :func:`fetch_adagents` with ``timeout`` and ``client``. + timeout: Per-fetch timeout for the default fetcher. + client: Optional shared client for the default fetcher. + max_domains: Upper bound on distinct publisher domains fetched. + max_concurrency: Concurrent fetches. + + Raises: + AdagentsValidationError: ``source_domain`` is not a registrable hostname. + ValueError: ``max_domains`` or ``max_concurrency`` is not positive. + """ + if max_domains < 1 or max_concurrency < 1: + raise ValueError("max_domains and max_concurrency must be positive") + source = _publisher_domain_host(source_domain) + fetch = fetcher or (lambda domain: fetch_adagents(domain, timeout=timeout, client=client)) + + # One verdict per input position: a property object, the reason it + # failed, or the publisher domains still to vouch for it. + verdicts: list[tuple[dict[str, Any], UnverifiedProperty | list[str] | None]] = [] + for prop in (p for p in properties if isinstance(p, dict)): + domains = _vouching_domains(prop, source) + if isinstance(domains, str): + verdicts.append((prop, UnverifiedProperty(prop, domains, "invalid_domain"))) + else: + verdicts.append((prop, domains)) + + wanted = (d for _, v in verdicts if isinstance(v, list) for d in v) + to_fetch = list(dict.fromkeys(wanted))[:max_domains] + semaphore = asyncio.Semaphore(max_concurrency) + + async def _authorized(domain: str) -> list[dict[str, Any]] | None: + async with semaphore: + try: + document = await fetch(domain) + return get_properties_by_agent(document, agent_url) + except Exception as exc: # every failure vouches for nothing + logger.info("adagents.json for %s did not load for vouching: %s", domain, exc) + return None + + fetched = dict(zip(to_fetch, await asyncio.gather(*(_authorized(d) for d in to_fetch)))) + + result = CrossDomainVerification() + for prop, verdict in verdicts: + if verdict is None: + result.domainless.append(prop) + continue + if isinstance(verdict, list): + verdict = _first_failure(prop, verdict, fetched) + if verdict is None: + result.verified.append(prop) + else: + result.unverified.append(verdict) + return result + + +def _first_failure( + prop: dict[str, Any], + domains: list[str], + fetched: dict[str, list[dict[str, Any]] | None], +) -> UnverifiedProperty | None: + """The first publisher domain that does not vouch for ``prop``, if any.""" + for domain in domains: + if domain not in fetched: + return UnverifiedProperty(prop, domain, "domain_limit") + authorized = fetched[domain] + if authorized is None: + return UnverifiedProperty(prop, domain, "fetch_failed") + if not _claim_covered(prop, authorized): + return UnverifiedProperty(prop, domain, "not_vouched") + return None + + +def _claim_covered(prop: dict[str, Any], authorized: list[dict[str, Any]]) -> bool: + identifiers = prop.get("identifiers") + if not isinstance(identifiers, list) or not identifiers: + return False + return all( + isinstance(identifier, dict) and _identifier_covered(identifier, authorized) + for identifier in identifiers + ) + + def _reference_renderer_catalog_errors( adagents_data: dict[str, Any], *, source_url: str | None ) -> list[AdagentsEntryError]: diff --git a/tests/test_adagents.py b/tests/test_adagents.py index 9c54ed30b..e39a05ed5 100644 --- a/tests/test_adagents.py +++ b/tests/test_adagents.py @@ -2557,6 +2557,7 @@ def test_resolve_properties_for_agent_permissive_bare_entry_uses_top_level_prope adagents_data, "https://agent1.example.com", mode="permissive", + allow_any_domain=True, ) assert [p["property_id"] for p in properties] == ["site1", "app1"] @@ -2583,6 +2584,7 @@ def test_resolve_properties_for_agent_permissive_still_requires_matching_agent(s adagents_data, "https://other-agent.example.com", mode="permissive", + allow_any_domain=True, ) assert properties == [] @@ -2610,6 +2612,7 @@ def test_resolve_properties_for_agent_permissive_does_not_override_selectors(sel adagents_data, "https://agent1.example.com", mode="permissive", + allow_any_domain=True, ) assert properties == [] @@ -2635,6 +2638,7 @@ def test_resolve_properties_for_agent_permissive_rejects_missing_authorized_for( adagents_data, "https://agent1.example.com", mode="permissive", + allow_any_domain=True, ) assert properties == [] @@ -2662,6 +2666,7 @@ def test_resolve_properties_for_agent_permissive_rejects_unknown_selector_fields adagents_data, "https://agent1.example.com", mode="permissive", + allow_any_domain=True, ) assert properties == [] @@ -2700,6 +2705,7 @@ def test_resolve_properties_for_agent_permissive_prefers_explicit_duplicate_matc adagents_data, "https://agent1.example.com", mode="permissive", + allow_any_domain=True, ) assert [p["property_id"] for p in properties] == ["site1"] @@ -2735,6 +2741,7 @@ def test_resolve_properties_for_agent_permissive_honors_revoked_domains(self): adagents_data, "https://agent1.example.com", mode="permissive", + allow_any_domain=True, ) assert [p["property_id"] for p in properties] == ["site1"] diff --git a/tests/test_adagents_fetch_api.py b/tests/test_adagents_fetch_api.py index 464df8eb8..88f757946 100644 --- a/tests/test_adagents_fetch_api.py +++ b/tests/test_adagents_fetch_api.py @@ -98,7 +98,12 @@ async def test_bare_entries_require_explicit_opt_out() -> None: "publisher.com", validate_structure=False, transport_factory=factory ) assert data == BARE - assert resolve_properties_for_agent(data, AGENT, mode="permissive") == BARE["properties"] + assert ( + resolve_properties_for_agent( + data, AGENT, mode="permissive", publisher_domain="publisher.com" + ) + == BARE["properties"] + ) cached = await fetch_adagents_with_cache( "publisher.com", validate_structure=False, transport_factory=factory ) diff --git a/tests/test_adagents_matching.py b/tests/test_adagents_matching.py index ffa80eb30..13ed89bb4 100644 --- a/tests/test_adagents_matching.py +++ b/tests/test_adagents_matching.py @@ -84,7 +84,10 @@ def test_invalid_sibling_cannot_break_lookup(url: Any) -> None: assert find_authorized_agent_entries(doc, AGENT) == [valid] assert verify_agent_authorization(doc, AGENT) assert get_properties_by_agent(doc, AGENT) == doc["properties"] - assert resolve_properties_for_agent(doc, AGENT, mode="permissive") == doc["properties"] + assert ( + resolve_properties_for_agent(doc, AGENT, mode="permissive", allow_any_domain=True) + == doc["properties"] + ) assert doc == before if not isinstance(url, str) or not url: assert any(error.kind == "missing_url" for error in validate_adagents_structure(doc).errors) diff --git a/tests/test_adagents_publisher_domain.py b/tests/test_adagents_publisher_domain.py new file mode 100644 index 000000000..b2d509168 --- /dev/null +++ b/tests/test_adagents_publisher_domain.py @@ -0,0 +1,532 @@ +"""Publisher-domain attribution, the permissive domain gate and cross-domain vouching (#1697). + +Every test that involves another publisher's domain is written from the +attacker's side: a document served at ``attacker.com`` (or a network +file) naming a domain it does not control must not authorize the agent for +it unless that domain vouches. +""" + +from __future__ import annotations + +import socket +from collections.abc import AsyncIterator, Callable +from contextlib import asynccontextmanager +from typing import Any + +import httpx +import pytest + +from adcp import ( + AdagentsTransportFactory, + CrossDomainVerification, + UnverifiedProperty, + fetch_adagents, + properties_for_publisher_domain, + resolve_properties_for_agent, + verify_cross_domain_properties, +) +from adcp.exceptions import AdagentsNotFoundError, AdagentsValidationError + +AGENT = "https://sales.agent.com/mcp" + + +def site(pid: str, *values: str, **extra: Any) -> dict[str, Any]: + return { + "property_id": pid, + "property_type": "website", + "name": pid, + "identifiers": [{"type": "domain", "value": v} for v in values], + **extra, + } + + +def app(pid: str, bundle: str, **extra: Any) -> dict[str, Any]: + return { + "property_id": pid, + "property_type": "mobile_app", + "name": pid, + "identifiers": [{"type": "bundle_id", "value": bundle}], + **extra, + } + + +def venue(pid: str, venue_id: str) -> dict[str, Any]: + return { + "property_id": pid, + "property_type": "dooh", + "name": pid, + "identifiers": [{"type": "venue_id", "value": venue_id}], + } + + +def inline_file(*props: dict[str, Any], agent: str = AGENT) -> dict[str, Any]: + return { + "authorized_agents": [ + { + "url": agent, + "authorized_for": "Display", + "authorization_type": "inline_properties", + "properties": list(props), + } + ] + } + + +def bare_file(*props: dict[str, Any]) -> dict[str, Any]: + return { + "authorized_agents": [{"url": AGENT, "authorized_for": "Everything"}], + "properties": list(props), + } + + +# --------------------------------------------------------------------------- +# 1. properties_for_publisher_domain +# --------------------------------------------------------------------------- + + +def test_shared_file_subdomains_belong_to_the_publisher() -> None: + """The reported case: property_ids authorize shop./ctv. under example.org's file.""" + props = [ + site("main", "example.org"), + site("shop", "shop.example.org"), + site("ctv", "ctv.example.org"), + site("wild", "*.example.org"), + site("mobile", "m.example.org"), + site("sister", "sister-paper.com"), + ] + attributed = properties_for_publisher_domain(props, "example.org") + assert [p["property_id"] for p in attributed] == ["main", "shop", "ctv", "wild", "mobile"] + assert [ + p["property_id"] for p in properties_for_publisher_domain(props, "sister-paper.com") + ] == ["sister"] + + +@pytest.mark.parametrize("entered", ["www.example.org", "WWW.Example.ORG.", " example.org "]) +def test_publisher_entered_as_www_is_the_apex(entered: str) -> None: + assert properties_for_publisher_domain([site("main", "example.org")], entered) + + +@pytest.mark.parametrize( + "value", + [ + "notexample.org", + "example.org.attacker.com", + "*.org", + "org", + "example.organic", + "https://example.org/", + "example.org:443", + "user@example.org", + "93.184.216.34", + "", + ], +) +def test_lookalike_and_malformed_identifiers_do_not_attribute(value: str) -> None: + assert properties_for_publisher_domain([site("x", value)], "example.org") == [] + + +def test_www_dot_tld_is_its_own_domain() -> None: + assert properties_for_publisher_domain([site("x", "www.com")], "www.com") + assert properties_for_publisher_domain([site("x", "com")], "www.com") == [] + + +def test_declared_publisher_domain_wins_over_identifiers() -> None: + """A network file attributing a property to its publisher is read as written.""" + props = [ + site("theirs", "example.org", publisher_domain="other.com"), + site("ours", "cdn.other.com", publisher_domain="www.example.org"), + site("garbage", "example.org", publisher_domain=None), + ] + assert [p["property_id"] for p in properties_for_publisher_domain(props, "example.org")] == [ + "ours" + ] + + +def test_domainless_properties_are_excluded_unless_requested() -> None: + props = [site("main", "example.org"), app("app", "com.example.app"), venue("v", "123")] + assert properties_for_publisher_domain(props, "example.org") == [props[0]] + assert properties_for_publisher_domain(props, "example.org", include_domainless=True) == props + # An app that declares its publisher_domain is attributable. + declared = app("app", "com.example.app", publisher_domain="example.org") + assert properties_for_publisher_domain([declared], "example.org") == [declared] + + +def test_idn_hosts_compare_in_a_label_form() -> None: + assert properties_for_publisher_domain([site("x", "shop.bücher.com")], "xn--bcher-kva.com") + assert properties_for_publisher_domain([site("x", "xn--bcher-kva.com")], "Bücher.com") + + +@pytest.mark.parametrize( + "bad", ["co.uk", "github.io", "localhost", "10.0.0.1", "", "*.example.org"] +) +def test_publisher_domain_must_be_registrable(bad: str) -> None: + """A public suffix would make every tenant beneath it a 'subdomain'.""" + with pytest.raises(AdagentsValidationError, match="registrable hostname"): + properties_for_publisher_domain([site("x", "victim.co.uk")], bad) + + +# --------------------------------------------------------------------------- +# 2. Permissive mode's domain gate +# --------------------------------------------------------------------------- + + +def test_permissive_requires_a_publisher_domain_or_an_explicit_opt_out() -> None: + doc = bare_file(site("own", "attacker.com")) + with pytest.raises(ValueError, match="requires publisher_domain"): + resolve_properties_for_agent(doc, AGENT, mode="permissive") + with pytest.raises(ValueError, match="not both"): + resolve_properties_for_agent( + doc, + AGENT, + mode="permissive", + publisher_domain="attacker.com", + allow_any_domain=True, + ) + with pytest.raises(ValueError, match="apply to mode='permissive'"): + resolve_properties_for_agent(doc, AGENT, publisher_domain="attacker.com") + with pytest.raises(ValueError, match="apply to mode='permissive'"): + resolve_properties_for_agent(doc, AGENT, allow_any_domain=True) + + +def test_bare_entry_cannot_claim_another_publishers_inventory() -> None: + """A file on attacker.com bare-lists the agent and claims a victim's + site, an app bundle and a DOOH venue; only its own domain survives.""" + own = site("own", "attacker.com") + own_sub = site("own-sub", "news.attacker.com", "*.attacker.com") + doc = bare_file( + own, + own_sub, + site("victim", "victim.com"), + site("mixed", "attacker.com", "victim.com"), + site("relabelled", "attacker.com", publisher_domain="victim.com"), + site("lookalike", "attacker.com.victim.com"), + app("bundle", "com.victim.app"), + { + **site("site-plus-app", "attacker.com"), + "identifiers": [ + {"type": "domain", "value": "attacker.com"}, + {"type": "bundle_id", "value": "com.victim.app"}, + ], + }, + venue("venue", "victim-venue-1"), + {"property_id": "no-ids", "property_type": "website", "name": "n"}, + ) + gated = resolve_properties_for_agent( + doc, AGENT, mode="permissive", publisher_domain="www.attacker.com" + ) + assert gated == [own, own_sub] + # The SDK 9 behavior stays available, explicitly. + ungated = resolve_properties_for_agent(doc, AGENT, mode="permissive", allow_any_domain=True) + assert ungated == doc["properties"] + + +def test_gate_applies_only_to_the_bare_fallback() -> None: + """Typed selectors are the publisher's explicit binding; the gate leaves them alone.""" + victim = site("victim", "victim.com") + doc = inline_file(victim) + assert resolve_properties_for_agent( + doc, AGENT, mode="permissive", publisher_domain="attacker.com" + ) == [victim] + + +def test_gate_still_honors_revocation_and_agent_matching() -> None: + doc = bare_file(site("own", "attacker.com", publisher_domain="attacker.com")) + doc["revoked_publisher_domains"] = [ + {"publisher_domain": "attacker.com", "revoked_at": "2026-10-01T00:00:00Z"} + ] + assert ( + resolve_properties_for_agent(doc, AGENT, mode="permissive", publisher_domain="attacker.com") + == [] + ) + other = bare_file(site("own", "attacker.com")) + assert ( + resolve_properties_for_agent( + other, + "https://other.agent.com", + mode="permissive", + publisher_domain="attacker.com", + ) + == [] + ) + + +def test_gate_rejects_an_unregistrable_publisher_domain() -> None: + with pytest.raises(AdagentsValidationError): + resolve_properties_for_agent( + bare_file(site("x", "victim.co.uk")), AGENT, mode="permissive", publisher_domain="co.uk" + ) + + +# --------------------------------------------------------------------------- +# 3. verify_cross_domain_properties +# --------------------------------------------------------------------------- + + +def fetcher_for( + documents: dict[str, dict[str, Any] | Exception], calls: list[str] | None = None +) -> Callable[[str], Any]: + async def fetch(domain: str) -> dict[str, Any]: + if calls is not None: + calls.append(domain) + if domain not in documents: + raise AdagentsNotFoundError(domain) + document = documents[domain] + if isinstance(document, Exception): + raise document + return document + + return fetch + + +async def verify( + props: list[dict[str, Any]], + documents: dict[str, dict[str, Any] | Exception], + calls: list[str] | None = None, + **kwargs: Any, +) -> CrossDomainVerification: + return await verify_cross_domain_properties( + props, + source_domain="network.com", + agent_url=AGENT, + fetcher=fetcher_for(documents, calls), + **kwargs, + ) + + +def reasons(result: CrossDomainVerification) -> dict[str, tuple[str, str]]: + return {u.prop["property_id"]: (u.domain, u.reason) for u in result.unverified} + + +async def test_own_domain_claims_need_no_fetch() -> None: + calls: list[str] = [] + props = [ + site("own", "network.com"), + site("sub", "ctv.network.com"), + site("declared", "network.com", publisher_domain="www.network.com"), + ] + result = await verify(props, {}, calls) + assert result.verified == props + assert result.unverified == [] and result.domainless == [] and calls == [] + + +async def test_claim_vouched_by_the_named_publishers_own_file() -> None: + claim = site("b", "b.com", publisher_domain="b.com") + result = await verify([claim], {"b.com": inline_file(site("b-own", "b.com"))}) + assert result.verified == [claim] + + +async def test_claim_is_refused_when_the_named_publisher_does_not_vouch() -> None: + claims = [ + site("unlisted", "b.com"), + site("other-agent", "c.com"), + site("other-site", "d.com"), + site("bare", "e.com"), + ] + documents: dict[str, dict[str, Any] | Exception] = { + "b.com": inline_file(site("x", "b.com"), agent="https://other.agent.com"), + "c.com": {"authorized_agents": []}, + "d.com": inline_file(site("x", "unrelated.d.com")), + # A bare entry in the vouching file binds nothing. + "e.com": bare_file(site("x", "e.com")), + } + result = await verify(claims, documents) + assert result.verified == [] + assert reasons(result) == { + "unlisted": ("b.com", "not_vouched"), + "other-agent": ("c.com", "not_vouched"), + "other-site": ("d.com", "not_vouched"), + "bare": ("e.com", "not_vouched"), + } + + +async def test_voucher_must_cover_every_identifier_of_the_claim() -> None: + """B authorizing its site does not vouch for a victim's bundle bolted onto the claim.""" + claim = site("b", "b.com") + claim["identifiers"].append({"type": "bundle_id", "value": "com.victim.app"}) + result = await verify([claim], {"b.com": inline_file(site("b", "b.com"))}) + assert reasons(result) == {"b": ("b.com", "not_vouched")} + + covered = await verify( + [claim], + {"b.com": inline_file(site("b", "b.com"), app("app", "com.victim.app"))}, + ) + assert covered.verified == [claim] + + +async def test_wildcard_and_subdomain_claims_follow_domain_matching() -> None: + sub = site("sub", "shop.b.com") + wild = site("wild", "*.b.com") + calls: list[str] = [] + narrow = await verify([sub, wild], {"b.com": inline_file(site("b", "b.com"))}, calls) + # A bare authorization of b.com covers www./m. only, never shop. or *. + assert reasons(narrow) == { + "sub": ("b.com", "not_vouched"), + "wild": ("b.com", "not_vouched"), + } + # Each claim's registrable domain is fetched, once. + assert calls == ["b.com"] + broad = await verify([sub, wild], {"b.com": inline_file(site("b", "*.b.com"))}) + assert broad.verified == [sub, wild] + + +async def test_a_relay_publisher_cannot_vouch_for_a_victims_domain() -> None: + """attacker.com's file says the victim's site lives under accomplice.com; + the victim's own domain must vouch too.""" + claim = site("victim", "victim.com", publisher_domain="accomplice.com") + documents: dict[str, dict[str, Any] | Exception] = { + "accomplice.com": inline_file(claim), + "victim.com": inline_file(site("x", "victim.com"), agent="https://real.agent.com"), + } + result = await verify([claim], documents) + assert reasons(result) == {"victim": ("victim.com", "not_vouched")} + + +async def test_every_named_publisher_must_vouch() -> None: + claim = site("both", "b.com", "c.com") + documents: dict[str, dict[str, Any] | Exception] = {"b.com": inline_file(claim)} + result = await verify([claim], documents) + assert reasons(result) == {"both": ("c.com", "fetch_failed")} + documents["c.com"] = inline_file(claim) + assert (await verify([claim], documents)).verified == [claim] + + +@pytest.mark.parametrize( + "failure", + [ + AdagentsNotFoundError("b.com"), + AdagentsValidationError("too large"), + httpx.ConnectError("refused"), + RuntimeError("fetcher bug"), + ], +) +async def test_any_fetch_failure_vouches_for_nothing(failure: Exception) -> None: + result = await verify([site("b", "b.com")], {"b.com": failure}) + assert reasons(result) == {"b": ("b.com", "fetch_failed")} + + +async def test_malformed_vouching_document_vouches_for_nothing() -> None: + result = await verify([site("b", "b.com")], {"b.com": {"not": "adagents"}}) + assert reasons(result) == {"b": ("b.com", "fetch_failed")} + + +async def test_unusable_domain_values_fail_closed() -> None: + props = [ + site("ip", "10.0.0.1"), + site("url", "https://b.com/"), + site("suffix", "co.uk"), + site("declared", "network.com", publisher_domain="b.com/path"), + ] + calls: list[str] = [] + result = await verify(props, {}, calls) + assert {k: v[1] for k, v in reasons(result).items()} == { + "ip": "invalid_domain", + "url": "invalid_domain", + "suffix": "invalid_domain", + "declared": "invalid_domain", + } + assert calls == [] + + +async def test_private_suffix_tenants_vouch_for_themselves_only() -> None: + """victim.github.io is vouched for by itself, not by github.io or a sibling tenant.""" + calls: list[str] = [] + await verify([site("v", "www.victim.github.io")], {}, calls) + assert calls == ["victim.github.io"] + + +async def test_fetch_fanout_is_bounded() -> None: + props = [site(f"p{i}", f"pub{i}.com") for i in range(5)] + documents: dict[str, dict[str, Any] | Exception] = { + f"pub{i}.com": inline_file(props[i]) for i in range(5) + } + calls: list[str] = [] + result = await verify(props, documents, calls, max_domains=2) + assert calls == ["pub0.com", "pub1.com"] + assert result.verified == props[:2] + assert [u.reason for u in result.unverified] == ["domain_limit"] * 3 + with pytest.raises(ValueError): + await verify(props, documents, max_domains=0) + + +async def test_domainless_properties_are_reported_separately_and_order_is_kept() -> None: + own = site("own", "network.com") + vouched = site("b", "b.com") + bundle = app("app", "com.network.app") + result = await verify([vouched, bundle, own], {"b.com": inline_file(vouched)}) + assert result.verified == [vouched, own] + assert result.domainless == [bundle] + assert isinstance(result.unverified, list) and not result.unverified + assert UnverifiedProperty(own, "x", "not_vouched").prop is own + + +# Delegation runs through fetch_adagents, which follows the named publisher's +# authoritative_location and, when it has no file, its ads.txt MANAGERDOMAIN. + + +@pytest.fixture +def public_dns(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr( + socket, + "getaddrinfo", + lambda host, port, *a, **kw: [ + (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("8.8.8.8", port)) + ], + ) + + +def http_fetcher(routes: dict[str, httpx.Response]) -> Callable[[str], Any]: + def handler(request: httpx.Request) -> httpx.Response: + url = str(request.url) + return routes[url] if url in routes else httpx.Response(404) + + @asynccontextmanager + async def factory(url: str, timeout: float) -> AsyncIterator[httpx.AsyncClient]: + async with httpx.AsyncClient(transport=httpx.MockTransport(handler), timeout=timeout) as c: + yield c + + transport_factory: AdagentsTransportFactory = factory + + async def fetch(domain: str) -> dict[str, Any]: + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: + return await fetch_adagents(domain, client=client, transport_factory=transport_factory) + + return fetch + + +NETWORK_FILE = "https://network.com/.well-known/adagents.json" + + +@pytest.mark.usefixtures("public_dns") +@pytest.mark.parametrize("delegation", ["authoritative_location", "managerdomain"]) +async def test_publisher_delegating_to_the_network_vouches(delegation: str) -> None: + claim = site("b", "b.com", publisher_domain="b.com") + network = inline_file(claim) + routes = {NETWORK_FILE: httpx.Response(200, json=network)} + if delegation == "authoritative_location": + routes["https://b.com/.well-known/adagents.json"] = httpx.Response( + 200, json={"authoritative_location": NETWORK_FILE} + ) + else: + routes["https://b.com/ads.txt"] = httpx.Response(200, text="MANAGERDOMAIN=network.com") + result = await verify_cross_domain_properties( + [claim], source_domain="network.com", agent_url=AGENT, fetcher=http_fetcher(routes) + ) + assert result.verified == [claim] + + +@pytest.mark.usefixtures("public_dns") +async def test_publisher_with_its_own_file_is_not_overridden_by_its_ads_txt() -> None: + """MANAGERDOMAIN is a fallback for a publisher with no file; a file that + does not authorize the agent is the publisher's answer.""" + claim = site("b", "b.com", publisher_domain="b.com") + routes = { + NETWORK_FILE: httpx.Response(200, json=inline_file(claim)), + "https://b.com/.well-known/adagents.json": httpx.Response( + 200, json=inline_file(site("x", "b.com"), agent="https://real.agent.com") + ), + "https://b.com/ads.txt": httpx.Response(200, text="MANAGERDOMAIN=network.com"), + } + result = await verify_cross_domain_properties( + [claim], source_domain="network.com", agent_url=AGENT, fetcher=http_fetcher(routes) + ) + assert reasons(result) == {"b": ("b.com", "not_vouched")} From 2dfd0a3cb7425797cc3f6c7df798b63fa14a3202 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Thu, 29 Oct 2026 21:46:01 +0000 Subject: [PATCH 2/3] test: record the new public names in the API snapshot Co-Authored-By: Claude Opus 5.5 (1M context) --- tests/fixtures/public_api_snapshot.json | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/tests/fixtures/public_api_snapshot.json b/tests/fixtures/public_api_snapshot.json index 8ac65a79a..4af9c8a94 100644 --- a/tests/fixtures/public_api_snapshot.json +++ b/tests/fixtures/public_api_snapshot.json @@ -72,6 +72,7 @@ "AdagentsCacheEntry": "adcp.adagents", "AdagentsEntryError": "adcp.adagents", "AdagentsFetchResult": "adcp.adagents", + "AdagentsFetcher": "union[instance:builtins:list,union[union[builtins:str,typing:Any]]]", "AdagentsHTTPError": "adcp.exceptions", "AdagentsNotFoundError": "adcp.exceptions", "AdagentsTimeoutError": "adcp.exceptions", @@ -184,6 +185,8 @@ "CreativeManifest": "adcp.types.canonical_creative", "CreativeStatus": "enums.creative_status", "CreativeVariant": "adcp.types.canonical_creative", + "CrossDomainFailureReason": "union[value:'domain_limit',value:'fetch_failed',value:'invalid_domain',value:'not_vouched']", + "CrossDomainVerification": "adcp.adagents", "CssContent": "core.assets.css_asset:CssAsset", "CursorStore": "adcp.registry_sync", "DEFAULT_ACCEPTANCE_CATALOG_CACHE_ENTRIES": "value:64", @@ -601,6 +604,7 @@ "UniversalMacroTranslationError": "adcp.substitution", "UniversalMacroTranslationErrorCode": "union[value:'unsafe_native_mapping']", "UnknownFieldPolicy": "adcp.validation.client_hooks", + "UnverifiedProperty": "adcp.adagents", "UpdateContentStandardsErrorResponse": "content_standards.update_content_standards_response:UpdateContentStandardsResponse2", "UpdateContentStandardsResponse1": "content_standards.update_content_standards_response", "UpdateContentStandardsSuccessResponse": "content_standards.update_content_standards_response:UpdateContentStandardsResponse1", @@ -719,6 +723,7 @@ "normalize_identifier": "adcp.identifiers", "parse_managerdomains": "adcp.adagents", "project_available_actions": "adcp.media_buy_actions", + "properties_for_publisher_domain": "adcp.adagents", "reassess_media_buy_action": "adcp.media_buy_actions", "resolve_properties_for_agent": "adcp.adagents", "route_media_buy_action": "adcp.media_buy_actions", @@ -746,7 +751,8 @@ "validate_publisher_properties_item": "adcp.validation.legacy", "validate_webhook_challenge_response": "adcp.webhooks", "verify_agent_authorization": "adcp.adagents", - "verify_agent_for_property": "adcp.adagents" + "verify_agent_for_property": "adcp.adagents", + "verify_cross_domain_properties": "adcp.adagents" } }, "adcp.reporting": { From e3e5f24044d381cbe35c07a734a928b1cdd64c93 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Thu, 29 Oct 2026 23:00:59 +0000 Subject: [PATCH 3/3] fix(adagents): keep the fetcher alias out of the top-level adcp namespace Its snapshot fingerprint differs on Python 3.10, where typing.Any is a _SpecialForm. It stays importable from adcp.adagents. Co-Authored-By: Claude Opus 5.5 (1M context) --- src/adcp/__init__.py | 3 --- tests/fixtures/public_api_snapshot.json | 1 - 2 files changed, 4 deletions(-) diff --git a/src/adcp/__init__.py b/src/adcp/__init__.py index e33732527..fc0909fc5 100644 --- a/src/adcp/__init__.py +++ b/src/adcp/__init__.py @@ -166,7 +166,6 @@ def _resolve_version() -> str: "DiscoveryMethod", "DivergenceReport", "EntryErrorKind", - "AdagentsFetcher", "CrossDomainFailureReason", "CrossDomainVerification", "PublisherDivergence", @@ -1612,7 +1611,6 @@ def get_adcp_version() -> str: "resolve_properties_for_agent", "properties_for_publisher_domain", "verify_cross_domain_properties", - "AdagentsFetcher", "CrossDomainFailureReason", "CrossDomainVerification", "UnverifiedProperty", @@ -1837,7 +1835,6 @@ def get_adcp_version() -> str: from adcp.adagents import ( AdagentsCacheEntry, AdagentsEntryError, - AdagentsFetcher, AdagentsFetchResult, AdagentsTransportFactory, AdagentsValidationReport, diff --git a/tests/fixtures/public_api_snapshot.json b/tests/fixtures/public_api_snapshot.json index 4af9c8a94..e50233214 100644 --- a/tests/fixtures/public_api_snapshot.json +++ b/tests/fixtures/public_api_snapshot.json @@ -72,7 +72,6 @@ "AdagentsCacheEntry": "adcp.adagents", "AdagentsEntryError": "adcp.adagents", "AdagentsFetchResult": "adcp.adagents", - "AdagentsFetcher": "union[instance:builtins:list,union[union[builtins:str,typing:Any]]]", "AdagentsHTTPError": "adcp.exceptions", "AdagentsNotFoundError": "adcp.exceptions", "AdagentsTimeoutError": "adcp.exceptions",