tegwick
f3b71b1521
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a10780-6d41-7952-9718-dfbc1f41236f
224 lines
13 KiB
Python
224 lines
13 KiB
Python
"""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, 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
|
|
|
|
|
|
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 set(row) != {"submitter_id", "memo_classes", "review_groups", "daily_limit"}
|
|
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 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
|
|
|
|
|
|
class IntakeService:
|
|
def __init__(self, store, authenticator, creator, *, submitters_path, groups_path):
|
|
self.store, self.authenticator, self.creator = store, authenticator, creator
|
|
self.submitters_path, self.groups_path = submitters_path, groups_path
|
|
|
|
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"]))
|
|
submission = validate_submission(envelope, submitter_id=grant.submitter_id, grant=grant)
|
|
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}
|
|
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"])
|
|
else:
|
|
raise IntakeError(409, "approval_create_unresolved")
|
|
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")
|
|
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:
|
|
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"])
|
|
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
|
|
|
|
def __call__(self, environ, start_response):
|
|
try:
|
|
if environ.get("PATH_INFO") != "/v1/memos":
|
|
raise IntakeError(404, "not_found")
|
|
if environ.get("REQUEST_METHOD") != "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 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 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]
|