tegwick
d6088e4e16
Add unified metadata-only audit.jsonl with secret-material guard, instrument sign/access/worker paths, and expose warden activity CLI. Surface broker hint when VAULT_TOKEN is unset, refresh INTENT/SCOPE docs, and add production integration checklists plus catalog lane promotion playbook.
237 lines
8.2 KiB
Python
237 lines
8.2 KiB
Python
"""CA backends for OpsWarden: LocalCA (ssh-keygen) and abstract base."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
from abc import ABC, abstractmethod
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import List, Optional
|
|
|
|
from warden.models import CertRecord, CertSpec, MAX_TTL_HOURS
|
|
|
|
|
|
class CAError(Exception):
|
|
"""Raised when a CA operation fails."""
|
|
|
|
|
|
class CABackend(ABC):
|
|
@abstractmethod
|
|
def sign(self, spec: CertSpec) -> CertRecord:
|
|
"""Sign the public key in spec and return a CertRecord."""
|
|
...
|
|
|
|
|
|
def _enforce_ttl(spec: CertSpec) -> None:
|
|
"""Raise CAError if spec.ttl_hours exceeds the type maximum (directive §2)."""
|
|
max_h = MAX_TTL_HOURS[spec.actor_type]
|
|
if spec.ttl_hours > max_h:
|
|
raise CAError(
|
|
f"TTL {spec.ttl_hours}h exceeds maximum {max_h}h for actor type "
|
|
f"{spec.actor_type.value!r} (AccessManagementDirective §2)"
|
|
)
|
|
|
|
|
|
def _evict_cert(actor_name: str, state_dir: Path) -> None:
|
|
"""Remove the existing cert for actor_name from state_dir, if present."""
|
|
cert_path = state_dir / f"{actor_name}-cert.pub"
|
|
cert_path.unlink(missing_ok=True)
|
|
|
|
|
|
def _append_signature_log(
|
|
record: CertRecord, spec: CertSpec, state_dir: Path, backend: str
|
|
) -> None:
|
|
"""Append one JSONL line to state_dir/signatures.log."""
|
|
entry = {
|
|
"timestamp": datetime.now(timezone.utc).isoformat(),
|
|
"actor": spec.actor_name,
|
|
"actor_type": spec.actor_type.value,
|
|
"identity": record.identity,
|
|
"principals": record.principals,
|
|
"ttl_hours": spec.ttl_hours,
|
|
"valid_before": record.valid_before.isoformat(),
|
|
"cert_path": str(record.cert_path),
|
|
"backend": backend,
|
|
}
|
|
if spec.policy_decision_id:
|
|
entry["policy_decision_id"] = spec.policy_decision_id
|
|
state_dir.mkdir(parents=True, exist_ok=True)
|
|
with (state_dir / "signatures.log").open("a") as f:
|
|
f.write(json.dumps(entry) + "\n")
|
|
try:
|
|
from warden.audit import record_event
|
|
|
|
record_event(
|
|
state_dir,
|
|
kind="sign",
|
|
action="issue",
|
|
subject=spec.actor_name,
|
|
target=spec.actor_name,
|
|
decision_id=spec.policy_decision_id,
|
|
outcome="ok",
|
|
source="sign",
|
|
actor_type=spec.actor_type.value,
|
|
backend=backend,
|
|
ttl_hours=spec.ttl_hours,
|
|
)
|
|
except Exception:
|
|
pass # audit must not block signing
|
|
|
|
|
|
def parse_cert_metadata(cert_path: Path) -> dict:
|
|
"""Parse ssh-keygen -L output into identity, valid_before, and principals.
|
|
|
|
Note: ssh-keygen displays timestamps without explicit timezone; we treat them
|
|
as UTC, consistent with how ssh-keygen internally stores certificate validity.
|
|
"""
|
|
result = subprocess.run(
|
|
["ssh-keygen", "-L", "-f", str(cert_path)],
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
if result.returncode != 0:
|
|
raise CAError(f"ssh-keygen -L failed: {result.stderr.strip()}")
|
|
|
|
identity: Optional[str] = None
|
|
valid_before: Optional[datetime] = None
|
|
valid_from: Optional[datetime] = None
|
|
principals: List[str] = []
|
|
in_principals = False
|
|
|
|
for line in result.stdout.splitlines():
|
|
stripped = line.strip()
|
|
if stripped.startswith("Key ID:"):
|
|
# Key ID: "agt-state-hub-bridge"
|
|
raw = stripped.split(":", 1)[1].strip()
|
|
identity = raw.strip('"')
|
|
elif stripped.startswith("Valid:"):
|
|
# Valid: from 2026-03-28T10:00:00 to 2026-03-29T10:00:00
|
|
parts = stripped.split(" to ", 1)
|
|
if len(parts) == 2:
|
|
ts_str = parts[1].strip()
|
|
try:
|
|
dt = datetime.fromisoformat(ts_str)
|
|
valid_before = dt.replace(tzinfo=timezone.utc)
|
|
except ValueError:
|
|
pass
|
|
from_parts = parts[0].split("from ", 1)
|
|
if len(from_parts) == 2:
|
|
try:
|
|
dt_from = datetime.fromisoformat(from_parts[1].strip())
|
|
valid_from = dt_from.replace(tzinfo=timezone.utc)
|
|
except ValueError:
|
|
pass
|
|
elif stripped == "Principals:":
|
|
in_principals = True
|
|
elif in_principals:
|
|
if stripped and ":" not in stripped and stripped != "(none)":
|
|
principals.append(stripped)
|
|
else:
|
|
in_principals = False
|
|
|
|
if valid_before is None:
|
|
raise CAError(
|
|
f"Could not parse valid_before from cert at {cert_path}. "
|
|
f"Ensure the cert has a valid TTL."
|
|
)
|
|
|
|
return {
|
|
"identity": identity or "",
|
|
"valid_before": valid_before,
|
|
"valid_from": valid_from,
|
|
"principals": principals,
|
|
}
|
|
|
|
|
|
class LocalCA(CABackend):
|
|
"""File-based CA using ssh-keygen. Requires the CA private key on disk."""
|
|
|
|
def __init__(self, ca_key: Path, state_dir: Path) -> None:
|
|
self._ca_key = Path(os.path.expanduser(str(ca_key)))
|
|
self._state_dir = Path(os.path.expanduser(str(state_dir)))
|
|
|
|
def sign(self, spec: CertSpec) -> CertRecord:
|
|
"""Sign the public key in spec. Returns a CertRecord; cert saved to state_dir."""
|
|
_enforce_ttl(spec)
|
|
pubkey = Path(os.path.expanduser(str(spec.pubkey_path)))
|
|
if not pubkey.exists():
|
|
raise CAError(f"Public key not found: {pubkey}")
|
|
if not self._ca_key.exists():
|
|
raise CAError(f"CA key not found: {self._ca_key}")
|
|
|
|
principals_str = ",".join(spec.principals)
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
tmpdir_path = Path(tmpdir)
|
|
pubkey_copy = tmpdir_path / "key.pub"
|
|
shutil.copy2(pubkey, pubkey_copy)
|
|
# ssh-keygen -s writes cert to <input_stem>-cert.pub
|
|
cert_path_tmp = tmpdir_path / "key-cert.pub"
|
|
|
|
cmd = [
|
|
"ssh-keygen",
|
|
"-s", str(self._ca_key),
|
|
"-I", spec.identity,
|
|
"-n", principals_str,
|
|
"-V", f"+{spec.ttl_hours}h",
|
|
str(pubkey_copy),
|
|
]
|
|
result = subprocess.run(cmd, capture_output=True, text=True)
|
|
if result.returncode != 0:
|
|
raise CAError(f"Signing failed: {result.stderr.strip()}")
|
|
|
|
if not cert_path_tmp.exists():
|
|
raise CAError(
|
|
f"Expected cert not written after signing: {cert_path_tmp}. "
|
|
f"ssh-keygen stderr: {result.stderr.strip()}"
|
|
)
|
|
|
|
meta = parse_cert_metadata(cert_path_tmp)
|
|
|
|
self._state_dir.mkdir(parents=True, exist_ok=True)
|
|
_evict_cert(spec.actor_name, self._state_dir)
|
|
dest = self._state_dir / f"{spec.actor_name}-cert.pub"
|
|
shutil.copy2(cert_path_tmp, dest)
|
|
os.chmod(dest, 0o600)
|
|
|
|
record = CertRecord(
|
|
identity=meta["identity"] or spec.identity,
|
|
valid_before=meta["valid_before"],
|
|
cert_path=dest,
|
|
signed_at=datetime.now(timezone.utc),
|
|
principals=meta["principals"],
|
|
actor_name=spec.actor_name,
|
|
)
|
|
_append_signature_log(record, spec, self._state_dir, "local")
|
|
return record
|
|
|
|
def generate_keypair(self, actor_name: str) -> tuple[Path, Path]:
|
|
"""Generate an ed25519 keypair for an actor.
|
|
|
|
Returns (privkey_path, pubkey_path). Overwrites existing files.
|
|
"""
|
|
key_dir = self._state_dir / "keys"
|
|
key_dir.mkdir(parents=True, exist_ok=True)
|
|
privkey = key_dir / f"{actor_name}_ed25519"
|
|
pubkey = key_dir / f"{actor_name}_ed25519.pub"
|
|
|
|
for p in (privkey, pubkey):
|
|
if p.exists():
|
|
p.unlink()
|
|
|
|
cmd = [
|
|
"ssh-keygen", "-t", "ed25519",
|
|
"-f", str(privkey),
|
|
"-N", "", # no passphrase
|
|
"-C", actor_name,
|
|
]
|
|
result = subprocess.run(cmd, capture_output=True, text=True)
|
|
if result.returncode != 0:
|
|
raise CAError(f"Key generation failed: {result.stderr.strip()}")
|
|
|
|
os.chmod(privkey, 0o600)
|
|
os.chmod(pubkey, 0o644)
|
|
return privkey, pubkey
|