tegwick
5149946a4c
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.
268 lines
9 KiB
Python
268 lines
9 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, MAX_TTL_HOURS
|
|
|
|
|
|
@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} — run 'warden cleanup'" if stale
|
|
else "no stale certs"
|
|
),
|
|
)
|
|
|
|
|
|
def check_ttl_policy(state_dir: Path, inventory: PrincipalsInventory) -> CheckResult:
|
|
"""Certs in state_dir must not exceed the type-max TTL (directive §2)."""
|
|
if not state_dir.exists():
|
|
return CheckResult("ttl_policy", passed=True, detail="no state dir")
|
|
|
|
violations = []
|
|
for cert_path in state_dir.glob("*-cert.pub"):
|
|
actor_name = cert_path.stem.replace("-cert", "")
|
|
entry = inventory.actors.get(actor_name)
|
|
if entry is None:
|
|
continue
|
|
try:
|
|
meta = parse_cert_metadata(cert_path)
|
|
except CAError:
|
|
continue
|
|
valid_from = meta.get("valid_from")
|
|
if valid_from is None:
|
|
continue
|
|
ttl_hours = (meta["valid_before"] - valid_from).total_seconds() / 3600
|
|
max_hours = MAX_TTL_HOURS[entry.actor_type]
|
|
if ttl_hours > max_hours + 0.01:
|
|
violations.append(
|
|
f"{actor_name!r}: {ttl_hours:.1f}h issued > {max_hours}h max"
|
|
)
|
|
|
|
return CheckResult(
|
|
name="ttl_policy",
|
|
passed=len(violations) == 0,
|
|
detail=(
|
|
"; ".join(violations) if violations
|
|
else "all certs within TTL policy"
|
|
),
|
|
)
|
|
|
|
|
|
def check_file_permissions(state_dir: Path) -> CheckResult:
|
|
"""Cert files and keys must not be group- or world-readable (mode 600/644)."""
|
|
if not state_dir.exists():
|
|
return CheckResult("file_permissions", passed=True, detail="no state dir")
|
|
|
|
bad = []
|
|
for cert_path in state_dir.glob("*-cert.pub"):
|
|
if cert_path.stat().st_mode & 0o044:
|
|
bad.append(cert_path.name)
|
|
|
|
keys_dir = state_dir / "keys"
|
|
if keys_dir.exists():
|
|
for key_path in keys_dir.iterdir():
|
|
if key_path.stat().st_mode & 0o044:
|
|
bad.append(f"keys/{key_path.name}")
|
|
|
|
return CheckResult(
|
|
name="file_permissions",
|
|
passed=len(bad) == 0,
|
|
detail=(
|
|
f"world/group readable files: {bad} — run 'chmod 600'" if bad
|
|
else "all cert/key files have restricted permissions"
|
|
),
|
|
)
|
|
|
|
|
|
def check_catalog_rotation_coverage() -> CheckResult:
|
|
"""Every active secret-vending catalog lane must carry rotation guidance (T06).
|
|
|
|
A lane an operator can obtain a *secret value* through must also tell them how
|
|
to renew or re-establish it. Scoped to ``vends_secret`` lanes: this exempts the
|
|
SSH lane (short-lived certs — renewal is re-issuance), ``login`` lanes (re-auth,
|
|
no stored value), and pure routing pointers (tunnel, principals, emission
|
|
sinks, policy checks) with no secret to rotate. Draft lanes are exempt until
|
|
promoted.
|
|
"""
|
|
try:
|
|
from warden.routing import load_catalog
|
|
catalog = load_catalog()
|
|
except Exception as e: # noqa: BLE001 — catalog missing/invalid is its own signal
|
|
return CheckResult(
|
|
name="catalog_rotation_coverage",
|
|
passed=False,
|
|
detail=f"could not load routing catalog: {e}",
|
|
)
|
|
|
|
missing = [
|
|
e.id
|
|
for e in catalog.entries
|
|
if e.is_active and e.vends_secret and not e.has_rotation
|
|
]
|
|
return CheckResult(
|
|
name="catalog_rotation_coverage",
|
|
passed=len(missing) == 0,
|
|
detail=(
|
|
f"active lanes lacking rotation guidance: {missing} — add a `rotation:` "
|
|
"block (see WARDEN-WP-0026 T06)"
|
|
if missing
|
|
else "all active lanes carry rotation guidance"
|
|
),
|
|
)
|
|
|
|
|
|
def check_organization_posture() -> CheckResult:
|
|
"""Surface declared organization lifecycle posture (WARDEN-WP-0029 T02).
|
|
|
|
Always informational PASS when descriptors load -- the check exists so operators
|
|
and agents see the posture in scorecard output without hunting config files.
|
|
"""
|
|
try:
|
|
from warden.posture import load_posture
|
|
|
|
cat = load_posture()
|
|
org = cat.organization_posture
|
|
except Exception as e: # noqa: BLE001
|
|
return CheckResult(
|
|
name="organization_posture",
|
|
passed=False,
|
|
detail=f"could not load organization posture: {e}",
|
|
)
|
|
relax = ", ".join(org.relaxations[:3])
|
|
if len(org.relaxations) > 3:
|
|
relax += ", ..."
|
|
summary = org.summary[:120]
|
|
if len(org.summary) > 120:
|
|
summary += "..."
|
|
relax_part = relax or "no relaxations listed"
|
|
return CheckResult(
|
|
name="organization_posture",
|
|
passed=True,
|
|
detail=f"{org.id} -- {summary} [{relax_part}]",
|
|
)
|
|
|
|
|
|
def check_catalog_freshness() -> CheckResult:
|
|
"""Warn when the CLI is using a bundled (stale-risk) catalog (WP-0029 T05)."""
|
|
try:
|
|
from warden.routing import load_catalog
|
|
|
|
fresh = load_catalog().freshness()
|
|
except Exception as e: # noqa: BLE001
|
|
return CheckResult(
|
|
name="catalog_freshness",
|
|
passed=False,
|
|
detail=f"could not load routing catalog: {e}",
|
|
)
|
|
if fresh.using_bundled:
|
|
nwarn = len(fresh.warnings)
|
|
return CheckResult(
|
|
name="catalog_freshness",
|
|
passed=False,
|
|
detail=(
|
|
f"bundled catalog hash={fresh.content_hash} - reinstall from checkout "
|
|
f"if lanes look missing ({nwarn} warnings)"
|
|
),
|
|
)
|
|
return CheckResult(
|
|
name="catalog_freshness",
|
|
passed=True,
|
|
detail=(
|
|
f"source={fresh.source} hash={fresh.content_hash} "
|
|
f"entries={fresh.active_count}/{fresh.entry_count} active "
|
|
f"newest_reviewed={fresh.newest_reviewed}"
|
|
),
|
|
)
|
|
|
|
|
|
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),
|
|
check_ttl_policy(state_dir, inventory),
|
|
check_file_permissions(state_dir),
|
|
check_catalog_rotation_coverage(),
|
|
check_organization_posture(),
|
|
check_catalog_freshness(),
|
|
]
|