Files
ops-warden/tests/test_scorecard.py
tegwick f3547acd0b feat(warden): WARDEN-WP-0003 — test coverage, permissions, status --state-dir
- File permissions: os.chmod(cert, 0o600) after every sign in LocalCA and
  VaultCA; chmod(privkey, 0o600) and chmod(pubkey, 0o644) after generate_keypair
- Scorecard: add check_file_permissions() that flags world/group-readable
  cert and key files; run_scorecard now returns 6 checks
- warden status --state-dir: bypasses config loading entirely for operators
  who have a cert but no warden.yaml installed
- tests/test_vault.py: 11 VaultCA unit tests covering success, HTTP 403,
  RequestError, missing token, missing role, missing pubkey, TTL enforcement,
  eviction, signatures log, and cert mode 600
- tests/test_ca.py: generate_keypair tests (paths, args, overwrite, error,
  permissions) and cert mode 600 assertion after sign
- tests/test_scorecard.py: file_permissions check tests (pass, fail cert,
  fail keys dir); scorecard count updated to 6
- tests/test_cli.py: covers sign, issue, status, scorecard, inventory, log,
  cleanup commands using CliRunner and tmp config/inventory files
- tests/test_integration.py: @pytest.mark.integration tests against real
  ssh-keygen; excluded from default suite via pyproject addopts
- pyproject.toml: addopts = "-m 'not integration'", integration marker declared

All 100 unit tests pass; 3 integration tests pass; ruff clean.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-05-15 17:05:38 +02:00

224 lines
7.3 KiB
Python

