Module adcp.canonical_formats.references

Fetch and cache immutable canonical-format reference documents.

AdCP 3.1 uses the same uri + sha256:<digest> reference shape for ProductFormatDeclaration.format_schema and platform_extensions. This module gives SDK adopters a hardened resolver for those references:

• HTTPS-only, redirect-free fetches.
• public-network SSRF validation before connect.
• DNS-rebinding defense via IP-pinned httpx transport by default.
• streaming body cap and digest verification.
• immutable cache keyed by uri@digest.
• optional JSON Schema validation for format_schema documents, including a conservative $ref sandbox.

The helper returns structured results instead of raising for normal negative outcomes so callers can surface stable conformance diagnostics without leaking internal network details.

Functions

def parse_canonical_reference(reference: CanonicalReference | Mapping[str, Any] | str | Any) ‑> tuple[CanonicalReference | None, CanonicalReferenceResult | None]

Expand source code

def parse_canonical_reference(
    reference: CanonicalReference | Mapping[str, Any] | str | Any,
) -> tuple[CanonicalReference | None, CanonicalReferenceResult | None]:
    """Parse supported reference inputs into a normalized dataclass."""

    if isinstance(reference, CanonicalReference):
        parsed = reference
    elif isinstance(reference, str):
        if _REFERENCE_MARKER not in reference:
            return None, _invalid_reference("reference must use uri@sha256:<digest> form")
        compact_uri, digest_hex = reference.rsplit(_REFERENCE_MARKER, 1)
        parsed = CanonicalReference(uri=compact_uri, digest=f"{_DIGEST_PREFIX}{digest_hex}")
    elif isinstance(reference, Mapping):
        mapped_uri = reference.get("uri")
        mapped_digest = reference.get("digest")
        if mapped_uri is None or mapped_digest is None:
            return None, _invalid_reference("reference mapping requires uri and digest")
        parsed = CanonicalReference(uri=str(mapped_uri), digest=str(mapped_digest))
    elif hasattr(reference, "uri") and hasattr(reference, "digest"):
        raw_uri = getattr(reference, "uri")
        raw_digest = getattr(reference, "digest")
        if raw_uri is None or raw_digest is None:
            return None, _invalid_reference("reference object requires uri and digest")
        parsed = CanonicalReference(uri=str(raw_uri), digest=str(raw_digest))
    else:
        return None, _invalid_reference("unsupported canonical reference type")

    if not _is_valid_digest(parsed.digest):
        return None, _invalid_reference("digest must be sha256:<64 lowercase hex characters>")
    if not parsed.uri:
        return None, _invalid_reference("reference URI is empty")
    return parsed, None

Parse supported reference inputs into a normalized dataclass.

Classes

class CanonicalReference (uri: str, digest: str)

Expand source code

@dataclass(frozen=True)
class CanonicalReference:
    """Normalized immutable reference.

    Callers may pass either this dataclass, a mapping with ``uri`` and
    ``digest`` keys, or a compact string in ``uri@sha256:<digest>`` form.
    """

    uri: str
    digest: str

    @property
    def cache_key(self) -> str:
        return f"{self.uri}@{self.digest}"

Normalized immutable reference.

Callers may pass either this dataclass, a mapping with uri and digest keys, or a compact string in uri@sha256:<digest> form.

Instance variables

prop cache_key : str

Expand source code

@property
def cache_key(self) -> str:
    return f"{self.uri}@{self.digest}"

var digest : str

var uri : str

class CanonicalReferenceResolver (*,
timeout: float = 5.0,
max_body_bytes: int = 1048576,
max_schema_refs: int = 256,
max_ref_depth: int = 8,
max_schema_keywords: int = 10000,
max_schema_depth: int = 128,
transport_factory: TransportFactory | None = None)

Expand source code

