hermes: bound SCM control calls with hard deadlines

Run every Gitea API and broker control exchange inside a killable helper
process whose connect, send, and read share one absolute wall-clock
deadline, and add a watchdog that force-closes streaming connections at
expiry. Redirects are rejected before authentication headers can move.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

services/hermes/scm-common/scripts/deadline_http.py

@@ -0,0 +1,302 @@
+#!/usr/bin/env python3
+"""Absolute wall-clock deadlines for every outbound Hermes HTTP exchange.
+
+Small control exchanges run inside a killable helper process, so connect,
+send, read, and DNS all share one hard bound that no blocking socket state
+can outlive.  Large streaming Git exchanges cannot cross a process boundary;
+they instead register their connection with a watchdog that force-closes the
+socket when the same kind of absolute deadline expires.
+"""
+
+from __future__ import annotations
+
+import base64
+import contextlib
+import email.message
+import http.client
+import io
+import json
+import os
+import signal
+import socket
+import subprocess
+import sys
+import threading
+import time
+import urllib.error
+import urllib.request
+from dataclasses import dataclass
+
+from gitea_api_policy import PolicyError
+
+MAX_CONTROL_BYTES = 128 * 1024
+MAX_PROTOCOL_BYTES = 4 * 1024 * 1024
+MAX_STREAM_SECONDS = 900.0
+READ_CHUNK = 64 * 1024
+
+
+class _RejectRedirect(urllib.request.HTTPRedirectHandler):
+    def redirect_request(self, req, fp, code, msg, headers, newurl):
+        raise urllib.error.URLError("redirect rejected")
+
+
+@dataclass(frozen=True)
+class Result:
+    """Minimal HTTP evidence returned across the process boundary."""
+
+    status: int
+    content_type: str
+    body: bytes
+
+
+def _read_response(response, maximum: int, deadline: float) -> bytes:
+    chunks: list[bytes] = []
+    total = 0
+    while True:
+        remaining = deadline - time.monotonic()
+        if remaining <= 0:
+            raise TimeoutError
+        raw = getattr(getattr(response, "fp", None), "raw", None)
+        sock = getattr(raw, "_sock", raw)
+        if hasattr(sock, "settimeout"):
+            sock.settimeout(remaining)
+        chunk = response.read(min(READ_CHUNK, maximum + 1 - total))
+        if not chunk:
+            break
+        chunks.append(chunk)
+        total += len(chunk)
+        if total > maximum:
+            raise ValueError("response too large")
+    return b"".join(chunks)
+
+
+def _child_exchange(control: dict[str, object]) -> Result:
+    url = control.get("url")
+    method = control.get("method")
+    headers = control.get("headers")
+    encoded = control.get("body")
+    maximum = control.get("maximum")
+    timeout = control.get("timeout")
+    if (
+        not isinstance(url, str)
+        or not isinstance(method, str)
+        or not isinstance(headers, dict)
+        or not all(isinstance(key, str) and isinstance(value, str) for key, value in headers.items())
+        or not isinstance(encoded, str)
+        or not isinstance(maximum, int)
+        or not 0 <= maximum <= 2 * 1024 * 1024
+        or not isinstance(timeout, (int, float))
+        or not 0 < timeout <= 120
+    ):
+        raise ValueError("invalid control")
+    body = base64.b64decode(encoded, validate=True) if encoded else None
+    if body is not None and len(body) > MAX_CONTROL_BYTES:
+        raise ValueError("request too large")
+    deadline = time.monotonic() + float(timeout)
+    request = urllib.request.Request(
+        url, data=body, method=method, headers=headers  # type: ignore[arg-type]
+    )
+    opener = urllib.request.build_opener(_RejectRedirect())
+    try:
+        response = opener.open(request, timeout=max(deadline - time.monotonic(), 0.001))
+    except urllib.error.HTTPError as error:
+        response = error
+    with response:
+        status = getattr(response, "status", response.getcode())
+        content_type = response.headers.get_content_type()
+        value = _read_response(response, maximum, deadline)
+    return Result(int(status), str(content_type), value)
+
+
+def _child_main() -> int:
+    try:
+        raw = sys.stdin.buffer.read(MAX_CONTROL_BYTES + 1)
+        if len(raw) > MAX_CONTROL_BYTES:
+            raise ValueError("control too large")
+        control = json.loads(raw)
+        if not isinstance(control, dict):
+            raise ValueError("invalid control")
+        result = _child_exchange(control)
+        output = {
+            "ok": True,
+            "status": result.status,
+            "content_type": result.content_type,
+            "body": base64.b64encode(result.body).decode("ascii"),
+        }
+    except Exception:
+        output = {"ok": False}
+    sys.stdout.write(json.dumps(output, separators=(",", ":")))
+    return 0 if output["ok"] else 1
+
+
+def _terminate(process: subprocess.Popen[bytes]) -> None:
+    with contextlib.suppress(ProcessLookupError):
+        os.killpg(process.pid, signal.SIGKILL)
+    process.communicate()
+
+
+def exchange(
+    request: urllib.request.Request,
+    *,
+    maximum: int,
+    timeout: float,
+    popen=subprocess.Popen,
+) -> Result:
+    """Execute a whole request under one monotonic deadline and size limit."""
+    if not 0 < timeout <= 120 or not 0 <= maximum <= 2 * 1024 * 1024:
+        raise PolicyError("HTTP control request bounds are invalid")
+    body = request.data or b""
+    if not isinstance(body, bytes) or len(body) > MAX_CONTROL_BYTES:
+        raise PolicyError("HTTP control request exceeds the safe size limit")
+    control = json.dumps(
+        {
+            "url": request.full_url,
+            "method": request.get_method(),
+            "headers": dict(request.header_items()),
+            "body": base64.b64encode(body).decode("ascii"),
+            "maximum": maximum,
+            "timeout": timeout,
+        },
+        separators=(",", ":"),
+    ).encode()
+    deadline = time.monotonic() + timeout
+    process = popen(
+        [sys.executable, os.path.realpath(__file__), "--child"],
+        stdin=subprocess.PIPE,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.DEVNULL,
+        start_new_session=True,
+        env={"LANG": "C.UTF-8", "PATH": "/usr/bin:/bin"},
+    )
+    try:
+        remaining = deadline - time.monotonic()
+        if remaining <= 0:
+            raise subprocess.TimeoutExpired(process.args, timeout)
+        output, _error = process.communicate(control, timeout=remaining)
+    except subprocess.TimeoutExpired as exc:
+        _terminate(process)
+        raise PolicyError("HTTP control request deadline exceeded") from exc
+    if process.returncode != 0 or len(output) > MAX_PROTOCOL_BYTES:
+        raise PolicyError("HTTP control request failed")
+    try:
+        result = json.loads(output)
+        if not isinstance(result, dict) or result.get("ok") is not True:
+            raise ValueError
+        status = result["status"]
+        content_type = result["content_type"]
+        encoded = result["body"]
+        if not isinstance(status, int) or not isinstance(content_type, str) or not isinstance(encoded, str):
+            raise ValueError
+        value = base64.b64decode(encoded, validate=True)
+    except (KeyError, ValueError, TypeError, json.JSONDecodeError) as exc:
+        raise PolicyError("HTTP control helper returned invalid evidence") from exc
+    if len(value) > maximum:
+        raise PolicyError("HTTP control response exceeds the safe size limit")
+    return Result(status, content_type, value)
+
+
+class _AdaptedResponse:
+    """Present one finished exchange through the urlopen reader interface."""
+
+    def __init__(self, result: Result):
+        self.status = result.status
+        self.headers = email.message.Message()
+        self.headers["Content-Type"] = result.content_type
+        self._stream = io.BytesIO(result.body)
+
+    def read(self, limit: int = -1) -> bytes:
+        return self._stream.read(limit)
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *_excinfo: object) -> bool:
+        return False
+
+
+def open_bounded(
+    request: urllib.request.Request,
+    *,
+    maximum: int,
+    timeout: float,
+    popen=subprocess.Popen,
+) -> _AdaptedResponse:
+    """Open one control exchange whose whole lifetime shares one deadline."""
+    return _AdaptedResponse(
+        exchange(request, maximum=maximum, timeout=timeout, popen=popen)
+    )
+
+
+class StreamDeadline:
+    """Force-close tracked connections once an absolute deadline passes."""
+
+    def __init__(self, timeout: float, *, timer=threading.Timer):
+        if not 0 < timeout <= MAX_STREAM_SECONDS:
+            raise PolicyError("HTTP stream deadline bounds are invalid")
+        self._deadline = time.monotonic() + timeout
+        self._lock = threading.Lock()
+        self._connections: list[http.client.HTTPConnection] = []
+        self.expired = False
+        self._timer = timer(timeout, self._expire)
+        self._timer.daemon = True
+        self._timer.start()
+
+    def remaining(self) -> float:
+        value = self._deadline - time.monotonic()
+        if value <= 0 or self.expired:
+            raise PolicyError("HTTP stream deadline exceeded")
+        return value
+
+    def _expire(self) -> None:
+        with self._lock:
+            self.expired = True
+            connections = list(self._connections)
+        for connection in connections:
+            # Shut the raw socket down first so a blocked send or recv wakes
+            # immediately; close alone leaves peers waiting on the old fd.
+            sock = getattr(connection, "sock", None)
+            if sock is not None:
+                with contextlib.suppress(OSError):
+                    sock.shutdown(socket.SHUT_RDWR)
+            connection.close()
+
+    def _track(self, connection: http.client.HTTPConnection):
+        with self._lock:
+            expired = self.expired
+            if not expired:
+                self._connections.append(connection)
+        if expired:
+            connection.close()
+            raise PolicyError("HTTP stream deadline exceeded")
+        return connection
+
+    def cancel(self) -> None:
+        self._timer.cancel()
+
+    def handlers(self) -> tuple[urllib.request.BaseHandler, ...]:
+        """Build urllib handlers whose connections obey this deadline."""
+        deadline = self
+
+        class GuardedHTTPHandler(urllib.request.HTTPHandler):
+            def http_open(self, req):
+                return self.do_open(
+                    lambda host, **kwargs: deadline._track(
+                        http.client.HTTPConnection(host, **kwargs)
+                    ),
+                    req,
+                )
+
+        class GuardedHTTPSHandler(urllib.request.HTTPSHandler):
+            def https_open(self, req):
+                return self.do_open(
+                    lambda host, **kwargs: deadline._track(
+                        http.client.HTTPSConnection(host, **kwargs)
+                    ),
+                    req,
+                )
+
+        return (GuardedHTTPHandler(), GuardedHTTPSHandler())
+
+
+if __name__ == "__main__":
+    raise SystemExit(_child_main() if sys.argv[1:] == ["--child"] else 2)

