93 lines
3 KiB
Python
93 lines
3 KiB
Python
|
"""flex-auth policy gate for SSH signing (opt-in via warden.yaml)."""
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import hashlib
|
||
|
|
import os
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
import httpx
|
||
|
|
|
||
|
|
from warden.ca import CAError
|
||
|
|
from warden.config import PolicyConfig
|
||
|
|
from warden.models import CertSpec
|
||
|
|
|
||
|
|
|
||
|
|
def pubkey_fingerprint(pubkey_path: Path) -> str:
|
||
|
|
"""SHA256 fingerprint of normalized pubkey text (for audit context)."""
|
||
|
|
text = pubkey_path.read_text().strip()
|
||
|
|
digest = hashlib.sha256(text.encode()).hexdigest()
|
||
|
|
return f"sha256:{digest}"
|
||
|
|
|
||
|
|
|
||
|
|
def _subject_id(cfg: PolicyConfig, spec: CertSpec) -> str:
|
||
|
|
return os.environ.get(cfg.subject_env, "").strip() or spec.actor_name
|
||
|
|
|
||
|
|
|
||
|
|
def check_sign_policy(cfg: PolicyConfig, spec: CertSpec) -> str | None:
|
||
|
|
"""Call flex-auth /v1/check before signing.
|
||
|
|
|
||
|
|
Returns decision id when policy is enabled and effect is allow.
|
||
|
|
Returns None when policy is disabled.
|
||
|
|
Raises CAError on deny or when fail_closed and flex-auth is unreachable.
|
||
|
|
"""
|
||
|
|
if not cfg.enabled:
|
||
|
|
return None
|
||
|
|
|
||
|
|
pubkey_path = Path(os.path.expanduser(str(spec.pubkey_path)))
|
||
|
|
if not pubkey_path.exists():
|
||
|
|
raise CAError(f"Public key not found: {pubkey_path}")
|
||
|
|
|
||
|
|
request = {
|
||
|
|
"subject": {
|
||
|
|
"id": _subject_id(cfg, spec),
|
||
|
|
"type": spec.actor_type.value,
|
||
|
|
"tenant": cfg.tenant,
|
||
|
|
},
|
||
|
|
"action": "sign",
|
||
|
|
"resource": {
|
||
|
|
"id": f"ssh-cert:actor/{spec.actor_name}",
|
||
|
|
"type": "ssh-certificate",
|
||
|
|
"system": cfg.system,
|
||
|
|
"tenant": cfg.tenant,
|
||
|
|
},
|
||
|
|
"context": {
|
||
|
|
"actor_name": spec.actor_name,
|
||
|
|
"actor_type": spec.actor_type.value,
|
||
|
|
"principals": spec.principals,
|
||
|
|
"ttl_hours": spec.ttl_hours,
|
||
|
|
"pubkey_fingerprint": pubkey_fingerprint(pubkey_path),
|
||
|
|
},
|
||
|
|
}
|
||
|
|
|
||
|
|
url = cfg.flex_auth_url.rstrip("/") + "/v1/check"
|
||
|
|
try:
|
||
|
|
response = httpx.post(url, json=request, timeout=10.0)
|
||
|
|
response.raise_for_status()
|
||
|
|
except httpx.HTTPStatusError as e:
|
||
|
|
if cfg.fail_closed:
|
||
|
|
raise CAError(
|
||
|
|
f"flex-auth denied or rejected sign policy check (HTTP {e.response.status_code})"
|
||
|
|
) from e
|
||
|
|
return None
|
||
|
|
except httpx.RequestError as e:
|
||
|
|
if cfg.fail_closed:
|
||
|
|
raise CAError(
|
||
|
|
f"flex-auth unreachable at {cfg.flex_auth_url!r} "
|
||
|
|
f"(fail_closed=true): {e}"
|
||
|
|
) from e
|
||
|
|
return None
|
||
|
|
|
||
|
|
try:
|
||
|
|
decision = response.json()
|
||
|
|
except ValueError as e:
|
||
|
|
raise CAError("flex-auth returned non-JSON decision") from e
|
||
|
|
|
||
|
|
effect = str(decision.get("effect", "")).lower()
|
||
|
|
decision_id = decision.get("id") or decision.get("request_id")
|
||
|
|
if effect != "allow":
|
||
|
|
reason = decision.get("reason") or "no reason provided"
|
||
|
|
raise CAError(f"flex-auth denied SSH sign for {spec.actor_name!r}: {reason}")
|
||
|
|
|
||
|
|
if not decision_id:
|
||
|
|
raise CAError("flex-auth allow decision missing id")
|
||
|
|
return str(decision_id)
|