"""Separate workload-only WSGI intake. No browser session or binding route. The creator and TokenReview transports are injected by the workload composition. An uncertain create is never automatically repeated: its durable reservation requires receipt recovery until approval-engine supports idempotent creation. """ from dataclasses import replace from datetime import datetime, timezone import hashlib import json from pathlib import Path import re import uuid from .intake import (IntakeRefused, PresentExisting, SubmitterGrant, canonical, idempotency_outcome, validate_submission) from .memo import Principal from .records import dumps from .store import Conflict AUDIENCE = "informed-decision-intake" MAX_BODY = 1024 * 1024 ROW_KEYS = {"submitter_id", "memo_classes", "review_groups", "daily_limit"} PRESENT_EXISTING_KEYS = {"approval_actor", "actions"} _ACTOR_RE = re.compile(r"service:[a-z0-9][a-z0-9-]{1,62}") _ACTION_RE = re.compile(r"[a-z0-9][a-z0-9-]{1,62}") _OUTCOME_PATH = re.compile(r"/v1/memos/([a-z0-9][a-z0-9-]{2,62})/outcome") class IntakeError(Exception): def __init__(self, status, code): self.status, self.code = status, code super().__init__(code) class TokenReview: """call(body) performs an authenticated Kubernetes TokenReview using pod credentials. The adapter must verify the API server's CA and suppress tokens and responses in its logs. TokenReview is performed afresh on every pose, including replay. """ def __init__(self, call): self.call = call def authenticate(self, token): try: result = self.call({"apiVersion": "authentication.k8s.io/v1", "kind": "TokenReview", "spec": {"token": token, "audiences": [AUDIENCE]}}) status = result["status"] user = status["user"] username = user["username"] if (status.get("authenticated") is not True or status.get("error") or AUDIENCE not in status.get("audiences", []) or not user.get("uid") or not isinstance(username, str) or not re.fullmatch(r"system:serviceaccount:[a-z0-9-]+:[a-z0-9-]+", username)): raise ValueError() return username except Exception: raise IntakeError(401, "workload_authentication_failed") from None def configuration(submitters_path, groups_path): """Reload reviewed deployment data on every request so removal revokes immediately. Empty shipped registries deliberately admit no workload or human recipient. Entries identify actual service accounts; drafted_by is only an assertion. """ try: submitters = json.loads(Path(submitters_path).read_text()) groups = json.loads(Path(groups_path).read_text()) if not isinstance(submitters, dict) or not isinstance(groups, dict): raise ValueError() for username, row in submitters.items(): if (not re.fullmatch(r"system:serviceaccount:[a-z0-9-]+:[a-z0-9-]+", username) or not isinstance(row, dict) or not ROW_KEYS <= set(row) <= ROW_KEYS | {"present_existing", "outcome_read"} or ("outcome_read" in row and (row["outcome_read"] is not True or "present_existing" in row)) or not isinstance(row["submitter_id"], str) or not row["submitter_id"] or type(row["daily_limit"]) is not int or row["daily_limit"] < 1): raise ValueError() for field in ("memo_classes", "review_groups"): if not isinstance(row[field], list) or any(not isinstance(x, str) for x in row[field]): raise ValueError() if "present_existing" in row: profile = row["present_existing"] if (not isinstance(profile, dict) or set(profile) != PRESENT_EXISTING_KEYS or not isinstance(profile["approval_actor"], str) or not _ACTOR_RE.fullmatch(profile["approval_actor"]) or profile["approval_actor"] == "service:informed-decision" or not isinstance(profile["actions"], list) or not profile["actions"] or any(not isinstance(a, str) or not _ACTION_RE.fullmatch(a) for a in profile["actions"])): raise ValueError() if len({r["submitter_id"] for r in submitters.values()}) != len(submitters): raise ValueError() for row in groups.values(): if (set(row) != {"subject", "display_name"} or not all(isinstance(x, str) and x for x in row.values())): raise ValueError() return submitters, groups except Exception: raise IntakeError(503, "intake_configuration_unavailable") from None def verify_existing(approval, submission, *, now=None): """present_existing (INFD-WP-0010-T01): may this existing approval be presented? Every check must hold; the first failure is the refusal code. Nothing here decides whether the act is permitted: it only refuses to present an object whose binding is not exactly the one the memo names, or that a human entry could no longer complete. """ memo = submission.memo expected = submission.approval_binding now = now or datetime.now(timezone.utc) if not isinstance(approval, dict) or approval.get("id") != memo.approval_id: return "approval_mismatch" binding = approval.get("binding") if not isinstance(binding, dict): return "approval_mismatch" if approval.get("status") != "requested": return "approval_not_requested" if approval.get("entries") != []: return "approval_has_entries" if binding.get("human_control") is not True: return "approval_not_human_controlled" if binding.get("actor") != expected["actor"]: return "approval_actor_mismatch" if binding.get("digest") != memo.approval_binding_digest: return "approval_digest_mismatch" if any(binding.get(k) != expected[k] for k in ("action", "principal", "purpose", "target")): return "approval_binding_mismatch" if approval.get("required_count") != 1: return "approval_required_count_unsupported" try: expires = datetime.fromisoformat(approval["validity"]["expires_at"].replace("Z", "+00:00")) if expires.tzinfo is None or expires <= now: return "approval_expired" except (KeyError, TypeError, AttributeError, ValueError): return "approval_expired" return None class IntakeService: def __init__(self, store, authenticator, creator, *, submitters_path, groups_path, reader=None, issuer=None): """``reader(approval_id) -> dict`` is a read-only GET used only by present_existing. ``issuer`` names the human sessions' OIDC issuer in outcome records; without it the outcome read is unavailable. """ self.store, self.authenticator, self.creator = store, authenticator, creator self.submitters_path, self.groups_path = submitters_path, groups_path self.reader, self.issuer = reader, issuer def outcome(self, token, memo_id): """INFD-WP-0010-T04: the answer to one memo this submitter posed, for the hub's own pull. Only rows with ``outcome_read: true`` (reviewed deployment data: State Hub's dispatch worker) may read, and only memos they posed. Read-only; no budget is spent. """ username = self.authenticator.authenticate(token) submitters, _ = configuration(self.submitters_path, self.groups_path) row = submitters.get(username) if row is None: raise IntakeError(403, "submitter_unknown") if row.get("outcome_read") is not True: raise IntakeError(403, "outcome_read_not_admitted") if self.issuer is None: raise IntakeError(503, "outcome_unavailable") from .outcome import outcome_for return outcome_for(self.store, row["submitter_id"], memo_id, issuer=self.issuer) def pose(self, token, envelope): username = self.authenticator.authenticate(token) submitters, groups = configuration(self.submitters_path, self.groups_path) row = submitters.get(username) if row is None: raise IntakeError(403, "submitter_unknown") grant = SubmitterGrant(row["submitter_id"], frozenset(row["memo_classes"]), frozenset(row["review_groups"])) profile = None if "present_existing" in row: profile = PresentExisting(row["present_existing"]["approval_actor"], frozenset(row["present_existing"]["actions"])) submission = validate_submission(envelope, submitter_id=grant.submitter_id, grant=grant, present_existing=profile) if profile is not None and self.reader is None: raise IntakeError(503, "present_existing_unavailable") recipient = groups.get(submission.review_group) if recipient is None: raise IntakeError(503, "review_group_unconfigured") metadata = {"memo_class": submission.memo_class, "review_group": submission.review_group, "drafted_by": submission.drafted_by, "serves": submission.serves} if profile is not None: # approval_action: the verified binding.action, for the PDP (compact-sitting v14). metadata.update(profile="present_existing", approval_requester=profile.approval_actor, approval_action=submission.approval_binding["action"]) recovered = None # Reserve before any network side effect. SQLite serializes concurrent poses. with self.store._transaction() as db: from .lifecycle_review import check_intake check_intake(db, envelope, submission.submitter_id) previous = db.execute("SELECT * FROM intake_ledger WHERE submitter=? AND key=?", (submission.submitter_id, submission.idempotency_key)).fetchone() if previous: idempotency_outcome(previous["digest"], submission.submission_digest) if previous["state"] == "ready": return 200, json.loads(previous["receipt"]) if previous["state"] == "created": recovered = json.loads(previous["receipt"]) elif profile is None: raise IntakeError(409, "approval_create_unresolved") # present_existing has no uncertain side effect: an interrupted read is re-verified. if not previous and (db.execute("SELECT 1 FROM memos WHERE id=?", (submission.memo.id,)).fetchone() or db.execute("SELECT 1 FROM intake_ledger WHERE memo_id=?", (submission.memo.id,)).fetchone()): raise IntakeError(409, "memo_id_taken") # Budget counts attempts as well as completed poses; retries do not spend twice. day = datetime.now(timezone.utc).date().isoformat() count = db.execute("SELECT COUNT(*) FROM intake_ledger WHERE submitter=? AND json_extract(metadata,'$.day')=?", (submission.submitter_id, day)).fetchone()[0] if not previous and count >= row["daily_limit"]: raise IntakeError(429, "submitter_budget_exhausted") if profile is not None and not previous: self._refuse_presented(db, submission.memo.approval_id) metadata["day"] = day if not previous: db.execute("INSERT INTO intake_ledger VALUES (?,?,?,?,?,?,NULL)", (submission.submitter_id, submission.idempotency_key, submission.submission_digest, submission.memo.id, "creating", dumps(metadata))) if recovered is None and profile is not None: # Read-only. No create, entry or consume is ever sent in this path. try: approval = self.reader(submission.memo.approval_id) except Exception: approval = None refusal = "approval_unreadable" if approval is None else verify_existing(approval, submission) if refusal is not None: # No engine side effect happened: release the reservation. with self.store._transaction() as db: db.execute("DELETE FROM intake_ledger WHERE submitter=? AND key=? AND state='creating'", (submission.submitter_id, submission.idempotency_key)) raise IntakeError(503 if refusal == "approval_unreadable" else 409, refusal) approval_id, digest = submission.memo.approval_id, submission.memo.approval_binding_digest receipt = {"memo_id": submission.memo.id, "approval_id": approval_id, "approval_binding_digest": digest} with self.store._transaction() as db: db.execute("UPDATE intake_ledger SET state='created',receipt=? WHERE submitter=? AND key=?", (dumps(receipt), submission.submitter_id, submission.idempotency_key)) elif recovered is None: try: approval = self.creator({"binding": submission.approval_binding, "human_control": True, "required_count": 1}) binding = approval["binding"] approval_id, digest = approval["id"], binding["digest"] if (not re.fullmatch(r"[0-9a-f]{8}(-[0-9a-f]{4}){3}-[0-9a-f]{12}", approval_id) or not re.fullmatch(r"sha256:[0-9a-f]{64}", digest) or binding.get("human_control") is not True or any(binding.get(k) != v for k, v in submission.approval_binding.items()) or approval.get("status") != "requested" or approval.get("entries")): raise ValueError() except Exception: raise IntakeError(503, "approval_create_unresolved") from None # Preserve the receipt before local load: recovery can locate this engine object. receipt = {"memo_id": submission.memo.id, "approval_id": approval_id, "approval_binding_digest": digest} with self.store._transaction() as db: db.execute("UPDATE intake_ledger SET state='created',receipt=? WHERE submitter=? AND key=?", (dumps(receipt), submission.submitter_id, submission.idempotency_key)) else: receipt = recovered approval_id, digest = receipt["approval_id"], receipt["approval_binding_digest"] memo = replace(submission.memo, approval_id=approval_id, approval_binding_digest=digest, binding=replace(submission.memo.binding, principal=Principal( recipient["subject"], "person", recipient["display_name"]))) with self.store._transaction() as db: check_intake(db, envelope, submission.submitter_id) current = db.execute("SELECT state,receipt FROM intake_ledger WHERE submitter=? AND key=?", (submission.submitter_id, submission.idempotency_key)).fetchone() if current["state"] == "ready": return 200, json.loads(current["receipt"]) if profile is not None: self._refuse_presented(db, approval_id, except_key=(submission.submitter_id, submission.idempotency_key)) for item in envelope["packet"]: data = submission.packet[item["item_id"]] document_digest = "sha256:" + hashlib.sha256(data).hexdigest() existing = db.execute("SELECT * FROM documents WHERE digest=?", (document_digest,)).fetchone() if existing and (existing["media_type"] != item["media_type"] or existing["content"] != data): raise IntakeError(409, "document_conflict") db.execute("INSERT OR IGNORE INTO documents VALUES (?,?,?)", (document_digest, item["media_type"], data)) db.execute("INSERT INTO memos VALUES (?,?,?)", (memo.id, memo.version, dumps(memo))) event_id = str(uuid.uuid4()) event = {"id": event_id, "type": "informed-decision.memo.posed", "source": "informed-decision", "tenant": "tenant:platform", "subject": memo.id, "correlation_id": approval_id, "occurred_at": datetime.now(timezone.utc).isoformat(), "data": {**receipt, "submitter": submission.submitter_id, "memo_class": submission.memo_class, "submission_digest": submission.submission_digest}} notification = {"memo_id": memo.id, "question": memo.question, "memo_class": submission.memo_class, "submitter": submission.submitter_id} for kind, payload in (("memo-posed", event), ("notification", notification)): db.execute("INSERT INTO intake_deliveries(id,kind,payload) VALUES (?,?,?)", (event_id if kind == "memo-posed" else str(uuid.uuid4()), kind, dumps(payload))) db.execute("UPDATE intake_ledger SET state='ready' WHERE submitter=? AND key=?", (submission.submitter_id, submission.idempotency_key)) return 201, receipt @staticmethod def _refuse_presented(db, approval_id, except_key=None): """One existing approval is presented by at most one memo.""" rows = db.execute("SELECT submitter,key FROM intake_ledger WHERE json_extract(receipt,'$.approval_id')=?", (approval_id,)).fetchall() if any((r["submitter"], r["key"]) != except_key for r in rows): raise IntakeError(409, "approval_already_presented") if db.execute("SELECT 1 FROM memos WHERE json_extract(body,'$.approval_id')=?", (approval_id,)).fetchone(): raise IntakeError(409, "approval_already_presented") def __call__(self, environ, start_response): from .outcome import OutcomeError try: path = environ.get("PATH_INFO", "") outcome = _OUTCOME_PATH.fullmatch(path) if path != "/v1/memos" and outcome is None: raise IntakeError(404, "not_found") if environ.get("REQUEST_METHOD") != ("GET" if outcome else "POST"): raise IntakeError(405, "method_not_allowed") authorization = environ.get("HTTP_AUTHORIZATION", "") if not authorization.startswith("Bearer ") or not authorization[7:] or len(authorization) > 16384: raise IntakeError(401, "workload_token_required") if outcome is not None: status, body = self.outcome(authorization[7:], outcome.group(1)) else: if environ.get("CONTENT_TYPE", "").split(";")[0] != "application/json": raise IntakeError(415, "json_required") length = int(environ.get("CONTENT_LENGTH") or 0) if length < 1 or length > MAX_BODY: raise IntakeError(413, "submission_too_large") raw = environ["wsgi.input"].read(length) if len(raw) != length: raise IntakeError(400, "invalid_json") status, body = self.pose(authorization[7:], json.loads(raw)) except (IntakeError, OutcomeError) as exc: status, body = exc.status, {"error": exc.code} except Conflict: status, body = 409, {"error": "review_handoff_conflict"} except IntakeRefused as exc: status = 409 if exc.reason == "idempotency_conflict" else 400 body = {"error": exc.reason} # No client-controlled field paths in logs/responses. except (ValueError, TypeError, KeyError, UnicodeError): status, body = 400, {"error": "invalid_submission"} except Exception: status, body = 503, {"error": "intake_unavailable"} from http import HTTPStatus raw = canonical(body) start_response(f"{status} {HTTPStatus(status).phrase}", [("Content-Type", "application/json"), ("Content-Length", str(len(raw))), ("Cache-Control", "no-store")]) return [raw]