31439f096a
The watchdog can only close a live socket, but getaddrinfo runs before any socket exists and ignores socket timeouts, so a slow resolver outlived the stream deadline. Resolve under the same absolute deadline in a joinable worker before the real connect. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
345 lines
12 KiB
Python
345 lines
12 KiB
Python
#!/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)
|
|
)
|
|
|
|
|
|
def _resolve_within(host: str, port: int, timeout: float) -> None:
|
|
"""Resolve one host under a hard timeout the socket layer never covers.
|
|
|
|
``getaddrinfo`` ignores socket timeouts and runs before any socket
|
|
exists, so a slow or hostile resolver would otherwise outlive the whole
|
|
stream deadline. Running it in a joinable worker bounds resolution by the
|
|
same wall clock; a stuck lookup leaks only one daemon thread.
|
|
"""
|
|
outcome: dict[str, object] = {}
|
|
|
|
def run() -> None:
|
|
try:
|
|
outcome["value"] = socket.getaddrinfo(host, port, type=socket.SOCK_STREAM)
|
|
except OSError as exc:
|
|
outcome["error"] = exc
|
|
|
|
worker = threading.Thread(target=run, daemon=True)
|
|
worker.start()
|
|
worker.join(timeout)
|
|
if worker.is_alive():
|
|
raise PolicyError("HTTP stream DNS resolution deadline exceeded")
|
|
if "error" in outcome:
|
|
raise outcome["error"] # type: ignore[misc]
|
|
|
|
|
|
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 _bounded_connection(self, connection: http.client.HTTPConnection):
|
|
"""Register one connection and bound its DNS phase by the deadline.
|
|
|
|
The watchdog can only close a live socket, but ``connect`` resolves
|
|
the host before any socket exists. Wrapping ``connect`` runs that
|
|
resolution under the same absolute deadline first, then hands off to
|
|
the real connect for the socket phases the watchdog already covers.
|
|
"""
|
|
original_connect = connection.connect
|
|
|
|
def connect() -> None:
|
|
_resolve_within(connection.host, connection.port, self.remaining())
|
|
original_connect()
|
|
|
|
connection.connect = connect # type: ignore[method-assign]
|
|
return self._track(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._bounded_connection(
|
|
http.client.HTTPConnection(host, **kwargs)
|
|
),
|
|
req,
|
|
)
|
|
|
|
class GuardedHTTPSHandler(urllib.request.HTTPSHandler):
|
|
def https_open(self, req):
|
|
return self.do_open(
|
|
lambda host, **kwargs: deadline._bounded_connection(
|
|
http.client.HTTPSConnection(host, **kwargs)
|
|
),
|
|
req,
|
|
)
|
|
|
|
return (GuardedHTTPHandler(), GuardedHTTPSHandler())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(_child_main() if sys.argv[1:] == ["--child"] else 2)
|