informed-decision/informed_decision/intake.py

"""Memo intake validation (INFD-WP-0005, docs/specs/FrictionlessDecisionIntake.md §3).

Pure and offline: validates a submission envelope from an authenticated
workload and returns the parsed draft plus its submission digest. It creates no
approval, writes no store row and binds nothing. The submitter id comes from
the caller's verified workload identity, never from the body.

Every refusal is an ``IntakeRefused`` with a stable reason code and a field
path. A secret-marker hit never echoes the matched text.
"""
from __future__ import annotations

import hashlib
import json
import re
from dataclasses import dataclass
from typing import Any, Mapping

from .memo import BindingLevel, Memo, StepKind
from .records import memo_from

KIND = "informed-decision.memo-intake"
VERSION = 1
DRAFT_PRINCIPAL = "pending-human-session"
REQUESTER_ACTOR = "informed-decision"
MAX_QUESTION = 400
MAX_PACKET_BYTES = 256 * 1024

ENVELOPE_KEYS = {"kind", "version", "idempotency_key", "memo_class", "review_group",
                 "drafted_by", "serves", "approval_intent", "memo", "packet"}
ENVELOPE_REQUIRED = ENVELOPE_KEYS - {"drafted_by", "serves"}
MEMO_KEYS = {"id", "version", "question", "requested_act", "binding_level", "brief",
             "binding", "step_kind", "packet", "highlights", "locale", "ui_release",
             "approval_id", "approval_binding_digest", "sealed"}
INTENT_KEYS = {"action", "principal", "purpose", "target"}
TARGET_KEYS = {"id", "type", "system"}
HIGHLIGHT_SEVERITIES = {"informational", "warning", "critical"}

