from __future__ import annotations import importlib.util import os import stat import subprocess import sys import tempfile import unittest from argparse import Namespace from pathlib import Path from unittest import mock REPO_DIR = Path(__file__).resolve().parents[1] SPEC = importlib.util.spec_from_file_location( "credential_helper", REPO_DIR / "scripts/credential.py" ) credential = importlib.util.module_from_spec(SPEC) assert SPEC.loader is not None sys.modules[SPEC.name] = credential SPEC.loader.exec_module(credential) def sample_grant() -> dict: return { "id": "ops-warden/warden-sign", "issuer": "openbao", "audience": "ops-warden", "credential_type": "openbao-token", "openbao": { "token_role": "warden-sign", "policies": ["warden-sign"], }, "ttl": {"default": "15m", "max": "1h"}, "actors": {"allowed_types": ["human-operator", "approved-agent"]}, "delivery": { "allowed": [ "exec-env", "response-wrap", "local-token-file", "kubernetes-auth", ], "kubernetes_auth": { "mount": "auth/kubernetes", "role": "credential-broker-warden-sign", "service_account_names": ["credential-broker"], "namespaces": ["openbao"], }, }, } class CredentialHelperTests(unittest.TestCase): def test_flex_auth_request_uses_canonical_contract_and_numeric_ttl(self) -> None: payload = credential.flex_auth_check_request( grant=sample_grant(), ttl="15m", purpose="flex-auth-openbao-smoke", delivery="exec-env", actor="codex:operator", actor_type="approved-agent", subject="agent:codex/railiance-platform", ) self.assertEqual(payload["tenant"], "tenant:platform") self.assertEqual( payload["subject"], {"id": "agent:codex/railiance-platform", "type": "Agent"}, ) self.assertEqual(payload["action"], "issue") self.assertEqual( payload["resource"], { "id": "credential-grant:ops-warden/warden-sign", "type": "credential-grant", "system": "railiance-platform", }, ) self.assertEqual(payload["context"]["requested_ttl_seconds"], 900) self.assertEqual( payload["context"]["bound_subject"], "agent:codex/railiance-platform", ) def test_authorize_request_reads_decision_envelope(self) -> None: args = Namespace( decision_id=None, dry_run=False, flex_auth_url="http://127.0.0.1:19099", flex_auth_path="/v1/check", require_flex_auth=True, actor="codex:operator", actor_type="approved-agent", subject="agent:codex/railiance-platform", http_timeout=1.0, ) response = { "id": "decision:credential-grant-allow", "effect": "allow", "reason": "credential_grant_allowed", "subject": { "id": "agent:codex/railiance-platform", "type": "Agent", }, "resource": { "id": "credential-grant:ops-warden/warden-sign", "type": "credential-grant", "system": "railiance-platform", }, "provenance": {"evaluator": "standalone", "mode": "embedded"}, } with mock.patch.object(credential, "post_json", return_value=response) as post: result = credential.authorize_request( args=args, grant=sample_grant(), ttl="15m", purpose="flex-auth-openbao-smoke", delivery="exec-env", ) self.assertEqual(result.mode, "flex-auth") self.assertEqual(result.decision_id, "decision:credential-grant-allow") endpoint, payload = post.call_args.args self.assertEqual(endpoint, "http://127.0.0.1:19099/v1/check") self.assertEqual(payload["context"]["requested_ttl_seconds"], 900) def test_authorize_request_fails_closed_on_non_allow_effect(self) -> None: args = Namespace( decision_id=None, dry_run=False, flex_auth_url="http://127.0.0.1:19099", flex_auth_path="/v1/check", require_flex_auth=True, actor="codex:operator", actor_type="approved-agent", subject="agent:codex/railiance-platform", http_timeout=1.0, ) with mock.patch.object( credential, "post_json", return_value={ "id": "decision:deny", "effect": "deny", "reason": "ttl_out_of_bounds", }, ): with self.assertRaisesRegex(SystemExit, "ttl_out_of_bounds"): credential.authorize_request( args=args, grant=sample_grant(), ttl="15m", purpose="flex-auth-openbao-smoke", delivery="exec-env", ) def test_authorize_request_rejects_unbound_allow_response(self) -> None: args = Namespace( decision_id=None, dry_run=False, flex_auth_url="http://127.0.0.1:19099", flex_auth_path="/v1/check", require_flex_auth=True, actor="codex:operator", actor_type="approved-agent", subject="agent:codex/railiance-platform", http_timeout=1.0, ) with mock.patch.object( credential, "post_json", return_value={"id": "decision:unbound", "effect": "allow"}, ): with self.assertRaisesRegex(SystemExit, "subject does not match"): credential.authorize_request( args=args, grant=sample_grant(), ttl="15m", purpose="flex-auth-openbao-smoke", delivery="exec-env", ) def test_flex_auth_default_path_is_canonical_check(self) -> None: args = credential.build_parser().parse_args( ["request", "--purpose", "flex-auth-openbao-smoke"] ) self.assertEqual(args.flex_auth_path, "/v1/check") def test_database_credential_response_is_parsed_without_emission(self) -> None: lease_id, username, password = credential.parse_database_credential( '{"lease_id":"database/creds/runtime/lease-1","data":{"username":"leased-user","password":"leased-secret"}}' ) self.assertEqual(lease_id, "database/creds/runtime/lease-1") self.assertEqual(username, "leased-user") self.assertEqual(password, "leased-secret") def test_ttl_over_max_is_rejected(self) -> None: with self.assertRaises(SystemExit): credential.validate_issue_request( sample_grant(), "2h", "purpose", "exec-env", "approved-agent" ) def test_actor_type_is_checked(self) -> None: with self.assertRaises(SystemExit): credential.validate_issue_request( sample_grant(), "15m", "purpose", "exec-env", "unknown-actor" ) def test_split_env_prefix_rejects_token_injection(self) -> None: with self.assertRaises(SystemExit): credential.split_env_prefix(["--", "VAULT_TOKEN=hvs.bad", "/bin/true"]) def test_split_env_prefix_accepts_safe_assignments(self) -> None: extra_env, command = credential.split_env_prefix( ["--", "SMOKE_VAULT=1", "/bin/true"] ) self.assertEqual(extra_env, {"SMOKE_VAULT": "1"}) self.assertEqual(command, ["/bin/true"]) def test_redaction_catches_bao_tokens_and_env_assignments(self) -> None: text = "token=hvb.abc123 VAULT_TOKEN=hvs.secret BAO_TOKEN=hvb.secret" redacted = credential.redact(text) self.assertNotIn("hvb.abc123", redacted) self.assertNotIn("hvs.secret", redacted) self.assertIn("[REDACTED]", redacted) def test_local_lease_is_mode_0600_and_cleanup_stays_in_lease_dir(self) -> None: with tempfile.TemporaryDirectory() as tmp: lease_dir = Path(tmp) / "leases" authz = credential.AuthorizationResult(True, "unit-test", "decision-1") payload = credential.write_local_lease( lease_dir=lease_dir, grant=sample_grant(), purpose="unit-test", ttl="15m", token="hvb.unit-test-secret", accessor="accessor-unit-test", authz=authz, ) token_file = Path(payload["token_file"]) metadata_file = Path(payload["metadata_file"]) self.assertEqual(stat.S_IMODE(token_file.stat().st_mode), 0o600) self.assertEqual(stat.S_IMODE(metadata_file.stat().st_mode), 0o600) removed = credential.remove_local_lease_files( lease_dir, "accessor-unit-test" ) self.assertIn(str(token_file), removed) self.assertIn(str(metadata_file), removed) self.assertFalse(token_file.exists()) self.assertFalse(metadata_file.exists()) def test_kubernetes_auth_payload_issues_no_token(self) -> None: authz = credential.AuthorizationResult(True, "dry-run-local", None) payload = credential.kubernetes_auth_payload( sample_grant(), "15m", "unit-test", authz ) self.assertEqual(payload["delivery_mode"], "kubernetes-auth") self.assertEqual(payload["openbao_auth_role"], "credential-broker-warden-sign") self.assertNotIn("token", payload) self.assertIn("service_account_names", payload) def test_lease_paths_are_gitignored(self) -> None: result = subprocess.run( ["git", "check-ignore", ".local/credential-leases/example.openbao-token"], cwd=REPO_DIR, capture_output=True, text=True, check=False, ) self.assertEqual(result.returncode, 0, result.stderr) if __name__ == "__main__": unittest.main()