"""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 re 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 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. #: #: This is not the KeyCape JWT tenant. ``service_auth.TENANT`` is #: ``tenant:coulomb``, which is the exact value this package denies. Whether #: those name one tenant or two layers is an open owner question -- see #: ``docs/tenant-alignment.md`` -- and is deliberately not resolved by reusing #: one constant for both. 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 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, ) -> 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}, } ) 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 #: 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 request_digest(request: object) -> str: canonical = digest_material(request) encoded = json.dumps( canonical, ensure_ascii=False, separators=(",", ":") ).encode("utf-8") return "sha256:" + hashlib.sha256(encoded).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: """Tie an approval claim to this exact request (FLEX-DEC-2026-007). ``binding.approval_binding_digest`` is the canonical request digest computed with ``context.approval`` removed, and it appears only when the request carried a claim there. It is the ONLY sound comparand for a claim's ``pdp_digest``: a digest recorded at issue time can never equal the ``request_digest`` of the request that carries the claim, because the claim is part of the hashed material. Comparing against ``request_digest`` fails closed forever; comparing against nothing fails open. It is deliberately NOT a replay identity -- two requests differing only in which approval was presented share it while their decisions differ -- so it is checked here in addition to ``request_digest``, never instead of it. When our own request is claim-free the field is absent by contract, and the identity already holds transitively: step 1 compared the claim's pdp_digest to this same canonical digest of the claim-free request. """ present = binding.get("approval_binding_digest") if present is not None: if not isinstance(present, str) or not DIGEST_RE.fullmatch(present): raise DecisionError("flex-auth approval binding digest is malformed") if present != approval_binding_digest(expected): raise DecisionError( "flex-auth approval binding digest does not match this request " "with the approval claim removed" ) if expected_digest: if present is None: context = expected.get("context") if not isinstance(context, dict) or APPROVAL_CONTEXT_KEY not in context: # Claim-free request: no approval_binding_digest is emitted and # step 1 already bound the claim to this canonical digest. return raise DecisionError( "request carried an approval claim but the decision records no " "approval_binding_digest; the claim cannot be tied to it" ) if present != expected_digest: raise DecisionError( "approval claim pdp digest does not match the decision's " "approval binding digest" ) #: Fields the evaluator's documented normalization may add to a ref, per #: canonical-request-digest.md "Normalization": the request tenant is copied onto #: subject and resource when they omit it, and a registry hit copies type, #: tenant, and selected attributes onto the refs the digest sees. _ENRICHABLE_REF_FIELDS = ("type", "tenant", "attributes") def _require_ref_corresponds(bound: Any, sent: dict[str, Any], tenant: str, name: str) -> None: """The bound ref must be our ref, plus only documented enrichment.""" if not isinstance(bound, dict): raise DecisionError(f"flex-auth decision binding {name} must be an object") for key, value in sent.items(): if key == "attributes": continue if bound.get(key) != value: raise DecisionError( f"flex-auth decision binding {name}.{key} does not match the " "proposed action" ) for key in bound: if key not in sent and key not in _ENRICHABLE_REF_FIELDS: raise DecisionError( f"flex-auth decision binding {name} carries unexpected field " f"'{key}'; enrichment may add only {_ENRICHABLE_REF_FIELDS}" ) # Normalization rule 1: an enriched tenant is the REQUEST's tenant. A ref # bearing some other tenant would be a cross-tenant binding wearing our # request's clothes. bound_tenant = bound.get("tenant") if bound_tenant is not None and tenant and bound_tenant != tenant: raise DecisionError( f"flex-auth decision binding {name}.tenant '{bound_tenant}' is not " f"the request tenant '{tenant}'" ) sent_attributes = sent.get("attributes") if isinstance(sent_attributes, dict): bound_attributes = bound.get("attributes") if not isinstance(bound_attributes, dict): raise DecisionError( f"flex-auth decision binding {name}.attributes must be an object" ) # Every attribute we proposed must survive unchanged. The registry may # add its own; it may not restate ours differently. stage, fields, # policy_targets and auth_targets are security-bearing and are ours. for key, value in sent_attributes.items(): if bound_attributes.get(key) != value: raise DecisionError( f"flex-auth decision binding {name}.attributes.{key} does " "not match the proposed action" ) def _require_binding_corresponds( binding: dict[str, Any], expected: dict[str, Any] ) -> None: """Compare structured binding fields to the proposed action. This is what canonical-request-digest.md tells a consumer to do, and why it is not a digest comparison: "A consumer that re-hashes the original unenriched request will not match a decision that turned on registry attributes. Compare structured binding fields to the proposed action, and treat request_digest as the evaluator's statement of what it hashed." The evaluator normalizes before hashing -- it copies the request tenant onto subject and resource, and a registry hit copies type, tenant and selected attributes onto the refs. Those additions are material we never sent and cannot reproduce, so byte-equality against our own request is unsatisfiable against every real decision. What must still hold, and is enforced here: everything we DID propose survives unchanged, enrichment appears only where the contract permits it, and an enriched tenant is our request's tenant rather than another one. """ tenant = str(expected.get("tenant", "") or "") if tenant and binding.get("tenant") != tenant: raise DecisionError("flex-auth decision binding tenant does not match request") if binding.get("action") != expected["action"]: raise DecisionError("flex-auth decision binding action does not match request") if binding.get("context", {}) != expected.get("context", {}): raise DecisionError("flex-auth decision binding context does not match request") _require_ref_corresponds(binding.get("subject"), expected["subject"], tenant, "subject") _require_ref_corresponds(binding.get("resource"), expected["resource"], tenant, "resource") def binding_tuple(binding: dict[str, Any]) -> dict[str, Any]: """The normalized tuple the binding carries, for independent re-hashing. canonical-request-digest.md: "To recompute independently, hash the same normalized tuple the binding carries (tenant, subject, action, resource, context)." """ tuple_: dict[str, Any] = {} if binding.get("tenant"): tuple_["tenant"] = binding["tenant"] tuple_.update( { "subject": binding.get("subject"), "action": binding.get("action"), "resource": binding.get("resource"), "context": binding.get("context", {}), } ) return tuple_ 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, ) -> 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") _require_binding_corresponds(binding, expected) # The digest is the evaluator's statement of what it hashed, so it is # verified for self-consistency against the tuple the binding carries -- # never against our unenriched request, which no real decision can match. bound_tuple = binding_tuple(binding) if binding.get("request_digest") != request_digest(bound_tuple): raise DecisionError( "flex-auth request digest does not match the binding it carries" ) _check_approval_binding_digest(binding, bound_tuple, expected_approval_binding_digest) # The envelope's top-level refs are enriched too, so they get the same # correspondence rule rather than byte-equality against what we sent. _tenant = str(expected.get("tenant", "") or "") _require_ref_corresponds( _subject_ref(envelope.get("subject")), expected["subject"], _tenant, "subject" ) _require_ref_corresponds( _resource_ref(envelope.get("resource")), expected["resource"], _tenant, "resource" ) current = (now or datetime.now(timezone.utc)).astimezone(timezone.utc) lifetime = _required_dict(envelope, "lifetime") expires = _parse_time(lifetime.get("expires_at"), "expires_at") if current >= expires: raise DecisionError("flex-auth decision lifetime has expired") if lifetime.get("not_before") is not None: if current < _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(), )