_KEY_RE = re.compile(r"^[A-Za-z0-9._:-]{8,128}$")
_MEMO_ID_RE = re.compile(r"^[a-z0-9][a-z0-9-]{2,62}$")
_SLUG_RE = re.compile(r"^[a-z0-9][a-z0-9-]{1,62}$")
_UUID_RE = re.compile(r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$")
_SHA_RE = re.compile(r"^sha256:[0-9a-f]{64}$")
_SLUG_KEY_RE = re.compile(r"^[a-z][a-z0-9_]{0,62}$")

#: Secret markers (§3.3.6). Shapes, not values; deliberately broad.
SECRET_MARKERS: tuple[tuple[str, re.Pattern[str]], ...] = (
    ("pem_private_key", re.compile(r"-----BEGIN [A-Z ]*PRIVATE KEY-----")),
    ("bearer_token", re.compile(r"\bBearer\s+[A-Za-z0-9._~+/=-]{16,}")),
    ("jwt", re.compile(r"\beyJ[A-Za-z0-9_-]{8,}\.eyJ[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,}")),
    ("vault_token", re.compile(r"\b(hvs|hvb|hvr)\.[A-Za-z0-9_-]{20,}")),
    ("provider_key", re.compile(r"\b(sk-(or-)?[A-Za-z0-9_-]{20,}|ghp_[A-Za-z0-9]{30,}|"
                                r"github_pat_[A-Za-z0-9_]{30,}|glpat-[A-Za-z0-9_-]{20,}|"
                                r"AKIA[0-9A-Z]{16}|xox[abpr]-[A-Za-z0-9-]{10,})")),
    ("credential_assignment", re.compile(
        r"(?i)\b(password|passwd|client_secret|secret|api[_-]?key|access[_-]?token|token)"
        r"\s*[:=]\s*['\"]?(?=[A-Za-z0-9_+.=-]*\d)[A-Za-z0-9_+.=-]{12,}(?=['\"\s,;)]|$)")),
    ("url_credentials", re.compile(r"://[^/@\s:]+:[^/@\s]+@")),
)


class IntakeRefused(ValueError):
    def __init__(self, reason: str, path: str = "") -> None:
        super().__init__(f"{reason} at {path or '$'}")
        self.reason = reason
        self.path = path


@dataclass(frozen=True)
class SubmitterGrant:
    """One row of deploy/intake-submitters.json, keyed by verified workload."""

    submitter_id: str
    memo_classes: frozenset[str]
    review_groups: frozenset[str]


@dataclass(frozen=True)
class PresentExisting:
    """Submitter profile ``present_existing`` (INFD-WP-0010-T01).

    The submitter's approval already exists in approval-engine, created by its
    own actor. Intake only verifies and presents it; it never creates one.
    """

    approval_actor: str
    actions: frozenset[str]


@dataclass(frozen=True)
class ValidatedSubmission:
    submitter_id: str
    idempotency_key: str
    submission_digest: str
    memo_class: str
    review_group: str
    drafted_by: str | None
    serves: dict | None
    approval_binding: dict
    memo: Memo
    packet: dict[str, bytes]
    profile: str = "create"


def _closed(value: Any, keys: set[str], required: set[str], path: str) -> dict:
    if not isinstance(value, dict):
        raise IntakeRefused("not_an_object", path)
    extra = set(value) - keys
    if extra:
        raise IntakeRefused("unknown_field", f"{path}.{sorted(extra)[0]}")
    missing = required - set(value)
    if missing:
        raise IntakeRefused("missing_field", f"{path}.{sorted(missing)[0]}")
    return value


def _text(value: Any, path: str, *, max_len: int = 4000, pattern: re.Pattern[str] | None = None) -> str:
    if not isinstance(value, str) or not value.strip() or len(value) > max_len:
        raise IntakeRefused("invalid_text", path)
    if pattern is not None and not pattern.match(value):
        raise IntakeRefused("invalid_format", path)
    return value


def _strings(value: Any, path: str):
    if isinstance(value, str):
        yield path, value
    elif isinstance(value, dict):
        for key, item in value.items():
            yield from _strings(item, f"{path}.{key}")
    elif isinstance(value, list):
        for i, item in enumerate(value):
            yield from _strings(item, f"{path}[{i}]")


def scan_secret_markers(value: Any, path: str = "$") -> None:
    """Refuse on the first secret-shaped string. Reports the path, never the text."""
    for where, text in _strings(value, path):
        for name, pattern in SECRET_MARKERS:
            if pattern.search(text):
                raise IntakeRefused(f"secret_marker:{name}", where)


def check_one_question(question: str, path: str = "$.memo.question") -> None:
    if "\n" in question or "\r" in question:
        raise IntakeRefused("question_multiline", path)
    if len(question) > MAX_QUESTION:
        raise IntakeRefused("question_too_long", path)
    if question.count("?") != 1 or not question.rstrip().endswith("?"):
        raise IntakeRefused("not_one_question", path)


def canonical(value: Any) -> bytes:
    return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False,
                      allow_nan=False).encode("utf-8")


def submission_digest(submitter_id: str, envelope: Mapping[str, Any]) -> str:
    body = {k: v for k, v in envelope.items() if k != "idempotency_key"}
    return "sha256:" + hashlib.sha256(canonical({"submitter": submitter_id, "envelope": body})).hexdigest()


def idempotency_outcome(stored_digest: str | None, digest: str) -> str:
    """``new`` | ``replay`` for the ledger key (submitter, idempotency_key)."""
    if stored_digest is None:
        return "new"
    if stored_digest == digest:
        return "replay"
    raise IntakeRefused("idempotency_conflict", "$.idempotency_key")


