131 lines
4.8 KiB
Python
131 lines
4.8 KiB
Python
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
|
|
|
|
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_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()
|