fix(signing): canonicalize both origins in the brand.json jwks gate (#998)

Two defects, and they are the same defect seen from two sides.

`_canonicalize_url` rebuilt the authority from `urlsplit(...).hostname`, which
returns IPv6 literals DE-bracketed. `https://[::1]/x` became `https://::1/x` --
a string with no parseable host -- and `https://[2001:db8::1]:8443/x` became
`https://2001:db8::1:8443/x`, where the port is no longer separable from the
address. Both are fed straight to the IP-pinned transport builder on every
redirect hop, so the first refused the URL and the second re-parsed wrongly.

`_default_jwks_uri` then compared a CANONICAL origin against a RAW one:
`final_brand_url` has been through the canonicalizer, `agent.url` is raw as
published. A publisher who spelled the same origin identically on both sides --
mixed case, an explicit :443, a trailing root dot, a U-label -- was told their
agent was on a different origin from their brand.json, which it was not. That
is precisely the confusing diagnostic the gate exists to avoid, produced by the
gate itself.

Fixing only the first would have made the second WORSE: a stronger
canonicalizer on one side of an asymmetric comparison widens the gap. Both
sides now run through one `_canonical_origin`.

It is built from `.hostname`, not `.netloc`. `.netloc` is the one accessor that
retains userinfo, so `https://user@brand.example` compared unequal to
`https://brand.example` -- a trust decision turning on a credential that is not
part of an origin at all.

The cross-origin fence is unmoved: an attacker-controlled brand.json naming an
agent on another origin is still rejected with `jwks_origin_mismatch`.
Canonicalizing closes spelling differences, not origin differences.

Also moves `build_async_ip_pinned_transport` inside the fetch try-block so a
transport-side SSRF refusal surfaces as `fetch_failed` rather than escaping the
resolver as a raw `SSRFValidationError`.

Fixes #989.

Author: KonstantinMirin

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
     ``<agent_origin>/.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",

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