generated from coulomb/repo-seed
Initial Commit
This commit is contained in:
98
src/warden/scorecard.py
Normal file
98
src/warden/scorecard.py
Normal file
@@ -0,0 +1,98 @@
|
||||
"""Compliance scorecard — cert-side checks (AccessManagementDirective §5)."""
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from pathlib import Path
|
||||
from typing import List
|
||||
|
||||
from warden.ca import CAError, parse_cert_metadata
|
||||
from warden.inventory import PrincipalsInventory
|
||||
from warden.models import ACTOR_PREFIX, ActorType
|
||||
|
||||
|
||||
@dataclass
|
||||
class CheckResult:
|
||||
name: str
|
||||
passed: bool
|
||||
detail: str = ""
|
||||
|
||||
|
||||
def check_actor_name_prefixes(inventory: PrincipalsInventory) -> CheckResult:
|
||||
"""All actor names must carry the prefix matching their type."""
|
||||
violations = []
|
||||
for name, entry in inventory.actors.items():
|
||||
expected = ACTOR_PREFIX[entry.actor_type]
|
||||
if not name.startswith(expected):
|
||||
violations.append(f"{name!r} should start with {expected!r}")
|
||||
return CheckResult(
|
||||
name="actor_name_prefixes",
|
||||
passed=len(violations) == 0,
|
||||
detail=(
|
||||
"; ".join(violations) if violations else "all actor names match prefix convention"
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def check_all_actors_have_principals(inventory: PrincipalsInventory) -> CheckResult:
|
||||
"""Every actor in inventory must have at least one principal."""
|
||||
missing = [name for name, e in inventory.actors.items() if not e.principals]
|
||||
return CheckResult(
|
||||
name="actors_have_principals",
|
||||
passed=len(missing) == 0,
|
||||
detail=f"missing principals: {missing}" if missing else "all actors have principals",
|
||||
)
|
||||
|
||||
|
||||
def check_no_expired_certs(state_dir: Path) -> CheckResult:
|
||||
"""No cert in state_dir should be currently expired."""
|
||||
if not state_dir.exists():
|
||||
return CheckResult("no_expired_certs", passed=True, detail="no state dir")
|
||||
|
||||
now = datetime.now(timezone.utc)
|
||||
expired = []
|
||||
for cert_path in state_dir.glob("*-cert.pub"):
|
||||
try:
|
||||
meta = parse_cert_metadata(cert_path)
|
||||
except CAError:
|
||||
continue
|
||||
if meta["valid_before"] < now:
|
||||
expired.append(cert_path.stem.replace("-cert", ""))
|
||||
|
||||
return CheckResult(
|
||||
name="no_expired_certs",
|
||||
passed=len(expired) == 0,
|
||||
detail=f"expired: {expired}" if expired else "no expired certs",
|
||||
)
|
||||
|
||||
|
||||
def check_no_stale_certs(state_dir: Path) -> CheckResult:
|
||||
"""Certs expired by more than 5 minutes should have been cleaned up."""
|
||||
if not state_dir.exists():
|
||||
return CheckResult("no_stale_certs", passed=True, detail="no state dir")
|
||||
|
||||
cutoff = datetime.now(timezone.utc) - timedelta(minutes=5)
|
||||
stale = []
|
||||
for cert_path in state_dir.glob("*-cert.pub"):
|
||||
try:
|
||||
meta = parse_cert_metadata(cert_path)
|
||||
except CAError:
|
||||
continue
|
||||
if meta["valid_before"] < cutoff:
|
||||
stale.append(cert_path.name)
|
||||
|
||||
return CheckResult(
|
||||
name="no_stale_certs",
|
||||
passed=len(stale) == 0,
|
||||
detail=f"stale certs present: {stale}" if stale else "no stale certs",
|
||||
)
|
||||
|
||||
|
||||
def run_scorecard(state_dir: Path, inventory: PrincipalsInventory) -> List[CheckResult]:
|
||||
"""Run all cert-side scorecard checks. Returns list of CheckResult."""
|
||||
return [
|
||||
check_actor_name_prefixes(inventory),
|
||||
check_all_actors_have_principals(inventory),
|
||||
check_no_expired_certs(state_dir),
|
||||
check_no_stale_certs(state_dir),
|
||||
]
|
||||
Reference in New Issue
Block a user