Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
128 changes: 128 additions & 0 deletions src/adcp/webhook_transport_hooks.py
Original file line number Diff line number Diff line change
Expand Up @@ -32,10 +32,18 @@

from __future__ import annotations

import ipaddress
from dataclasses import dataclass
from typing import Protocol
from urllib.parse import urlsplit, urlunsplit

#: Characters that, interpolated into a netloc, move the boundary between
#: authority / userinfo / port / path / query / fragment -- i.e. rewrite the
#: signed URL into a different URL. `%` covers RFC 6874 IPv6 zone IDs.
#: `:` is deliberately absent: IPv6 literals are made of them, so a stray port
#: is caught after the IP-literal parse instead.
_STRUCTURAL_CHARS = frozenset('/@?#\\[]%"<>^`{|}')


class TransportHook(Protocol):
"""Rewrite the destination URL before SSRF runs.
Expand Down Expand Up @@ -77,10 +85,130 @@ class DockerLocalhostRewrite:
The check happens via :meth:`validate_for_sender`, called by
:meth:`WebhookSender._from_strategy` (and ``__init__``) when
``transport_hooks`` is set.

``rewrite_to`` is validated and canonicalized at construction: it
must be a hostname or IP literal, is lower-cased, and bare IPv6
literals are bracketed automatically (``"::1"`` is stored as
``"[::1]"``) so the assembled authority is unambiguous with a port.
IPv4-mapped IPv6 is re-formatted to its canonical compressed form
(``"::ffff:127.0.0.1"`` becomes ``"[::ffff:7f00:1]"`` — the same
address, spelled canonically). Non-ASCII hostnames are IDNA-encoded
to A-labels. IPv6 zone IDs (``"fe80::1%eth0"``) are rejected: RFC
6874 requires the ``%`` be percent-encoded inside a URI, and
silently emitting an invalid authority is worse than failing at
wiring time.

The validation is deliberately **structural**, not a hostname-syntax
check. ASCII names are accepted as-is once they cannot alter the
URL's shape, because Docker Compose service names legally contain
underscores (``my_service``, ``host_gateway``) which RFC 952/1123
and IDNA both reject — and Docker's embedded DNS resolves them.
Enforcing hostname syntax here would refuse the exact configuration
this class exists to serve. Whether the name resolves is the
resolver's business; whether it rewrites the signed URL is ours.
"""

rewrite_to: str = "host.docker.internal"

def __post_init__(self) -> None:
# Deferred import: ``adcp.signing``'s package __init__ is an
# order of magnitude heavier than this leaf module, and both
# real consumers (webhook_sender, webhooks) already import it.
# This runs once per hook construction, never per delivery.
from adcp.signing._idna_canonicalize import canonicalize_host

value = self.rewrite_to
if not value:
raise ValueError(
"DockerLocalhostRewrite(rewrite_to=...) must be a hostname or IP "
"literal; got an empty string"
)

# Accept an already-bracketed IPv6 literal by unwrapping it first;
# the brackets are re-applied below from the canonical form.
inner = value[1:-1] if value.startswith("[") and value.endswith("]") else value
if not inner:
# `"[]"` survives the non-empty check above and empties here. An
# empty host is the one outcome this guard exists to prevent: it
# assembles to `https://:9000/hook`, an authority with a port and
# no host -- exactly the shape @target-uri canonicalization rejects.
raise ValueError(
f"DockerLocalhostRewrite(rewrite_to=...) must be a hostname or IP "
f"literal; got {value!r}, which has no host"
)

# Structural rejection comes FIRST and is the actual point of this
# guard: `rewrite_to` is interpolated straight into the netloc, so any
# character that can move the boundary between authority, path, query,
# fragment or userinfo rewrites the signed URL into a different URL.
# `%` is here for RFC 6874 IPv6 zone IDs, which are not representable
# in a URI authority without percent-encoding.
bad = {ch for ch in inner if ch in _STRUCTURAL_CHARS or ord(ch) < 0x21 or ord(ch) == 0x7F}
if bad:
raise ValueError(
f"DockerLocalhostRewrite(rewrite_to=...) must be a hostname or IP "
f"literal; got {value!r}, which contains {sorted(bad)!r} and would "
f"change the structure of the signed URL"
)

