From ce501a926f7348976ba8c827799137dfa865eee0 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Thu, 29 Oct 2026 22:54:22 +0000 Subject: [PATCH 1/2] feat(adagents): attribute and verify publisher property claims --- AGENTS.md | 12 + docs/adagents-discovery.md | 117 +++ docs/adagents-fetch.md | 8 +- src/adcp/__init__.py | 6 + src/adcp/adagents.py | 299 +++++++- tests/fixtures/public_api_snapshot.json | 4 +- tests/test_adagents_property_resolution.py | 685 ++++++++++++++++++ .../adagents_property_resolution.py | 25 + 8 files changed, 1138 insertions(+), 18 deletions(-) create mode 100644 tests/test_adagents_property_resolution.py create mode 100644 tests/type_checks/adagents_property_resolution.py diff --git a/AGENTS.md b/AGENTS.md index da71320c5..2b2c9ac69 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -267,6 +267,18 @@ from adcp.testing import build_asgi_app, build_test_client, make_request_context ## Property Lists and Identifier Resolution +For publisher manifest sync, resolve the agent's selectors with +`get_properties_by_agent` or `resolve_properties_for_agent`, then call +`properties_for_publisher_domain(properties, publisher_domain)` to attribute +inventory. It keeps publisher subdomains and supports wildcard domain inventory. +Pass `include_unattributed=False` to exclude inventory without a publisher or +domain anchor. When using bare-entry permissive resolution, pass +`publisher_domain=` from the trusted fetch origin to require a matching domain +identifier. Use `await verify_cross_domain_properties(properties, source_domain, +agent_url)` before trusting external publisher claims; it follows secure +publisher discovery and checks strict inventory authorization and revocations. +All three helpers are exported from `adcp`. See `docs/adagents-discovery.md`. + Import `MCPPropertyListFetcher`, `ProductPropertyPolicy`, and `check_property_list_reference` from `adcp`. The default seller fetcher resolves lists with MCP `get_property_list`, follows pagination, and caches complete diff --git a/docs/adagents-discovery.md b/docs/adagents-discovery.md index 42dca3a66..45d67f929 100644 --- a/docs/adagents-discovery.md +++ b/docs/adagents-discovery.md @@ -20,3 +20,120 @@ 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 + +After resolving the agent's selectors, use `properties_for_publisher_domain` +to select a publisher's inventory from a shared manifest: + +```python +from adcp import get_properties_by_agent, properties_for_publisher_domain + +authorized = get_properties_by_agent(document, agent_url) +properties = properties_for_publisher_domain(authorized, "www.example.org") +``` + +An explicit property `publisher_domain` takes precedence over identifiers and +must name the requested publisher. Otherwise, a `type="domain"` or +`type="subdomain"` identifier must +name that publisher or any subdomain. For example, `shop.example.org` and +`ctv.example.org` both belong to `example.org`. Host case, trailing dots, IDNA +and a leading `www.` alias within the same registrable domain are normalized. +The alias rule respects both public and private suffixes, so `www.com` cannot +become `com` and `www.github.io` cannot become `github.io`. Domain matching uses +`domain_matches`, including an explicit `*.example.org` pattern. +Wildcard domain inventory such as `*.example.org` belongs to `example.org`; +publisher-domain arguments must be concrete hostnames. + +Properties with neither a publisher anchor nor a domain identifier, such as +bundle-only apps, are kept by default. Pass `include_unattributed=False` to +exclude them. Invalid domain claims are excluded. The helper returns original +objects in input order without changing the document. Attribution trusts the +manifest's claims; it does not establish cross-domain authority. + +## Gating bare-entry resolution + +When using permissive resolution, supply the domain from the fetch origin: + +```python +from adcp import fetch_adagents, resolve_properties_for_agent + +document = await fetch_adagents(domain, validate_structure=False) +properties = resolve_properties_for_agent( + document, agent_url, mode="permissive", publisher_domain=domain +) +``` + +For bare-entry fallback, each property must have a domain identifier naming +the publisher or a subdomain. The property's own `publisher_domain` cannot +satisfy this gate. Apps, venues and other inventory without a matching domain +identifier are excluded. Typed authorization selectors keep their existing +behavior in both modes. Omitting `publisher_domain` preserves the existing +permissive behavior; strict resolution remains the default. + +## Verifying cross-domain claims + +Use `verify_cross_domain_properties` on resolved inventory before trusting +claims about other domains: + +```python +from adcp import ( + fetch_adagents, + get_properties_by_agent, + properties_for_publisher_domain, + verify_cross_domain_properties, +) + +source_domain = "network.example" +document = await fetch_adagents(source_domain) +authorized = get_properties_by_agent(document, agent_url) +verified = await verify_cross_domain_properties( + authorized, source_domain, agent_url +) +publisher_properties = properties_for_publisher_domain(verified, "publisher.example") +``` + +The source domain must come from the trusted fetch origin. For every external +publisher anchor and domain or subdomain identifier, the helper fetches that domain's own +manifest and strictly resolves the same agent's selectors. The remote +authorization must cover the claimed inventory and be attributable to the +checked domain. A matching agent URL alone is insufficient. All claimed typed +identifiers must match an authorized property of the same type, using +`identifiers_match`; a matching property ID cannot override different domain +or bundle identifiers. When identifiers are absent, a publisher-scoped +property ID is used. + +A verified publisher anchor also covers that publisher's subdomain identifiers. +Parent domains and unrelated domains require their own verification. Wildcard +inventory is verified at its base hostname, and Unicode and IDNA domain +identities compare equivalently. The permissive gate continues to require +`type="domain"`; attribution and verification also recognize `type="subdomain"`. + +The default fetcher is `fetch_adagents`, which applies the existing URL/DNS, +response size and redirect protections, follows `authoritative_location`, and +discovers ads.txt `MANAGERDOMAIN` delegation. Delegation still requires the +resolved manifest to authorize inventory attributable to the checked publisher; +an unanchored app in a shared manager file is insufficient. Remote revocations +override authorization. A missing, invalid or unavailable manifest excludes +that domain's claims and logs the failure at debug level. Each external domain +is fetched once per call, sequentially, without recursively verifying its +other claims. There is no persistent verification cache, so later calls can +observe revocations. + +Local inventory, including subdomains of the source, passes through. Inventory +with no domain identifier or publisher anchor also passes through because +there is no remote domain to check. Use `include_unattributed=False` during +attribution to exclude it, and use the domain gate above for bare entries. + +For custom timeouts, deterministic tests or caller-controlled discovery, pass +an async `fetcher(domain) -> dict`. It must provide secure discovery and follow +publisher delegation. For example: + +```python +async def fetch_publisher(domain: str) -> dict: + return await fetch_adagents(domain, timeout=5.0) + +verified = await verify_cross_domain_properties( + authorized, source_domain, agent_url, fetcher=fetch_publisher +) +``` diff --git a/docs/adagents-fetch.md b/docs/adagents-fetch.md index c1c88728c..1ce206040 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 ``` +The domain gate restricts bare-entry fallback to properties with a matching +domain identifier. See [publisher agent discovery](adagents-discovery.md) for +publisher attribution and cross-domain verification. + 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..c0ae78440 100644 --- a/src/adcp/__init__.py +++ b/src/adcp/__init__.py @@ -180,11 +180,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", @@ -1600,10 +1602,12 @@ def get_adcp_version() -> str: "domain_matches", "identifiers_match", "parse_managerdomains", + "properties_for_publisher_domain", "get_all_properties", "get_all_tags", "get_properties_by_agent", "resolve_properties_for_agent", + "verify_cross_domain_properties", # Test helpers "test_agent", "test_agent_a2a", @@ -1851,11 +1855,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..f5a449239 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 @@ -34,7 +34,11 @@ AdagentsValidationError, ) from adcp.signing._strict_json import parse_strict_json -from adcp.signing.etld import registrable_domain, same_registrable_domain +from adcp.signing.etld import ( + registrable_domain, + same_development_brand_domain, + same_registrable_domain, +) from adcp.types.base import AdCPBaseModel from adcp.validation import ValidationError, validate_adagents @@ -616,6 +620,119 @@ def identifiers_match( return False +def _publisher_attribution_domain(domain: object) -> str: + """Canonical hostname for publisher attribution, including the www alias.""" + from adcp.identifiers import IdentifierValidationError, validate_publisher_domain + + if not isinstance(domain, str): + raise AdagentsValidationError("Invalid publisher domain for property attribution") + try: + canonical = validate_publisher_domain(domain) + except IdentifierValidationError as exc: + raise AdagentsValidationError("Invalid publisher domain for property attribution") from exc + if canonical.startswith("www."): + alias = canonical.removeprefix("www.") + if same_registrable_domain(canonical, alias) or same_development_brand_domain( + canonical, alias + ): + return alias + return canonical + + +def _domain_is_under_publisher(domain: object, publisher_domain: str) -> bool: + from adcp.identifiers import IdentifierValidationError, validate_publisher_domain + + if not isinstance(domain, str): + return False + try: + domain = validate_publisher_domain(domain) + except IdentifierValidationError: + return False + return domain_matches(domain, publisher_domain) or domain_matches( + domain, f"*.{publisher_domain}" + ) + + +def _property_domain_identifiers( + prop: dict[str, Any], *, include_subdomains: bool = False +) -> list[dict[str, Any]]: + identifiers = prop.get("identifiers") + if not isinstance(identifiers, list): + return [] + domain_types = {"domain", "subdomain"} if include_subdomains else {"domain"} + return [ + i + for i in identifiers + if isinstance(i, dict) and isinstance(i.get("type"), str) and i["type"] in domain_types + ] + + +def _property_identity_identifier(identifier: Any) -> dict[str, str] | None: + """Canonicalize domain identities while preserving opaque identifier values.""" + from adcp.identifiers import IdentifierValidationError, normalize_domain + + if ( + not isinstance(identifier, dict) + or not isinstance(identifier.get("type"), str) + or not isinstance(identifier.get("value"), str) + or not identifier["value"] + ): + return None + kind, value = identifier["type"], identifier["value"] + if kind in {"domain", "subdomain"}: + try: + value = normalize_domain(value, allow_wildcard=kind == "domain").value + except IdentifierValidationError: + return None + return {"type": kind, "value": value} + + +def _domain_identifier_is_under_publisher(identifier: dict[str, Any], domain: str) -> bool: + normalized = _property_identity_identifier(identifier) + return normalized is not None and _domain_is_under_publisher( + normalized["value"].removeprefix("*."), domain + ) + + +def properties_for_publisher_domain( + properties: list[dict[str, Any]], + publisher_domain: str, + *, + include_unattributed: bool = True, +) -> list[dict[str, Any]]: + """Attribute already-resolved properties to a publisher without network I/O. + + An explicit ``publisher_domain`` takes precedence and must name the same + publisher (case, trailing dots and a leading ``www.`` are normalized). + Otherwise, a domain or subdomain identifier must name the publisher or any subdomain, + using :func:`domain_matches`. Properties without either kind of domain, + such as bundle-only apps, are kept unless ``include_unattributed=False``. + Invalid domain claims are excluded rather than treated as unattributed. + + Returns the original property objects in input order without mutation. + This is attribution of trusted claims, not proof of authorization; use + :func:`verify_cross_domain_properties` to verify cross-domain claims. + + Raises AdagentsValidationError if ``publisher_domain`` is not a hostname. + """ + domain = _publisher_attribution_domain(publisher_domain) + selected: list[dict[str, Any]] = [] + for prop in properties: + if "publisher_domain" in prop: + try: + if _publisher_attribution_domain(prop["publisher_domain"]) == domain: + selected.append(prop) + except AdagentsValidationError: + pass + continue + domain_ids = _property_domain_identifiers(prop, include_subdomains=True) + if any(_domain_identifier_is_under_publisher(i, domain) for i in domain_ids) or ( + not domain_ids and include_unattributed + ): + selected.append(prop) + return selected + + def verify_agent_authorization( adagents_data: dict[str, Any], agent_url: str, @@ -2099,6 +2216,7 @@ def resolve_properties_for_agent( agent_url: str, *, mode: PropertyResolutionMode = "strict", + publisher_domain: str | None = None, ) -> list[dict[str, Any]]: """Resolve properties for an agent with an explicit strict/permissive mode. @@ -2115,11 +2233,18 @@ def resolve_properties_for_agent( explicit or unknown selector field, or has multiple same-URL entries, the resolver still returns the strict result. + ``publisher_domain`` gates only bare-entry fallback: a property must have + a domain identifier naming that publisher or a subdomain. Self-declared + ``publisher_domain`` values cannot satisfy the gate; apps and venues with + no matching domain identifier are excluded. Typed selectors are unchanged. + 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: Optional hostname attested by the caller's fetch + origin, used to restrict permissive bare-entry fallback. Returns: List of properties for the specified agent. @@ -2128,19 +2253,17 @@ def resolve_properties_for_agent( AdagentsValidationError: If adagents_data is malformed ValueError: If mode is not ``"strict"`` or ``"permissive"`` """ - if mode == "strict": - return _resolve_properties_for_agent( - adagents_data, - agent_url, - permissive_bare_top_level=False, - ) - if mode == "permissive": - return _resolve_properties_for_agent( - adagents_data, - agent_url, - permissive_bare_top_level=True, - ) - raise ValueError("mode must be 'strict' or 'permissive'") + if mode not in {"strict", "permissive"}: + raise ValueError("mode must be 'strict' or 'permissive'") + domain = ( + _publisher_attribution_domain(publisher_domain) if publisher_domain is not None else None + ) + return _resolve_properties_for_agent( + adagents_data, + agent_url, + permissive_bare_top_level=mode == "permissive", + publisher_domain=domain, + ) def _resolve_properties_for_agent( @@ -2148,6 +2271,7 @@ def _resolve_properties_for_agent( agent_url: str, *, permissive_bare_top_level: bool, + publisher_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 +2309,154 @@ 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 ( + publisher_domain is None + or any( + _domain_identifier_is_under_publisher(i, publisher_domain) + for i in _property_domain_identifiers(p) + ) + ) + ] return [] +def _same_authorized_property(claim: dict[str, Any], authorized: dict[str, Any]) -> bool: + """Compare inventory identity without trusting file-local IDs over identifiers.""" + if claim.get("property_type") != authorized.get("property_type"): + return False + identifiers = claim.get("identifiers") + if identifiers is not None and not isinstance(identifiers, list): + return False + if isinstance(identifiers, list) and identifiers: + authorized_ids = authorized.get("identifiers") + if not isinstance(authorized_ids, list): + return False + valid_ids = [ + normalized + for i in authorized_ids + if (normalized := _property_identity_identifier(i)) is not None + ] + return all( + (normalized := _property_identity_identifier(i)) is not None + and identifiers_match([normalized], valid_ids) + for i in identifiers + ) + property_id = claim.get("property_id") + return ( + isinstance(property_id, str) + and bool(property_id) + and property_id == authorized.get("property_id") + ) + + +async def verify_cross_domain_properties( + properties: list[dict[str, Any]], + publisher_domain: str, + agent_url: str, + *, + fetcher: Callable[[str], Awaitable[dict[str, Any]]] | None = None, +) -> list[dict[str, Any]]: + """Return the subset of resolved properties vouched for by their domains. + + ``publisher_domain`` is the trusted origin of the source manifest, not a + value read from that manifest. First resolve its agent selectors with + :func:`get_properties_by_agent` or :func:`resolve_properties_for_agent`. + + Every external ``publisher_domain``, domain or subdomain identifier must + authorize the same inventory for ``agent_url`` in strict resolution. Domain + identifiers take precedence over file-local property IDs when comparing + inventory. Inventory with no identifiers uses its publisher anchor and + property ID. Remote authorizations must be attributable to the checked + domain; unanchored apps in a shared manager file cannot vouch for it. + Local properties (including subdomains) and unanchored non-domain inventory + pass through; the latter cannot be verified by this helper. + + The default fetcher is :func:`fetch_adagents`, including its SSRF gates, + ``authoritative_location`` following and ads.txt MANAGERDOMAIN discovery. + Delegated manifests must still authorize the particular property; merely + listing an agent or delegating to a manager is insufficient. A custom async + ``fetcher(domain)`` must provide the same secure discovery semantics. + + Each external domain is fetched once per call, sequentially. Missing, + invalid, revoked or unavailable authorizations are excluded, with failures + logged at debug level. No recursive property verification is performed. + Returns original objects in input order without mutation. + """ + source = _publisher_attribution_domain(publisher_domain) + fetch = fetcher if fetcher is not None else fetch_adagents + authorizations: dict[str, list[dict[str, Any]]] = {} + verified: list[dict[str, Any]] = [] + + for prop in properties: + try: + anchor = ( + _publisher_attribution_domain(prop["publisher_domain"]) + if "publisher_domain" in prop + else None + ) + domains = [anchor] if anchor is not None else [] + for identifier in _property_domain_identifiers(prop, include_subdomains=True): + normalized = _property_identity_identifier(identifier) + if normalized is None: + raise AdagentsValidationError("Invalid property domain identifier") + domain = _publisher_attribution_domain(normalized["value"].removeprefix("*.")) + if domain not in domains: + domains.append(domain) + except AdagentsValidationError: + continue + + for domain in domains: + if _domain_is_under_publisher(domain, source): + continue + # The anchor is checked first. Its verified authorization covers + # its own subdomains, matching the source-domain trust boundary. + if ( + anchor is not None + and domain != anchor + and _domain_is_under_publisher(domain, anchor) + ): + continue + if domain not in authorizations: + try: + data = await fetch(domain) + revoked = { + _publisher_attribution_domain(d) + for d in _get_revoked_publisher_domains(data) + } + try: + candidates = get_properties_by_agent(data, agent_url) + except (TypeError, ValueError) as exc: + raise AdagentsValidationError( + "Malformed remote property authorization" + ) from exc + authorizations[domain] = ( + [] + if domain in revoked + else properties_for_publisher_domain( + candidates, domain, include_unattributed=False + ) + ) + except ( + AdagentsNotFoundError, + AdagentsValidationError, + AdagentsTimeoutError, + httpx.HTTPError, + OSError, + ) as exc: + logger.debug("Cross-domain authorization failed for %s: %s", domain, exc) + authorizations[domain] = [] + if not any(_same_authorized_property(prop, p) for p in authorizations[domain]): + break + else: + verified.append(prop) + return verified + + def _reference_renderer_catalog_errors( adagents_data: dict[str, Any], *, source_url: str | None ) -> list[AdagentsEntryError]: diff --git a/tests/fixtures/public_api_snapshot.json b/tests/fixtures/public_api_snapshot.json index 6e9c3d0ff..c17644746 100644 --- a/tests/fixtures/public_api_snapshot.json +++ b/tests/fixtures/public_api_snapshot.json @@ -719,6 +719,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 +747,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": { diff --git a/tests/test_adagents_property_resolution.py b/tests/test_adagents_property_resolution.py new file mode 100644 index 000000000..450f75133 --- /dev/null +++ b/tests/test_adagents_property_resolution.py @@ -0,0 +1,685 @@ +"""Publisher attribution and verification without live network dependencies.""" + +import asyncio +import subprocess +import sys +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager +from copy import deepcopy +from typing import Any +from unittest.mock import AsyncMock + +import httpx +import pytest + +from adcp import ( + fetch_adagents, + get_properties_by_agent, + properties_for_publisher_domain, + resolve_properties_for_agent, + verify_cross_domain_properties, +) +from adcp.exceptions import ( + AdagentsNotFoundError, + AdagentsTimeoutError, + AdagentsValidationError, +) + +AGENT = "https://sales.example.com/mcp" + + +def site(domain: str, property_id: str = "site", **extra: Any) -> dict[str, Any]: + return { + "property_id": property_id, + "property_type": "website", + "name": domain, + "identifiers": [{"type": "domain", "value": domain}], + **extra, + } + + +def manifest(properties: list[dict[str, Any]], *, bare: bool = False) -> dict[str, Any]: + entry: dict[str, Any] = {"url": AGENT, "authorized_for": "Sales"} + if not bare: + entry.update( + authorization_type="property_ids", property_ids=[p["property_id"] for p in properties] + ) + return {"authorized_agents": [entry], "properties": properties} + + +@pytest.mark.parametrize("publisher", ["example.org", "www.example.org", " WWW.EXAMPLE.ORG. "]) +def test_attribution_keeps_authorized_subdomains(publisher: str) -> None: + props = [ + site(d, d) + for d in [ + "example.org", + "www.example.org", + "m.example.org", + "mobile.example.org", + "shop.example.org", + "ctv.example.org", + "deep.shop.example.org", + "other.org", + "notexample.org", + "example.org.attacker.org", + ] + ] + document = manifest(props) + before = deepcopy(document) + resolved = get_properties_by_agent(document, AGENT) + selected = properties_for_publisher_domain(resolved, publisher) + assert selected == props[:7] + assert all(a is b for a, b in zip(selected, props)) + assert document == before + + +def test_attribution_explicit_publisher_takes_precedence() -> None: + selected = site("elsewhere.org", publisher_domain="WWW.EXAMPLE.ORG.") + wrong = site("example.org", publisher_domain="other.org") + child_publisher = site("example.org", publisher_domain="shop.example.org") + app = { + "property_type": "mobile_app", + "identifiers": [{"type": "bundle_id", "value": "org.app"}], + } + props = [wrong, selected, child_publisher, app] + assert properties_for_publisher_domain(props, "example.org") == [selected, app] + assert properties_for_publisher_domain(props, "example.org", include_unattributed=False) == [ + selected + ] + + +@pytest.mark.parametrize("bad", [None, "", 1, [], "https://example.org", "bad..org"]) +def test_invalid_claims_are_not_unattributed(bad: Any) -> None: + assert ( + properties_for_publisher_domain([site("example.org", publisher_domain=bad)], "example.org") + == [] + ) + assert properties_for_publisher_domain([site(bad)], "example.org") == [] + + +@pytest.mark.parametrize( + "bad", ["", "https://example.org", "*.example.org", "example.org:443", "bad..org"] +) +def test_invalid_publisher_arguments_raise(bad: str) -> None: + with pytest.raises(AdagentsValidationError): + properties_for_publisher_domain([], bad) + with pytest.raises(AdagentsValidationError): + resolve_properties_for_agent( + manifest([], bare=True), AGENT, mode="permissive", publisher_domain=bad + ) + + +def test_attribution_handles_multi_label_suffixes_and_idna() -> None: + uk = site("shop.publisher.co.uk") + idn = site("shop.xn--bcher-kva.example") + assert properties_for_publisher_domain([uk], "www.publisher.co.uk") == [uk] + assert properties_for_publisher_domain([idn], "bücher.example") == [idn] + + +def test_permissive_gate_uses_only_domain_identifiers() -> None: + good = site("ctv.example.org", publisher_domain="other.org") + forged = site("victim.org", publisher_domain="example.org") + app = { + "property_id": "app", + "publisher_domain": "example.org", + "identifiers": [{"type": "bundle_id", "value": "org.app"}], + } + venue = {"property_id": "venue", "identifiers": [{"type": "venue_id", "value": "example.org"}]} + props = [ + good, + forged, + app, + venue, + site("example.org.attacker.org"), + site("https://shop.example.org"), + ] + document = manifest(props, bare=True) + before = deepcopy(document) + assert resolve_properties_for_agent(document, AGENT, mode="permissive") == props + assert resolve_properties_for_agent( + document, AGENT, mode="permissive", publisher_domain="www.example.org" + ) == [good] + assert document == before + + +@pytest.mark.parametrize("mode", ["strict", "permissive"]) +def test_domain_gate_does_not_change_typed_selectors(mode: str) -> None: + props = [site("other.org")] + assert ( + resolve_properties_for_agent( + manifest(props), AGENT, mode=mode, publisher_domain="example.org" + ) + == props + ) + + +def test_gate_does_not_expand_unlisted_or_ambiguous_authorizations() -> None: + document = manifest([site("example.org")], bare=True) + assert ( + resolve_properties_for_agent( + document, AGENT + "/other", mode="permissive", publisher_domain="example.org" + ) + == [] + ) + document["authorized_agents"] *= 2 + assert ( + resolve_properties_for_agent( + document, AGENT, mode="permissive", publisher_domain="example.org" + ) + == [] + ) + document["authorized_agents"] = [ + {"url": AGENT, "authorized_for": "Sales", "authorization_type": "unknown"} + ] + assert ( + resolve_properties_for_agent( + document, AGENT, mode="permissive", publisher_domain="example.org" + ) + == [] + ) + + +def test_gate_honors_existing_revocations() -> None: + prop = site("ctv.example.org", publisher_domain="example.org") + document = manifest([prop], bare=True) + document["revoked_publisher_domains"] = [{"publisher_domain": "example.org"}] + assert ( + resolve_properties_for_agent( + document, AGENT, mode="permissive", publisher_domain="example.org" + ) + == [] + ) + + +async def test_verification_keeps_local_and_unanchored_inventory_without_fetching() -> None: + props = [ + site("example.org"), + site("ctv.example.org"), + {"property_id": "app", "identifiers": [{"type": "bundle_id", "value": "org.app"}]}, + ] + fetcher = AsyncMock() + assert ( + await verify_cross_domain_properties(props, "WWW.EXAMPLE.ORG.", AGENT, fetcher=fetcher) + == props + ) + fetcher.assert_not_awaited() + + +async def test_verification_fetches_once_per_domain_and_retains_input_order() -> None: + first = site("remote.example", "first") + second = site("remote.example", "second") + local = site("publisher.example", "local") + props = [first, local, second] + before = deepcopy(props) + fetcher = AsyncMock(return_value=manifest([first, second])) + result = await verify_cross_domain_properties( + props, "publisher.example", AGENT, fetcher=fetcher + ) + assert result == props + assert all(a is b for a, b in zip(result, props)) + fetcher.assert_awaited_once_with("remote.example") + assert props == before + + +@pytest.mark.parametrize( + "authorization_type", + ["property_ids", "property_tags", "inline_properties", "publisher_properties"], +) +async def test_remote_selectors_resolve_inventory_independently_of_file_local_ids( + authorization_type: str, +) -> None: + claim = site("remote.example", "parent-id") + remote_property = site( + "remote.example", "child-id", publisher_domain="remote.example", tags=["sales"] + ) + remote = manifest([remote_property]) + entry = remote["authorized_agents"][0] + entry.pop("property_ids") + entry["authorization_type"] = authorization_type + entry.update( + { + "property_ids": {"property_ids": ["child-id"]}, + "property_tags": {"property_tags": ["sales"]}, + "inline_properties": {"properties": [remote_property]}, + "publisher_properties": { + "publisher_properties": [ + {"publisher_domain": "remote.example", "selection_type": "all"} + ] + }, + }[authorization_type] + ) + assert await verify_cross_domain_properties( + [claim], "publisher.example", AGENT, fetcher=AsyncMock(return_value=remote) + ) == [claim] + + +async def test_verification_refreshes_between_calls_to_observe_revocation() -> None: + prop = site("remote.example") + remote = manifest([prop]) + revoked = {**remote, "revoked_publisher_domains": [{"publisher_domain": "remote.example"}]} + fetcher = AsyncMock(side_effect=[remote, revoked]) + assert await verify_cross_domain_properties( + [prop], "publisher.example", AGENT, fetcher=fetcher + ) == [prop] + assert ( + await verify_cross_domain_properties([prop], "publisher.example", AGENT, fetcher=fetcher) + == [] + ) + assert fetcher.await_count == 2 + + +@pytest.mark.parametrize( + "remote", + [ + {"authorized_agents": []}, + manifest([site("remote.example")], bare=True), + manifest([site("elsewhere.example")]), + manifest([site("remote.example", property_type="mobile_app")]), + { + "authorized_agents": [ + { + "url": AGENT, + "authorized_for": "Sales", + "authorization_type": "property_ids", + "property_ids": [], + } + ] + }, + { + "authorized_agents": [ + { + "url": "https://other.example", + "authorization_type": "inline_properties", + "properties": [site("remote.example")], + } + ] + }, + ], +) +async def test_listing_agent_or_reusing_property_id_is_insufficient(remote: dict[str, Any]) -> None: + fetcher = AsyncMock(return_value=remote) + assert ( + await verify_cross_domain_properties( + [site("remote.example")], "publisher.example", AGENT, fetcher=fetcher + ) + == [] + ) + + +async def test_every_claimed_identifier_must_be_authorized() -> None: + claim = site("remote.example") + claim["identifiers"].append({"type": "bundle_id", "value": "org.forged"}) + assert ( + await verify_cross_domain_properties( + [claim], + "publisher.example", + AGENT, + fetcher=AsyncMock(return_value=manifest([site("remote.example")])), + ) + == [] + ) + + +async def test_forged_local_publisher_anchor_does_not_bypass_external_domain_check() -> None: + prop = site("victim.example", publisher_domain="publisher.example") + fetcher = AsyncMock(return_value=manifest([], bare=True)) + assert ( + await verify_cross_domain_properties([prop], "publisher.example", AGENT, fetcher=fetcher) + == [] + ) + fetcher.assert_awaited_once_with("victim.example") + + +async def test_every_external_domain_must_vouch() -> None: + prop = site("first.example") + prop["identifiers"].append({"type": "domain", "value": "second.example"}) + fetcher = AsyncMock(side_effect=[manifest([prop]), {"authorized_agents": []}]) + assert ( + await verify_cross_domain_properties([prop], "publisher.example", AGENT, fetcher=fetcher) + == [] + ) + assert [c.args[0] for c in fetcher.await_args_list] == ["first.example", "second.example"] + + +async def test_bundle_only_inventory_uses_explicit_publisher_anchor() -> None: + prop = { + "property_id": "app", + "property_type": "mobile_app", + "publisher_domain": "remote.example", + "identifiers": [{"type": "bundle_id", "value": "org.app"}], + } + fetcher = AsyncMock(return_value=manifest([prop])) + assert await verify_cross_domain_properties( + [prop], "publisher.example", AGENT, fetcher=fetcher + ) == [prop] + fetcher.assert_awaited_once_with("remote.example") + # Matching a file-local ID cannot override a different bundle identifier. + remote = {**prop, "identifiers": [{"type": "bundle_id", "value": "org.other"}]} + assert ( + await verify_cross_domain_properties( + [prop], "publisher.example", AGENT, fetcher=AsyncMock(return_value=manifest([remote])) + ) + == [] + ) + + # An app with no publisher anchor in a shared manager file cannot vouch for B. + unanchored = {k: v for k, v in prop.items() if k != "publisher_domain"} + assert ( + await verify_cross_domain_properties( + [prop], + "publisher.example", + AGENT, + fetcher=AsyncMock(return_value=manifest([unanchored])), + ) + == [] + ) + + +async def test_publisher_scoped_property_id_fallback() -> None: + prop = {"property_id": "venue", "publisher_domain": "remote.example"} + assert await verify_cross_domain_properties( + [prop], "publisher.example", AGENT, fetcher=AsyncMock(return_value=manifest([prop])) + ) == [prop] + other = {**prop, "publisher_domain": "elsewhere.example"} + assert ( + await verify_cross_domain_properties( + [prop], "publisher.example", AGENT, fetcher=AsyncMock(return_value=manifest([other])) + ) + == [] + ) + + +@pytest.mark.parametrize("inline", [False, True]) +async def test_remote_revocation_overrides_authorization(inline: bool) -> None: + prop = site("remote.example", publisher_domain="remote.example") + remote = manifest([prop]) + if inline: + remote["authorized_agents"][0] = { + "url": AGENT, + "authorization_type": "inline_properties", + "properties": [prop], + } + remote["revoked_publisher_domains"] = [{"publisher_domain": "WWW.REMOTE.EXAMPLE."}] + assert ( + await verify_cross_domain_properties( + [prop], "publisher.example", AGENT, fetcher=AsyncMock(return_value=remote) + ) + == [] + ) + + +@pytest.mark.parametrize( + "error", + [ + AdagentsNotFoundError("remote.example"), + AdagentsTimeoutError("remote.example", 10), + AdagentsValidationError("bad manifest"), + httpx.ConnectError("unavailable"), + OSError("DNS unavailable"), + ], +) +async def test_fetch_failure_is_closed_and_cached(error: Exception) -> None: + props = [ + site("remote.example", "one"), + site("remote.example", "two"), + site("publisher.example", "local"), + ] + fetcher = AsyncMock(side_effect=error) + assert ( + await verify_cross_domain_properties(props, "publisher.example", AGENT, fetcher=fetcher) + == props[-1:] + ) + fetcher.assert_awaited_once() + + +@pytest.mark.parametrize("selector", ["property_ids", "property_tags"]) +async def test_malformed_remote_inventory_excludes_only_that_origin(selector: str) -> None: + from adcp import validate_adagents_structure + + remote = manifest([site("remote.example")]) + if selector == "property_ids": + remote["properties"][0]["property_id"] = [] + else: + remote["authorized_agents"][0].pop("property_ids") + remote["authorized_agents"][0].update( + authorization_type="property_tags", property_tags=["sales"] + ) + remote["properties"][0]["tags"] = None + # Fetch-time structural validation checks entries, not the inventory catalog. + assert validate_adagents_structure(remote).schema_valid + props = [site("remote.example"), site("publisher.example", "local")] + assert ( + await verify_cross_domain_properties( + props, "publisher.example", AGENT, fetcher=AsyncMock(return_value=remote) + ) + == props[-1:] + ) + + +async def test_fetcher_programming_errors_are_not_hidden() -> None: + with pytest.raises(TypeError, match="fetcher bug"): + await verify_cross_domain_properties( + [site("remote.example")], + "publisher.example", + AGENT, + fetcher=AsyncMock(side_effect=TypeError("fetcher bug")), + ) + + +async def test_cancellation_is_propagated() -> None: + with pytest.raises(asyncio.CancelledError): + await verify_cross_domain_properties( + [site("remote.example")], + "publisher.example", + AGENT, + fetcher=AsyncMock(side_effect=asyncio.CancelledError), + ) + + +@pytest.mark.parametrize("bad", [None, "https://remote.example", "", []]) +async def test_invalid_cross_domain_claims_are_excluded_without_fetch(bad: Any) -> None: + fetcher = AsyncMock() + assert ( + await verify_cross_domain_properties( + [site(bad)], "publisher.example", AGENT, fetcher=fetcher + ) + == [] + ) + assert ( + await verify_cross_domain_properties( + [site("publisher.example", publisher_domain=bad)], + "publisher.example", + AGENT, + fetcher=fetcher, + ) + == [] + ) + fetcher.assert_not_awaited() + + +@pytest.mark.parametrize("module", ["adcp", "adcp.adagents"]) +def test_helpers_import_in_a_fresh_interpreter(module: str) -> None: + code = ( + f"from {module} import fetch_adagents, properties_for_publisher_domain, " + "verify_cross_domain_properties; " + "assert properties_for_publisher_domain([], 'publisher.example') == []" + ) + result = subprocess.run( + [sys.executable, "-c", code], capture_output=True, text=True, timeout=60 + ) + assert result.returncode == 0, result.stderr + + +@pytest.mark.parametrize( + ("publisher", "victim"), + [ + ("www.com", "victim.com"), + ("www.co.uk", "victim.co.uk"), + ("www.github.io", "victim.github.io"), + ], +) +async def test_www_alias_never_broadens_authority_to_public_or_private_suffix( + publisher: str, victim: str +) -> None: + local = site(publisher, "local") + remote = site(victim, "victim") + assert properties_for_publisher_domain([local, remote], publisher) == [local] + assert resolve_properties_for_agent( + manifest([local, remote], bare=True), AGENT, mode="permissive", publisher_domain=publisher + ) == [local] + fetcher = AsyncMock(return_value={"authorized_agents": []}) + assert await verify_cross_domain_properties( + [local, remote], publisher, AGENT, fetcher=fetcher + ) == [local] + fetcher.assert_awaited_once_with(victim) + + +async def test_verified_publisher_anchor_covers_its_own_subdomain_inventory() -> None: + prop = site("ctv.publisher.example", publisher_domain="publisher.example") + fetcher = AsyncMock(return_value=manifest([prop])) + assert await verify_cross_domain_properties( + [prop], "network.example", AGENT, fetcher=fetcher + ) == [prop] + fetcher.assert_awaited_once_with("publisher.example") + assert ( + await verify_cross_domain_properties( + [prop], + "network.example", + AGENT, + fetcher=AsyncMock(return_value={"authorized_agents": []}), + ) + == [] + ) + + +async def test_child_publisher_cannot_vouch_for_parent_domain() -> None: + prop = site("publisher.example", publisher_domain="ctv.publisher.example") + fetcher = AsyncMock(side_effect=[manifest([prop]), {"authorized_agents": []}]) + assert ( + await verify_cross_domain_properties([prop], "network.example", AGENT, fetcher=fetcher) + == [] + ) + assert [c.args[0] for c in fetcher.await_args_list] == [ + "ctv.publisher.example", + "publisher.example", + ] + + +async def test_subdomain_identifier_requires_external_authorization() -> None: + prop = site("ctv.victim.example") + prop["identifiers"][0]["type"] = "subdomain" + fetcher = AsyncMock(return_value={"authorized_agents": []}) + assert ( + await verify_cross_domain_properties([prop], "attacker.example", AGENT, fetcher=fetcher) + == [] + ) + fetcher.assert_awaited_once_with("ctv.victim.example") + assert properties_for_publisher_domain([prop], "victim.example") == [prop] + assert properties_for_publisher_domain([prop], "attacker.example") == [] + assert await verify_cross_domain_properties( + [prop], "network.example", AGENT, fetcher=AsyncMock(return_value=manifest([prop])) + ) == [prop] + # The bare-entry gate continues to require the documented domain type. + assert ( + resolve_properties_for_agent( + manifest([prop], bare=True), AGENT, mode="permissive", publisher_domain="victim.example" + ) + == [] + ) + + +@pytest.mark.parametrize("kind", ["domain", "subdomain"]) +async def test_unicode_and_idna_inventory_identities_match(kind: str) -> None: + prop = site("bücher.example") + prop["identifiers"][0]["type"] = kind + remote = site("xn--bcher-kva.example") + remote["identifiers"][0]["type"] = kind + fetcher = AsyncMock(return_value=manifest([remote])) + assert await verify_cross_domain_properties( + [prop], "network.example", AGENT, fetcher=fetcher + ) == [prop] + fetcher.assert_awaited_once_with("xn--bcher-kva.example") + + +async def test_wildcard_inventory_is_attributed_gated_and_verified_at_its_base() -> None: + prop = site("*.publisher.example") + assert properties_for_publisher_domain([prop], "publisher.example") == [prop] + assert resolve_properties_for_agent( + manifest([prop], bare=True), AGENT, mode="permissive", publisher_domain="publisher.example" + ) == [prop] + fetcher = AsyncMock(return_value=manifest([prop])) + assert await verify_cross_domain_properties( + [prop], "network.example", AGENT, fetcher=fetcher + ) == [prop] + fetcher.assert_awaited_once_with("publisher.example") + # A single authorized subdomain cannot vouch for every subdomain in a wildcard. + assert ( + await verify_cross_domain_properties( + [prop], + "network.example", + AGENT, + fetcher=AsyncMock(return_value=manifest([site("ctv.publisher.example")])), + ) + == [] + ) + with pytest.raises(AdagentsValidationError): + properties_for_publisher_domain([prop], "*.publisher.example") + + +async def test_default_fetcher_is_sdk_discovery(monkeypatch: pytest.MonkeyPatch) -> None: + prop = site("remote.example") + fetcher = AsyncMock(return_value=manifest([prop])) + monkeypatch.setattr("adcp.adagents.fetch_adagents", fetcher) + assert await verify_cross_domain_properties([prop], "publisher.example", AGENT) == [prop] + fetcher.assert_awaited_once_with("remote.example") + + +@pytest.mark.parametrize( + "delegation", ["direct", "authoritative_location", "ads_txt_managerdomain"] +) +@pytest.mark.parametrize("authorized", [True, False]) +async def test_real_discovery_requires_publisher_vouching( + delegation: str, authorized: bool +) -> None: + prop = site("remote.example", publisher_domain="remote.example") + parent = manifest([prop]) + no_vouch = manifest([site("unrelated.example")]) + requested: list[str] = [] + + def handler(request: httpx.Request) -> httpx.Response: + url = str(request.url) + requested.append(url) + if url == "https://remote.example/.well-known/adagents.json": + if delegation == "authoritative_location": + return httpx.Response( + 200, + json={ + "authoritative_location": "https://manager.example/.well-known/adagents.json" + }, + ) + if delegation == "ads_txt_managerdomain": + return httpx.Response(404) + return httpx.Response(200, json=parent if authorized else no_vouch) + if url == "https://remote.example/ads.txt": + return httpx.Response(200, text="MANAGERDOMAIN=manager.example\n") + if url == "https://manager.example/.well-known/adagents.json": + return httpx.Response(200, json=parent if authorized else no_vouch) + raise AssertionError(f"Unexpected fetch: {url}") + + @asynccontextmanager + async def transport(url: str, timeout: float) -> AsyncIterator[httpx.AsyncClient]: + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: + yield client + + async def fetcher(domain: str) -> dict[str, Any]: + return await fetch_adagents(domain, transport_factory=transport) + + resolved = get_properties_by_agent(parent, AGENT) + assert await verify_cross_domain_properties( + resolved, "manager.example", AGENT, fetcher=fetcher + ) == ([prop] if authorized else []) + assert requested[0] == "https://remote.example/.well-known/adagents.json" + if delegation == "ads_txt_managerdomain": + assert "https://remote.example/ads.txt" in requested + if delegation != "direct": + assert "https://manager.example/.well-known/adagents.json" in requested diff --git a/tests/type_checks/adagents_property_resolution.py b/tests/type_checks/adagents_property_resolution.py new file mode 100644 index 000000000..54b3ef80a --- /dev/null +++ b/tests/type_checks/adagents_property_resolution.py @@ -0,0 +1,25 @@ +"""Publisher sync helpers accept typed custom discovery and preserve dictionaries.""" + +from typing import Any + +from adcp import ( + fetch_adagents, + properties_for_publisher_domain, + resolve_properties_for_agent, + verify_cross_domain_properties, +) + + +async def fetch_publisher(domain: str) -> dict[str, Any]: + return await fetch_adagents(domain, timeout=5.0) + + +async def sync_publisher(domain: str, agent_url: str) -> list[dict[str, Any]]: + document = await fetch_adagents(domain, validate_structure=False) + authorized = resolve_properties_for_agent( + document, agent_url, mode="permissive", publisher_domain=domain + ) + verified = await verify_cross_domain_properties( + authorized, domain, agent_url, fetcher=fetch_publisher + ) + return properties_for_publisher_domain(verified, domain, include_unattributed=False) From 6df4dacaf8c51251b99f1b737e465d4c42937303 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Thu, 29 Oct 2026 23:43:37 +0000 Subject: [PATCH 2/2] fix(adagents): explain invalid publisher claim exclusions --- src/adcp/adagents.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/adcp/adagents.py b/src/adcp/adagents.py index f5a449239..c199f947f 100644 --- a/src/adcp/adagents.py +++ b/src/adcp/adagents.py @@ -723,7 +723,8 @@ def properties_for_publisher_domain( if _publisher_attribution_domain(prop["publisher_domain"]) == domain: selected.append(prop) except AdagentsValidationError: - pass + # An invalid explicit publisher claim cannot fall back to identifiers. + logger.debug("Skipping property with invalid publisher_domain") continue domain_ids = _property_domain_identifiers(prop, include_subdomains=True) if any(_domain_identifier_is_under_publisher(i, domain) for i in domain_ids) or (