class CanonicalReferenceResolver:
    """Resolve immutable ``format_schema`` and ``platform_extensions`` refs.

    The resolver owns an in-memory cache keyed by ``uri@digest``. Construct one
    per adopter component or request context; no process-global mutable
    configuration is used.
    """

    def __init__(
        self,
        *,
        timeout: float = DEFAULT_REFERENCE_TIMEOUT_SECONDS,
        max_body_bytes: int = DEFAULT_REFERENCE_BODY_LIMIT_BYTES,
        max_schema_refs: int = DEFAULT_MAX_SCHEMA_REFS,
        max_ref_depth: int = DEFAULT_MAX_REF_DEPTH,
        max_schema_keywords: int = DEFAULT_MAX_SCHEMA_KEYWORDS,
        max_schema_depth: int = DEFAULT_MAX_SCHEMA_DEPTH,
        transport_factory: TransportFactory | None = None,
    ) -> None:
        self._timeout = timeout
        self._max_body_bytes = max_body_bytes
        self._max_schema_refs = max_schema_refs
        self._max_ref_depth = max_ref_depth
        self._max_schema_keywords = max_schema_keywords
        self._max_schema_depth = max_schema_depth
        self._transport_factory = transport_factory
        self._cache: dict[str, CanonicalReferenceResult] = {}

    @property
    def cache(self) -> Mapping[str, CanonicalReferenceResult]:
        """Read-only view of resolved immutable references."""

        return MappingProxyType({key: _copy_result(value) for key, value in self._cache.items()})

    def resolve_platform_extension(
        self,
        reference: CanonicalReference | Mapping[str, Any] | str,
    ) -> CanonicalReferenceResult:
        """Fetch and digest-verify a ``platform_extensions`` reference."""

        parsed, error = parse_canonical_reference(reference)
        if error is not None:
            return error
        assert parsed is not None
        cached = self._cache.get(parsed.cache_key)
        if cached is not None:
            return _copy_result(cached, from_cache=True)

        fetch_result = self._fetch(parsed)
        if fetch_result.status is CanonicalReferenceStatus.RESOLVED:
            self._cache[parsed.cache_key] = _copy_result(fetch_result)
        return fetch_result

    def resolve_format_schema(
        self,
        reference: CanonicalReference | Mapping[str, Any] | str,
    ) -> CanonicalReferenceResult:
        """Fetch, digest-verify, and validate a ``format_schema`` document."""

        parsed, error = parse_canonical_reference(reference)
        if error is not None:
            return error
        assert parsed is not None
        cached = self._cache.get(parsed.cache_key)
        if cached is not None:
            if cached.document is None and cached.body is not None:
                schema_result = self._validate_schema(parsed, cached.body)
                if schema_result.status is CanonicalReferenceStatus.RESOLVED:
                    self._cache[parsed.cache_key] = _copy_result(schema_result)
                return _copy_result(schema_result, from_cache=True)
            return _copy_result(cached, from_cache=True)

        fetch_result = self._fetch(parsed)
        if fetch_result.status is not CanonicalReferenceStatus.RESOLVED:
            return fetch_result
        assert fetch_result.body is not None
        self._cache[parsed.cache_key] = _copy_result(fetch_result)

        schema_result = self._validate_schema(parsed, fetch_result.body)
        if schema_result.status is CanonicalReferenceStatus.RESOLVED:
            self._cache[parsed.cache_key] = _copy_result(schema_result)
        return schema_result

    def _fetch(self, reference: CanonicalReference) -> CanonicalReferenceResult:
        resolved, unsafe = _resolve_public_https(reference.uri)
        if unsafe is not None:
            return CanonicalReferenceResult(
                status=CanonicalReferenceStatus.BLOCKED_UNSAFE_URL,
                reference=reference,
                message=unsafe,
            )
        assert resolved is not None

        transport = (
            self._transport_factory(resolved.host, resolved.resolved_ip)
            if self._transport_factory is not None
            else IpPinnedTransport(hostname=resolved.host, resolved_ip=resolved.resolved_ip)
        )
        try:
            with httpx.Client(
                transport=transport,
                timeout=self._timeout,
                follow_redirects=False,
                trust_env=False,
            ) as client:
                with client.stream(
                    "GET",
                    reference.uri,
                    headers={"Accept": "application/schema+json, application/json"},
                ) as response:
                    if 300 <= response.status_code < 400:
                        return CanonicalReferenceResult(
                            status=CanonicalReferenceStatus.BLOCKED_UNSAFE_URL,
                            reference=reference,
                            message="redirects are not allowed for canonical references",
                        )
                    if response.status_code != 200:
                        return CanonicalReferenceResult(
                            status=CanonicalReferenceStatus.NETWORK_ERROR,
                            reference=reference,
                            message="reference fetch failed",
                        )
                    body = _read_capped_body(response, self._max_body_bytes)
        except _BodyTooLargeError:
            return CanonicalReferenceResult(
                status=CanonicalReferenceStatus.BODY_TOO_LARGE,
                reference=reference,
                message="reference body exceeded configured size cap",
            )
        except httpx.HTTPError:
            return CanonicalReferenceResult(
                status=CanonicalReferenceStatus.NETWORK_ERROR,
                reference=reference,
                message="reference fetch failed",
            )

        actual_digest = _DIGEST_PREFIX + hashlib.sha256(body).hexdigest()
        if actual_digest != reference.digest:
            return CanonicalReferenceResult(
                status=CanonicalReferenceStatus.DIGEST_MISMATCH,
                reference=reference,
                message="reference digest mismatch",
            )
        return CanonicalReferenceResult(
            status=CanonicalReferenceStatus.RESOLVED,
            reference=reference,
            body=body,
        )

    def _validate_schema(
        self,
        reference: CanonicalReference,
        body: bytes,
    ) -> CanonicalReferenceResult:
        try:
            document = json.loads(body.decode("utf-8"))
        except (UnicodeDecodeError, ValueError):
            return CanonicalReferenceResult(
                status=CanonicalReferenceStatus.INVALID_SCHEMA,
                reference=reference,
                body=body,
                message="format_schema is not valid UTF-8 JSON",
            )
        if not isinstance(document, dict):
            return CanonicalReferenceResult(
                status=CanonicalReferenceStatus.INVALID_SCHEMA,
                reference=reference,
                body=body,
                message="format_schema root must be a JSON object",
            )

        ref_error = _validate_schema_refs(
            document,
            base_uri=reference.uri,
            max_refs=self._max_schema_refs,
            max_ref_depth=self._max_ref_depth,
            max_keywords=self._max_schema_keywords,
            max_depth=self._max_schema_depth,
        )
        if ref_error is not None:
            return CanonicalReferenceResult(
                status=CanonicalReferenceStatus.INVALID_SCHEMA,
                reference=reference,
                body=body,
                message=ref_error,
            )

        validator_class, draft_error = _validator_class_for_schema(document)
        if draft_error is not None:
            return CanonicalReferenceResult(
                status=CanonicalReferenceStatus.INVALID_SCHEMA,
                reference=reference,
                body=body,
                message=draft_error,
            )
        try:
            validator_class.check_schema(document)
        except (jsonschema.exceptions.SchemaError, RecursionError):
            return CanonicalReferenceResult(
                status=CanonicalReferenceStatus.INVALID_SCHEMA,
                reference=reference,
                body=body,
                message="format_schema failed JSON Schema validation",
            )

        return CanonicalReferenceResult(
            status=CanonicalReferenceStatus.RESOLVED,
            reference=reference,
            body=body,
            document=document,
        )

