"""Fail-closed consumer validation for flex-auth action authorizations. The canonical contract is flex-auth revision c473f19. State Hub does not yet provide the durable authoritative endpoint, so this module validates supplied objects but does not resolve or enable production actions by itself. """ from __future__ import annotations import hashlib import json import uuid from dataclasses import dataclass from datetime import datetime, timezone from typing import Any from secrets_engine.catalog import CatalogEntry from secrets_engine.errors import DecisionError SCHEMA_VERSION = "0.1" AUTHORITY = "state-hub" @dataclass(frozen=True) class ValidatedActionAuthorization: authorization_id: str decision_id: str action: str subject_id: str expires_at: str def build_action_request( entry: CatalogEntry, action: str, *, subject_id: str, subject_type: str, purpose: str, fields: list[str] | tuple[str, ...] = (), policy_targets: list[str] | tuple[str, ...] = (), auth_targets: list[str] | tuple[str, ...] = (), request_id: str = "", ) -> dict[str, Any]: """Build the exact normalized secrets-engine profile for flex-auth.""" if not action or not subject_id or not subject_type or not purpose: raise DecisionError( "action request requires action, subject id/type, and purpose" ) request: dict[str, Any] = {} if request_id: request["id"] = request_id request.update( { "subject": {"id": subject_id, "type": subject_type}, "action": action, "resource": { "id": f"catalog:{entry.id}", "type": "secret-catalog-lane", "system": "secrets-engine", "attributes": { "stage": entry.stage, "fields": sorted(set(fields)), "policy_targets": sorted(set(policy_targets)), "auth_targets": sorted(set(auth_targets)), }, }, "context": {"purpose": purpose}, } ) return request def _required_dict(container: dict[str, Any], name: str) -> dict[str, Any]: value = container.get(name) if not isinstance(value, dict): raise DecisionError(f"action authorization requires object '{name}'") return value def _required_text(container: dict[str, Any], name: str) -> str: value = container.get(name) if not isinstance(value, str) or not value: raise DecisionError(f"action authorization requires non-empty '{name}'") return value def _parse_time(value: object, name: str) -> datetime: if not isinstance(value, str): raise DecisionError(f"action authorization requires timestamp '{name}'") try: parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) except ValueError as e: raise DecisionError(f"action authorization has invalid timestamp '{name}'") from e if parsed.tzinfo is None: raise DecisionError(f"action authorization timestamp '{name}' needs timezone") return parsed.astimezone(timezone.utc) def _sorted_map(value: object) -> dict[str, Any]: if not isinstance(value, dict): return {} return {key: _canonical_map_value(value[key]) for key in sorted(value)} def _canonical_map_value(value: Any) -> Any: if isinstance(value, dict): return _sorted_map(value) if isinstance(value, list): return [_canonical_map_value(item) for item in value] return value def _subject_ref(value: object) -> dict[str, Any]: if not isinstance(value, dict): raise DecisionError("action authorization subject must be an object") subject: dict[str, Any] = {"id": _required_text(value, "id")} for name in ("type", "tenant"): if value.get(name): subject[name] = _required_text(value, name) if value.get("attributes") is not None: subject["attributes"] = _sorted_map(value.get("attributes")) return subject def _resource_ref(value: object) -> dict[str, Any]: if not isinstance(value, dict): raise DecisionError("action authorization resource must be an object") resource: dict[str, Any] = {"id": _required_text(value, "id")} for name in ("type", "system", "tenant"): if value.get(name): resource[name] = _required_text(value, name) if value.get("attributes") is not None: resource["attributes"] = _sorted_map(value.get("attributes")) return resource def canonical_check_request(request: object) -> dict[str, Any]: """Match Go encoding/json field order used by flex-auth request digests.""" if not isinstance(request, dict): raise DecisionError("action authorization request must be an object") canonical: dict[str, Any] = {} if request.get("id"): canonical["id"] = _required_text(request, "id") if request.get("tenant"): canonical["tenant"] = _required_text(request, "tenant") canonical["subject"] = _subject_ref(request.get("subject")) canonical["action"] = _required_text(request, "action") canonical["resource"] = _resource_ref(request.get("resource")) if request.get("context") is not None: canonical["context"] = _sorted_map(request.get("context")) if request.get("caring_context") is not None: canonical["caring_context"] = _canonical_map_value( request.get("caring_context") ) if request.get("policy_version"): canonical["policy_version"] = _required_text(request, "policy_version") return canonical def request_digest(request: object) -> str: canonical = canonical_check_request(request) encoded = json.dumps( canonical, ensure_ascii=False, separators=(",", ":") ).encode("utf-8") return "sha256:" + hashlib.sha256(encoded).hexdigest() def _require_exact_target_sets(request: dict[str, Any]) -> None: resource = _required_dict(request, "resource") attributes = resource.get("attributes", {}) if not isinstance(attributes, dict): raise DecisionError("action authorization resource attributes must be an object") for name in ("fields", "policy_targets", "auth_targets"): values = attributes.get(name, []) if not isinstance(values, list) or not all( isinstance(item, str) and item for item in values ): raise DecisionError(f"action authorization target set '{name}' is invalid") if values != sorted(set(values)): raise DecisionError( f"action authorization target set '{name}' must be sorted and unique" ) def validate_action_authorization( envelope: object, expected_request: object, *, accepted_policy_packages: set[str], accepted_policy_versions: set[str], minimum_approval_count: int = 1, now: datetime | None = None, ) -> ValidatedActionAuthorization: """Validate exact request binding and dual control; never parse prose.""" if minimum_approval_count < 1: raise DecisionError("minimum approval count must be positive") if not accepted_policy_packages or not accepted_policy_versions: raise DecisionError("accepted flex-auth policy package/version is required") if not isinstance(envelope, dict): raise DecisionError("action authorization must be an object") if envelope.get("schema_version") != SCHEMA_VERSION: raise DecisionError("unsupported action authorization schema version") authorization_id = _required_text(envelope, "id") try: parsed_authorization_id = uuid.UUID(authorization_id) except ValueError as e: raise DecisionError("action authorization id must be a canonical UUID") from e if str(parsed_authorization_id) != authorization_id: raise DecisionError("action authorization id must be a canonical UUID") if envelope.get("status") != "approved": raise DecisionError("action authorization status is not approved") if envelope.get("superseded_by"): raise DecisionError("action authorization is superseded") provenance = _required_dict(envelope, "provenance") if provenance.get("authority") != AUTHORITY: raise DecisionError("action authorization authority is not State Hub") request = canonical_check_request(envelope.get("request")) expected = canonical_check_request(expected_request) _require_exact_target_sets(request) _require_exact_target_sets(expected) if request != expected: raise DecisionError("action authorization request does not exactly match action") validity = _required_dict(envelope, "validity") expires = _parse_time(validity.get("expires_at"), "expires_at") not_before = ( _parse_time(validity.get("not_before"), "not_before") if validity.get("not_before") is not None else None ) current = (now or datetime.now(timezone.utc)).astimezone(timezone.utc) if not_before is not None and current < not_before: raise DecisionError("action authorization window has not started") if current >= expires: raise DecisionError("action authorization has expired") approvals = _required_dict(envelope, "approvals") required_count = approvals.get("required_count") entries = approvals.get("entries") if not isinstance(required_count, int) or required_count < 1: raise DecisionError("action authorization approval count is invalid") if required_count < minimum_approval_count: raise DecisionError("action authorization approval threshold is insufficient") if not isinstance(entries, list): raise DecisionError("action authorization approval entries are invalid") approvers: set[str] = set() for entry in entries: if not isinstance(entry, dict): raise DecisionError("action authorization approval entry is invalid") subject_id = _required_text(entry, "subject_id") approved_at = _parse_time(entry.get("approved_at"), "approved_at") if approved_at > current or approved_at >= expires: raise DecisionError("action authorization approval time is outside window") if not_before is not None and approved_at < not_before: raise DecisionError("action authorization approval time is outside window") if subject_id in approvers: raise DecisionError("action authorization contains duplicate approver") approvers.add(subject_id) if len(approvers) < required_count: raise DecisionError("action authorization has insufficient distinct approvals") decision = _required_dict(envelope, "decision") if decision.get("effect") != "allow": raise DecisionError("flex-auth decision effect is not allow") decision_id = _required_text(decision, "id") if request.get("id") and decision.get("request_id") != request["id"]: raise DecisionError("flex-auth decision request id does not match request") binding = _required_dict(decision, "binding") bound_request: dict[str, Any] = {} if binding.get("tenant"): bound_request["tenant"] = binding["tenant"] bound_request.update( { "subject": binding.get("subject"), "action": binding.get("action"), "resource": binding.get("resource"), "context": binding.get("context", {}), } ) expected_bound: dict[str, Any] = {} if request.get("tenant"): expected_bound["tenant"] = request["tenant"] expected_bound.update( { "subject": request["subject"], "action": request["action"], "resource": request["resource"], "context": request.get("context", {}), } ) if canonical_check_request(bound_request) != canonical_check_request( expected_bound ): raise DecisionError("flex-auth decision binding does not match request") if binding.get("request_digest") != request_digest(request): raise DecisionError("flex-auth request digest does not match request") if _subject_ref(decision.get("subject")) != request["subject"]: raise DecisionError("flex-auth decision subject does not match request") if _resource_ref(decision.get("resource")) != request["resource"]: raise DecisionError("flex-auth decision resource does not match request") decision_provenance = _required_dict(decision, "provenance") if decision_provenance.get("policy_package") not in accepted_policy_packages: raise DecisionError("flex-auth policy package is not accepted") if decision_provenance.get("policy_version") not in accepted_policy_versions: raise DecisionError("flex-auth policy version is not accepted") return ValidatedActionAuthorization( authorization_id=authorization_id, decision_id=decision_id, action=request["action"], subject_id=request["subject"]["id"], expires_at=expires.isoformat(), )