"""Fail-closed consumer validation for flex-auth action authorizations. Replay identity follows FLEX-DEC-2026-012. The authenticated evaluator owns registry enrichment; the consumer binds its exact submitted request and checks policy, lifetime and approval correspondence before consumption. """ from __future__ import annotations import copy import hashlib import json import math import re import time import uuid from decimal import Decimal from dataclasses import dataclass from datetime import datetime, timezone from typing import Any from secrets_engine.catalog import CatalogEntry, human_control_required from secrets_engine.errors import DecisionError from secrets_engine.execution_contract import ExecutionContract, attach_execution_contract from secrets_engine.application_time import validity_bounds DIGEST_RE = re.compile(r"^sha256:[0-9a-f]{64}$") #: The single tenant the published policy package scopes this engine to. #: #: `secrets-engine.catalog-lane.lifecycle` v2 reads #: ``request_tenant := object.get(input, "tenant", "")`` and allows only #: ``known_tenant := "tenant:platform"``; anything else, **an absent tenant #: included**, matches its ``wrong_tenant`` first-denial branch. Sending no #: tenant is therefore not a neutral omission, it is a denial. #: #: v1 had no tenant rule at all and failed open -- a `rotate` sent under #: ``tenant:coulomb`` returned allow against the deployed package. That is why #: v1 is superseded rather than amended, and why this value is pinned against #: the vendored replay fixtures in ``tests/test_decision_replay.py`` instead of #: being left to a deployment to supply correctly. #: #: Operator clarification, 2026-09-27: the platform infrastructure identity #: also belongs to tenant:platform. Keep audience-specific profile validation; #: matching tenant strings do not grant cross-audience or workload authority. REQUEST_TENANT = "tenant:platform" SCHEMA_VERSION = "0.1" CONTRACT_VERSION = "flex-auth.decision-record.v1" @dataclass(frozen=True) class ValidatedDecision: decision_id: str action: str subject_id: str expires_at: str request_digest: str submitted_request_digest: 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 = "", tenant: str = REQUEST_TENANT, execution_contract: ExecutionContract | None = None, ) -> dict[str, Any]: """Build the exact normalized secrets-engine profile for flex-auth. ``tenant`` is required and defaults to the package's published ``known_tenant``. It is hashed into the request digest and compared by the package's ``wrong_tenant`` branch, so an omitted tenant is a denial rather than a field the evaluator ignores. """ 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" ) if not tenant: raise DecisionError( "action request requires a tenant; the policy package denies an " "absent tenant as wrong_tenant rather than ignoring it" ) request: dict[str, Any] = {} if request_id: request["id"] = request_id request.update( { "tenant": tenant, "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, # Bind the actual non-secret custody/delivery specification, not # just names which can be reused for a different backend target. # Approval identifiers are excluded: adding the newly issued # approval must not change its own approval-free binding. "catalog_target": copy.deepcopy({ name: getattr(entry, name) for name in ( "kind", "org", "repo", "mount", "path", "fields", "mount_management", "consumers", "delivery_modes", "delivery_auth", "delivery_config", "auth_capability", "workload_delivery", ) }), }, } ) attach_execution_contract(request, execution_contract) if human_control_required(entry): # Bind the requested control to the exact PDP submission. This is a # requirement on the observed approval, not an authorization verdict. request["context"]["human_control"] = True if action == "exec": from secrets_engine.exec_owner import owner_digest digest = owner_digest(entry) if digest is not None: request["context"]["exec_owner_sha256"] = digest 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 {} if not all(isinstance(key, str) for key in value): raise DecisionError("canonical request object keys must be strings") 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"): 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"): 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") if request.get("purpose"): canonical["purpose"] = _required_text(request, "purpose") return canonical #: Correlation-only or separately-hashed fields, excluded from the digest #: material by docs/canonical-request-digest.md "What is hashed". _UNHASHED_FIELDS = ("id", "policy_version", "caring_context") def digest_material(request: object) -> dict[str, Any]: """The exact tuple flex-auth hashes: tenant, subject, action, resource, context. ``id`` is correlation only, ``policy_version`` is recorded in provenance, and ``caring_context`` is hashed separately as ``provenance.input_claim_digests.caring_context``. Including any of them produces a digest that matches no real DecisionEnvelope, which fails closed against every correctly issued decision. """ canonical = canonical_check_request(request) material = {k: v for k, v in canonical.items() if k not in _UNHASHED_FIELDS} # flex-auth tags Context `json:"context,omitempty"`, which drops the key for # an empty map as well as a nil one. Keeping an empty object here would hash # different material than the evaluator did. if material.get("context") == {}: del material["context"] return material def _go_json_encode(value: Any) -> str: """Go encoding/json wire profile, including decoded binary64 numbers. Integers outside the exactly representable binary64-safe interval fail closed instead of silently hashing Go's rounded HTTP-decoded value. """ if value is None: return "null" if isinstance(value, bool): return "true" if value else "false" if isinstance(value, str): try: value.encode("utf-8", errors="strict") except UnicodeEncodeError as error: raise DecisionError("canonical request requires valid Unicode scalars") from error encoded = json.dumps(value, ensure_ascii=False) for character, escaped in (("<", "\\u003c"), (">", "\\u003e"), ("&", "\\u0026"), ("\u2028", "\\u2028"), ("\u2029", "\\u2029")): encoded = encoded.replace(character, escaped) return encoded if isinstance(value, int): if abs(value) > 2**53 - 1: raise DecisionError("canonical request integer exceeds safe binary64 interval") return str(value) if isinstance(value, float): if not math.isfinite(value): raise DecisionError("canonical request number must be finite") if value == 0: return "-0" if math.copysign(1, value) < 0 else "0" # Both runtimes use shortest round-tripping binary64 digits. Go uses # fixed notation for [1e-6, 1e21), scientific notation otherwise. text = repr(value) magnitude = abs(value) if 1e-6 <= magnitude < 1e21: fixed = format(Decimal(text), "f") return fixed.rstrip("0").rstrip(".") if "." in fixed else fixed mantissa, exponent = text.lower().split("e") if mantissa.endswith(".0"): mantissa = mantissa[:-2] number = int(exponent) return mantissa + "e" + ("+" if number >= 0 else "-") + str(abs(number)) if isinstance(value, list): return "[" + ",".join(_go_json_encode(item) for item in value) + "]" if isinstance(value, dict): if not all(isinstance(key, str) for key in value): raise DecisionError("canonical request object keys must be strings") return "{" + ",".join(_go_json_encode(key) + ":" + _go_json_encode(item) for key, item in value.items()) + "}" raise DecisionError("canonical request contains an unsupported JSON value") def request_digest(request: object) -> str: encoded = _go_json_encode(digest_material(request)) return "sha256:" + hashlib.sha256(encoded.encode("utf-8")).hexdigest() #: The context key an approval claim travels on, per flex-auth's #: ``ApprovalContextKey``. APPROVAL_CONTEXT_KEY = "approval" def approval_binding_digest(request: object) -> str: """flex-auth's ``binding.approval_binding_digest`` for this request. The same canonical digest with ``context.approval`` removed, so it is stable across attaching the claim. A request that carries no claim has no separate binding digest, and flex-auth returns the plain request digest there. This exists because a ``pdp_digest`` recorded at approval-issue time can never equal the ``request_digest`` of the request that later carries the claim: the claim is part of the hashed context. See ``flex-auth/docs/canonical-request-digest.md`` "The approval-binding digest" and FLEX-DEC-2026-007. It is deliberately NOT a replay identity. """ canonical = canonical_check_request(request) context = canonical.get("context") if not isinstance(context, dict) or APPROVAL_CONTEXT_KEY not in context: return request_digest(request) stripped = dict(canonical) stripped["context"] = { k: v for k, v in context.items() if k != APPROVAL_CONTEXT_KEY } return request_digest(stripped) 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 _check_approval_binding_digest( binding: dict[str, Any], expected: dict[str, Any], expected_digest: str, ) -> None: """Compare evaluator-origin values; never reproduce registry enrichment.""" context = expected.get("context", {}) carried = APPROVAL_CONTEXT_KEY in context present = binding.get("approval_binding_digest") if not carried: if present is not None or expected_digest: raise DecisionError("approval binding requires a carried approval claim") return if present is None: raise DecisionError("request carried an approval claim but the decision records no approval_binding_digest") if not isinstance(present, str) or not DIGEST_RE.fullmatch(present): raise DecisionError("flex-auth approval binding digest is malformed") claim = context[APPROVAL_CONTEXT_KEY] claim_binding = claim.get("binding", {}) if isinstance(claim, dict) else {} if not isinstance(claim_binding, dict): raise DecisionError("carried approval claim binding must be an object") recorded = claim_binding.get("pdp_digest") if (claim_binding.get("pdp_path") is not True or not isinstance(recorded, str) or not DIGEST_RE.fullmatch(recorded)): raise DecisionError("carried approval claim requires a canonical pdp_path binding") if present != recorded or (expected_digest and present != expected_digest): raise DecisionError("approval claim pdp digest does not match the decision's approval binding digest") def validate_decision_envelope( envelope: object, expected_request: object, *, accepted_policy_packages: set[str], accepted_policy_versions: set[str], expected_approval_binding_digest: str = "", now: datetime | None = None, time_window=None, ) -> ValidatedDecision: """Validate a flex-auth DecisionEnvelope against the proposed action. This is step 2 of GH-DEC-2026-003. It owns exactly the decision-layer checks: effect, exact CheckRequest binding, canonical request digest, lifetime, and the policy package/version pin. The approval fact -- validity, supersession, consumption, distinct approvers -- belongs to the approval-engine claim and is NOT re-checked here (GH-DEC-2026-005: a PIP must not republish the PDP's decision, and neither layer republishes the other's data). There is deliberately no authority constant. State Hub is a read model and holds no runtime approval authority; requiring it fails closed against every correctly issued record. """ 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("decision envelope must be an object") contract = envelope.get("contract_version") if contract is not None and contract != CONTRACT_VERSION: raise DecisionError("unsupported decision envelope contract version") if envelope.get("effect") != "allow": raise DecisionError("flex-auth decision effect is not allow") decision_id = _required_text(envelope, "id") expected = canonical_check_request(expected_request) _require_exact_target_sets(expected) if expected.get("id") and envelope.get("request_id") not in (None, expected["id"]): raise DecisionError("flex-auth decision request id does not match request") binding = _required_dict(envelope, "binding") submitted = binding.get("submitted_request_digest") if submitted != request_digest(expected): raise DecisionError("flex-auth submitted request digest is missing or does not match the proposed action") # Keep the evaluator's enriched digest as audit/consume evidence. It is not # locally computable, and registry facts may override caller attributes. evaluated = binding.get("request_digest") if not isinstance(evaluated, str) or not DIGEST_RE.fullmatch(evaluated): raise DecisionError("flex-auth evaluated request digest is malformed") _check_approval_binding_digest(binding, expected, expected_approval_binding_digest) lower, upper = validity_bounds(now, time_window) lifetime = _required_dict(envelope, "lifetime") expires = _parse_time(lifetime.get("expires_at"), "expires_at") if upper >= expires: raise DecisionError("flex-auth decision lifetime has expired") if lifetime.get("not_before") is not None: if lower < _parse_time(lifetime.get("not_before"), "not_before"): raise DecisionError("flex-auth decision lifetime has not started") provenance = _required_dict(envelope, "provenance") if provenance.get("policy_package") not in accepted_policy_packages: raise DecisionError("flex-auth policy package is not accepted") if provenance.get("policy_version") not in accepted_policy_versions: raise DecisionError("flex-auth policy version is not accepted") return ValidatedDecision( decision_id=decision_id, action=expected["action"], subject_id=expected["subject"]["id"], expires_at=expires.isoformat(), request_digest=evaluated, submitted_request_digest=submitted, ) def wait_for_decision_start(envelope: object, *, clock=None, sleeper=None) -> None: """Wait at most two seconds for an owner's future start, never waive it. Full decision/claim validation still runs afterward against the real clock. Longer skew refuses rather than holding an approval claim until it is stale. """ current_time = clock or (lambda: datetime.now(timezone.utc)) sleep = sleeper or time.sleep if not isinstance(envelope, dict): return lifetime = _required_dict(envelope, "lifetime") if lifetime.get("not_before") is None: return start = _parse_time(lifetime["not_before"], "not_before") delay = (start - current_time()).total_seconds() if delay > 2: raise DecisionError("flex-auth decision start exceeds bounded wait") if delay > 0: sleep(delay)