Resolve immutable format_schema and platform_extensions refs.

The resolver owns an in-memory cache keyed by uri@digest. Construct one per adopter component or request context; no process-global mutable configuration is used.

Instance variables

prop cache : Mapping[str, CanonicalReferenceResult]

Expand source code

@property
def cache(self) -> Mapping[str, CanonicalReferenceResult]:
    """Read-only view of resolved immutable references."""

    return MappingProxyType({key: _copy_result(value) for key, value in self._cache.items()})

Read-only view of resolved immutable references.

Methods

def resolve_format_schema(self,
reference: CanonicalReference | Mapping[str, Any] | str) ‑> CanonicalReferenceResult

Expand source code

def resolve_format_schema(
    self,
    reference: CanonicalReference | Mapping[str, Any] | str,
) -> CanonicalReferenceResult:
    """Fetch, digest-verify, and validate a ``format_schema`` document."""

    parsed, error = parse_canonical_reference(reference)
    if error is not None:
        return error
    assert parsed is not None
    cached = self._cache.get(parsed.cache_key)
    if cached is not None:
        if cached.document is None and cached.body is not None:
            schema_result = self._validate_schema(parsed, cached.body)
            if schema_result.status is CanonicalReferenceStatus.RESOLVED:
                self._cache[parsed.cache_key] = _copy_result(schema_result)
            return _copy_result(schema_result, from_cache=True)
        return _copy_result(cached, from_cache=True)

    fetch_result = self._fetch(parsed)
    if fetch_result.status is not CanonicalReferenceStatus.RESOLVED:
        return fetch_result
    assert fetch_result.body is not None
    self._cache[parsed.cache_key] = _copy_result(fetch_result)

    schema_result = self._validate_schema(parsed, fetch_result.body)
    if schema_result.status is CanonicalReferenceStatus.RESOLVED:
        self._cache[parsed.cache_key] = _copy_result(schema_result)
    return schema_result

