informed-decision/informed_decision/intake_service.py

"""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}")


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 set(row) not in (ROW_KEYS, ROW_KEYS | {"present_existing"})
                    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):
        """``reader(approval_id) -> dict`` is a read-only GET used only by present_existing."""
        self.store, self.authenticator, self.creator = store, authenticator, creator
        self.submitters_path, self.groups_path = submitters_path, groups_path
        self.reader = reader

    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:
            metadata.update(profile="present_existing", approval_requester=profile.approval_actor)
        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):
        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]