"""Zone-aware flex-auth policy gates for OpsWarden.""" from __future__ import annotations import hashlib import json import os from pathlib import Path import httpx from warden.ca import CAError from warden.caller_identity import CallerIdentityError, caller_auth_headers 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 _caller_headers(cfg: PolicyConfig, *, fail_closed: bool) -> dict[str, str]: """Bearer header identifying ops-warden itself to flex-auth (FLEX-WP-0016). When the token cannot be obtained we refuse the call under the selected zone's ``fail_closed`` behavior rather than silently falling back to an unauthenticated request — an unauthenticated call is exactly what keeps the flex-auth pin in ``warn``. """ try: return caller_auth_headers(cfg.caller_auth) except CallerIdentityError as e: if fail_closed: raise CAError(f"flex-auth caller identity unavailable: {e}") from e return {} def _resource_zone(cfg: PolicyConfig, resource_id: str) -> str: """Read a compiled resource zone; absence or ambiguity is always unknown.""" if cfg.zone_registry_path is None: return "unknown" try: registry = json.loads(cfg.zone_registry_path.read_text()) resources = registry["resource_manifests"][0]["resources"] resource = next(item for item in resources if item.get("id") == resource_id) attributes = resource.get("attributes") or {} if attributes.get("security_zone_admission") == "not-applicable": return "not-applicable" zone = str(attributes.get("security_zone") or "unknown") return zone if zone in cfg.failure_modes else "unknown" except (OSError, ValueError, KeyError, StopIteration, TypeError): return "unknown" def _is_fail_closed(cfg: PolicyConfig, zone: str) -> bool: return cfg.failure_modes.get(zone, cfg.failure_modes["unknown"]) == "fail_closed" def _evaluator_failure( message: str, *, fail_closed: bool, cause: Exception | None = None, spec: CertSpec | None = None, ) -> None: if fail_closed: if spec is not None: spec.policy_outcome = "fail_closed" if cause is None: raise CAError(message) raise CAError(message) from cause if spec is not None: spec.policy_outcome = "fail_open" 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 a decision id on ``allow`` or ``audit_only``. A deny always blocks. Evaluator failures use the PEP-owned failure mode for the target workload's compiled zone; absent resolution is the explicit ``unknown`` profile. """ resource_id = f"ssh-cert:actor/{spec.actor_name}" zone = _resource_zone(cfg, resource_id) fail_closed = _is_fail_closed(cfg, zone) spec.policy_zone = zone spec.policy_failure_mode = "fail_closed" if fail_closed else "fail_open" if cfg.flex_auth_url is None: _evaluator_failure( f"flex-auth URL is not configured for security zone {zone!r}", fail_closed=fail_closed, spec=spec, ) 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" headers = _caller_headers(cfg, fail_closed=fail_closed) try: response = httpx.post(url, json=request, headers=headers, timeout=10.0) response.raise_for_status() except httpx.HTTPStatusError as e: _evaluator_failure( f"flex-auth rejected sign policy check (HTTP {e.response.status_code}) " f"for security zone {zone!r}", fail_closed=fail_closed, cause=e, spec=spec, ) return None except httpx.RequestError as e: _evaluator_failure( f"flex-auth unreachable at {cfg.flex_auth_url!r} for security zone {zone!r}", fail_closed=fail_closed, cause=e, spec=spec, ) return None try: decision = response.json() except ValueError as e: _evaluator_failure( f"flex-auth returned a non-JSON decision for security zone {zone!r}", fail_closed=fail_closed, cause=e, spec=spec, ) return None effect = str(decision.get("effect", "")).lower() decision_id = decision.get("id") or decision.get("request_id") if effect not in {"allow", "audit_only"}: spec.policy_outcome = "deny" 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: _evaluator_failure( f"flex-auth {effect} decision missing id for security zone {zone!r}", fail_closed=fail_closed, spec=spec, ) return None spec.policy_outcome = effect return str(decision_id) def check_fetch_policy( cfg: PolicyConfig, *, need_id: str, owner_repo: str, domain: str | None ) -> str | None: """Call flex-auth /v1/check before proxying a non-SSH credential fetch (WP-0014). The action is ``read`` on a ``secret`` resource owned by another subsystem — ops-warden is the conduit, not the owner. Unresolved target workload identity selects the explicit ``unknown`` profile; no secret value enters the request. """ zone = "unknown" fail_closed = _is_fail_closed(cfg, zone) if cfg.flex_auth_url is None: _evaluator_failure( "flex-auth URL is not configured for security zone 'unknown'", fail_closed=fail_closed, ) return None subject_id = os.environ.get(cfg.subject_env, "").strip() or "operator" request = { "subject": {"id": subject_id, "type": "operator", "tenant": cfg.tenant}, "action": "read", "resource": { "id": f"secret:{need_id}" + (f"/{domain}" if domain else ""), "type": "secret", "system": owner_repo, "tenant": cfg.tenant, }, "context": {"need_id": need_id, "owner_repo": owner_repo, "domain": domain}, } url = cfg.flex_auth_url.rstrip("/") + "/v1/check" headers = _caller_headers(cfg, fail_closed=fail_closed) try: response = httpx.post(url, json=request, headers=headers, timeout=10.0) response.raise_for_status() except httpx.HTTPStatusError as e: _evaluator_failure( f"flex-auth rejected fetch policy check (HTTP {e.response.status_code})", fail_closed=fail_closed, cause=e, ) return None except httpx.RequestError as e: _evaluator_failure( f"flex-auth unreachable at {cfg.flex_auth_url!r} for security zone 'unknown'", fail_closed=fail_closed, cause=e, ) return None try: decision = response.json() except ValueError as e: _evaluator_failure( "flex-auth returned a non-JSON decision for security zone 'unknown'", fail_closed=fail_closed, cause=e, ) return None effect = str(decision.get("effect", "")).lower() decision_id = decision.get("id") or decision.get("request_id") if effect not in {"allow", "audit_only"}: reason = decision.get("reason") or "no reason provided" raise CAError(f"flex-auth denied secret read for {need_id!r}: {reason}") if not decision_id: _evaluator_failure( f"flex-auth {effect} decision missing id for security zone 'unknown'", fail_closed=fail_closed, ) return None return str(decision_id)