def _check_memo(memo: Any, packet_hashes: dict[str, str], *, pinned: bool = False) -> Memo:
    memo = _closed(memo, MEMO_KEYS, MEMO_KEYS - {"locale", "ui_release"}, "$.memo")
    _text(memo["id"], "$.memo.id", max_len=63, pattern=_MEMO_ID_RE)
    if memo["version"] != 1 or isinstance(memo["version"], bool):
        raise IntakeRefused("draft_version_must_be_1", "$.memo.version")
    if memo["sealed"] is not False:
        raise IntakeRefused("draft_must_be_unsealed", "$.memo.sealed")
    if pinned:
        # present_existing: the memo must name the existing approval exactly.
        _text(memo["approval_id"], "$.memo.approval_id", max_len=36, pattern=_UUID_RE)
        _text(memo["approval_binding_digest"], "$.memo.approval_binding_digest", max_len=71, pattern=_SHA_RE)
    else:
        for field in ("approval_id", "approval_binding_digest"):
            if memo[field] is not None:
                raise IntakeRefused("approval_pinned_by_submitter", f"$.memo.{field}")
    check_one_question(_text(memo["question"], "$.memo.question"))
    _text(memo["requested_act"], "$.memo.requested_act", max_len=40, pattern=_SLUG_RE)
    _text(memo["brief"], "$.memo.brief", max_len=20000)
    try:
        BindingLevel(memo["binding_level"])
        StepKind(memo["step_kind"])
    except ValueError:
        raise IntakeRefused("unknown_enum", "$.memo.step_kind") from None
    binding = memo["binding"]
    if not isinstance(binding, dict) or not isinstance(binding.get("principal"), dict):
        raise IntakeRefused("invalid_binding", "$.memo.binding")
    if binding["principal"].get("id") != DRAFT_PRINCIPAL:
        raise IntakeRefused("principal_prebound", "$.memo.binding.principal.id")

    items = memo["packet"]
    if not isinstance(items, list) or not items:
        raise IntakeRefused("packet_required", "$.memo.packet")
    seen: set[str] = set()
    for i, item in enumerate(items):
        item = _closed(item, {"item_id", "label", "hash"}, {"item_id", "label", "hash"}, f"$.memo.packet[{i}]")
        if item["item_id"] in seen:
            raise IntakeRefused("duplicate_packet_item", f"$.memo.packet[{i}]")
        seen.add(item["item_id"])
        if item["item_id"] not in packet_hashes:
            raise IntakeRefused("packet_content_missing", f"$.memo.packet[{i}]")
        if item["hash"] != packet_hashes[item["item_id"]]:
            raise IntakeRefused("packet_digest_mismatch", f"$.memo.packet[{i}].hash")
    orphans = set(packet_hashes) - seen
    if orphans:
        raise IntakeRefused("orphan_packet_content", "$.packet")

    highlights = memo["highlights"]
    if not isinstance(highlights, list):
        raise IntakeRefused("invalid_highlights", "$.memo.highlights")
    ids: set[str] = set()
    work_records: set[tuple] = set()
    for i, h in enumerate(highlights):
        if not isinstance(h, dict) or h.get("id") in ids or not h.get("id"):
            raise IntakeRefused("duplicate_highlight", f"$.memo.highlights[{i}]")
        ids.add(h["id"])
        if h.get("item_id") not in seen:
            raise IntakeRefused("highlight_item_unknown", f"$.memo.highlights[{i}].item_id")
        if h.get("severity", "informational") not in HIGHLIGHT_SEVERITIES:
            raise IntakeRefused("unknown_enum", f"$.memo.highlights[{i}].severity")
        loc = h.get("locator") or {}
        if loc.get("kind") == "work_record":
            work_records.add((loc.get("workplan"), loc.get("task")))
    if len(work_records) > 1:
        raise IntakeRefused("bundled_acts", "$.memo.highlights")
    try:
        return memo_from(memo)
    except (TypeError, ValueError, KeyError):
        raise IntakeRefused("memo_unparseable", "$.memo") from None


def _existing_target(value: Any, path: str) -> dict:
    """An existing approval's target may have any flat shape (BEA uses admission id/digest)."""
    if not isinstance(value, dict) or not 1 <= len(value) <= 8:
        raise IntakeRefused("invalid_target", path)
    for key, item in value.items():
        if not isinstance(key, str) or not _SLUG_KEY_RE.match(key):
            raise IntakeRefused("invalid_target", path)
        _text(item, f"{path}.{key}", max_len=200)
    return dict(value)