"""Tests for warden.scorecard."""
from pathlib import Path
from warden.inventory import ActorEntry, PrincipalsInventory
from warden.models import ActorType
from unittest.mock import patch
from datetime import datetime, timezone
from warden.scorecard import (
check_actor_name_prefixes,
check_all_actors_have_principals,
check_file_permissions,
check_no_stale_certs,
check_no_expired_certs,
check_ttl_policy,
run_scorecard,
)
def make_inventory(*actors):
inv = PrincipalsInventory()
for name, atype, principals in actors:
inv.actors[name] = ActorEntry(
name=name, actor_type=atype, principals=principals, ttl_hours=24
)
return inv
# ---------------------------------------------------------------------------
# check_actor_name_prefixes
# ---------------------------------------------------------------------------
def test_prefix_check_pass():
inv = make_inventory(
("adm-bernd", ActorType.ADM, ["adm-full"]),
("agt-bridge", ActorType.AGT, ["agt-task-bridge"]),
("atm-cron", ActorType.ATM, ["atm-cron"]),
)
result = check_actor_name_prefixes(inv)
assert result.passed
def test_prefix_check_fail_bad_name():
# Bypass validate_actor_name by inserting directly
inv = PrincipalsInventory()
inv.actors["bad-name"] = ActorEntry(
name="bad-name", actor_type=ActorType.AGT, principals=["x"], ttl_hours=24
)
result = check_actor_name_prefixes(inv)
assert not result.passed
assert "bad-name" in result.detail
# ---------------------------------------------------------------------------
# check_all_actors_have_principals
# ---------------------------------------------------------------------------
def test_principals_check_pass():
inv = make_inventory(("agt-bridge", ActorType.AGT, ["agt-task-bridge"]))
result = check_all_actors_have_principals(inv)
assert result.passed
def test_principals_check_fail_empty():
inv = PrincipalsInventory()
inv.actors["agt-bridge"] = ActorEntry(
name="agt-bridge", actor_type=ActorType.AGT, principals=[], ttl_hours=24
)
result = check_all_actors_have_principals(inv)
assert not result.passed
assert "agt-bridge" in result.detail
# ---------------------------------------------------------------------------
# check_no_stale_certs
# ---------------------------------------------------------------------------
def test_no_stale_certs_nonexistent_dir():
result = check_no_stale_certs(Path("/nonexistent/state/dir"))
assert result.passed
def test_no_stale_certs_empty_dir(tmp_path):
result = check_no_stale_certs(tmp_path)
assert result.passed
def test_no_expired_certs_empty_dir(tmp_path):
result = check_no_expired_certs(tmp_path)
assert result.passed
# ---------------------------------------------------------------------------
# run_scorecard
# ---------------------------------------------------------------------------
def test_run_scorecard_clean(tmp_path):
inv = make_inventory(
("agt-bridge", ActorType.AGT, ["agt-task-bridge"]),
)
results = run_scorecard(tmp_path, inv)
assert all(r.passed for r in results)
assert len(results) == 6
# ---------------------------------------------------------------------------
# check_ttl_policy
# ---------------------------------------------------------------------------
def test_ttl_policy_no_state_dir():
inv = make_inventory(("agt-bridge", ActorType.AGT, ["agt-task-bridge"]))
result = check_ttl_policy(Path("/nonexistent/state"), inv)
assert result.passed
def test_ttl_policy_empty_dir(tmp_path):
inv = make_inventory(("agt-bridge", ActorType.AGT, ["agt-task-bridge"]))
result = check_ttl_policy(tmp_path, inv)
assert result.passed
def test_ttl_policy_pass(tmp_path):
inv = make_inventory(("agt-bridge", ActorType.AGT, ["agt-task-bridge"]))
cert_path = tmp_path / "agt-bridge-cert.pub"
cert_path.write_text("fake")
# 24h window — exactly at AGT max
meta = {
"identity": "agt-bridge",
"valid_before": datetime(2026, 3, 29, 10, 0, 0, tzinfo=timezone.utc),
"valid_from": datetime(2026, 3, 28, 10, 0, 0, tzinfo=timezone.utc),
"principals": ["agt-task-bridge"],
}
with patch("warden.scorecard.parse_cert_metadata", return_value=meta):
result = check_ttl_policy(tmp_path, inv)
assert result.passed
def test_ttl_policy_fail(tmp_path):
inv = make_inventory(("agt-bridge", ActorType.AGT, ["agt-task-bridge"]))
cert_path = tmp_path / "agt-bridge-cert.pub"
cert_path.write_text("fake")
# 48h window — exceeds AGT max of 24h
meta = {
"identity": "agt-bridge",
"valid_before": datetime(2026, 3, 30, 10, 0, 0, tzinfo=timezone.utc),
"valid_from": datetime(2026, 3, 28, 10, 0, 0, tzinfo=timezone.utc),
"principals": ["agt-task-bridge"],
}
with patch("warden.scorecard.parse_cert_metadata", return_value=meta):
result = check_ttl_policy(tmp_path, inv)
assert not result.passed
assert "agt-bridge" in result.detail
def test_ttl_policy_skips_unknown_actor(tmp_path):
inv = PrincipalsInventory() # empty — no actors
cert_path = tmp_path / "agt-unknown-cert.pub"
cert_path.write_text("fake")
result = check_ttl_policy(tmp_path, inv)
assert result.passed # unknown actor skipped, not a violation
# ---------------------------------------------------------------------------
# check_file_permissions
# ---------------------------------------------------------------------------
def test_file_permissions_no_state_dir():
result = check_file_permissions(Path("/nonexistent/state/dir"))
assert result.passed
def test_file_permissions_empty_dir(tmp_path):
result = check_file_permissions(tmp_path)
assert result.passed
def test_file_permissions_pass(tmp_path):
cert = tmp_path / "agt-bridge-cert.pub"
cert.write_text("fake")
cert.chmod(0o600)
result = check_file_permissions(tmp_path)
assert result.passed
def test_file_permissions_fail_world_readable(tmp_path):
cert = tmp_path / "agt-bridge-cert.pub"
cert.write_text("fake")
cert.chmod(0o644)
result = check_file_permissions(tmp_path)
assert not result.passed
assert "agt-bridge-cert.pub" in result.detail
def test_file_permissions_keys_dir(tmp_path):
keys_dir = tmp_path / "keys"
keys_dir.mkdir()
key = keys_dir / "agt-test_ed25519"
key.write_text("fake key")
key.chmod(0o644)
result = check_file_permissions(tmp_path)
assert not result.passed
assert "keys/agt-test_ed25519" in result.detail
# ---------------------------------------------------------------------------
# (continuation)
# ---------------------------------------------------------------------------
def test_stale_certs_detail_suggests_cleanup(tmp_path):
cert_path = tmp_path / "agt-bridge-cert.pub"
cert_path.write_text("fake")
# Expired well over 5 minutes ago
meta = {
"identity": "agt-bridge",
"valid_before": datetime(2020, 1, 1, tzinfo=timezone.utc),
"valid_from": datetime(2019, 12, 31, tzinfo=timezone.utc),
"principals": [],
}
with patch("warden.scorecard.parse_cert_metadata", return_value=meta):
result = check_no_stale_certs(tmp_path)
assert not result.passed
assert "warden cleanup" in result.detail