Bound approval transport and refuse redirects and proxies

Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a126e1-76f8-7643-8cf3-e93336539ade

tests/test_confirmation_transport.py

@@ -0,0 +1,199 @@
+from contextlib import contextmanager
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+import os
+from pathlib import Path
+import threading
+import time
+
+import pytest
+
+from approval_engine.execution_confirmation import ExecutionConfirmationClient
+from approval_engine.pep import ApprovalHTTPClient, ApprovalProtocolError
+
+
+@contextmanager
+def server(handle):
+    class Handler(BaseHTTPRequestHandler):
+        def do_GET(self):
+            handle(self)
+        def do_POST(self):
+            handle(self)
+        def log_message(self, *args):
+            pass
+    http = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
+    worker = threading.Thread(target=http.serve_forever, daemon=True)
+    worker.start()
+    try:
+        yield f"http://127.0.0.1:{http.server_port}"
+    finally:
+        http.shutdown()
+        http.server_close()
+        worker.join()
+
+
+def credential(tmp_path):
+    path = tmp_path / "token"
+    path.write_text("synthetic-test-only")
+    path.chmod(0o600)
+    return path
+
+
+def reply(req, body=b'{"ok":true}', content="application/json", status=200):
+    req.send_response(status)
+    req.send_header("Content-Type", content)
+    req.send_header("Content-Length", str(len(body)))
+    req.end_headers()
+    try:
+        req.wfile.write(body)
+    except (BrokenPipeError, ConnectionResetError):
+        pass
+
+
+@pytest.mark.parametrize("status", [301, 302, 303, 307, 308])
+@pytest.mark.parametrize("method", ["GET", "POST"])
+def test_default_transport_never_forwards_redirect_credentials(tmp_path, status, method):
+    received = []
+    def second(req):
+        received.append(req.headers.get("Authorization"))
+        reply(req)
+    with server(second) as foreign:
+        def first(req):
+            req.send_response(status)
+            req.send_header("Location", foreign + "/stolen")
+            req.end_headers()
+        with server(first) as origin:
+            c = ExecutionConfirmationClient(origin, credential(tmp_path))
+            with pytest.raises(ApprovalProtocolError):
+                c._request(method, "/v1/approvals/synthetic", {} if method == "POST" else None)
+    assert received == []
+
+
+def test_no_environment_proxy_receives_bearer(tmp_path, monkeypatch):
+    received = []
+    with server(lambda req: (received.append(req.headers.get("Authorization")), reply(req))) as proxy:
+        for name in ("HTTP_PROXY", "HTTPS_PROXY", "ALL_PROXY", "http_proxy", "https_proxy", "all_proxy"):
+            monkeypatch.setenv(name, proxy)
+        monkeypatch.setenv("NO_PROXY", "")
+        monkeypatch.setenv("no_proxy", "")
+        with server(reply) as origin:
+            assert ApprovalHTTPClient(origin, credential(tmp_path)).claim("synthetic") == {"ok": True}
+    assert received == []
+
+
+@pytest.mark.parametrize("body,content", [
+    (b'{"ok":true,"ok":false}', "application/json"),
+    (b'{"nested":{"ok":true,"ok":false}}', "application/json"),
+    (b'{"value":NaN}', "application/json"), (b'{"value":Infinity}', "application/json"),
+    (b'{"value":1e9999}', "application/json"),
+    (b'{"ok":true}', "text/html"), (b'{"ok":true}', "application/json; charset=latin1"),
+    (b'\xff', "application/json"), (b'{', "application/json"),
+    (b'[]', "application/json"), (b' ' * (256 * 1024 + 1), "application/json"),
+])
+def test_oversized_or_malformed_content_refuses(tmp_path, body, content):
+    with server(lambda req: reply(req, body, content)) as origin:
+        with pytest.raises(ApprovalProtocolError) as error:
+            ApprovalHTTPClient(origin, credential(tmp_path)).claim("synthetic")
+        assert "synthetic-test-only" not in str(error.value)
+        assert repr(body) not in str(error.value)
+
+
+@pytest.mark.parametrize("status", [201, 204, 401, 403, 404, 409, 500, 503])
+def test_non_200_has_safe_closed_error_without_error_body(tmp_path, status):
+    with server(lambda req: reply(req, b'private error details', status=status)) as origin:
+        with pytest.raises(ApprovalProtocolError) as error:
+            ApprovalHTTPClient(origin, credential(tmp_path)).claim("synthetic")
+        assert "private" not in str(error.value)
+
+
+@pytest.mark.parametrize("phase", ["headers", "body", "errorbody"])
+def test_entire_transport_deadline_in_worker_thread(tmp_path, phase):
+    arrived = threading.Event()
+    def slow(req):
+        arrived.set()
+        if phase == "headers":
+            time.sleep(1.5)
+        req.send_response(500 if phase == "errorbody" else 200)
+        req.send_header("Content-Type", "application/json")
+        req.send_header("Content-Length", "100")
+        req.end_headers()
+        if phase != "errorbody":
+            for _ in range(30):
+                try:
+                    req.wfile.write(b" ")
+                    req.wfile.flush()
+                except (BrokenPipeError, ConnectionResetError):
+                    break
+                time.sleep(0.1)
+    with server(slow) as origin:
+        elapsed = []
+        errors = []
+        def caller():
+            start = time.monotonic()
+            try:
+                ApprovalHTTPClient(origin, credential(tmp_path), timeout_seconds=0.4).claim("synthetic")
+            except ApprovalProtocolError as exc:
+                errors.append(str(exc))
+            elapsed.append(time.monotonic() - start)
+        worker = threading.Thread(target=caller)
+        worker.start()
+        worker.join(timeout=1.2)
+        assert not worker.is_alive()
+        assert arrived.is_set()
+        assert len(errors) == 1
+        assert elapsed[0] < 1.2
+
+
+@pytest.mark.parametrize("kind", ["symlink", "fifo", "directory", "public", "group_write", "oversized", "malformed", "hardlink", "missing"])
+def test_credential_file_guards_refuse_before_network(tmp_path, kind):
+    path = credential(tmp_path)
+    if kind in {"symlink", "fifo", "directory", "missing"}:
+        path.unlink()
+    if kind == "symlink":
+        other = tmp_path / "other"
+        other.write_text("synthetic-test-only")
+        path.symlink_to(other)
+    elif kind == "fifo":
+        os.mkfifo(path, 0o600)
+    elif kind == "directory":
+        path.mkdir()
+    elif kind == "public":
+        path.chmod(0o644)
+    elif kind == "group_write":
+        path.chmod(0o660)
+    elif kind == "oversized":
+        path.write_text("x" * 16385)
+    elif kind == "malformed":
+        path.write_bytes(b'\xff')
+    elif kind == "hardlink":
+        os.link(path, tmp_path / "linked")
+    with pytest.raises(ApprovalProtocolError, match="credential is unavailable"):
+        ApprovalHTTPClient("http://127.0.0.1:1", path).claim("synthetic")
+
+
+@pytest.mark.parametrize("url", ["http://user:password@host", "http://host?target=x", "http://host#target", " http://host", "http://host:0", "http://host:wrong", "ftp://host"])
+def test_invalid_origin_refuses(tmp_path, url):
+    with pytest.raises(ApprovalProtocolError):
+        ApprovalHTTPClient(url, credential(tmp_path))
+
+
+def test_safe_injected_opener_close_failure_is_closed(tmp_path):
+    class Response:
+        def getcode(self):
+            return 200
+        def read(self, size):
+            return b'{"ok":true}'
+        def close(self):
+            raise RuntimeError("synthetic credential-bearing exception")
+    with pytest.raises(ApprovalProtocolError) as error:
+        ApprovalHTTPClient("http://fixed-origin", credential(tmp_path), opener=lambda *_a, **_k: Response()).claim("synthetic")
+    assert "credential-bearing" not in str(error.value)
+
+
+def test_symlinked_credential_parent_refuses(tmp_path):
+    real = tmp_path / "real"
+    real.mkdir()
+    token = credential(real)
+    link = tmp_path / "link"
+    link.symlink_to(real, target_is_directory=True)
+    with pytest.raises(ApprovalProtocolError, match="credential is unavailable"):
+        ApprovalHTTPClient("http://127.0.0.1:1", link / token.name).claim("synthetic")

tests/test_execution_confirmation.py

@@ -19,6 +19,7 @@ FREE = "sha256:" + "cd" * 32
 def client(tmp_path, opener):
     token = tmp_path / "test-token"
     token.write_text("test-token")
+    token.chmod(0o600)
     return ExecutionConfirmationClient("http://approval-engine", token, opener=opener)
 
 

tests/test_pep.py

@@ -58,6 +58,7 @@ class Response:
 def _token(tmp_path, value="test-token"):
     path = tmp_path / "token"
     path.write_text(value)
+    path.chmod(0o600)
     return path