def validate_submission(envelope: Any, *, submitter_id: str, grant: SubmitterGrant | None,
                        present_existing: PresentExisting | None = None) -> ValidatedSubmission:
    """Validate one intake envelope for an already-authenticated submitter.

    With a ``present_existing`` profile the memo must pin an existing approval
    and the intent names that approval's binding; the actor is the profile's,
    never the body's. Without it, behaviour is unchanged: the memo must not pin
    an approval and intake will create one as ``informed-decision``.
    """
    if grant is None or grant.submitter_id != submitter_id:
        raise IntakeRefused("submitter_unknown", "$")
    env = _closed(envelope, ENVELOPE_KEYS, ENVELOPE_REQUIRED, "$")
    if env["kind"] != KIND or env["version"] != VERSION:
        raise IntakeRefused("unsupported_kind", "$.kind")
    key = _text(env["idempotency_key"], "$.idempotency_key", max_len=128, pattern=_KEY_RE)
    memo_class = _text(env["memo_class"], "$.memo_class", max_len=63, pattern=_SLUG_RE)
    review_group = _text(env["review_group"], "$.review_group", max_len=63, pattern=_SLUG_RE)
    drafted_by = env.get("drafted_by")
    if drafted_by is not None:
        _text(drafted_by, "$.drafted_by", max_len=200)

    # Secret scan before anything is parsed further or echoed anywhere.
    scan_secret_markers(env)

    serves = env.get("serves")
    if serves is not None:
        serves = _closed(serves, {"hub_decision_id", "resolves"}, {"hub_decision_id", "resolves"}, "$.serves")
        _text(serves["hub_decision_id"], "$.serves.hub_decision_id", max_len=36, pattern=_UUID_RE)
        if serves["resolves"] not in ("on-answer", "never"):
            raise IntakeRefused("unknown_enum", "$.serves.resolves")

    intent = _closed(env["approval_intent"], INTENT_KEYS, INTENT_KEYS, "$.approval_intent")
    for field in ("action", "principal", "purpose"):
        _text(intent[field], f"$.approval_intent.{field}", max_len=400)
    if present_existing is not None:
        target = _existing_target(intent["target"], "$.approval_intent.target")
        if intent["action"] not in present_existing.actions:
            raise IntakeRefused("action_not_admitted", "$.approval_intent.action")
    else:
        target = _closed(intent["target"], TARGET_KEYS, TARGET_KEYS, "$.approval_intent.target")
        for field in TARGET_KEYS:
            _text(target[field], f"$.approval_intent.target.{field}", max_len=200)

    packet = env["packet"]
    if not isinstance(packet, list) or not packet:
        raise IntakeRefused("packet_required", "$.packet")
    contents: dict[str, bytes] = {}
    hashes: dict[str, str] = {}
    for i, item in enumerate(packet):
        item = _closed(item, {"item_id", "media_type", "content"}, {"item_id", "media_type", "content"}, f"$.packet[{i}]")
        if item["media_type"] not in ("text/markdown", "text/plain", "application/json"):
            raise IntakeRefused("unsupported_media_type", f"$.packet[{i}].media_type")
        if not isinstance(item["content"], str):
            raise IntakeRefused("invalid_text", f"$.packet[{i}].content")
        data = item["content"].encode("utf-8")
        if len(data) > MAX_PACKET_BYTES:
            raise IntakeRefused("packet_too_large", f"$.packet[{i}]")
        if item["item_id"] in contents:
            raise IntakeRefused("duplicate_packet_item", f"$.packet[{i}]")
        contents[item["item_id"]] = data
        hashes[item["item_id"]] = "sha256:" + hashlib.sha256(data).hexdigest()

    memo = _check_memo(env["memo"], hashes, pinned=present_existing is not None)

    if memo_class not in grant.memo_classes:
        raise IntakeRefused("memo_class_not_admitted", "$.memo_class")
    if review_group not in grant.review_groups:
        raise IntakeRefused("review_group_not_admitted", "$.review_group")

    return ValidatedSubmission(
        submitter_id=submitter_id,
        idempotency_key=key,
        submission_digest=submission_digest(submitter_id, env),
        memo_class=memo_class,
        review_group=review_group,
        drafted_by=drafted_by,
        serves=dict(serves) if serves else None,
        approval_binding={**{k: intent[k] for k in ("action", "principal", "purpose")},
                          "actor": present_existing.approval_actor if present_existing else REQUESTER_ACTOR,
                          "target": dict(target)},
        memo=memo,
        packet=contents,
        profile="present_existing" if present_existing else "create",
    )