diff --git a/src/adcp/signing/brand_jwks.py b/src/adcp/signing/brand_jwks.py index 18af2b7a..3e82d441 100644 --- a/src/adcp/signing/brand_jwks.py +++ b/src/adcp/signing/brand_jwks.py @@ -45,7 +45,9 @@ from urllib.parse import urlsplit, urlunsplit import httpx +import idna +from adcp.signing._idna_canonicalize import canonicalize_host from adcp.signing.jwks import ( AsyncCachingJwksResolver, AsyncJwksFetcher, @@ -566,7 +568,17 @@ async def _fetch_brand_json( if client_factory is not None: client_cm = client_factory(url) else: - transport = build_async_ip_pinned_transport(url, allow_private=allow_private) + # The transport builder resolves + validates the host up + # front, so it — not the later request — is where an SSRF + # refusal surfaces. It must sit inside the same handler as + # the request or the refusal escapes the resolver's + # documented error contract as a raw SSRFValidationError. + try: + transport = build_async_ip_pinned_transport(url, allow_private=allow_private) + except SSRFValidationError as exc: + raise BrandJsonResolverError( + "fetch_failed", f"brand.json URL failed SSRF check: {exc}" + ) from exc client_cm = httpx.AsyncClient( transport=transport, timeout=timeout_seconds, @@ -691,6 +703,19 @@ def _canonicalize_url(raw: str, *, allow_private: bool) -> str: * Scheme lowercased. * Host lowercased (``urlsplit`` does NOT do this — we do it). + * Trailing FQDN-root dot stripped (``brand.example.`` and + ``brand.example`` are the same host). + * Unicode U-labels encoded to A-labels (``bücher.example`` → + ``xn--bcher-kva.example``), via the package-wide UTS#46 + convention in :mod:`adcp.signing._idna_canonicalize`. A host the + IDNA encoder refuses (e.g. an underscore label) is rejected with + ``invalid_url`` rather than passed through. + * IP literals normalized to their canonical form, and IPv6 + literals re-bracketed. ``urlsplit(...).hostname`` returns IPv6 + hosts *de-bracketed*, so rebuilding the authority from it + without re-adding brackets emits ``https://::1/x`` — a string + with no parseable host, and with a non-default port a string + whose port can no longer be separated from the address. * Default port (443 for https, 80 for http) stripped. * Fragments stripped — they aren't sent on the wire and must not smuggle loop-detection aliases into ``seen``. @@ -699,6 +724,10 @@ def _canonicalize_url(raw: str, *, allow_private: bool) -> str: ``https://x.example/`` as distinct strings; the JS-side resolver canonicalizes both via ``new URL``, so a Python-only deployment would fail open where JS fails closed. + + The returned string is required to be a URL the transport layer + accepts: it is fed straight to ``build_async_ip_pinned_transport`` + and ``client.get`` on every redirect hop. """ try: parts = urlsplit(raw) @@ -711,9 +740,21 @@ def _canonicalize_url(raw: str, *, allow_private: bool) -> str: scheme = parts.scheme.lower() if scheme != "https" and not (allow_private and scheme == "http"): raise BrandJsonResolverError("invalid_url", "brand.json URL must use https://") - host = parts.hostname or "" - if not host: + raw_host = parts.hostname or "" + if not raw_host: raise BrandJsonResolverError("invalid_url", "brand.json URL has no host") + try: + host = canonicalize_host(raw_host) + except (idna.IDNAError, UnicodeError) as exc: + raise BrandJsonResolverError( + "invalid_url", f"brand.json URL host is not a valid IDNA name: {exc}" + ) from exc + # ``canonicalize_host`` returns IPv6 literals UNBRACKETED (its step + # 3 short-circuits to ``str(ipaddress.ip_address(...))``); putting + # the brackets back is the caller's job. A canonicalized DNS name + # never contains ':', so this is an unambiguous IPv6 test. + if ":" in host: + host = f"[{host}]" port = parts.port if port is not None and port == _DEFAULT_PORTS.get(scheme): port = None @@ -874,6 +915,40 @@ def _pick_agent( return _SelectedAgent(url=url, jwks_uri=jwks_uri) +def _canonical_origin(raw: str, label: str) -> str: + """`scheme://host[:port]` in the same canonical form on both sides. + + Shares its host handling with :func:`_canonicalize_url` -- same + `canonicalize_host`, same re-bracketing, same default-port elision -- so an + origin equality test cannot be defeated by spelling. Deliberately built + from `.hostname` rather than `.netloc`: `.netloc` is the one accessor that + retains userinfo, and `https://user@brand.example` must compare equal to + `https://brand.example` rather than pivoting trust onto a different string. + """ + try: + parts = urlsplit(raw) + except ValueError as exc: + raise BrandJsonResolverError("invalid_url", f"{label} is not a valid URL") from exc + if not parts.scheme or not parts.netloc: + raise BrandJsonResolverError("invalid_url", f"{label} is not a valid URL") + raw_host = parts.hostname or "" + if not raw_host: + raise BrandJsonResolverError("invalid_url", f"{label} has no host") + try: + host = canonicalize_host(raw_host) + except (idna.IDNAError, UnicodeError) as exc: + raise BrandJsonResolverError( + "invalid_url", f"{label} host is not a valid IDNA name: {exc}" + ) from exc + if ":" in host: # canonicalize_host returns IPv6 unbracketed + host = f"[{host}]" + scheme = parts.scheme.lower() + port = parts.port + if port is not None and port == _DEFAULT_PORTS.get(scheme): + port = None + return f"{scheme}://{host}" if port is None else f"{scheme}://{host}:{port}" + + def _default_jwks_uri(agent_url: str, final_brand_url: str) -> str: """Spec fallback: when ``agent.jwks_uri`` is absent, default to ``/.well-known/jwks.json``. @@ -886,15 +961,16 @@ def _default_jwks_uri(agent_url: str, final_brand_url: str) -> str: agent on a different origin from their brand.json MUST declare an explicit ``jwks_uri``. """ - try: - agent_parts = urlsplit(agent_url) - except ValueError as exc: - raise BrandJsonResolverError("invalid_url", "agent.url is not a valid URL") from exc - if not agent_parts.scheme or not agent_parts.netloc: - raise BrandJsonResolverError("invalid_url", "agent.url is not a valid URL") - brand_parts = urlsplit(final_brand_url) - agent_origin = f"{agent_parts.scheme}://{agent_parts.netloc}" - brand_origin = f"{brand_parts.scheme}://{brand_parts.netloc}" + agent_origin = _canonical_origin(agent_url, "agent.url") + # ``final_brand_url`` has already been through ``_canonicalize_url``, but + # canonicalizing it again is required rather than merely tidy: the two + # sides of this comparison MUST be produced by the same function or the + # check compares a canonical string to a raw one. That asymmetry is the + # defect -- a publisher spelling the same origin identically on both sides + # (a U-label, a trailing root dot, a default port) got told their agent was + # on a different origin from their brand.json, which it was not. + # Re-canonicalizing is idempotent, so this costs nothing. + brand_origin = _canonical_origin(final_brand_url, "brand.json URL") if agent_origin != brand_origin: raise BrandJsonResolverError( "jwks_origin_mismatch", diff --git a/tests/conformance/signing/test_brand_jwks_canonicalization.py b/tests/conformance/signing/test_brand_jwks_canonicalization.py new file mode 100644 index 00000000..233ba884 --- /dev/null +++ b/tests/conformance/signing/test_brand_jwks_canonicalization.py @@ -0,0 +1,161 @@ +"""Proves ``_canonicalize_url`` emits a URL the transport layer accepts. + +The brand.json walk canonicalizes every hop URL before it is used for +three things: the outbound fetch, the redirect-loop ``seen`` key, and +the origin gate in ``_default_jwks_uri``. All three break when the +canonicalizer rebuilds the authority from the *de-bracketed* IPv6 +literal that ``urlsplit(...).hostname`` returns: + +* ``https://[::1]/x`` became ``https://::1/x`` — a string with no + parseable host, which the IP-pinned transport builder refuses. +* ``https://[2001:db8::1]:8443/x`` became ``https://2001:db8::1:8443/x`` + — re-parsing that raises because the port is no longer separable. +* A brand.json and an ``agent.url`` on the same IPv6 host compared + unequal because one side was bracketed and the other was not. + +These tests pin the round-trip contract: canonicalizer output must be +a well-formed URL that re-parses to the same host and port, and that +the SSRF-validating transport builder accepts. + +They also pin the fetch path's exception contract — a transport-side +SSRF refusal must surface as ``BrandJsonResolverError("fetch_failed")``, +not as a raw ``SSRFValidationError`` leaking through the resolver. +""" + +from __future__ import annotations + +from urllib.parse import urlsplit + +import pytest + +from adcp.signing.brand_jwks import ( + BrandJsonResolverError, + _canonicalize_url, + _default_jwks_uri, + _fetch_brand_json, +) +from adcp.signing.ip_pinned_transport import build_async_ip_pinned_transport + +# ----- IPv6 bracket preservation ----- + + +def test_canonicalize_preserves_ipv6_brackets() -> None: + assert _canonicalize_url("https://[::1]/x", allow_private=True) == "https://[::1]/x" + + +def test_canonicalize_ipv6_with_port_stays_unambiguous() -> None: + result = _canonicalize_url("https://[2001:DB8:0:0::1]:8443/x", allow_private=True) + assert result == "https://[2001:db8::1]:8443/x" + reparsed = urlsplit(result) + assert reparsed.hostname == "2001:db8::1" + assert reparsed.port == 8443 + + +def test_canonicalize_ipv6_strips_default_port() -> None: + assert ( + _canonicalize_url("https://[2001:db8::1]:443/x", allow_private=True) + == "https://[2001:db8::1]/x" + ) + + +def test_canonicalized_ipv6_url_is_accepted_by_the_transport_builder() -> None: + url = _canonicalize_url("https://[::1]/x", allow_private=True) + # Must not raise: the canonicalizer's output is fed straight to this + # builder at every redirect hop. + build_async_ip_pinned_transport(url, allow_private=True) + + +def test_canonicalize_ipv6_same_origin_agent_gets_default_jwks_uri() -> None: + brand = _canonicalize_url("https://[::1]/.well-known/brand.json", allow_private=True) + assert _default_jwks_uri("https://[::1]/agent", brand) == "https://[::1]/.well-known/jwks.json" + + +# ----- host aliasing closed by routing through the shared canonicalizer ----- + + +def test_canonicalize_strips_trailing_root_dot() -> None: + assert _canonicalize_url("https://Brand.Example./x", allow_private=False) == ( + "https://brand.example/x" + ) + + +def test_canonicalize_encodes_u_label_to_a_label() -> None: + assert _canonicalize_url("https://bücher.example/x", allow_private=False) == ( + "https://xn--bcher-kva.example/x" + ) + + +def test_canonicalize_rejects_host_the_idna_encoder_refuses() -> None: + with pytest.raises(BrandJsonResolverError) as exc: + _canonicalize_url("https://foo_bar.example/x", allow_private=False) + assert exc.value.code == "invalid_url" + + +# ----- fetch path exception contract ----- + + +async def test_fetch_brand_json_maps_transport_ssrf_refusal_to_resolver_error() -> None: + with pytest.raises(BrandJsonResolverError) as exc: + await _fetch_brand_json( + start_url="https://localhost/.well-known/brand.json", + current_etag=None, + max_redirects=3, + allow_private=False, + timeout_seconds=1.0, + ) + assert exc.value.code == "fetch_failed" + + +# ---------- the origin gate must compare like with like ---------- + + +@pytest.mark.parametrize( + ("agent_url", "brand_url"), + [ + ("https://Brand.Example/agent", "https://brand.example/brand.json"), + ("https://brand.example:443/agent", "https://brand.example/brand.json"), + ("https://brand.example./agent", "https://brand.example/brand.json"), + ("https://user@brand.example/agent", "https://brand.example/brand.json"), + ("https://bücher.example/agent", "https://bücher.example/brand.json"), + ("https://[::1]/agent", "https://[::1]/brand.json"), + ], + ids=["case", "default-port", "root-dot", "userinfo", "u-label", "ipv6"], +) +def test_same_origin_spelled_differently_still_defaults_the_jwks_uri( + agent_url: str, brand_url: str +) -> None: + """The gate compares origins, not spellings. + + ``final_brand_url`` arrives already canonicalized while ``agent.url`` is + raw as published, so building one side with ``.netloc`` and the other from + the canonicalizer compares a canonical string to a raw one. Every row here + is a publisher who spelled the SAME origin on both sides and was told their + agent lives somewhere else — the confusing diagnostic this gate exists to + avoid, produced by the gate itself. + + ``userinfo`` is in the list for a sharper reason: ``.netloc`` is the one + accessor that retains it, so ``https://user@brand.example`` compared + unequal to ``https://brand.example`` and the trust decision turned on a + credential that is not part of the origin at all. + """ + assert _default_jwks_uri(agent_url, brand_url).endswith("/.well-known/jwks.json") + + +def test_cross_origin_agent_is_still_rejected() -> None: + """The fence the gate exists for, unmoved by the canonicalization above. + + An attacker-controlled brand.json naming an agent on another origin must + not make that origin's JWKS authoritative. Canonicalizing both sides closes + spelling differences; it must not close genuine origin differences. + """ + with pytest.raises(BrandJsonResolverError) as exc: + _default_jwks_uri("https://evil.example/agent", "https://brand.example/brand.json") + assert exc.value.code == "jwks_origin_mismatch" + + +def test_ipv6_agent_origin_keeps_its_brackets_in_the_defaulted_uri() -> None: + """The defaulted jwks_uri is fetched, so it must re-parse to the same host.""" + uri = _default_jwks_uri("https://[2001:db8::1]:8443/agent", "https://[2001:db8::1]:8443/b.json") + assert uri == "https://[2001:db8::1]:8443/.well-known/jwks.json" + assert urlsplit(uri).hostname == "2001:db8::1" + assert urlsplit(uri).port == 8443