99 lines
3.3 KiB
Python
99 lines
3.3 KiB
Python
|
"""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),
|
||
|
|
]
|