diff --git a/src/webhook_manager.py b/src/webhook_manager.py index af28fe2a7..ca3e454df 100644 --- a/src/webhook_manager.py +++ b/src/webhook_manager.py @@ -7,10 +7,12 @@ import ipaddress import json import logging import re +import ssl from datetime import datetime, timezone from typing import Optional from urllib.parse import urlparse +import httpcore import httpx from src.database import SessionLocal, Webhook @@ -125,6 +127,128 @@ def validate_webhook_url(url: str) -> str: return url +def _validated_public_ips(url: str) -> list: + """Resolve *url*'s host and return its IPs, raising ValueError if any is + private/internal. + + ``validate_webhook_url`` resolves the host to decide accept/reject, but the + subsequent ``httpx`` connect re-resolves independently — so a DNS record + that flips between the two lookups (rebinding) can slip an internal IP past + the check. Callers pin the delivery connection to the IP this function + returns, closing that TOCTOU. Fail closed: an unresolvable or partly-private + result raises rather than returning a usable IP. + """ + parsed = urlparse(url) + hostname = (parsed.hostname or "").strip() + if not hostname: + raise ValueError("URL must have a hostname") + try: + literal = ipaddress.ip_address(hostname) + except ValueError: + literal = None + if literal is not None: + if _ip_is_private(literal): + raise ValueError("URL must not point to private/internal addresses") + return [literal] + addrs = _resolve_hostname_ips(hostname) + if not addrs or any(_ip_is_private(a) for a in addrs): + raise ValueError("URL must not point to private/internal addresses") + return addrs + + +# httpcore raises its own exception hierarchy; map the ones a simple POST can +# surface back to their httpx equivalents so callers' `except httpx.*` blocks +# (and sanitize_error) behave exactly as they did with the default transport. +_HTTPCORE_TO_HTTPX_EXC = { + httpcore.ConnectError: httpx.ConnectError, + httpcore.ConnectTimeout: httpx.ConnectTimeout, + httpcore.NetworkError: httpx.NetworkError, + httpcore.PoolTimeout: httpx.PoolTimeout, + httpcore.ProtocolError: httpx.ProtocolError, + httpcore.ReadError: httpx.ReadError, + httpcore.ReadTimeout: httpx.ReadTimeout, + httpcore.RemoteProtocolError: httpx.RemoteProtocolError, + httpcore.TimeoutException: httpx.TimeoutException, + httpcore.WriteError: httpx.WriteError, + httpcore.WriteTimeout: httpx.WriteTimeout, +} + + +class _PinnedAsyncBackend(httpcore.AsyncNetworkBackend): + """Async network backend that routes every TCP connect to a fixed IP. + + httpcore derives TLS SNI and the ``Host`` header from the request URL, not + from the connect host, so pinning only the socket destination keeps + certificate validation and vhost routing pointed at the original hostname. + """ + + def __init__(self, ip: ipaddress._BaseAddress): + self._ip = str(ip) + self._real = httpcore.AnyIOBackend() + + async def connect_tcp(self, host, port, timeout=None, local_address=None, + socket_options=None): + return await self._real.connect_tcp( + self._ip, port, timeout, local_address, socket_options + ) + + async def connect_unix_socket(self, path, timeout=None, socket_options=None): + return await self._real.connect_unix_socket(path, timeout, socket_options) + + async def sleep(self, seconds: float) -> None: + return await self._real.sleep(seconds) + + +class _PinnedAsyncTransport(httpx.AsyncBaseTransport): + """httpx transport that pins the TCP connect to a pre-resolved public IP. + + Uses only public ``httpcore`` / ``httpx`` APIs. The request URL is passed + through unchanged (Host + SNI stay the original hostname); only the socket + destination is pinned, closing the DNS-rebinding TOCTOU between the SSRF + check and the connect. HTTP/1.1 only — webhook deliveries are small POSTs. + """ + + def __init__(self, ip: ipaddress._BaseAddress): + self._pool = httpcore.AsyncConnectionPool( + ssl_context=ssl.create_default_context(), + http1=True, + http2=False, + network_backend=_PinnedAsyncBackend(ip), + ) + + async def handle_async_request(self, request: httpx.Request) -> httpx.Response: + core_req = httpcore.Request( + method=request.method, + url=httpcore.URL( + scheme=request.url.raw_scheme, + host=request.url.raw_host, + port=request.url.port, + target=request.url.raw_path, + ), + headers=request.headers.raw, + content=request.stream, + extensions=request.extensions, + ) + try: + core_resp = await self._pool.handle_async_request(core_req) + content = b"".join([chunk async for chunk in core_resp.aiter_stream()]) + await core_resp.aclose() + except Exception as exc: + mapped = _HTTPCORE_TO_HTTPX_EXC.get(type(exc)) + if mapped is not None: + raise mapped(str(exc)) from exc + raise + return httpx.Response( + status_code=core_resp.status, + headers=core_resp.headers, + content=content, + extensions=core_resp.extensions, + ) + + async def aclose(self) -> None: + await self._pool.aclose() + + def validate_events(events_str: str) -> str: """Validate comma-separated event names. Returns cleaned string.""" events = [e.strip() for e in events_str.split(",") if e.strip()] @@ -198,8 +322,11 @@ def sanitize_error(error: str, max_len: int = 200) -> str: class WebhookManager: def __init__(self, api_key_manager=None): - # Disable redirects to prevent SSRF via redirect chains - self._client = httpx.AsyncClient(timeout=10, follow_redirects=False) + # No shared client: each delivery builds a short-lived client whose + # transport is pinned to the SSRF-approved IP (see _deliver / + # _send_request), so a single reusable client can't be pointed at + # different pinned hosts. Redirects stay disabled on every delivery + # client to prevent SSRF via redirect chains. self._loop: Optional[asyncio.AbstractEventLoop] = None self._api_key_manager = api_key_manager # Strong references to in-flight fire-and-forget tasks. asyncio only @@ -262,11 +389,28 @@ class WebhookManager: decrypted = self._decrypt_secret(encrypted_secret) await self._deliver(webhook_id, url, decrypted, "webhook.test", {"message": "Test ping from Odysseus"}) + async def _send_request(self, url: str, body: str, headers: dict, + ip: ipaddress._BaseAddress) -> httpx.Response: + """POST *body* to *url* with the TCP connect pinned to *ip*. + + Overridable seam: tests replace this to avoid real sockets. Redirects + are disabled so a 3xx can't bounce the delivery to another host. + """ + transport = _PinnedAsyncTransport(ip) + async with httpx.AsyncClient( + timeout=10, follow_redirects=False, transport=transport, + ) as client: + return await client.post(url, content=body, headers=headers) + async def _deliver(self, webhook_id: str, url: str, secret: Optional[str], event: str, payload: dict): """Internal delivery. Never call directly from outside this class (use deliver_test).""" - # Re-validate URL at delivery time in case DB was tampered with + # Re-validate URL at delivery time in case DB was tampered with, and + # capture the exact IPs that passed the check so the connect can be + # pinned to one of them (closes the DNS-rebinding TOCTOU: the check + # below and the socket connect no longer resolve independently). try: validate_webhook_url(url) + pinned_ips = _validated_public_ips(url) except ValueError as e: logger.warning(f"Webhook {webhook_id} has invalid URL, skipping: {e}") return @@ -283,7 +427,7 @@ class WebhookManager: db = SessionLocal() try: - resp = await self._client.post(url, content=body, headers=headers) + resp = await self._send_request(url, body, headers, pinned_ips[0]) db.query(Webhook).filter(Webhook.id == webhook_id).update({ "last_triggered_at": _utcnow(), "last_status_code": resp.status_code, @@ -305,4 +449,7 @@ class WebhookManager: db.close() async def close(self): - await self._client.aclose() + # Delivery clients are per-request and closed via their async context + # manager, so there is no long-lived client to tear down here. Kept for + # API compatibility with callers (e.g. app shutdown). + return None diff --git a/tests/test_webhook_dns_rebinding_pin.py b/tests/test_webhook_dns_rebinding_pin.py new file mode 100644 index 000000000..144a19e96 --- /dev/null +++ b/tests/test_webhook_dns_rebinding_pin.py @@ -0,0 +1,156 @@ +"""Regression: webhook delivery must pin the TCP connect to the SSRF-approved IP. + +validate_webhook_url resolves the host to accept/reject, but the delivery +connect previously re-resolved independently — a DNS record flipping between +the two lookups (rebinding) could slip an internal IP past the check. _deliver +now resolves+validates once via _validated_public_ips and pins the connect to +that IP through _PinnedAsyncTransport. These tests drive the real transport +against local servers so the pin is exercised end-to-end, not mocked away. +""" +import asyncio +import http.server +import ipaddress +import socketserver +import threading + +import pytest + +from tests.helpers.import_state import clear_module, preserve_import_state + +import os +import sys +from unittest.mock import patch + +with patch.dict(os.environ, {"DATABASE_URL": "sqlite:///:memory:"}), \ + preserve_import_state("src.database", "core.database"): + clear_module("src.database") + _core_database = sys.modules.get("core.database") + if _core_database is not None and not getattr(_core_database, "__file__", None): + del sys.modules["core.database"] + import src.webhook_manager as wm + + +# --------------------------------------------------------------------------- +# _validated_public_ips +# --------------------------------------------------------------------------- + +def test_validated_public_ips_rejects_metadata_literal(): + with pytest.raises(ValueError): + wm._validated_public_ips("http://169.254.169.254/") + + +def test_validated_public_ips_rejects_loopback_literal(): + with pytest.raises(ValueError): + wm._validated_public_ips("http://127.0.0.1/") + + +def test_validated_public_ips_returns_public_literal(): + ips = wm._validated_public_ips("http://93.184.216.34/") + assert ips == [ipaddress.ip_address("93.184.216.34")] + + +def test_validated_public_ips_rejects_hostname_resolving_private(monkeypatch): + # Rebinding shape: a hostname that (now) resolves into loopback space. + monkeypatch.setattr(wm, "_resolve_hostname_ips", + lambda h: [ipaddress.ip_address("127.0.0.1")]) + with pytest.raises(ValueError): + wm._validated_public_ips("http://evil.rebind.example/") + + +# --------------------------------------------------------------------------- +# End-to-end: the pinned transport actually routes to the pinned IP +# --------------------------------------------------------------------------- + +def _serve(handler): + srv = socketserver.TCPServer(("127.0.0.1", 0), handler) + port = srv.server_address[1] + threading.Thread(target=srv.serve_forever, daemon=True).start() + return srv, port + + +def test_pinned_transport_connects_to_pinned_ip(): + """A request whose URL host is a throwaway hostname is still delivered to + the pinned loopback IP — proving the socket destination comes from the pin, + not from resolving the URL host.""" + hits = [] + + class _Handler(http.server.BaseHTTPRequestHandler): + def do_POST(self): # noqa: N802 + length = int(self.headers.get("Content-Length", 0)) + self.rfile.read(length) + hits.append(self.path) + self.send_response(204) + self.end_headers() + + def log_message(self, *a): + pass + + srv, port = _serve(_Handler) + try: + ip = ipaddress.ip_address("127.0.0.1") + transport = wm._PinnedAsyncTransport(ip) + + async def go(): + async with __import__("httpx").AsyncClient( + transport=transport, timeout=5, follow_redirects=False, + ) as client: + # Host "unresolvable.invalid" would never resolve; the pin is + # what makes this reach the loopback server on `port`. + return await client.post( + f"http://unresolvable.invalid:{port}/hook", content=b"{}", + ) + + resp = asyncio.run(go()) + assert resp.status_code == 204 + assert hits == ["/hook"] + finally: + srv.shutdown() + + +def test_deliver_pins_to_validated_ip_end_to_end(monkeypatch): + """Full _deliver path: a hostname that validation resolves to loopback is + pinned to loopback and the local server receives the signed POST.""" + received = {} + + class _Handler(http.server.BaseHTTPRequestHandler): + def do_POST(self): # noqa: N802 + length = int(self.headers.get("Content-Length", 0)) + received["body"] = self.rfile.read(length) + received["event"] = self.headers.get("X-Odysseus-Event") + self.send_response(200) + self.end_headers() + + def log_message(self, *a): + pass + + srv, port = _serve(_Handler) + + class _Query: + def filter(self, *a, **k): return self + def update(self, values): return None + + class _Db: + def query(self, _m): return _Query() + def commit(self): pass + def rollback(self): pass + def close(self): pass + + # Make both the validation resolve and the pin target loopback, and treat + # loopback as allowed for this test (production blocks it — here we only + # want to prove the pin routes to the validated IP). + monkeypatch.setattr(wm, "SessionLocal", lambda: _Db()) + monkeypatch.setattr(wm, "_is_private_url", lambda url: False) + monkeypatch.setattr(wm, "_resolve_hostname_ips", + lambda h: [ipaddress.ip_address("127.0.0.1")]) + monkeypatch.setattr(wm, "_ip_is_private", lambda a: False) + + manager = wm.WebhookManager() + try: + asyncio.run(manager._deliver( + "hook-1", f"http://webhook.test:{port}/cb", "s3cret", + "webhook.test", {"ok": True}, + )) + assert received.get("event") == "webhook.test" + assert b'"ok": true' in received["body"] + finally: + srv.shutdown() diff --git a/tests/test_webhook_ssrf_resilience.py b/tests/test_webhook_ssrf_resilience.py index e02f17a25..ca82a7595 100644 --- a/tests/test_webhook_ssrf_resilience.py +++ b/tests/test_webhook_ssrf_resilience.py @@ -96,26 +96,29 @@ async def test_webhook_delivery_uses_naive_utc_timestamps(monkeypatch): class _Response: status_code = 204 - class _Client: - def __init__(self): - self.content = "" - - async def post(self, _url, content, headers): - self.content = content - assert headers["X-Odysseus-Event"] == "webhook.test" - return _Response() - db = _Db() - client = _Client() monkeypatch.setattr(wm, "SessionLocal", lambda: db) manager = wm.WebhookManager() - await manager._client.aclose() - manager._client = client + + # Replace the pinned-transport send seam so no real socket is opened. The + # public-IP literal below still exercises _validated_public_ips (which pins + # the connect); the captured content proves the body/headers are built. + captured = {} + + async def _fake_send(url, body, headers, ip): + captured["content"] = body + captured["ip"] = str(ip) + assert headers["X-Odysseus-Event"] == "webhook.test" + return _Response() + + monkeypatch.setattr(manager, "_send_request", _fake_send) await manager._deliver("hook-1", "http://93.184.216.34/", None, "webhook.test", {"ok": True}) - body = json.loads(client.content) + # The delivery must have pinned to the literal public IP from the URL. + assert captured["ip"] == "93.184.216.34" + body = json.loads(captured["content"]) payload_timestamp = datetime.fromisoformat(body["timestamp"]) assert payload_timestamp.tzinfo is None assert db.updates[0]["last_triggered_at"].tzinfo is None