"""Consume Flex Auth decisions for this surface; never evaluate local policy. The submitted digest follows Flex Auth's published Go wire contract. It is not the approval act digest, which this package only carries unchanged. """ from dataclasses import dataclass from datetime import datetime import hashlib import json import re import time import uuid from .http_transport import JSONTransport, TransportError, fixed_origin from .oidc import HumanSession DIGEST = re.compile(r"sha256:[0-9a-f]{64}") ACTIONS = frozenset({"read", "acknowledge", "accept", "return", "discuss", "decline"}) CONTRACT = "flex-auth.decision-record.v1" class PolicyError(RuntimeError): """Only bounded codes, never tokens or upstream diagnostic text.""" def _sorted(value): if isinstance(value, dict): return {k: _sorted(value[k]) for k in sorted(value)} if isinstance(value, list): return [_sorted(v) for v in value] if value is None or type(value) in (str, bool): return value if type(value) is int and abs(value) <= 2**53: return value raise ValueError("request profile contains an unsupported value") def submitted_digest(request): """Exact submitted tuple: Go struct order, sorted maps, HTML escaping. Supports this consumer's closed JSON profile (no floats or CARING input). Real Flex Auth tests pin this against api.RequestDigest and enrichment. """ def ref(value, fields): return {k: _sorted(value[k]) for k in fields if value.get(k)} material = {"tenant": request["tenant"], "subject": ref(request["subject"], ("id", "type", "tenant", "attributes")), "action": request["action"], "resource": ref(request["resource"], ("id", "type", "system", "tenant", "attributes"))} if request.get("context"): material["context"] = _sorted(request["context"]) encoded = json.dumps(material, ensure_ascii=False, separators=(",", ":"), allow_nan=False) for char, escaped in (("<", r"\u003c"), (">", r"\u003e"), ("&", r"\u0026"), ("\u2028", r"\u2028"), ("\u2029", r"\u2029")): encoded = encoded.replace(char, escaped) return "sha256:" + hashlib.sha256(encoded.encode()).hexdigest() def build_request(session: HumanSession, memo, action, policy_version): if action not in ACTIONS: raise ValueError("unsupported review action") return {"id": str(uuid.uuid4()), "tenant": "tenant:platform", "subject": {"id": session.subject, "type": "human", "tenant": session.tenant.value, "attributes": {"tenant_source": session.tenant.route.value, "principal_type_source": session.principal_type.route.value, "assurance": dict(session.assurance), "roles": list(session.roles)}}, "action": action, "resource": {"id": "memo:" + memo.id, "type": "decision-memo", "system": "informed-decision", "tenant": "tenant:platform"}, "context": {"memo_version": memo.version, "approval_id": memo.approval_id, "approval_binding_digest": memo.approval_binding_digest}, "policy_version": policy_version} def _timestamp(value): if not isinstance(value, str): raise ValueError("timestamp required") parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) if parsed.tzinfo is None: raise ValueError("timezone required") return parsed.timestamp() @dataclass(frozen=True) class Observation: request_json: str decision_json: str | None outcome: str usable_until: float response_sha256: str | None def require_current(self, now): if self.outcome != "allow": raise PolicyError(self.outcome) if now >= self.usable_until: raise PolicyError("decision_expired") class PolicyClient: def __init__(self, origin, token_provider, *, package, version, package_digest, transport=None, clock=time.time, allow_internal_http=False): self.origin = fixed_origin(origin, allow_internal_http=allow_internal_http) if (not isinstance(package, str) or not package or not isinstance(version, str) or not version or not isinstance(package_digest, str) or not DIGEST.fullmatch(package_digest)): raise ValueError("exact policy package, version and digest pins are required") self.package, self.version, self.package_digest = package, version, package_digest self.token_provider = token_provider self.transport = transport or JSONTransport(allow_internal_http=allow_internal_http) self.clock = clock def check(self, request): # Round-trip before dispatch: the observation always records the exact # submitted object, even if another caller mutates its input later. raw = json.dumps(request, ensure_ascii=False, separators=(",", ":"), allow_nan=False) request = json.loads(raw) submitted = submitted_digest(request) started = self.clock() try: token = self.token_provider() # Workload credential, NOT the browser token. if (not isinstance(token, str) or not token or not token.isascii() or len(token) > 32768 or any(ord(c) < 33 or ord(c) > 126 for c in token)): raise ValueError() except (OSError, ValueError): return Observation(raw, None, "caller_unavailable", 0, None) try: status, data = self.transport.request("POST", self.origin + "/v1/check", headers={"Authorization": "Bearer " + token, "Content-Type": "application/json"}, body=raw.encode()) except TransportError: return Observation(raw, None, "policy_unavailable", 0, None) # Retain no diagnostic/reason text, including on transport/auth failures. try: response_digest = "sha256:" + hashlib.sha256(json.dumps(data, sort_keys=True, ensure_ascii=False, separators=(",", ":"), allow_nan=False).encode()).hexdigest() except (TypeError, ValueError, UnicodeError): return Observation(raw, None, "invalid_decision", 0, None) if status != 200: return Observation(raw, None, "caller_refused" if status in (401, 403) else "policy_unavailable", 0, response_digest) try: if data.get("contract_version") != CONTRACT or data.get("request_id") != request["id"]: raise ValueError() if not isinstance(data.get("id"), str) or not data["id"] or len(data["id"]) > 256: raise ValueError() binding, provenance = data["binding"], data["provenance"] if (binding["submitted_request_digest"] != submitted or not DIGEST.fullmatch(binding["request_digest"]) or binding["tenant"] != request["tenant"] or binding["action"] != request["action"] or binding.get("context", {}) != request.get("context", {})): raise ValueError() # Registry may enrich attributes; core actor/resource identity must # still name this request. Never reproduce registry enrichment. for key, names in (("subject", ("id", "type", "tenant")), ("resource", ("id", "type", "system", "tenant"))): for name in names: if binding[key].get(name) != request[key].get(name) or data[key].get(name) != binding[key].get(name): raise ValueError() if (provenance["policy_package"] != self.package or provenance["policy_version"] != self.version or provenance["policy_package_digest"] != self.package_digest or data.get("matched_policy_version") != self.version or not DIGEST.fullmatch(provenance["registry_snapshot_digest"]) or not isinstance(provenance.get("evaluator"), str) or not provenance["evaluator"].startswith("flex-auth/")): raise ValueError() now = self.clock() decided = _timestamp(provenance["decision_time"]) if decided > now + 30 or decided < started - 30: raise ValueError() safe = {k: data[k] for k in ("id", "contract_version", "request_id", "effect", "matched_policy_version", "subject", "resource", "binding", "provenance")} effect = data.get("effect") if effect not in {"allow", "deny", "redact", "audit_only", "not_applicable"}: raise ValueError() until = 0 outcome = "policy_denied" if effect == "deny" else "unsupported_effect" if effect == "allow": life = data["lifetime"] until = min(_timestamp(life["expires_at"]), started + 30) if life["kind"] != "ttl" or now >= until or _timestamp(life.get("not_before", provenance["decision_time"])) > now: raise ValueError() safe["lifetime"] = life outcome = "allow" if data.get("obligations", []) == [] else "unsupported_obligations" return Observation(raw, json.dumps(safe, ensure_ascii=False, separators=(",", ":")), outcome, until, response_digest) except (KeyError, TypeError, ValueError, AttributeError): return Observation(raw, None, "invalid_decision", 0, response_digest)