generated from coulomb/repo-seed
feat(warden): implement WARDEN-WP-0002 correctness and operational completeness
T1 — TTL max enforcement: - models.py: MAX_TTL_HOURS policy constant - ca.py: _enforce_ttl() raises CAError when spec.ttl_hours > type max - Called at top of LocalCA.sign() and VaultCA.sign() - scorecard.py: check_ttl_policy() — flags certs with issued TTL > type max - run_scorecard() now returns 5 checks T2 — Stale cert cleanup: - ca.py: _evict_cert() removes existing cert before writing new one (no accumulation) - cli.py: warden cleanup [actor] [--dry-run] command - check_no_stale_certs detail suggests 'warden cleanup' when stale certs found T3 — Outgoing signatures log: - ca.py: _append_signature_log() writes JSONL to state_dir/signatures.log - Called after every successful sign() in LocalCA and VaultCA - cli.py: warden log [actor] [--last N] [--json] command - parse_cert_metadata now also returns valid_from (needed for TTL policy check) 61 tests passing, ruff clean. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -4,11 +4,15 @@ 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_no_stale_certs,
|
||||
check_no_expired_certs,
|
||||
check_ttl_policy,
|
||||
run_scorecard,
|
||||
)
|
||||
|
||||
@@ -96,4 +100,77 @@ def test_run_scorecard_clean(tmp_path):
|
||||
)
|
||||
results = run_scorecard(tmp_path, inv)
|
||||
assert all(r.passed for r in results)
|
||||
assert len(results) == 4
|
||||
assert len(results) == 5
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 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
|
||||
|
||||
|
||||
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
|
||||
|
||||
Reference in New Issue
Block a user