Fetch, digest-verify, and validate a format_schema document.

def resolve_platform_extension(self,
reference: CanonicalReference | Mapping[str, Any] | str) ‑> CanonicalReferenceResult

Expand source code

def resolve_platform_extension(
    self,
    reference: CanonicalReference | Mapping[str, Any] | str,
) -> CanonicalReferenceResult:
    """Fetch and digest-verify a ``platform_extensions`` reference."""

    parsed, error = parse_canonical_reference(reference)
    if error is not None:
        return error
    assert parsed is not None
    cached = self._cache.get(parsed.cache_key)
    if cached is not None:
        return _copy_result(cached, from_cache=True)

    fetch_result = self._fetch(parsed)
    if fetch_result.status is CanonicalReferenceStatus.RESOLVED:
        self._cache[parsed.cache_key] = _copy_result(fetch_result)
    return fetch_result

Fetch and digest-verify a platform_extensions reference.

class CanonicalReferenceResult (status: CanonicalReferenceStatus,
reference: CanonicalReference | None = None,
body: bytes | None = None,
document: Any | None = None,
message: str | None = None,
from_cache: bool = False)

Expand source code

@dataclass(frozen=True)
class CanonicalReferenceResult:
    """Structured resolver result.

    ``body`` is populated only on successful fetches. ``document`` is populated
    for successful ``format_schema`` resolution after JSON parsing. ``message``
    is intentionally coarse and should be safe to show in diagnostics.
    """

    status: CanonicalReferenceStatus
    reference: CanonicalReference | None = None
    body: bytes | None = None
    document: Any | None = None
    message: str | None = None
    from_cache: bool = False

    @property
    def resolved(self) -> bool:
        return self.status is CanonicalReferenceStatus.RESOLVED

Structured resolver result.

body is populated only on successful fetches. document is populated for successful format_schema resolution after JSON parsing. message is intentionally coarse and should be safe to show in diagnostics.

Instance variables

var body : bytes | None

var document : typing.Any | None

var from_cache : bool

var message : str | None

var reference : CanonicalReference | None

prop resolved : bool

Expand source code

@property
def resolved(self) -> bool:
    return self.status is CanonicalReferenceStatus.RESOLVED

var status : CanonicalReferenceStatus

class CanonicalReferenceStatus (*args, **kwds)

Expand source code

class CanonicalReferenceStatus(str, Enum):
    """Stable resolver outcomes for conformance diagnostics."""

    RESOLVED = "resolved"
    INVALID_REFERENCE = "invalid_reference"
    BLOCKED_UNSAFE_URL = "blocked_unsafe_url"
    NETWORK_ERROR = "network_error"
    BODY_TOO_LARGE = "body_too_large"
    DIGEST_MISMATCH = "digest_mismatch"
    INVALID_SCHEMA = "invalid_schema"

Stable resolver outcomes for conformance diagnostics.

Ancestors

• builtins.str
• enum.Enum

Class variables

var BLOCKED_UNSAFE_URL

var BODY_TOO_LARGE

var DIGEST_MISMATCH

var INVALID_REFERENCE

var INVALID_SCHEMA

var NETWORK_ERROR

var RESOLVED