llm-ingestion-pipeline-security/tests/inbox_frontend.py

"""OKF inbox front-end — stage 1 of the two-stage upload showcase (PLAN §247).

Reads the files a human actually drops into an inbox and *materializes* them into
an OKF bundle ``{concept_path: document_text}`` with provenance, so the stage-2
guard (:func:`llm_ingestion_guard.okf.import_bundle`) can validate every concept.
This stage owns the container/format threats; the guard owns the text/structural
threats.

**Placement.** This lives in the test tree, not ``src/``: the extraction parsers
are showcase/dev-scoped (``python-docx``/``python-pptx`` in the ``dev`` extra,
never core ``dependencies``), and the shippable core stays stdlib-only. It is an
in-repo demonstration a consumer reads and adapts, not v1 shipped code.

Slice 2a covers the text formats — ``.txt`` and ``.md`` (stdlib only). ``.zip``
(zip-slip / zip-bomb), ``.csv`` (formula injection), folders, ``.docx`` and
``.pptx`` land in later slices.
"""
from __future__ import annotations

from dataclasses import dataclass
from pathlib import Path

from llm_ingestion_guard.okf import import_bundle, Origin, Channel, BundleResult
from llm_ingestion_guard.disposition import Disposition


# Aggregate disposition -> inbox verdict. Fail-secure by default: a front-end
# refusal (a container threat the guard never sees) is itself a REJECT.
_VERDICT = {
    Disposition.WARN: "ADMIT",
    Disposition.QUARANTINE_REVIEW: "HOLD",
    Disposition.FAIL_SECURE: "REJECT",
}

# Where materialized uploads live inside the bundle. An upload named ``index.*``
# thus lands on the reserved ``uploads/index.md`` and is refused by the path gate.
_MATERIALIZE_PREFIX = "uploads"

# The text formats this slice reads directly (no parser dependency).
_TEXT_SUFFIXES = {".txt", ".md"}


@dataclass(frozen=True)
class Provenance:
    """Where one materialized concept came from — the audit record (brief §6)."""

    concept_path: str
    source_name: str
    source_type: str


@dataclass(frozen=True)
class InboxExtract:
    """The stage-1 output: the OKF bundle, its provenance, and front-end refusals.

    ``rejected`` holds ``(source_name, reason)`` for drops the front-end refuses
    outright (a container threat the guard never gets to see, e.g. a zip bomb).
    Empty for the text-format slice.
    """

    bundle: dict
    provenance: tuple
    rejected: tuple


def _materialize_path(source_name: str) -> str:
    """Assign an OKF concept path to a dropped file: ``<prefix>/<stem>.md``."""
    stem = Path(source_name).stem
    return f"{_MATERIALIZE_PREFIX}/{stem}.md"


def extract_inbox(paths) -> InboxExtract:
    """Read dropped files and materialize them into an OKF bundle + provenance.

    ``paths`` is an iterable of file paths. Each ``.txt`` / ``.md`` becomes one
    concept; a ``.md`` keeps its OKF frontmatter verbatim, a ``.txt`` becomes a
    body-only concept. Non-text suffixes are not handled in this slice.
    """
    bundle: dict = {}
    provenance: list = []
    for path in paths:
        path = Path(path)
        suffix = path.suffix.lower()
        if suffix not in _TEXT_SUFFIXES:
            raise ValueError(f"unsupported upload format in this slice: {path.name!r}")
        concept_path = _materialize_path(path.name)
        bundle[concept_path] = path.read_text(encoding="utf-8")
        provenance.append(
            Provenance(concept_path, path.name, suffix.lstrip("."))
        )
    return InboxExtract(bundle, tuple(provenance), ())


def receive(paths) -> tuple[InboxExtract, BundleResult, str]:
    """The full two-stage inbox: extract & materialize, then guard, then verdict.

    Returns ``(extracted, guard_result, verdict)``. A front-end refusal forces a
    REJECT regardless of the guard's aggregate — the guard never saw that drop.
    """
    extracted = extract_inbox(paths)
    result = import_bundle(extracted.bundle, origin=Origin.EXTERNAL, channel=Channel.AUTOMATIC)
    verdict = "REJECT" if extracted.rejected else _VERDICT[result.disposition]
    return extracted, result, verdict