secrets-engine/tests/test_service_jwt_boundary.py

import importlib.util
from pathlib import Path
from types import SimpleNamespace

import pytest

from secrets_engine.errors import BackendError

spec = importlib.util.spec_from_file_location("boundary", Path(__file__).parents[1] / "tools/verify_service_jwt_boundary.py")
boundary = importlib.util.module_from_spec(spec)
spec.loader.exec_module(boundary)


class Session:
    client = SimpleNamespace()
    closed = False

    def close(self):
        self.closed = True

    def evidence(self):
        return {"secret_source": "openbao-oidc-in-memory", "established": True,
                "operator_session_handle": "0123456789ab", "session_handle": "abcdef012345",
                "operator_revocation_attempted": True, "operator_revocation_succeeded": True,
                "revocation_attempted": self.closed, "revocation_succeeded": self.closed}


@pytest.mark.parametrize("extra_identity", [False, True])
def test_batch_denial_checks_identity_policy_and_always_cleans(extra_identity):
    session = Session()
    paths = list(boundary.EXPECTED_GUARDS + boundary.ADMIN_DENIES)
    calls = []

    def request(client, path, payload=None):
        calls.append((path, payload))
        if path == "auth/token/lookup-self":
            return {"policies": ["secrets-engine-login-self"], "identity_policies": ["admin"] if extra_identity else []}
        return {p: ["deny"] for p in paths}

    result = boundary.verify_session(session, {"expected_policy": "secrets-engine-login-self", "denied_paths": paths}, request)
    assert result["native_verified"] is (not extra_identity)
    assert session.closed
    assert len(calls) == 2
    assert calls[1][1] == {"paths": paths}
    assert not result["protected_operation_adoption_proved"]


@pytest.mark.parametrize("failure", ["request", "missing-deny", "revoke"])
def test_failed_or_partial_acceptance_never_passes(failure):
    session = Session()
    paths = list(boundary.EXPECTED_GUARDS + boundary.ADMIN_DENIES)

    def request(client, path, payload=None):
        if failure == "request":
            raise BackendError("sensitive backend error")
        if path == "auth/token/lookup-self":
            return {"policies": ["secrets-engine-login-self"]}
        return {p: ["deny"] for p in paths[:-1] if failure == "missing-deny"} if failure == "missing-deny" else {p: ["deny"] for p in paths}

    if failure == "revoke":
        def close():
            raise BackendError("sensitive backend error")
        session.close = close
    result = boundary.verify_session(session, {"expected_policy": "secrets-engine-login-self", "denied_paths": paths}, request)
    assert not result["native_verified"]
    assert "sensitive backend error" not in str(result)
    if failure != "revoke":
        assert session.closed


@pytest.mark.parametrize("fault", ["oversize-header", "oversize-body", "malformed", "wrong-shape", "wrong-status", "wrong-content", "no-response", "unknown-error"])
def test_transport_failures_suppress_body_and_cleanup(monkeypatch, fault):
    from email.message import Message
    headers = Message()
    headers["Content-Type"] = "text/plain" if fault == "wrong-content" else "application/json"
    if fault == "oversize-header":
        headers["Content-Length"] = str(boundary.MAX_RESPONSE_BYTES + 1)
    response = SimpleNamespace(status=503 if fault == "wrong-status" else 200, headers=headers)
    raw = b'{"data": {}}'
    if fault == "oversize-body":
        raw = b'x' * (boundary.MAX_RESPONSE_BYTES + 1)
    if fault == "malformed":
        raw = b'private-body'
    if fault == "wrong-shape":
        raw = b'{"data": []}'

    class Response:
        status = response.status
        headers = response.headers
        def __enter__(self): return self
        def __exit__(self, *args): pass
        def read(self, count):
            assert count == boundary.MAX_RESPONSE_BYTES + 1
            return raw

    def opening(req, timeout):
        if fault in ("no-response", "unknown-error"):
            raise RuntimeError("private-body")
        return Response()
    monkeypatch.setattr(boundary, "build_opener", lambda *args: SimpleNamespace(open=opening))
    session = Session()
    session.client = SimpleNamespace(addr="http://127.0.0.1:18200", token="sentinel")
    plan = {"expected_policy": "secrets-engine-login-self", "denied_paths": list(boundary.EXPECTED_GUARDS + boundary.ADMIN_DENIES)}
    result = boundary.verify_session(session, plan, lambda c,p,d=None: boundary.request_json(c,p,d,expected_addr=c.addr))
    assert not result["native_verified"]
    assert session.closed
    assert "private-body" not in str(result)


