Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.

Expand Down
2 changes: 1 addition & 1 deletion docs/guide.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
25 changes: 23 additions & 2 deletions src/mnfst/outbound.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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
Expand Down
20 changes: 15 additions & 5 deletions src/mnfst/response_capture.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
from __future__ import annotations

import io
import time
import warnings
import zlib
from itertools import chain
Expand All @@ -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
Expand Down Expand Up @@ -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
Expand All @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down
23 changes: 21 additions & 2 deletions src/mnfst/url_filter.py
Original file line number Diff line number Diff line change
Expand Up @@ -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]:
Expand Down Expand Up @@ -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)
12 changes: 11 additions & 1 deletion tests/fixtures/url-filter.json
Original file line number Diff line number Diff line change
Expand Up @@ -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 },
Expand Down Expand Up @@ -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 }
]
}
106 changes: 106 additions & 0 deletions tests/test_review_findings.py
Original file line number Diff line number Diff line change
@@ -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
Loading