services/hermes/scm-common/scripts/gitea_api.py

@@ -15,6 +15,7 @@ import urllib.request
 from collections.abc import Callable
 from pathlib import Path
 
+import deadline_http
 from gitea_api_policy import (
     PolicyError,
     _draft_title,
@@ -40,19 +41,16 @@ MAX_API_PATH_LENGTH = 512
 RAW_API_TARGET_RE = re.compile(r"[A-Za-z0-9/_.?&=-]+\Z")
 
 
-class RejectRedirectHandler(urllib.request.HTTPRedirectHandler):
-    """Reject every redirect before urllib can copy authentication headers."""
-
-    def redirect_request(self, req, fp, code, msg, headers, newurl):
-        raise PolicyError("Forgejo redirects are not allowed")
-
-
-_SAFE_OPENER = urllib.request.build_opener(RejectRedirectHandler())
-
-
 def _safe_urlopen(request: urllib.request.Request, timeout: int):
-    """Open one request using a handler that never follows redirects."""
+    """Open one exchange under a killable absolute wall-clock deadline.
-    return _SAFE_OPENER.open(request, timeout=timeout)
+
+    The helper process rejects redirects before urllib can copy
+    authentication headers, and its whole connect/send/read lifetime
+    shares one hard deadline.
+    """
+    return deadline_http.open_bounded(
+        request, maximum=MAX_RESPONSE_BYTES, timeout=timeout
+    )
 
 
 def read_token(path: Path = DEFAULT_TOKEN_FILE) -> str:

services/hermes/scm-common/scripts/scm_broker_client.py

@@ -7,24 +7,22 @@ import json
 import urllib.request
 from collections.abc import Callable
 
+import deadline_http
 from gitea_api_policy import PolicyError
 
 BROKER_ORIGIN = "http://hermes-scm-broker.hermes-scm.svc.cluster.local:9081"
 MAX_RESPONSE_BYTES = 2 * 1024 * 1024
 
 
-class RejectRedirectHandler(urllib.request.HTTPRedirectHandler):
-    """Keep every broker request on its fixed in-cluster origin."""
-
-    def redirect_request(self, req, fp, code, msg, headers, newurl):
-        raise PolicyError("SCM broker redirects are not allowed")
-
-
-_OPENER = urllib.request.build_opener(RejectRedirectHandler())
-
-
 def _open(request: urllib.request.Request, timeout: int):
-    return _OPENER.open(request, timeout=timeout)
+    """Open one broker exchange under a killable absolute deadline.
+
+    The helper process keeps every request on its fixed in-cluster origin by
+    rejecting redirects before they are followed.
+    """
+    return deadline_http.open_bounded(
+        request, maximum=MAX_RESPONSE_BYTES, timeout=timeout
+    )
 
 
 def request(

testing/tests/test_hermes_deadline_http.py

@@ -0,0 +1,309 @@
+"""Absolute-deadline contracts for the killable HTTP control helper."""
+
+from __future__ import annotations
+
+import base64
+import io
+import json
+import subprocess
+import threading
+import time
+import urllib.request
+from http.server import BaseHTTPRequestHandler, HTTPServer
+
+import pytest
+
+from testing.tests.test_hermes_scm_broker_support import _load
+
+
+class _Handler(BaseHTTPRequestHandler):
+    def _respond(self):
+        if self.path == "/slow":
+            time.sleep(10)
+        if self.path == "/redirect":
+            self.send_response(302)
+            self.send_header("Location", "http://127.0.0.1:1/evil")
+            self.send_header("Content-Length", "0")
+            self.end_headers()
+            return
+        status = 404 if self.path == "/missing" else 200
+        body = b"x" * 64 if self.path == "/large" else b'{"ok":true}'
+        self.send_response(status)
+        self.send_header("Content-Type", "application/json")
+        self.send_header("Content-Length", str(len(body)))
+        self.end_headers()
+        self.wfile.write(body)
+
+    do_GET = _respond
+    do_POST = _respond
+
+    def log_message(self, *_args):
+        return
+
+
+@pytest.fixture(scope="module")
+def server():
+    value = HTTPServer(("127.0.0.1", 0), _Handler)
+    thread = threading.Thread(target=value.serve_forever, daemon=True)
+    thread.start()
+    yield f"http://127.0.0.1:{value.server_port}"
+    value.shutdown()
+
+
+def _control(url: str, **overrides):
+    value = {
+        "url": url,
+        "method": "GET",
+        "headers": {"Accept": "application/json"},
+        "body": "",
+        "maximum": 1024,
+        "timeout": 10,
+    }
+    value.update(overrides)
+    return value
+
+
+def test_child_exchange_round_trips_and_reports_http_errors(server):
+    module = _load("deadline_http")
+    result = module._child_exchange(_control(server + "/ok"))
+    assert (result.status, result.content_type) == (200, "application/json")
+    assert result.body == b'{"ok":true}'
+
+    posted = module._child_exchange(
+        _control(
+            server + "/ok",
+            method="POST",
+            body=base64.b64encode(b'{"a":1}').decode(),
+        )
+    )
+    assert posted.status == 200
+
+    missing = module._child_exchange(_control(server + "/missing"))
+    assert missing.status == 404
+
+
+def test_child_exchange_rejects_redirects_and_oversized_responses(server):
+    module = _load("deadline_http")
+    with pytest.raises(Exception):
+        module._child_exchange(_control(server + "/redirect"))
+    with pytest.raises(ValueError, match="too large"):
+        module._child_exchange(_control(server + "/large", maximum=8))
+
+
+@pytest.mark.parametrize(
+    "overrides",
+    [
+        {"url": 1},
+        {"method": None},
+        {"headers": ["not", "dict"]},
+        {"headers": {"Accept": 5}},
+        {"body": b"raw"},
+        {"maximum": "big"},
+        {"maximum": -1},
+        {"maximum": 4 * 1024 * 1024},
+        {"timeout": "soon"},
+        {"timeout": 0},
+        {"timeout": 121},
+    ],
+)
+def test_child_exchange_rejects_invalid_control_documents(overrides):
+    module = _load("deadline_http")
+    control = _control("http://127.0.0.1:1/")
+    control.update(overrides)
+    with pytest.raises(ValueError, match="invalid control"):
+        module._child_exchange(control)
+
+
+def test_child_exchange_rejects_bad_or_oversized_encoded_bodies():
+    module = _load("deadline_http")
+    with pytest.raises(ValueError):
+        module._child_exchange(_control("http://127.0.0.1:1/", body="!!not-b64!!"))
+    huge = base64.b64encode(b"x" * (module.MAX_CONTROL_BYTES + 1)).decode()
+    with pytest.raises(ValueError, match="request too large"):
+        module._child_exchange(_control("http://127.0.0.1:1/", body=huge))
+
+
+def test_read_response_enforces_deadline_and_size():
+    module = _load("deadline_http")
+
+    class Reader:
+        def __init__(self, body: bytes):
+            self.stream = io.BytesIO(body)
+
+        def read(self, limit: int):
+            return self.stream.read(limit)
+
+    with pytest.raises(TimeoutError):
+        module._read_response(Reader(b"abc"), 10, time.monotonic() - 1)
+    with pytest.raises(ValueError, match="too large"):
+        module._read_response(Reader(b"abcdef"), 4, time.monotonic() + 5)
+    value = module._read_response(Reader(b"abc"), 10, time.monotonic() + 5)
+    assert value == b"abc"
+
+
+def _run_child_main(module, monkeypatch, raw: bytes) -> dict:
+    stdin = type("Stdin", (), {"buffer": io.BytesIO(raw)})()
+    stdout = io.StringIO()
+    monkeypatch.setattr(module.sys, "stdin", stdin)
+    monkeypatch.setattr(module.sys, "stdout", stdout)
+    code = module._child_main()
+    return code, json.loads(stdout.getvalue())
+
+
+def test_child_main_reports_success_and_failure_documents(monkeypatch, server):
+    module = _load("deadline_http")
+    code, output = _run_child_main(
+        module, monkeypatch, json.dumps(_control(server + "/ok")).encode()
+    )
+    assert code == 0 and output["ok"] is True
+    assert base64.b64decode(output["body"]) == b'{"ok":true}'
+
+    for raw in (
+        b"x" * (module.MAX_CONTROL_BYTES + 1),
+        b"not json",
+        b"[1,2]",
+    ):
+        code, output = _run_child_main(module, monkeypatch, raw)
+        assert code == 1 and output == {"ok": False}
+
+
+class _FakeProcess:
+    def __init__(self, output: bytes, returncode: int = 0):
+        self.args = ["fake-child"]
+        self.pid = 0
+        self.returncode = returncode
+        self.output = output
+        self.stdin_payload = None
+
+    def communicate(self, payload=None, timeout=None):
+        self.stdin_payload = payload
+        return self.output, b""
+
+
+def _fake_popen(output: bytes, returncode: int = 0):
+    def popen(*_args, **_kwargs):
+        return _FakeProcess(output, returncode)
+
+    return popen
+
+
+def _success_output(body: bytes, **overrides) -> bytes:
+    value = {
+        "ok": True,
+        "status": 200,
+        "content_type": "application/json",
+        "body": base64.b64encode(body).decode("ascii"),
+    }
+    value.update(overrides)
+    return json.dumps(value).encode()
+
+
+def test_exchange_validates_bounds_and_request_body():
+    module = _load("deadline_http")
+    request = urllib.request.Request("https://scm.bstein.dev/api/v1/x")
+    with pytest.raises(module.PolicyError, match="bounds are invalid"):
+        module.exchange(request, maximum=1, timeout=0)
+    with pytest.raises(module.PolicyError, match="bounds are invalid"):
+        module.exchange(request, maximum=-1, timeout=10)
+    request.data = "not-bytes"
+    with pytest.raises(module.PolicyError, match="safe size"):
+        module.exchange(request, maximum=1, timeout=10)
+    request.data = b"x" * (module.MAX_CONTROL_BYTES + 1)
+    with pytest.raises(module.PolicyError, match="safe size"):
+        module.exchange(request, maximum=1, timeout=10)
+
+
+def test_exchange_returns_result_and_enforces_response_limit():
+    module = _load("deadline_http")
+    request = urllib.request.Request(
+        "https://scm.bstein.dev/api/v1/x", data=b"{}", method="POST"
+    )
+    result = module.exchange(
+        request, maximum=64, timeout=10, popen=_fake_popen(_success_output(b"body"))
+    )
+    assert (result.status, result.content_type, result.body) == (
+        200,
+        "application/json",
+        b"body",
+    )
+    with pytest.raises(module.PolicyError, match="safe size"):
+        module.exchange(
+            request,
+            maximum=2,
+            timeout=10,
+            popen=_fake_popen(_success_output(b"body")),
+        )
+
+
+@pytest.mark.parametrize(
+    ("output", "returncode", "match"),
+    [
+        (b"", 1, "request failed"),
+        (b"x" * (4 * 1024 * 1024 + 1), 0, "request failed"),
+        (b"not json", 0, "invalid evidence"),
+        (b'{"ok":false}', 0, "invalid evidence"),
+        (b"[1]", 0, "invalid evidence"),
+        (b'{"ok":true}', 0, "invalid evidence"),
+        (b'{"ok":true,"status":"200","content_type":"a","body":""}', 0, "invalid evidence"),
+        (b'{"ok":true,"status":200,"content_type":"a","body":"!!"}', 0, "invalid evidence"),
+    ],
+)
+def test_exchange_rejects_invalid_helper_evidence(output, returncode, match):
+    module = _load("deadline_http")
+    request = urllib.request.Request("https://scm.bstein.dev/api/v1/x")
+    with pytest.raises(module.PolicyError, match=match):
+        module.exchange(
+            request, maximum=64, timeout=10, popen=_fake_popen(output, returncode)
+        )
+
+
+def test_exchange_kills_the_helper_when_the_deadline_expires(server):
+    module = _load("deadline_http")
+    request = urllib.request.Request(server + "/slow")
+    started = time.monotonic()
+    with pytest.raises(module.PolicyError, match="deadline exceeded"):
+        module.exchange(request, maximum=64, timeout=1.0)
+    assert time.monotonic() - started < 5
+
+
+def test_exchange_fails_before_spawn_wait_when_time_is_exhausted(monkeypatch):
+    module = _load("deadline_http")
+    request = urllib.request.Request("https://scm.bstein.dev/api/v1/x")
+    killed = []
+    monkeypatch.setattr(module.os, "killpg", lambda pid, sig: killed.append((pid, sig)))
+    clock = iter([0.0, 1_000_000.0, 1_000_000.0, 1_000_000.0])
+    monkeypatch.setattr(module.time, "monotonic", lambda: next(clock))
+    with pytest.raises(module.PolicyError, match="deadline exceeded"):
+        module.exchange(
+            request, maximum=64, timeout=10, popen=_fake_popen(_success_output(b""))
+        )
+    assert killed
+
+
+def test_terminate_tolerates_already_finished_helpers():
+    module = _load("deadline_http")
+    process = subprocess.Popen(
+        ["/bin/true"], start_new_session=True, stdout=subprocess.DEVNULL
+    )
+    process.wait()
+    module._terminate(process)
+    assert process.returncode == 0
+
+
+def test_child_entrypoint_round_trip_through_real_subprocess(server):
+    module = _load("deadline_http")
+    request = urllib.request.Request(server + "/ok")
+    result = module.exchange(request, maximum=1024, timeout=20)
+    assert result.status == 200 and result.body == b'{"ok":true}'
+
+
+def test_open_bounded_adapts_result_to_the_urlopen_interface():
+    module = _load("deadline_http")
+    request = urllib.request.Request("https://scm.bstein.dev/api/v1/x")
+    with module.open_bounded(
+        request, maximum=64, timeout=10, popen=_fake_popen(_success_output(b"body"))
+    ) as response:
+        assert response.status == 200
+        assert response.headers.get_content_type() == "application/json"
+        assert response.read(2) == b"bo"
+        assert response.read() == b"dy"

testing/tests/test_hermes_gitea_internal_coverage.py

@@ -37,16 +37,21 @@ def test_runtime_token_reader_and_fixed_origin(tmp_path: Path, monkeypatch):
         module.configured_base_url()
 
 
-def test_safe_urlopen_delegates_only_to_redirect_rejecting_opener(monkeypatch):
+def test_safe_urlopen_delegates_only_to_deadline_bounded_exchange(monkeypatch):
     module = _load()
     sentinel = object()
     monkeypatch.setattr(
-        module._SAFE_OPENER,
+        module.deadline_http,
-        "open",
+        "open_bounded",
-        lambda request, timeout: (request, timeout, sentinel),
+        lambda request, *, maximum, timeout: (request, maximum, timeout, sentinel),
     )
     request = object()
-    assert module._safe_urlopen(request, 7) == (request, 7, sentinel)
+    assert module._safe_urlopen(request, 7) == (
+        request,
+        module.MAX_RESPONSE_BYTES,
+        7,
+        sentinel,
+    )
 
 
 def test_api_target_and_request_body_edge_paths(monkeypatch):

testing/tests/test_hermes_gitea_pr_client.py

@@ -2,21 +2,22 @@
 
 from __future__ import annotations
 
+import urllib.error
 import urllib.request
 
 import pytest
 
 from testing.tests.test_hermes_gitea_support import HEAD_SHA, Response, _load
 
-def test_redirect_handler_rejects_cross_origin_with_sentinel_authorization():
+def test_default_opener_rejects_redirects_under_an_absolute_deadline(monkeypatch):
     client = _load()
     source = urllib.request.Request(
         "https://scm.bstein.dev/api/v1/repos/atlas/cassandra",
         headers={"Authorization": "token redirect-sentinel"},
     )
 
-    with pytest.raises(client.PolicyError, match="redirects are not allowed") as exc:
+    with pytest.raises(urllib.error.URLError) as exc:
-        client.RejectRedirectHandler().redirect_request(
+        client.deadline_http._RejectRedirect().redirect_request(
             source,
             None,
             302,
@@ -26,10 +27,16 @@ def test_redirect_handler_rejects_cross_origin_with_sentinel_authorization():
         )
 
     assert "redirect-sentinel" not in str(exc.value)
-    assert any(
+    seen = []
-        isinstance(handler, client.RejectRedirectHandler)
+    monkeypatch.setattr(
-        for handler in client._SAFE_OPENER.handlers
+        client.deadline_http,
+        "open_bounded",
+        lambda request, *, maximum, timeout: seen.append(
+            (request, maximum, timeout)
+        ),
     )
+    client._safe_urlopen(source, 30)
+    assert seen == [(source, client.MAX_RESPONSE_BYTES, 30)]
 
 
 @pytest.mark.parametrize(

testing/tests/test_hermes_scm_broker_client_coverage.py

@@ -9,17 +9,20 @@ import pytest
 from testing.tests.test_hermes_scm_broker_support import Response, _load
 
 
-def test_client_redirect_and_default_opener_are_fail_closed(monkeypatch):
+def test_client_default_opener_uses_deadline_bounded_exchange(monkeypatch):
     client = _load("scm_broker_client")
-    with pytest.raises(client.PolicyError, match="redirects"):
-        client.RejectRedirectHandler().redirect_request(
-            object(), None, 302, "redirect", {}, "http://elsewhere.invalid"
-        )
     sentinel = object()
     monkeypatch.setattr(
-        client._OPENER, "open", lambda request, timeout: (request, timeout, sentinel)
+        client.deadline_http,
+        "open_bounded",
+        lambda request, *, maximum, timeout: (request, maximum, timeout, sentinel),
+    )
+    assert client._open("request", 4) == (
+        "request",
+        client.MAX_RESPONSE_BYTES,
+        4,
+        sentinel,
     )
-    assert client._open("request", 4) == ("request", 4, sentinel)
 
 
 @pytest.mark.parametrize(