try:
ip = ipaddress.ip_address(inner)
except ValueError:
pass
else:
# Bracket v6 so the assembled authority is unambiguous with a port
# -- the defect this guard exists to close. `str(ip)` also folds the
# literal to its canonical compressed form.
canonical = f"[{ip}]" if ip.version == 6 else str(ip)
object.__setattr__(self, "rewrite_to", canonical)
return

if ":" in inner:
# Not an IP literal, so a colon is a port -- which `rewrite_url`
# re-appends itself, and which `apply_hooks`' port guard would then
# see as a port change.
raise ValueError(
f"DockerLocalhostRewrite(rewrite_to=...) must be a hostname or IP "
f"literal without a port; got {value!r}"
)

if inner.isascii():
# Deliberately NOT routed through `canonicalize_host`: this is a
# Docker helper, and Docker Compose service names legally contain
# underscores (`my_service`, `host_gateway`), which IDNA rejects
# under RFC 952/1123. Docker's embedded DNS resolves them, so
# refusing them here would break the case this class exists for.
# Structural safety is already established above; anything further
# is the resolver's business, not ours.
# One trailing root dot, matching `canonicalize_host` -- `rstrip`
# would eat every dot, so `"."` and `".."` normalized to the empty
# host rather than being rejected.
ascii_host = inner.lower()
if ascii_host.endswith("."):
ascii_host = ascii_host[:-1]
if not ascii_host or any(label == "" for label in ascii_host.split(".")):
# Catches `"."`, `".."` and `"a..b"`. An empty label is not a
# host, and `".."` in particular survives a single-dot strip as
# `"."` -- non-empty, but still no host.
raise ValueError(
f"DockerLocalhostRewrite(rewrite_to=...) must be a hostname or IP "
f"literal; got {value!r}, which normalizes to an empty label"
)
object.__setattr__(self, "rewrite_to", ascii_host)
return

# Non-ASCII: convert to A-labels so the netloc is wire-legal. Failure
# here is a genuinely unusable host, not a naming-convention quibble.
try:
canonical = canonicalize_host(inner)
except (UnicodeError, ValueError) as exc:
# ``idna.IDNAError`` subclasses ``UnicodeError``.
raise ValueError(
f"DockerLocalhostRewrite(rewrite_to=...) must be a hostname or IP "
f"literal; got {value!r} ({exc})"
) from exc
object.__setattr__(self, "rewrite_to", canonical)

def rewrite_url(self, url: str) -> str | None:
parsed = urlsplit(url)
# ``hostname`` lower-cases and strips brackets from IPv6 — match
Expand Down
103 changes: 103 additions & 0 deletions tests/conformance/signing/test_webhook_transport_hooks.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@
import pytest

from adcp.signing import SSRFValidationError
from adcp.signing.canonical import canonicalize_authority
from adcp.webhook_sender import WebhookSender
from adcp.webhook_transport_hooks import (
DockerLocalhostRewrite,
Expand Down Expand Up @@ -65,6 +66,108 @@ def test_rewrite_preserves_query_and_fragment() -> None:
)


# ---------- rewrite_to validation ----------


def test_bare_ipv6_rewrite_to_is_bracketed() -> None:
"""A bare IPv6 ``rewrite_to`` yields an unbracketed authority that RFC 3986
makes ambiguous with a port. Bracket it at construction — the operator's
intent is unambiguous and the docstring encourages IP-literal values."""
hook = DockerLocalhostRewrite(rewrite_to="::1")
assert hook.rewrite_url("https://localhost:9000/hook") == "https://[::1]:9000/hook"


def test_bare_ipv6_rewrite_survives_apply_hooks_and_signing_canonicalization() -> None:
"""The bracketing must hold all the way through the framework's port guard
and into signing canonicalization. Unbracketed, ``urlsplit(...).port`` inside
``apply_hooks`` raises an opaque stdlib ValueError about a port the operator
never configured."""
hook = DockerLocalhostRewrite(rewrite_to="2001:db8::1")
out = apply_hooks("https://localhost:9000/hook", (hook,))
assert out == "https://[2001:db8::1]:9000/hook"
assert canonicalize_authority(out) == "[2001:db8::1]:9000"


