approval-engine/tests/test_confirmation_transport.py

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")