Prepare reviewed service JWT boundary verification and retain execution adoption gates
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a126e1-76f8-7643-8cf3-e93336539ade
This commit is contained in:
parent
443c2698c3
commit
4378ec1fc2
6 changed files with 551 additions and 7 deletions
212
tests/test_service_jwt_boundary.py
Normal file
212
tests/test_service_jwt_boundary.py
Normal file
|
|
@ -0,0 +1,212 @@
|
|||
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)
|
||||
Loading…
Add table
Add a link
Reference in a new issue