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

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