hermes: bound streaming git DNS resolution
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>
This commit is contained in:
parent
6c1123201e
commit
31439f096a
@ -227,6 +227,31 @@ def open_bounded(
|
|||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
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:
|
class StreamDeadline:
|
||||||
"""Force-close tracked connections once an absolute deadline passes."""
|
"""Force-close tracked connections once an absolute deadline passes."""
|
||||||
|
|
||||||
@ -270,6 +295,23 @@ class StreamDeadline:
|
|||||||
raise PolicyError("HTTP stream deadline exceeded")
|
raise PolicyError("HTTP stream deadline exceeded")
|
||||||
return connection
|
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:
|
def cancel(self) -> None:
|
||||||
self._timer.cancel()
|
self._timer.cancel()
|
||||||
|
|
||||||
@ -280,7 +322,7 @@ class StreamDeadline:
|
|||||||
class GuardedHTTPHandler(urllib.request.HTTPHandler):
|
class GuardedHTTPHandler(urllib.request.HTTPHandler):
|
||||||
def http_open(self, req):
|
def http_open(self, req):
|
||||||
return self.do_open(
|
return self.do_open(
|
||||||
lambda host, **kwargs: deadline._track(
|
lambda host, **kwargs: deadline._bounded_connection(
|
||||||
http.client.HTTPConnection(host, **kwargs)
|
http.client.HTTPConnection(host, **kwargs)
|
||||||
),
|
),
|
||||||
req,
|
req,
|
||||||
@ -289,7 +331,7 @@ class StreamDeadline:
|
|||||||
class GuardedHTTPSHandler(urllib.request.HTTPSHandler):
|
class GuardedHTTPSHandler(urllib.request.HTTPSHandler):
|
||||||
def https_open(self, req):
|
def https_open(self, req):
|
||||||
return self.do_open(
|
return self.do_open(
|
||||||
lambda host, **kwargs: deadline._track(
|
lambda host, **kwargs: deadline._bounded_connection(
|
||||||
http.client.HTTPSConnection(host, **kwargs)
|
http.client.HTTPSConnection(host, **kwargs)
|
||||||
),
|
),
|
||||||
req,
|
req,
|
||||||
|
|||||||
@ -107,6 +107,93 @@ def test_handlers_build_guarded_connections_without_network(monkeypatch):
|
|||||||
guard.cancel()
|
guard.cancel()
|
||||||
|
|
||||||
|
|
||||||
|
def test_dns_resolution_is_bounded_by_the_stream_deadline(monkeypatch):
|
||||||
|
module = _load("deadline_http")
|
||||||
|
|
||||||
|
def hang(*_args, **_kwargs):
|
||||||
|
time.sleep(30)
|
||||||
|
|
||||||
|
monkeypatch.setattr(module.socket, "getaddrinfo", hang)
|
||||||
|
started = time.monotonic()
|
||||||
|
with pytest.raises(module.PolicyError, match="DNS resolution deadline"):
|
||||||
|
module._resolve_within("example.invalid", 443, 0.4)
|
||||||
|
assert time.monotonic() - started < 3
|
||||||
|
|
||||||
|
|
||||||
|
def test_resolve_within_returns_on_success_and_propagates_errors(monkeypatch):
|
||||||
|
module = _load("deadline_http")
|
||||||
|
calls = []
|
||||||
|
monkeypatch.setattr(
|
||||||
|
module.socket, "getaddrinfo", lambda *a, **k: calls.append((a, k))
|
||||||
|
)
|
||||||
|
module._resolve_within("127.0.0.1", 80, 5)
|
||||||
|
assert calls
|
||||||
|
|
||||||
|
def boom(*_a, **_k):
|
||||||
|
raise OSError("name resolution failed")
|
||||||
|
|
||||||
|
monkeypatch.setattr(module.socket, "getaddrinfo", boom)
|
||||||
|
with pytest.raises(OSError, match="name resolution failed"):
|
||||||
|
module._resolve_within("127.0.0.1", 80, 5)
|
||||||
|
|
||||||
|
|
||||||
|
def test_bounded_connection_wraps_connect_to_resolve_first(monkeypatch):
|
||||||
|
module = _load("deadline_http")
|
||||||
|
guard = module.StreamDeadline(30, timer=_FakeTimer)
|
||||||
|
order = []
|
||||||
|
monkeypatch.setattr(
|
||||||
|
module,
|
||||||
|
"_resolve_within",
|
||||||
|
lambda host, port, timeout: order.append(("resolve", host, port)),
|
||||||
|
)
|
||||||
|
|
||||||
|
class FakeConnection:
|
||||||
|
host = "scm.bstein.dev"
|
||||||
|
port = 443
|
||||||
|
|
||||||
|
def __init__(self):
|
||||||
|
self.sock = None
|
||||||
|
|
||||||
|
def connect(self):
|
||||||
|
order.append(("connect", self.host))
|
||||||
|
|
||||||
|
def close(self):
|
||||||
|
order.append(("close",))
|
||||||
|
|
||||||
|
connection = guard._bounded_connection(FakeConnection())
|
||||||
|
assert connection in guard._connections
|
||||||
|
connection.connect()
|
||||||
|
assert order == [("resolve", "scm.bstein.dev", 443), ("connect", "scm.bstein.dev")]
|
||||||
|
guard.cancel()
|
||||||
|
|
||||||
|
|
||||||
|
def test_bounded_connection_connect_fails_closed_once_expired(monkeypatch):
|
||||||
|
module = _load("deadline_http")
|
||||||
|
guard = module.StreamDeadline(30, timer=_FakeTimer)
|
||||||
|
resolved = []
|
||||||
|
monkeypatch.setattr(
|
||||||
|
module, "_resolve_within", lambda *a, **k: resolved.append(a)
|
||||||
|
)
|
||||||
|
|
||||||
|
class FakeConnection:
|
||||||
|
host = "scm.bstein.dev"
|
||||||
|
port = 443
|
||||||
|
sock = None
|
||||||
|
|
||||||
|
def connect(self):
|
||||||
|
resolved.append("connected")
|
||||||
|
|
||||||
|
def close(self):
|
||||||
|
pass
|
||||||
|
|
||||||
|
connection = guard._bounded_connection(FakeConnection())
|
||||||
|
guard._expire()
|
||||||
|
with pytest.raises(module.PolicyError, match="deadline exceeded"):
|
||||||
|
connection.connect()
|
||||||
|
assert resolved == []
|
||||||
|
guard.cancel()
|
||||||
|
|
||||||
|
|
||||||
class _TrickleHandler(BaseHTTPRequestHandler):
|
class _TrickleHandler(BaseHTTPRequestHandler):
|
||||||
def do_GET(self):
|
def do_GET(self):
|
||||||
time.sleep(10)
|
time.sleep(10)
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user