"""Approval credentials must never select the OpenBao or operator identity.""" import base64 import copy import json import threading from dataclasses import replace from datetime import datetime, timezone from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from io import BytesIO from pathlib import Path from types import SimpleNamespace from urllib.error import HTTPError from urllib.parse import parse_qs from urllib.request import Request import pytest from secrets_engine import approval_auth, cli from secrets_engine.approval_auth import ( KeyCapeApprovalAuthConfig, approval_token, credential_urlopen, require_approval_address, ) from secrets_engine.catalog import validate_entry from secrets_engine.config import Config from secrets_engine.errors import BackendError, DecisionError, ProvisioningError from secrets_engine.service_auth import KeyCapeServiceAuthConfig, preflight_service_jwt from tests.authorization_stub import AuthorizationStub from tests.test_catalog import VALID from tests.test_service_auth import _jwt ISSUER = "https://keycape.example.test" def _cfg(tmp_path, **changes): secret = tmp_path / "approval-client.secret" secret.write_text("synthetic-client-secret") secret.chmod(0o600) return SimpleNamespace( **({"approval_client_secret_file": secret, "approval_token_file": None, "keycape_issuer": ISSUER, "keycape_token_url": ISSUER + "/token", "approval_url": "http://127.0.0.1:18081"} | changes) ) def _token(_requested_scope="approval:read", **overrides): now = int(datetime.now(timezone.utc).timestamp()) return _jwt(**({"iat": now, "exp": now + 900, "aud": "approval-engine", "tenant": "tenant:platform", "scope": _requested_scope} | overrides)) def _transport(seen, **overrides): def send(request, *, timeout): assert request.full_url == ISSUER + "/token" auth = base64.b64decode(request.get_header("Authorization").split()[1]).decode() assert auth == "secrets-engine-approval:synthetic-client-secret" body = parse_qs(request.data.decode()) assert body["grant_type"] == ["client_credentials"] scope = body["scope"][0] seen.append(scope) payload = {"access_token": _token(scope, **overrides), "token_type": "Bearer", "expires_in": 900} response = BytesIO(json.dumps(payload).encode()) response.status = 200 return response return send def test_read_and_consume_each_exchange_their_own_minimal_scope(tmp_path, monkeypatch): cfg, seen = _cfg(tmp_path), [] monkeypatch.setattr(approval_auth, "credential_urlopen", _transport(seen)) for scope in ("approval:read", "approval:consume", "approval:read"): result = approval_token(cfg, scope=scope) claims = json.loads(base64.urlsafe_b64decode(result.split(".")[1] + "==")) assert claims["scope"] == scope assert seen == ["approval:read", "approval:consume", "approval:read"] assert sorted(p.name for p in tmp_path.iterdir()) == ["approval-client.secret"] @pytest.mark.parametrize("claims", [ {"aud": "secrets-engine-openbao"}, {"tenant": "tenant:coulomb"}, {"sub": "service:approval-engine-operator"}, {"roles": ["secrets-engine", "admin"]}, {"scope": "approval:read approval:consume"}, {"principal_type": "human"}, {"exp": 1}, {"iss": "https://another-issuer.test"}, {"assurance": {"aal": "AAL1", "method": "client_secret", "mfa": False, "source": "key-cape"}}, {"assurance": {"level": "aal1", "methods": ["client_secret", "password"], "mfa": False, "source": "key-cape"}}, ]) def test_wrong_returned_identity_is_refused_without_fallback(tmp_path, monkeypatch, claims): cfg = _cfg(tmp_path) # An OpenBao identity credential exists, but must never be read here. cfg.keycape_client_secret_file = Path("/must/not/read/openbao-client") monkeypatch.setenv("BAO_TOKEN", "synthetic-forbidden-fallback") monkeypatch.setattr(approval_auth, "credential_urlopen", _transport([], **claims)) with pytest.raises(BackendError): approval_token(cfg, scope="approval:read") def test_exchange_failure_is_bounded_and_never_falls_back(tmp_path, monkeypatch): cfg = _cfg(tmp_path) seen = [] def refuse(request, *, timeout): seen.append(request.full_url) raise HTTPError(request.full_url, 401, "private error", {}, BytesIO(b"secret-like error body")) monkeypatch.setattr(approval_auth, "credential_urlopen", refuse) with pytest.raises(BackendError, match="HTTP 401") as error: approval_token(cfg, scope="approval:consume") assert "secret-like" not in str(error.value) assert len(seen) == 1 def test_mixed_providers_refused_before_either_file_read(tmp_path): cfg = _cfg(tmp_path, approval_token_file=Path("/not/read/token")) with pytest.raises(DecisionError, match="conflict"): approval_token(cfg, scope="approval:read") def test_secret_file_protection_is_required(tmp_path): cfg = _cfg(tmp_path) cfg.approval_client_secret_file.chmod(0o644) with pytest.raises(ProvisioningError, match="accessible"): approval_token(cfg, scope="approval:read") @pytest.mark.parametrize("url", [ "http://keycape.example.test/token", "https://other.test/token", ISSUER + "/token?next=elsewhere", ISSUER + "/token#fragment", ]) def test_exchange_url_must_be_the_issuer_endpoint(url): with pytest.raises(BackendError, match="HTTPS /token"): KeyCapeApprovalAuthConfig(token_url=url, issuer=ISSUER, client_secret_file=Path("/unused")) @pytest.mark.parametrize("url", [ "http://approval.example", "http://localhost:8000", "https://approval-engine.svc.cluster.local", "https://user:password@approval.test", "http://127.0.0.1:8000?next=other", "https://approval.test/#fragment", ]) def test_unsafe_approval_destinations_refused_before_exchange(url): with pytest.raises(DecisionError): require_approval_address(url) def test_approval_profile_cannot_relax_the_openbao_contract(): with pytest.raises(BackendError): KeyCapeServiceAuthConfig(token_url=ISSUER + "/token", issuer=ISSUER, client_secret_file=Path("/unused"), audience="approval-engine") cfg = KeyCapeApprovalAuthConfig(token_url=ISSUER + "/token", issuer=ISSUER, client_secret_file=Path("/unused")) for changes in ({"client_id": "approval-engine-operator"}, {"scope": "approval:create"}, {"tenant": "tenant:coulomb"}): with pytest.raises(BackendError): replace(cfg, **changes) now = datetime.now(timezone.utc) stamp = int(now.timestamp()) preflight_service_jwt(_token(iat=stamp + 30, exp=stamp + 930), cfg, now=now) for changes in ({"iat": stamp + 31, "exp": stamp + 931}, {"exp": stamp}, {"exp": stamp + 901}): with pytest.raises(BackendError): preflight_service_jwt(_token(**changes), cfg, now=now) @pytest.mark.parametrize("method", ["GET", "POST"]) def test_redirect_cannot_forward_credentials(method): seen = [] class Redirect(BaseHTTPRequestHandler): def log_message(self, *_args): pass def do_GET(self): seen.append(self.path) if self.path == "/start": self.send_response(302) self.send_header("Location", "/sink") else: self.send_response(200) self.end_headers() do_POST = do_GET server = ThreadingHTTPServer(("127.0.0.1", 0), Redirect) thread = threading.Thread(target=server.serve_forever) thread.start() try: req = Request(f"http://127.0.0.1:{server.server_port}/start", method=method, headers={"Authorization": "Bearer synthetic-token"}) with pytest.raises(HTTPError) as error: credential_urlopen(req, timeout=2) assert error.value.code == 302 assert seen == ["/start"] finally: server.shutdown() server.server_close() thread.join() def test_environment_selects_separate_approval_secret(monkeypatch): monkeypatch.setenv("SECRETS_ENGINE_APPROVAL_CLIENT_SECRET_FILE", "/protected/approval.secret") monkeypatch.setenv("SECRETS_ENGINE_KEYCAPE_CLIENT_SECRET_FILE", "/protected/openbao.secret") cfg = Config.load() assert cfg.approval_client_secret_file == Path("/protected/approval.secret") assert cfg.keycape_client_secret_file == Path("/protected/openbao.secret") @pytest.mark.parametrize("failure", [None, "exchange", "conflict", "wrong-action"]) def test_cli_gate_exchanges_for_claim_and_consume(tmp_path, monkeypatch, failure): raw = copy.deepcopy(VALID) raw["stage"] = "prod" raw["approval"] = {"model": "bootstrap-only", "authorization_id": "approval-test", "purpose": "synthetic proof"} entry = validate_entry(raw) cfg = _cfg(tmp_path) cfg.authorization_subject_id = "secrets-engine" cfg.authorization_subject_type = "service" cfg.authorization_policy_package = "secrets-engine.catalog-lane.lifecycle" cfg.authorization_policy_version = "v2" cfg.evidence_dir = tmp_path / "evidence" cfg.pdp_token_file = tmp_path / "pdp.token" cfg.pdp_token_file.write_text("synthetic-pdp-token") cfg.pdp_token_file.chmod(0o600) monkeypatch.delenv("SECRETS_ENGINE_UNSAFE_DEMO", raising=False) stub = AuthorizationStub(approval_id="approval-test", package=cfg.authorization_policy_package, version="v2").start() seen = [] transport = _transport(seen) def exchange(request, *, timeout): if failure == "exchange": raise HTTPError(request.full_url, 401, "denied", {}, BytesIO(b"")) return transport(request, timeout=timeout) monkeypatch.setattr(approval_auth, "credential_urlopen", exchange) cfg.approval_url = cfg.pdp_url = stub.url from secrets_engine.approval_consume import _expected_request stub.bind_request(_expected_request(cfg, entry, "apply")) if failure == "conflict": stub.consume_status = 409 try: if failure: with pytest.raises((DecisionError, BackendError)): cli._require_lane_approval(cfg, entry, "destroy" if failure == "wrong-action" else "apply") assert not stub.consumed else: cli._require_lane_approval(cfg, entry, "apply") assert stub.calls == ["claim", "check", "consume"] assert seen == ["approval:read", "approval:consume"] assert stub.consumed finally: stub.stop()