llm-ingestion-okf/tests/test_read_path_containment.py

"""Every read path into a bundle is contained, not just the one that fetches.

`okf_fetch` has resolved a concept through `connectors.safe_resolve` since the
server was written. The other three ways into the same bytes did not: `okf
consume` and `okf_ask` reach `consume.build_payload`, and `okf_describe`
reaches `mcp_server.card`, and both built the concept path by joining the
bundle-relative name onto the root. `consume._join` refuses a `..` segment and
an absolute target, but it is a STRING rule -- a symlink is a fact about the
filesystem, and no amount of reading the index text can see one.

So the index could name `leak.md`, that name could be a link to a file outside
the bundle, and the file's contents came back in the answer. The pair of checks
the module docstring promises held for one tool out of three.

The named pipe is the same check and the other failure mode: a path that is
neither a regular file nor a refusal is a path that blocks the server for as
long as nobody writes to it. Those cases run in a SUBPROCESS with a deadline,
because a test that hangs is not a red test.
"""

from __future__ import annotations

import json
import os
import subprocess
import sys
import textwrap
from pathlib import Path

import pytest

from llm_ingestion_okf import consume, mcp_server

TOOLS = Path(__file__).resolve().parents[1] / "tools"
sys.path.insert(0, str(TOOLS))

import okf_mcp_gate as gate  # noqa: E402

#: A token that occurs NOWHERE in the bundle, only in the file outside it. An
#: answer carrying it read something the bundle never held.
SECRET = "SEKRETMARKOER"

QUESTION = f"hva sier notatet om {SECRET.lower()} og krav"


def _outside_concept(path: Path) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(
        gate.CONCEPT.format(
            title=f"Hemmelig notat {SECRET}",
            source="hemmelig.txt",
            bundle_id="not-this-bundle",
            segment="s9",
            body=f"Dette notatet ligger UTENFOR samlingen og inneholder {SECRET}.",
        ),
        encoding="utf-8",
    )


def _bundle_with_escape(tmp_path: Path, *, link: str) -> Path:
    """A bundle whose index names one honest concept and one escaping link.

    `link` is `"file"` (a link to a file outside) or `"directory"` (a link to a
    directory outside, with the index naming a file through it).
    """
    outside = tmp_path / "outside"
    _outside_concept(outside / "hemmelig.md")
    root = tmp_path / "root"
    gate.write_bundle(root, "escape-bundle", [("krav", "Krav", "Et krav om krav og notat.")])
    if link == "file":
        (root / "lekkasje.md").symlink_to(outside / "hemmelig.md")
        entry = "- [Lekkasje](lekkasje.md) — adjudication: proposed"
    else:
        (root / "lenket").symlink_to(outside, target_is_directory=True)
        entry = "- [Lekkasje](lenket/hemmelig.md) — adjudication: proposed"
    index = root / "index.md"
    index.write_text(index.read_text(encoding="utf-8") + entry + "\n", encoding="utf-8")
    return root


def _surface(root: Path) -> mcp_server.Surface:
    return mcp_server.build_surface(bundle=root, roots=())


def _refusal(root: Path, tool: str, arguments: dict[str, object]) -> str:
    """The envelope a client actually sees: `refused (<code>): <message>`."""
    result = mcp_server.handle(_surface(root), "tools/call", {"name": tool, "arguments": arguments})
    text = result["content"][0]["text"]
    assert result["isError"], f"{tool} answered instead of refusing: {text[:400]}"
    return text


@pytest.mark.parametrize("link", ["file", "directory"])
def test_okf_consume_refuses_a_symlink_out_of_the_bundle(tmp_path: Path, link: str) -> None:
    root = _bundle_with_escape(tmp_path, link=link)
    with pytest.raises(consume.ConsumeError) as raised:
        consume.build_payload(root, question=QUESTION, k=8, limit=120_000)
    assert raised.value.code == "path_escape"
    assert SECRET not in str(raised.value)


@pytest.mark.parametrize("link", ["file", "directory"])
def test_okf_ask_refuses_a_symlink_out_of_the_bundle(tmp_path: Path, link: str) -> None:
    root = _bundle_with_escape(tmp_path, link=link)
    text = _refusal(root, "okf_ask", {"question": QUESTION})
    assert "path_escape" in text
    assert SECRET not in text


@pytest.mark.parametrize("link", ["file", "directory"])
def test_okf_describe_refuses_a_symlink_out_of_the_bundle(tmp_path: Path, link: str) -> None:
    root = _bundle_with_escape(tmp_path, link=link)
    text = _refusal(root, "okf_describe", {})
    assert "path_escape" in text
    assert SECRET not in text