@pytest.mark.parametrize("location", ["http://127.0.0.1:9/private", "/relative"])
def test_real_redirect_refused_and_proxy_environment_ignored(monkeypatch, location):
    import threading
    from http.server import BaseHTTPRequestHandler, HTTPServer
    observed = []

    class Handler(BaseHTTPRequestHandler):
        def do_GET(self):
            observed.append(self.path)
            self.send_response(302)
            self.send_header("Location", location)
            self.end_headers()
        def log_message(self, *args): pass

    server = HTTPServer(("127.0.0.1", 0), Handler)
    worker = threading.Thread(target=server.serve_forever, daemon=True)
    worker.start()
    # Invalid proxy would prevent the first request if environment proxies were used.
    monkeypatch.setenv("http_proxy", "http://127.0.0.1:1")
    monkeypatch.setenv("HTTP_PROXY", "http://127.0.0.1:1")
    monkeypatch.setenv("no_proxy", "")
    monkeypatch.setenv("NO_PROXY", "")
    addr = f"http://127.0.0.1:{server.server_port}"
    session = Session()
    session.client = SimpleNamespace(addr=addr, token="synthetic-token-only")
    try:
        result = boundary.verify_session(session, {"expected_policy":"secrets-engine-login-self", "denied_paths": list(boundary.EXPECTED_GUARDS + boundary.ADMIN_DENIES)},
            lambda c,p,d=None: boundary.request_json(c,p,d,expected_addr=addr))
    finally:
        server.shutdown()
        server.server_close()
        worker.join()
    assert observed == ["/v1/auth/token/lookup-self"]
    assert session.closed
    assert not result["native_verified"]


def test_origin_mismatch_refused_before_request(monkeypatch):
    monkeypatch.setattr(boundary, "build_opener", lambda *args: pytest.fail("must not send token"))
    client = SimpleNamespace(addr="http://evil.invalid", token="sentinel")
    with pytest.raises(BackendError, match="response suppressed"):
        boundary.request_json(client, "auth/token/lookup-self", expected_addr="http://127.0.0.1:18200")


def test_unknown_evidence_and_private_failure_not_published():
    session = Session()
    session.evidence = lambda: {"private_token": "private-body", "session_handle": "private-body", "secret_source": "private-body", "operator_revocation_succeeded": True, "revocation_succeeded": True}
    def fail(*args): raise RuntimeError("private-body")
    result = boundary.verify_session(session, {"expected_policy":"secrets-engine-login-self", "denied_paths": list(boundary.EXPECTED_GUARDS)}, fail)
    assert session.closed
    assert not result["native_verified"]
    assert "private-body" not in str(result)
    assert "private_token" not in result


@pytest.mark.parametrize("missing", list(boundary.SAFE_EVIDENCE))
def test_every_contract_evidence_field_required(missing):
    session = Session()
    original = session.evidence
    def evidence():
        result = original()
        del result[missing]
        return result
    session.evidence = evidence
    paths = list(boundary.EXPECTED_GUARDS + boundary.ADMIN_DENIES)
    def request(client, path, payload=None):
        return {"policies": ["secrets-engine-login-self"]} if path == "auth/token/lookup-self" else {p:["deny"] for p in paths}
    result = boundary.verify_session(session, {"expected_policy":"secrets-engine-login-self", "denied_paths": paths}, request)
    assert session.closed
    assert not result["native_verified"]


def test_revoke_exception_cannot_be_overridden_by_stale_evidence():
    session = Session()
    session.closed = True
    def close(): raise RuntimeError("private failure")
    session.close = close
    paths = list(boundary.EXPECTED_GUARDS + boundary.ADMIN_DENIES)
    def request(client, path, payload=None):
        return {"policies": ["secrets-engine-login-self"]} if path == "auth/token/lookup-self" else {p:["deny"] for p in paths}
    result = boundary.verify_session(session, {"expected_policy":"secrets-engine-login-self", "denied_paths":paths}, request)
    assert result["revocation_succeeded"] is True
    assert not result["native_verified"]
    assert "private failure" not in str(result)


@pytest.mark.parametrize("errors", [[], ["private backend failure"], None])
def test_errors_and_data_response_never_accepted(monkeypatch, errors):
    import json
    from email.message import Message
    headers = Message()
    headers["Content-Type"] = "application/json"
    class Response:
        status = 200
        def __enter__(self): return self
        def __exit__(self, *args): pass
        def read(self, count): return json.dumps({"data": {}, "errors": errors}).encode()
    response = Response()
    response.headers = headers
    monkeypatch.setattr(boundary, "build_opener", lambda *args: SimpleNamespace(open=lambda *a,**k: response))
    client = SimpleNamespace(addr="http://127.0.0.1:18200", token="synthetic-only")
    with pytest.raises(BackendError, match="response suppressed"):
        boundary.request_json(client, "auth/token/lookup-self", expected_addr=client.addr)