diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 73d21f3..3bbc826 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -56,10 +56,11 @@ The SDK works with: - `httpx` and `httpx2`, sync and async - `requests` - `aiohttp` 3.13 and later +- `urllib.request` - Python 3.10 through 3.14 - `mnfst run ` or a `manifest()` call at startup -Custom transports, `urllib` and other clients are not covered. See +Custom transports and other clients are not covered. See [the coverage details](docs/guide.md). ## License diff --git a/docs/guide.md b/docs/guide.md index f09c132..4c0d20d 100644 --- a/docs/guide.md +++ b/docs/guide.md @@ -74,7 +74,7 @@ Then send a JSON or `application/x-www-form-urlencoded` request that your test A ## Behavior and limits -- Standard httpx and httpx2 transports, requests adapters and aiohttp 3.13+ sessions are instrumented process-wide, including existing clients. aiohttp is covered through its client middleware chain; the SDK's middleware runs innermost, so your own middlewares and `raise_for_status` see the result of the heal. Custom transports, `urllib`, aiohttp before 3.13, websocket handshakes, browsers and other languages are not covered. +- Standard httpx and httpx2 transports, requests adapters, aiohttp 3.13+ sessions and `urllib.request` openers are instrumented process-wide, including existing clients. aiohttp is covered through its client middleware chain; the SDK's middleware runs innermost, so your own middlewares and `raise_for_status` see the result of the heal. A `urllib.request` failure is captured before urllib raises its `HTTPError`, so a healed call returns normally. Custom transports and handlers, aiohttp before 3.13, websocket handshakes, browsers and other languages are not covered. - Any 4xx is sent to Manifest for healing except 401, 402, 403 and 429. Successful calls, those four, and every 5xx are never healed: authentication, billing, rate limiting and server faults are not repaired by editing the request. They are tracked as metadata only (see "Data sent to Manifest"). A network failure before an HTTP response passes through untracked. - Manifest selects repairs. The SDK retries at most once per captured failure. The original error response is returned if healing is unavailable, no repair can be applied, or the retry has a transport error. - A successful streaming retry remains streamed. Error capture reads a bounded prefix and preserves the original response bytes for the caller. Error reads use the caller's read timeout; healing adds up to 60 seconds, and the retry uses the caller's timeout. aiohttp's total timeout covers the whole call, so there healing takes at most half of the time left and the rest is kept for the retry. diff --git a/src/mnfst/outbound.py b/src/mnfst/outbound.py index 89ed868..19a1a9c 100644 --- a/src/mnfst/outbound.py +++ b/src/mnfst/outbound.py @@ -1,4 +1,5 @@ -"""Auto-instrumentation of outbound HTTP clients (httpx, httpx2, requests and aiohttp). +"""Auto-instrumentation of outbound HTTP clients (httpx, httpx2, requests, +aiohttp and urllib.request). Patches at TRANSPORT level: one hook per client library, so every client — including ones created before manifest() ran — is covered, and redirects, retries and streaming stay the client's business. The internal_call guard @@ -10,6 +11,7 @@ from __future__ import annotations import atexit +import copy import platform import time import uuid @@ -26,7 +28,8 @@ from .gate import should_capture from .heal_api import NOT_ATTEMPTED, NOT_SENT, AsyncHealApi, HealApi, HealEvent, internal_call from .merge import merge_healed_body -from .response_capture import capture_aiohttp, capture_httpx, capture_httpx_async, capture_requests +from .response_capture import (capture_aiohttp, capture_httpx, capture_httpx_async, + capture_requests, capture_urllib) from .tracking import CallBuffer from .wire import capped_response_body, heal_payload, safe_error_text, tracked_url, traveling_body @@ -261,6 +264,7 @@ def install_outbound(config: Config, heal_api: Optional[HealApi] = None, _install_httpx(mod, config, sync_api, async_api) install_requests(config, sync_api) install_aiohttp(config, async_api) + install_urllib(config, sync_api) # Announce the install, so that silence stops being ambiguous: a healthy # app and a broken one are otherwise the same nothing on the dashboard. # Once per process, fire-and-forget — it must never delay startup, and a @@ -382,6 +386,7 @@ def uninstall_outbound() -> None: mod.AsyncHTTPTransport.handle_async_request = _originals.pop(async_key) uninstall_requests() uninstall_aiohttp() + uninstall_urllib() _installed = False _installed_config = None @@ -611,3 +616,91 @@ async def mnfst_middleware(request, handler): return response # nothing in here may reach the caller return mnfst_middleware + + +# --- urllib.request ------------------------------------------------------------ +# AbstractHTTPHandler.do_open is urllib's one hop on the wire, shared by the +# http and https handlers. It returns the raw response for any status, before +# urllib's processors turn a 4xx into an HTTPError and follow redirects. + +def install_urllib(config: Config, heal_api: HealApi) -> None: + import urllib.request + if "urllib_do_open" in _originals: + return + + _originals["urllib_do_open"] = urllib.request.AbstractHTTPHandler.do_open + + def patched_do_open(self, http_class, req, **http_conn_args): + original = _originals["urllib_do_open"] + started_at = time.time() + started = time.monotonic() + response = original(self, http_class, req, **http_conn_args) + elapsed_ms = int((time.monotonic() - started) * 1000) + if internal_call.get() or _excluded(config, req.full_url): + return response + method = req.get_method() + try: + if not should_capture(response.status) or not heal_api.healing_enabled(): + _track(method, req.full_url, response.status, started_at, elapsed_ms) + return response + try: + response, raw, incomplete = capture_urllib(response) + except Exception: + return response + # A file or iterable body is not bytes: capture without it, so an + # upload is never consumed. + data = req.data + content = data if isinstance(data, bytes) else (b"" if data is None else None) + capture = _Capture(method, req.full_url, dict(req.header_items()), content, + response.status, raw, elapsed_ms, incomplete) + result = heal_api.heal(capture.payload) + if result is NOT_SENT: # every heal slot busy: track it, never lose it + _track(method, req.full_url, response.status, started_at, elapsed_ms) + return response + retry = _decide(config, heal_api, capture, result) + if retry is None: + return response + try: + retried = original(self, http_class, _retarget_urllib(req, retry), **http_conn_args) + retry_body, truncated = None, False + if retried.status >= 400: + retried, raw, incomplete = capture_urllib(retried) + retry_body, truncated = capped_response_body(raw) + truncated = truncated or incomplete + except Exception as exc: + _report(heal_api, result, 0, safe_error_text(exc)) + _emit(config, capture, result, None) + return response + _report(heal_api, result, retried.status, retry_body, truncated) + _emit(config, capture, result, retried.status) + try: + response.close() + except Exception: + pass + return retried + except Exception: + return response # nothing in here may reach the caller + + urllib.request.AbstractHTTPHandler.do_open = patched_do_open + + +def uninstall_urllib() -> None: + if "urllib_do_open" not in _originals: + return + import urllib.request + urllib.request.AbstractHTTPHandler.do_open = _originals.pop("urllib_do_open") + + +def _retarget_urllib(req, retry: _Retry): + """A copy of the request with the retry applied. The processors that set + Content-Length and Host already ran on the original; http.client sizes a + bytes body itself, and Host rides in the captured headers.""" + rebuilt = copy.copy(req) + rebuilt.full_url = retry.url # same origin: _apply refuses anything else + rebuilt.host = req.host # the proxy, when there is one + if req.has_proxy(): + rebuilt.selector = rebuilt.full_url + rebuilt.headers, rebuilt.unredirected_hdrs = dict(retry.headers), {} + rebuilt.method = req.get_method() # else urllib infers POST from any body + rebuilt.data = retry.content + return rebuilt diff --git a/src/mnfst/response_capture.py b/src/mnfst/response_capture.py index ccaf9f7..08066ad 100644 --- a/src/mnfst/response_capture.py +++ b/src/mnfst/response_capture.py @@ -8,6 +8,7 @@ import io import warnings import zlib +from itertools import chain from types import ModuleType from typing import Iterator @@ -151,7 +152,6 @@ def close(self): def capture_requests(response): - from itertools import chain from urllib3.response import HTTPResponse if response._content is not False: return response, response.content[:RESPONSE_BODY_CAP + 1] @@ -212,3 +212,40 @@ async def capture_aiohttp(response): if getattr(stream, 'total_compressed_bytes', None) is None: raw = _decode(raw, response.headers.get('content-encoding', '')) return raw, incomplete + + +def capture_urllib(response): + """Read up to the cap from a stdlib http.client failure, then give the + same response object a body of those bytes followed by the unread rest. + It stays an HTTPResponse, so an HTTPError built on it, or a caller + without urllib's error processing, reads the whole body. Returns + (response, decoded prefix, incomplete).""" + rest = _clone(response) # reads the connection from here on + chunks, size = [], 0 + incomplete = True + iterator = iter(lambda: rest.read(65536), b'') + try: + for chunk in iterator: + chunks.append(chunk) + size += len(chunk) + if size > RESPONSE_BODY_CAP: + break + else: + incomplete = False + except Exception as exc: + error = exc + def failed(): + raise error + yield # pragma: no cover + iterator = failed() + # The replacement body is already de-chunked and has no known length. + response.fp = io.BufferedReader(_Reader(chain(chunks, iterator), rest)) + response.chunked, response.length = False, None + raw = b''.join(chunks)[:RESPONSE_BODY_CAP + 1] + return response, _decode(raw, response.headers.get('content-encoding', '')), incomplete + + +def _clone(response): + twin = type(response).__new__(type(response)) + twin.__dict__.update(response.__dict__) + return twin diff --git a/tests/test_outbound_urllib.py b/tests/test_outbound_urllib.py new file mode 100644 index 0000000..4ef2a00 --- /dev/null +++ b/tests/test_outbound_urllib.py @@ -0,0 +1,249 @@ +"""urllib.request through the real patch: the SDK wraps the HTTP handlers' +do_open, one hop on the wire, before urllib turns a 4xx into an HTTPError.""" +import gzip +import io +import json +import urllib.error +import urllib.parse +import urllib.request + +import pytest + +from mnfst import outbound +from mnfst.outbound import uninstall_outbound +from mnfst.bodies import is_form +from mnfst.wire import RESPONSE_BODY_CAP +from tests.helpers import wait_for +from tests.test_outbound_httpx import BIG_ERROR_CHUNKS, build_rig, header + + +@pytest.fixture +def rig(): + provider, stub = build_rig() + yield provider, stub + uninstall_outbound() + stub.stop() + provider.stop() + + +@pytest.fixture +def denied(): + provider, stub = build_rig(denylist="127.0.0.1") + yield provider, stub + uninstall_outbound() + stub.stop() + provider.stop() + + +def post_json(url, body, headers=None): + request = urllib.request.Request(url, data=json.dumps(body).encode(), + headers={"Content-Type": "application/json", + **(headers or {})}) + return urllib.request.urlopen(request, timeout=5) + + +def test_json_post_heals(rig): + provider, stub = rig + with post_json(f"{provider.url}/v1/generate", {"model": "m", "temperature": 0.2}) as response: + assert response.status == 200 + assert json.loads(response.read())["received"] == {"model": "m"} + assert len(stub.heals) == 1 + heal = stub.heals[0] + assert heal["request"]["method"] == "POST" + assert heal["request"]["body"] == {"model": "m", "temperature": 0.2} + assert heal["response"]["statusCode"] == 400 + assert "temperature unsupported" in json.dumps(heal["response"]["body"]) + + +def test_success_is_untouched(rig): + provider, stub = rig + with post_json(f"{provider.url}/v1/generate", {"model": "m"}) as response: + assert response.status == 200 + assert stub.heals == [] + + +def test_streamed_200_reaches_the_caller_unread(rig): + provider, stub = rig + with post_json(f"{provider.url}/v1/stream", {"model": "m"}) as response: + assert [response.read(6) for _ in range(3)] == [b"chunk0", b"chunk1", b"chunk2"] + assert stub.heals == [] + + +def test_unhealed_failure_raises_http_error_with_its_body(rig): + provider, stub = rig + stub.result = None # no_patch + with pytest.raises(urllib.error.HTTPError) as caught: + post_json(f"{provider.url}/v1/generate", {"temperature": 1}) + error = caught.value + assert error.code == 400 + assert error.reason == "Bad Request" + assert json.loads(error.read())["error"]["message"] == "temperature unsupported" + assert len(stub.heals) == 1 + + +def test_an_opener_without_error_processing_gets_the_raw_response(rig): + provider, stub = rig + stub.result = None + opener = urllib.request.OpenerDirector() + opener.add_handler(urllib.request.HTTPHandler()) + request = urllib.request.Request(f"{provider.url}/v1/generate", + data=b'{"temperature": 1}', + headers={"Content-Type": "application/json"}) + with opener.open(request, timeout=5) as response: + assert response.status == 400 + assert response.msg == "Bad Request" + assert response.getheader("content-type") == "application/json" + assert response.readline().startswith(b'{"error"') + assert len(stub.heals) == 1 + + +def test_healed_headers_and_url_apply(rig): + provider, stub = rig + stub.result = {"status": "patched", "issueId": "i1", "healAttemptId": "a1", + "healedRequest": {"url": f"{provider.url}/v1/generate", + "headers": {"X-Beta": "on", "x-bad": None}}} + with post_json(f"{provider.url}/v1/old", {"needs_beta": True}, + headers={"X-Bad": "1", "Authorization": "Bearer sk"}) as response: + assert response.status == 200 + assert response.geturl() == f"{provider.url}/v1/generate" + route, headers, body = provider.received[-1] + assert route == "/v1/generate" + assert header(headers, "x-beta") == "on" + assert header(headers, "x-bad") is None + assert header(headers, "authorization") == "Bearer sk" + assert body == {"needs_beta": True} + assert stub.heals[0]["request"]["headers"]["authorization"] == "REDACTED" + + +def test_a_replay_through_a_proxy_goes_through_the_proxy(rig): + """The provider stands in for an http proxy: requests reach it in + absolute form, and so must the retry.""" + provider, stub = rig + opener = urllib.request.build_opener(urllib.request.ProxyHandler({"http": provider.url})) + request = urllib.request.Request("http://api.example.test/v1/generate", + data=b'{"model": "m", "temperature": 0.2}', + headers={"Content-Type": "application/json"}) + with opener.open(request, timeout=5) as response: + assert response.status == 200 + assert [route for route, _ in provider.requests] == [ + "http://api.example.test/v1/generate"] * 2 + assert stub.heals[0]["request"]["url"] == "http://api.example.test/v1/generate" + + +def test_cross_origin_url_heal_is_refused(rig): + provider, stub = rig + stub.result = {"status": "patched", "issueId": "i1", "healAttemptId": "a1", + "healedRequest": {"url": "http://evil.example/v1/generate"}} + with pytest.raises(urllib.error.HTTPError) as caught: + post_json(f"{provider.url}/v1/old", {"model": "m"}) + assert caught.value.code == 404 + assert wait_for(lambda: stub.outcomes), "no outcome report arrived" + assert stub.outcomes[0][1]["failure"]["kind"] == "not_attempted" + + +def test_form_body_is_healed_and_replayed_as_a_form(rig): + provider, stub = rig + stub.result = {"status": "patched", "issueId": "i1", "healAttemptId": "a1", + "healedRequest": {"body": {"model": "m"}}} + data = urllib.parse.urlencode({"model": "m", "temperature": "0.2"}).encode() + with urllib.request.urlopen(f"{provider.url}/v1/generate", data=data, timeout=5) as response: + assert response.status == 200 + assert stub.heals[0]["request"]["body"] == {"model": "m", "temperature": "0.2"} + route, raw = provider.requests[-1] + assert raw == b"model=m" + headers = provider.received[-1][1] + assert is_form(header(headers, "content-type")) + assert header(headers, "content-length") == str(len(raw)) + + +def test_a_file_body_is_captured_without_being_read(rig): + provider, stub = rig + stub.result = None + body = io.BytesIO(b'{"model": "m"}') + request = urllib.request.Request(f"{provider.url}/v1/old", data=body, + headers={"Content-Type": "application/json", + "Content-Length": "14"}) + with pytest.raises(urllib.error.HTTPError): + urllib.request.urlopen(request, timeout=5) + assert stub.heals[0]["request"]["body"] is None + + +def test_failed_get_is_captured_with_no_body(rig): + provider, stub = rig + with pytest.raises(urllib.error.HTTPError) as caught: + urllib.request.urlopen(f"{provider.url}/v1/generate", timeout=5) + assert caught.value.code == 404 + assert stub.heals[0]["request"]["method"] == "GET" + assert stub.heals[0]["request"]["body"] is None + assert len(provider.gets) == 2 # the replay kept its method + assert provider.requests == [] + + +def test_replay_exception_reports_outcome_and_serves_the_original(rig): + provider, stub = rig + with pytest.raises(urllib.error.HTTPError) as caught: + post_json(f"{provider.url}/v1/flaky", {"model": "m", "temperature": 0.2}) + assert json.loads(caught.value.read())["error"]["message"] == "temperature unsupported" + assert wait_for(lambda: stub.outcomes), "no outcome report arrived" + assert stub.outcomes[0][1]["failure"]["kind"] == "transport_error" + + +def test_crash_in_heal_branch_serves_the_original_response(rig, monkeypatch): + provider, stub = rig + + def boom(*args, **kwargs): + raise RuntimeError("merge exploded") + + monkeypatch.setattr("mnfst.outbound._apply", boom) + with pytest.raises(urllib.error.HTTPError) as caught: + post_json(f"{provider.url}/v1/generate", {"model": "m", "temperature": 0.2}) + assert json.loads(caught.value.read())["error"]["message"] == "temperature unsupported" + + +def test_a_gzip_error_body_is_decoded_for_manifest_and_left_raw_for_the_caller(rig): + provider, stub = rig + stub.result = None + with pytest.raises(urllib.error.HTTPError) as caught: + post_json(f"{provider.url}/v1/gzip", {"model": "m"}) + body = gzip.decompress(caught.value.read()) + assert json.loads(body)["error"]["message"] == "temperature unsupported" + assert "temperature unsupported" in json.dumps(stub.heals[0]["response"]["body"]) + + +def test_an_error_body_over_the_cap_reaches_the_caller_whole(rig): + provider, stub = rig + with pytest.raises(urllib.error.HTTPError) as caught: + post_json(f"{provider.url}/v1/big", {"model": "m"}) + body = caught.value.read() + assert body == b"x" * (16384 * BIG_ERROR_CHUNKS) + assert len(body) > RESPONSE_BODY_CAP + assert stub.heals[0]["response"]["truncated"] is True + assert len(provider.requests) == 1 # a truncated capture is never replayed + + +def test_calls_are_tracked(rig): + provider, stub = rig + post_json(f"{provider.url}/v1/generate", {"model": "m"}).close() + outbound._tracker.flush(5) + assert [call["statusCode"] for call in stub.tracked] == [200] + assert stub.tracked[0]["method"] == "POST" + + +def test_a_denied_host_is_neither_healed_nor_tracked(denied): + provider, stub = denied + with pytest.raises(urllib.error.HTTPError): + post_json(f"{provider.url}/v1/generate", {"model": "m", "temperature": 0.2}) + outbound._tracker.flush(5) + assert stub.heals == [] and stub.tracked == [] + + +def test_uninstall_restores_the_handler(): + original = urllib.request.AbstractHTTPHandler.do_open + provider, stub = build_rig() + try: + assert urllib.request.AbstractHTTPHandler.do_open is not original + finally: + uninstall_outbound() + stub.stop() + provider.stop() + assert urllib.request.AbstractHTTPHandler.do_open is original