@pytest.mark.parametrize("link", ["file", "directory"])
def test_okf_fetch_refuses_the_same_link_and_is_the_control(tmp_path: Path, link: str) -> None:
    """The arm that was already contained. Without it, a suite that refused
    everything for some unrelated reason would look like a fix."""
    root = _bundle_with_escape(tmp_path, link=link)
    concept = "lekkasje" if link == "file" else "lenket/hemmelig"
    text = _refusal(root, "okf_fetch", {"concept_id": concept})
    assert "path_escape" in text
    assert SECRET not in text


def test_the_honest_concept_of_that_same_bundle_is_still_delivered(tmp_path: Path) -> None:
    """The known-positive for all four rows above: the refusal is the LINK's,
    never the bundle's. A rule that refused every bundle holding a link would
    pass every assertion above and destroy the tool."""
    root = tmp_path / "root"
    gate.write_bundle(root, "clean-bundle", [("krav", "Krav", "Et krav om krav og notat.")])
    payload = consume.build_payload(root, question=QUESTION, k=8, limit=120_000)
    assert payload["excerpts"], "the control question delivered nothing"
    card = mcp_server.card(root, profile=consume.DEFAULT_PROFILE)
    assert card["concept_count"] == 1


# --------------------------------------------------------------------------
# The named pipe, and the dead index link, both in a subprocess with a deadline
# --------------------------------------------------------------------------

_DRIVER = """
import json, sys
from pathlib import Path
from llm_ingestion_okf import consume, mcp_server

root = Path(sys.argv[1])
what = sys.argv[2]
codes = {}
try:
    consume.build_payload(root, question="krav notat", k=8, limit=120_000)
    codes["consume"] = "ANSWERED"
except consume.ConsumeError as error:
    codes["consume"] = error.code
surface = mcp_server.build_surface(bundle=root, roots=())
for tool, arguments in (("okf_describe", {}), ("okf_ask", {"question": "krav notat"})):
    result = mcp_server.handle(surface, "tools/call", {"name": tool, "arguments": arguments})
    codes[tool] = result["content"][0]["text"] if result.get("isError") else "ANSWERED"
print(json.dumps(codes))
"""


def _drive(root: Path, what: str) -> dict[str, str]:
    """Run the three read paths out of process, with a deadline.

    A named pipe with no writer blocks the reader forever, so an in-process
    assertion would hang the suite instead of failing it. `TimeoutExpired`
    surfaces here as the failure it is.
    """
    run = subprocess.run(
        [sys.executable, "-c", textwrap.dedent(_DRIVER), str(root), what],
        capture_output=True,
        text=True,
        timeout=60,
    )
    assert run.returncode == 0, f"driver crashed: {run.stderr[-2000:]}"
    return json.loads(run.stdout.strip().splitlines()[-1])


def test_a_named_pipe_in_the_index_is_refused_and_never_read(tmp_path: Path) -> None:
    root = tmp_path / "root"
    gate.write_bundle(root, "pipe-bundle", [("krav", "Krav", "Et krav om krav og notat.")])
    os.mkfifo(root / "roer.md")
    index = root / "index.md"
    index.write_text(
        index.read_text(encoding="utf-8") + "- [Roer](roer.md) — adjudication: proposed\n",
        encoding="utf-8",
    )
    codes = _drive(root, "fifo")
    assert codes["consume"] == "path_escape", codes
    assert "path_escape" in codes["okf_describe"], codes
    assert "path_escape" in codes["okf_ask"], codes


def test_a_dead_index_link_is_refused_by_code_and_concept_id_not_by_path(tmp_path: Path) -> None:
    """The refusal names what the client can act on.

    A `FileNotFoundError` escaping into the envelope wrote the SERVER's absolute
    path into an answer -- the reader's own directory layout, handed to whoever
    asked, over one index entry naming a file nobody wrote.
    """
    root = tmp_path / "root"
    gate.write_bundle(root, "dead-bundle", [("krav", "Krav", "Et krav om krav og notat.")])
    index = root / "index.md"
    index.write_text(
        index.read_text(encoding="utf-8") + "- [Borte](borte.md) — adjudication: proposed\n",
        encoding="utf-8",
    )
    codes = _drive(root, "dead")
    for key in ("okf_describe", "okf_ask"):
        assert "borte" in codes[key], codes[key]
        assert str(root) not in codes[key], codes[key]
        assert str(tmp_path) not in codes[key], codes[key]
    assert codes["consume"] != "ANSWERED", codes