diff --git a/README.md b/README.md index 9d65910..37d557b 100644 --- a/README.md +++ b/README.md @@ -108,6 +108,7 @@ MNFST_DENYLIST=stripe.com/v1/charges,internal.example.com # never these - A path matches whole segments: `/v1/charges` covers `/v1/charges/ch_123`, not `/v1/charges_export`. Paths are case-sensitive. - A scheme, port, query or fragment in an entry is ignored. `*` in a path is not supported yet: the entry is skipped with a warning, and an allowlist made only of skipped entries lets nothing through. - The denylist wins over the allowlist. With no allowlist, every call is eligible. +- Paths are compared decoded, with `.` and `..` resolved, so `/%70rivate` and `/public/..%2Fprivate` both match `/private`. A patched retry is filtered too: a repair never moves a call onto an excluded path. Or in code: `manifest(denylist=["stripe.com/v1/charges"])`. An option overrides its environment variable. With `mnfst run`, set the environment variables: a later `manifest()` call cannot change the configuration. diff --git a/docs/guide.md b/docs/guide.md index 4c0d20d..6f2fa0f 100644 --- a/docs/guide.md +++ b/docs/guide.md @@ -77,7 +77,7 @@ Then send a JSON or `application/x-www-form-urlencoded` request that your test A - 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. +- A successful streaming retry remains streamed. Error capture reads a bounded prefix and preserves the original response bytes for the caller. Error capture stops after one second, and the rest of the body stays with the caller; 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. - The sync heal worker pool permits eight concurrent calls. Excess calls fail open. Timed-out workers can continue in the background within that bound. Outcome reporting permits 64 concurrent reports per reporter. - JSON and `application/x-www-form-urlencoded` request bodies are parsed, including nested form keys such as `line_items[0][price]`. Both are limited to 256 KiB and depth 64. Multipart, binary, streamed, oversized and invalid bodies travel as `null`; they are not generally repairable. A form retry is re-encoded from the parsed structure, so a repeated key such as `expand=a&expand=b` returns as `expand[0]=a&expand[1]=b`. Response metadata is limited to 64 KiB; truncated errors are reported without retry. Gzip and deflate error prefixes are decoded within that limit; unsupported content encodings provide no body evidence. - Retries can repeat side effects. Use APIs with safe retry semantics and caller-managed idempotency keys. Existing credentials and idempotency headers are retained unless explicitly changed by the repair. URL repairs must stay on the same origin. diff --git a/src/mnfst/outbound.py b/src/mnfst/outbound.py index 19a1a9c..69ab4c3 100644 --- a/src/mnfst/outbound.py +++ b/src/mnfst/outbound.py @@ -15,6 +15,7 @@ import platform import time import uuid +import warnings from datetime import datetime, timezone from typing import Any, Mapping, Optional from urllib.parse import urlsplit @@ -141,7 +142,10 @@ def _apply(capture: _Capture, healed: dict) -> Optional[_Retry]: if not _same_origin(url, capture.url): return None # a URL heal may move the path, never the host: the # retry carries the caller's credentials (CONTRACT §4) - headers = {k: v for k, v in capture.headers.items() if k.lower() != "content-length"} + # The retry's body is concrete bytes: the client frames it anew, so neither the old + # length nor chunked framing may ride along (both at once is a malformed request). + headers = {k: v for k, v in capture.headers.items() + if k.lower() not in ("content-length", "transfer-encoding")} for name, value in (healed.get("headers") or {}).items(): headers = {k: v for k, v in headers.items() if k.lower() != str(name).lower()} if value is not None: @@ -209,6 +213,9 @@ def _decide(config: Config, api, capture: _Capture, result: Optional[dict]) -> O retry = _apply(capture, healed) if healed else None except Exception: retry = None + # A patched path is filtered like any other call: a retry never goes where the lists forbid. + if retry is not None and _excluded(config, retry.url): + retry = None if retry is None: _report(api, result, 0, NOT_ATTEMPTED) _emit(config, capture, result, None, replay_attempted=False) @@ -248,9 +255,23 @@ def _rebuild(request, retry: _Retry, mod=httpx): def install_outbound(config: Config, heal_api: Optional[HealApi] = None, async_heal_api: Optional[AsyncHealApi] = None) -> None: - global _installed, _installed_config, _tracker if _installed: return + # Instrumenting must never stop the app: a failure (a malformed proxy variable + # breaking client construction, say) undoes what was patched and leaves it running. + try: + _install(config, heal_api, async_heal_api) + except Exception as exc: + try: + uninstall_outbound() + except Exception: + pass + warnings.warn(f"mnfst: install failed; mnfst is disabled: {type(exc).__name__}", stacklevel=3) + + +def _install(config: Config, heal_api: Optional[HealApi], + async_heal_api: Optional[AsyncHealApi]) -> None: + global _installed, _installed_config, _tracker sync_api = heal_api or HealApi(config) async_api = async_heal_api or AsyncHealApi(config) _installed = True diff --git a/src/mnfst/response_capture.py b/src/mnfst/response_capture.py index 08066ad..33a7882 100644 --- a/src/mnfst/response_capture.py +++ b/src/mnfst/response_capture.py @@ -6,6 +6,7 @@ from __future__ import annotations import io +import time import warnings import zlib from itertools import chain @@ -16,6 +17,10 @@ from .wire import RESPONSE_BODY_CAP +# Past this, capture stops reading and hands the rest of the stream back to the +# caller: a server that trickles a failed response must not stall it indefinitely. +CAPTURE_DEADLINE_SECONDS = 1.0 + # httpx and httpx2 are separate packages with the same API. A restored # response must be built from the package that produced the original, and # its stream must subclass that package's byte stream, so the replay @@ -77,11 +82,12 @@ def capture_httpx(response, mod: ModuleType = httpx): chunks, size = [], 0 incomplete = True iterator = iter(response.stream) + deadline = time.monotonic() + CAPTURE_DEADLINE_SECONDS try: for chunk in iterator: chunks.append(chunk) size += len(chunk) - if size > RESPONSE_BODY_CAP: + if size > RESPONSE_BODY_CAP or time.monotonic() > deadline: break else: incomplete = False @@ -108,11 +114,12 @@ async def capture_httpx_async(response, mod: ModuleType = httpx): chunks, size = [], 0 incomplete = True iterator = response.stream.__aiter__() + deadline = time.monotonic() + CAPTURE_DEADLINE_SECONDS try: async for chunk in iterator: chunks.append(chunk) size += len(chunk) - if size > RESPONSE_BODY_CAP: + if size > RESPONSE_BODY_CAP or time.monotonic() > deadline: break else: incomplete = False @@ -159,11 +166,12 @@ def capture_requests(response): chunks, size = [], 0 incomplete = True iterator = original.stream(amt=65536, decode_content=False) + deadline = time.monotonic() + CAPTURE_DEADLINE_SECONDS try: for chunk in iterator: chunks.append(chunk) size += len(chunk) - if size > RESPONSE_BODY_CAP: + if size > RESPONSE_BODY_CAP or time.monotonic() > deadline: break else: incomplete = False @@ -191,8 +199,9 @@ async def capture_aiohttp(response): stream = response.content chunks, size = [], 0 incomplete = True + deadline = time.monotonic() + CAPTURE_DEADLINE_SECONDS try: - while size <= RESPONSE_BODY_CAP: + while size <= RESPONSE_BODY_CAP and time.monotonic() <= deadline: chunk = await stream.readany() if not chunk: incomplete = False @@ -224,11 +233,12 @@ def capture_urllib(response): chunks, size = [], 0 incomplete = True iterator = iter(lambda: rest.read(65536), b'') + deadline = time.monotonic() + CAPTURE_DEADLINE_SECONDS try: for chunk in iterator: chunks.append(chunk) size += len(chunk) - if size > RESPONSE_BODY_CAP: + if size > RESPONSE_BODY_CAP or time.monotonic() > deadline: break else: incomplete = False diff --git a/src/mnfst/url_filter.py b/src/mnfst/url_filter.py index 07f1d20..7ff6ede 100644 --- a/src/mnfst/url_filter.py +++ b/src/mnfst/url_filter.py @@ -32,7 +32,26 @@ def parse_rule(entry: str) -> Optional[Rule]: # `*` inside a path is reserved for a future segment wildcard, so it is refused today. if not _HOST.match(host) or "*" in path: return None - return Rule(host.strip("[]"), path or None) + canonical = canonical_path(path) if path else "/" + return Rule(host.strip("[]"), None if canonical == "/" else canonical) + + +_ESCAPE = re.compile(r"%[0-9a-fA-F]{2}") + + +def canonical_path(path: str) -> str: + """The path a server is likely to route: every percent-escape decoded (so `/%70rivate` + and `/private%2Fitem` read as `/private…`) and `.`/`..` segments resolved. Rules compare + against it, so an encoded spelling cannot slip past a denylist or into an allowlist.""" + decoded = _ESCAPE.sub(lambda m: chr(int(m.group(0)[1:], 16)), path) + out: List[str] = [] + for segment in decoded.split("/")[1:]: + if segment == "..": + if out: + out.pop() + elif segment != ".": + out.append(segment) + return "/" + "/".join(out) def _entries(value: RuleList) -> List[str]: @@ -65,7 +84,7 @@ def is_excluded(allow: Optional[Rules], deny: Rules, url: str) -> bool: return False if not host: return False - path = parts.path or "/" + path = canonical_path(parts.path or "/") if any(_covers(rule, host, path) for rule in deny): return True return allow is not None and not any(_covers(rule, host, path) for rule in allow) diff --git a/tests/fixtures/url-filter.json b/tests/fixtures/url-filter.json index de89acc..cda1680 100644 --- a/tests/fixtures/url-filter.json +++ b/tests/fixtures/url-filter.json @@ -14,6 +14,8 @@ { "entry": "https://api.stripe.com", "rule": { "host": "api.stripe.com", "path": null } }, { "entry": "localhost:3000/api", "rule": { "host": "localhost", "path": "/api" } }, { "entry": "[::1]:8080/x", "rule": { "host": "::1", "path": "/x" } }, + { "entry": "example.com/%70rivate", "rule": { "host": "example.com", "path": "/private" } }, + { "entry": "example.com/a/../b", "rule": { "host": "example.com", "path": "/b" } }, { "entry": "", "rule": null, "invalid": false }, { "entry": " ", "rule": null, "invalid": false }, { "entry": "stripe.com/v1/customers/*/sources", "rule": null, "invalid": true }, @@ -51,6 +53,14 @@ { "allow": "stripe.com/v1/*", "url": "https://api.stripe.com/v1/charges", "excluded": true }, { "allow": " , ", "url": "https://api.openai.com/", "excluded": false }, { "url": "https://api.openai.com/", "excluded": false }, - { "allow": "stripe.com", "url": "not a url", "excluded": false } + { "allow": "stripe.com", "url": "not a url", "excluded": false }, + { "deny": "example.com/private", "url": "https://example.com/%70rivate", "excluded": true }, + { "deny": "example.com/private", "url": "https://example.com/private%2Fitem", "excluded": true }, + { "deny": "example.com/private", "url": "https://example.com/public/..%2Fprivate/x", "excluded": true }, + { "deny": "example.com/private", "url": "https://example.com/public/%2E%2E/private", "excluded": true }, + { "deny": "example.com/private", "url": "https://example.com/privatefoo%2Fx", "excluded": false }, + { "allow": "example.com/public", "url": "https://example.com/public/..%2Fprivate", "excluded": true }, + { "allow": "example.com/public", "url": "https://example.com/p%75blic/x", "excluded": false }, + { "deny": "example.com/a%20b", "url": "https://example.com/a%20b/c", "excluded": true } ] } diff --git a/tests/test_review_findings.py b/tests/test_review_findings.py new file mode 100644 index 0000000..0cd1773 --- /dev/null +++ b/tests/test_review_findings.py @@ -0,0 +1,106 @@ +"""Regressions from the SDK review: filtering, retry framing, capture deadline, fail-open install.""" +import json +import threading +import time +import warnings +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + +import httpx +import pytest + +from mnfst import outbound, response_capture +from mnfst.config import resolve_config +from mnfst.outbound import _Capture, _apply, _rebuild, install_outbound, uninstall_outbound +from tests.helpers import wait_for +from tests.test_outbound_httpx import build_rig + + +@pytest.fixture +def denied_target(): + provider, stub = build_rig(denylist="127.0.0.1/v1/generate") + yield provider, stub + uninstall_outbound() + stub.stop() + provider.stop() + + +def test_a_patch_never_moves_the_retry_onto_a_denied_route(denied_target): + provider, stub = denied_target + stub.result = {"status": "patched", "issueId": "i1", "healAttemptId": "a1", + "healedRequest": {"url": f"{provider.url}/v1/generate"}} + response = httpx.post(f"{provider.url}/v1/old", json={"model": "m"}) + assert response.status_code == 404 + assert len(stub.heals) == 1 + assert [r[0] for r in provider.received] == ["/v1/old"] + assert wait_for(lambda: len(stub.outcomes) == 1) + + +def test_a_retry_with_a_new_body_drops_chunked_framing(): + capture = _Capture("POST", "https://api.example.com/v1/x", + {"content-type": "application/json", "transfer-encoding": "chunked"}, + b'{"model": "m", "temperature": 2}', 400, b"{}", 5) + retry = _apply(capture, {"body": {"model": "m"}}) + assert retry is not None + assert "transfer-encoding" not in {name.lower() for name in retry.headers} + rebuilt = _rebuild(httpx.Request("POST", capture.url), retry) + assert "transfer-encoding" not in rebuilt.headers + assert rebuilt.headers["content-length"] == str(len(retry.content)) + + +class _Trickle(BaseHTTPRequestHandler): + """A 400 whose body arrives one byte at a time, far slower than any capture should wait.""" + protocol_version = "HTTP/1.1" + + def log_message(self, *args): + pass + + def do_POST(self): + self.rfile.read(int(self.headers.get("content-length", 0))) + self.send_response(400) + self.send_header("content-type", "application/json") + self.send_header("transfer-encoding", "chunked") + self.end_headers() + try: + for _ in range(40): + self.wfile.write(b"1\r\n \r\n") + self.wfile.flush() + time.sleep(0.1) + self.wfile.write(b"0\r\n\r\n") + except OSError: + pass + + +def test_capturing_a_trickling_failure_stops_at_its_deadline(monkeypatch): + monkeypatch.setattr(response_capture, "CAPTURE_DEADLINE_SECONDS", 0.3) + server = ThreadingHTTPServer(("127.0.0.1", 0), _Trickle) + threading.Thread(target=server.serve_forever, daemon=True).start() + provider, stub = build_rig() + stub.result = {"status": "no_patch"} + try: + started = time.monotonic() + with httpx.Client() as client: + with client.stream("POST", f"http://127.0.0.1:{server.server_address[1]}/x", + json={"temperature": 2}) as response: + handed_back = time.monotonic() - started + assert response.status_code == 400 + assert handed_back < 2.0 + assert len(stub.heals) == 1 + assert stub.heals[0]["response"]["truncated"] is True + finally: + uninstall_outbound() + stub.stop() + provider.stop() + server.shutdown() + server.server_close() + + +def test_an_install_failure_leaves_the_app_running_uninstrumented(monkeypatch): + original = httpx.HTTPTransport.handle_request + monkeypatch.setenv("HTTPS_PROXY", "bad://proxy") + config = resolve_config(api_key="mnfx_k", url="https://manifest.example") + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + install_outbound(config) + assert any("install failed" in str(w.message) for w in caught) + assert outbound.installed_config() is None + assert httpx.HTTPTransport.handle_request is original