@pytest.mark.parametrize(
"bad",
["", "attacker.com/path", "user@evil.example", "host.docker.internal:1234", "ho st"],
)
def test_rewrite_to_rejects_non_host_values(bad: str) -> None:
"""``rewrite_to`` is interpolated into the netloc; anything that is not a
hostname or IP literal changes the signed URL in ways apply_hooks' scheme/
port guard does not catch (path injection, userinfo injection, raw spaces)."""
with pytest.raises(ValueError, match="rewrite_to"):
DockerLocalhostRewrite(rewrite_to=bad)


def test_rewrite_to_rejects_ipv6_zone_id() -> None:
"""``ipaddress`` accepts scoped addresses, but RFC 6874 requires the ``%``
be percent-encoded as ``%25`` inside a URI. Rather than emit an authority
that is invalid on the wire, reject the zone-ID form at construction."""
with pytest.raises(ValueError, match="rewrite_to"):
DockerLocalhostRewrite(rewrite_to="fe80::1%eth0")


def test_bracketed_ipv6_rewrite_to_accepted_unchanged() -> None:
hook = DockerLocalhostRewrite(rewrite_to="[::1]")
assert hook.rewrite_url("https://localhost:9000/hook") == "https://[::1]:9000/hook"


def test_rewrite_to_hostname_and_ipv4_still_accepted() -> None:
assert (
DockerLocalhostRewrite().rewrite_url("http://localhost:8080/webhook")
== "http://host.docker.internal:8080/webhook"
)
assert (
DockerLocalhostRewrite(rewrite_to="172.17.0.1").rewrite_url("http://localhost/x")
== "http://172.17.0.1/x"
)


@pytest.mark.parametrize("bad", ["[]", "[.]", ".", "..", "...", "a..b"])
def test_rewrite_to_rejects_values_that_normalize_to_no_host(bad: str) -> None:
"""An empty host is the outcome this guard exists to prevent.

These reach the empty host by two different routes the earlier checks each
miss: ``"[]"`` is non-empty until the brackets come off, and ``"."`` /
``".."`` are non-empty until the trailing root dot is stripped. Both used
to assemble to ``https://:9000/hook`` — an authority with a port and no
host, which is precisely the shape ``@target-uri`` canonicalization
rejects, produced by the hook meant to keep the authority well-formed.

``"a..b"`` is here because the fix is stated as "no empty label" rather
than "not empty", and an interior empty label is the same defect.
"""
with pytest.raises(ValueError, match="rewrite_to"):
DockerLocalhostRewrite(rewrite_to=bad)


@pytest.mark.parametrize(
"name", ["my_service", "host_gateway", "docker_host.local", "_dns-sd._udp.local"]
)
def test_rewrite_to_accepts_docker_legal_service_names(name: str) -> None:
"""Underscored names must keep working -- this is a Docker helper.

Docker Compose service names legally contain underscores and Docker's
embedded DNS resolves them, but RFC 952/1123 and IDNA both reject them.
Validating ``rewrite_to`` as a *hostname* rather than structurally would
refuse the exact configuration this class exists to serve, and would do it
at construction -- turning a working deployment into a startup crash on
upgrade. The guard only has to prove the value cannot restructure the URL.
"""
assert DockerLocalhostRewrite(rewrite_to=name).rewrite_to == name


def test_rewrite_to_is_normalized_at_construction() -> None:
"""Canonicalization is visible on the field: case-folded, trailing FQDN
root dot dropped, IDN encoded to A-labels, IPv6 bracketed."""
assert DockerLocalhostRewrite(rewrite_to="HOST.Docker.Internal.").rewrite_to == (
"host.docker.internal"
)
assert DockerLocalhostRewrite(rewrite_to="bücher.example").rewrite_to == "xn--bcher-kva.example"
assert DockerLocalhostRewrite(rewrite_to="[2001:DB8::1]").rewrite_to == "[2001:db8::1]"


# ---------- apply_hooks (framework) ----------


Expand Down
Loading