ops-warden/tests/test_scorecard.py
tegwick 5149946a4c
All checks were successful
CI Smoke / host-smoke (push) Successful in 1s
CI Smoke / container-smoke (push) Successful in 2s
WARDEN-WP-0029: implement plan front door, org posture, desk, freshness
Ship posture-aware access planning: organization_posture=build (axis C),
catalog freshness warnings, warden plan verdicts, localhost founder desk,
and playbook/agent guidance that retire /tmp file-drop patterns.

Compose route catalog + handoff rather than a second routing layer.
2026-07-18 16:59:37 +02:00

228 lines
7.5 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)
# cert-side checks + catalog_rotation_coverage (WP-0026 T06)
# + organization_posture + catalog_freshness (WP-0029)
assert len(results) == 9
names = {r.name for r in results}
assert "organization_posture" in names
assert "catalog_freshness" in names
# ---------------------------------------------------------------------------
# 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