railiance-platform/tests/test_credential_helper.py

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