Gitea-Tools/tests/test_issue_956_threat_model.py

"""Validation tooling for the remote-MCP threat model (#956).

#956 requires that "every boundary claim [is] traceable to a file and line
anchor that resolves at the reviewed commit". A prose document cannot enforce
that about itself, and #930 demonstrated the failure mode: its inventory cited
``gitea_mcp_server.py`` anchors generated at ``7bf4f125`` which no longer point
at the described code at ``aad5c8b4``. Nothing failed, because nothing checked.

These tests are that check. They enforce, in both directions:

* every ``file.py:NNN`` anchor cited in the prose is declared in the fixture;
* every declared anchor resolves — the file exists, the line exists, and the
  source line actually contains the substring the fixture claims for it;
* the document's structural obligations (assets, adversaries, boundaries,
  credential rows, the co-residency ruling, and the child mapping) are present
  and internally consistent.

A refactor that shifts a line number therefore breaks the suite instead of
silently rotting the security documentation.
"""

import json
import os
import re
import unittest

REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
DOC_PATH = os.path.join(REPO_ROOT, "docs", "remote-mcp", "threat-model.md")
FIXTURE_PATH = os.path.join(
    REPO_ROOT, "docs", "remote-mcp", "threat-model-anchors.json"
)

# ``module.py:123`` as it appears inside markdown inline code spans.
ANCHOR_RE = re.compile(r"`([A-Za-z0-9_./-]+\.py):(\d+)`")

# The epic children this document must map to a boundary (#929 children 2-10).
REQUIRED_CHILDREN = [931, 932, 933, 934, 935, 936, 937, 938, 939]

# The adversaries #956 names explicitly.
REQUIRED_ADVERSARIES = [
    "compromised LLM client",
    "prompt injection",
    "malicious tool arguments",
    "network attacker",
    "curious operator",
]


def _read(path):
    with open(path, "r", encoding="utf-8") as fh:
        return fh.read()


def _heading_re(title):
    """Match a level-2 heading by title, with or without section numbering.

    The document numbers its sections ('## 6. Decomposition ruling'), so an
    exact-substring assertion would break on renumbering without the document
    having actually lost anything.
    """
    return re.compile(
        r"^##\s+(?:\d+\.\s+)?" + re.escape(title), re.MULTILINE
    )


def _section_body(doc, title):
    """Return the text of section *title*, bounded by the next level-2 heading.

    Bounding matters: an unbounded slice runs to end-of-document, so the
    walkthrough tables in a later section leak into the child-to-boundary
    mapping and satisfy its coverage check with rows that assign no owner.
    """
    match = _heading_re(title).search(doc)
    if match is None:
        return None
    rest = doc[match.end():]
    nxt = re.search(r"^##\s", rest, re.MULTILINE)
    return rest[: nxt.start()] if nxt else rest


def _source_line(rel_path, lineno):
    """Return the 1-based *lineno* of *rel_path*, or None if out of range."""
    abs_path = os.path.join(REPO_ROOT, rel_path)
    if not os.path.exists(abs_path):
        return None
    with open(abs_path, "r", encoding="utf-8", errors="replace") as fh:
        for idx, line in enumerate(fh, start=1):
            if idx == lineno:
                return line
    return None


class ThreatModelFixtureTests(unittest.TestCase):
    """The fixture itself must be well-formed before it can prove anything."""

    def setUp(self):
        self.fixture = json.loads(_read(FIXTURE_PATH))

    def test_fixture_declares_a_generation_commit(self):
        sha = self.fixture.get("generated_against_commit") or ""
        self.assertRegex(
            sha,
            r"^[0-9a-f]{40}$",
            "the fixture must record the full commit its anchors were taken at",
        )

    def test_fixture_anchors_are_unique_and_well_formed(self):
        seen = set()
        for entry in self.fixture["anchors"]:
            anchor = entry["anchor"]
            self.assertNotIn(anchor, seen, f"duplicate anchor entry: {anchor}")
            seen.add(anchor)
            self.assertRegex(anchor, r"^[A-Za-z0-9_./-]+\.py:[1-9]\d*$", anchor)
            self.assertTrue(
                (entry.get("expect") or "").strip(),
                f"anchor {anchor} declares no 'expect' substring, so it proves nothing",
            )


class ThreatModelAnchorResolutionTests(unittest.TestCase):
    """#956 required positive test: every anchor resolves at the reviewed commit."""

    def setUp(self):
        self.fixture = json.loads(_read(FIXTURE_PATH))
        self.doc = _read(DOC_PATH)

    def test_every_declared_anchor_resolves_to_the_claimed_source_line(self):
        failures = []
        for entry in self.fixture["anchors"]:
            rel_path, _, raw_lineno = entry["anchor"].partition(":")
            lineno = int(raw_lineno)
            line = _source_line(rel_path, lineno)
            if line is None:
                failures.append(f"{entry['anchor']}: file or line does not exist")
                continue
            if entry["expect"] not in line:
                failures.append(
                    f"{entry['anchor']}: expected {entry['expect']!r}, "
                    f"found {line.strip()!r}"
                )
        self.assertEqual(
            [], failures, "unresolved threat-model anchors:\n" + "\n".join(failures)
        )

    def test_every_anchor_cited_in_the_document_is_declared_in_the_fixture(self):
        declared = {e["anchor"] for e in self.fixture["anchors"]}
        cited = {f"{m.group(1)}:{m.group(2)}" for m in ANCHOR_RE.finditer(self.doc)}
        undeclared = sorted(cited - declared)
        self.assertEqual(
            [],
            undeclared,
            "document cites anchors that no test verifies: " + ", ".join(undeclared),
        )

    def test_the_document_actually_cites_anchors(self):
        cited = {f"{m.group(1)}:{m.group(2)}" for m in ANCHOR_RE.finditer(self.doc)}
        self.assertGreaterEqual(
            len(cited),
            30,
            "a boundary document with almost no anchors is not traceable",
        )

    def test_unresolvable_anchor_is_detected(self):
        """Negative control: the checker must fail on a deliberately bad anchor.

        Without this, a checker that silently passed everything would look
        identical to a correct one.
        """
        self.assertIsNone(_source_line("gitea_config.py", 10**9))
        self.assertIsNone(_source_line("no_such_module_for_956.py", 1))
        real = _source_line("gitea_config.py", 54)
        self.assertIsNotNone(real)
        self.assertNotIn("this substring is not on that line", real)


class ThreatModelStructureTests(unittest.TestCase):
    """The document must contain what #956's acceptance criteria demand."""

    def setUp(self):
        self.doc = _read(DOC_PATH)

    def test_records_the_commit_it_was_generated_against(self):
        fixture = json.loads(_read(FIXTURE_PATH))
        self.assertIn(
            fixture["generated_against_commit"],
            self.doc,
            "the document must state the commit its anchors resolve at",
        )

    def test_names_every_required_adversary(self):
        low = self.doc.lower()
        for adversary in REQUIRED_ADVERSARIES:
            self.assertIn(adversary.lower(), low, f"adversary not covered: {adversary}")

    def test_maps_every_epic_child_from_two_through_ten(self):
        for number in REQUIRED_CHILDREN:
            self.assertIn(
                f"#{number}",
                self.doc,
                f"epic child #{number} is not mapped to a boundary",
            )

    def test_credential_rows_declare_holder_boundary_and_blast_radius(self):
        for column in ("Holder", "Boundary", "Blast radius"):
            self.assertIn(
                column,
                self.doc,
                f"the credential inventory must state each credential's {column.lower()}",
            )

    def test_states_an_explicit_co_residency_ruling(self):
        """AC3/AC5: an explicit ruling, not an implication."""
        self.assertIsNotNone(
            _heading_re("Decomposition ruling").search(self.doc),
            "the document must contain an explicit decomposition-ruling section",
        )
        for service in ("Jenkins", "GlitchTip", "Sentry", "database"):
            self.assertIn(service, self.doc, f"ruling does not address {service}")
        self.assertRegex(
            self.doc,
            r"D1\b.*must not",
            "the ruling must state the prohibition, not merely discuss it",
        )

    def test_contains_the_compromised_client_walkthrough(self):
        """#956 required negative/adversarial test."""
        self.assertIsNotNone(
            _heading_re("Adversarial walkthrough").search(self.doc),
            "the required compromised-client walkthrough is missing",
        )
        self.assertIn("Before the migration", self.doc)
        self.assertIn("After the migration", self.doc)

    def test_every_boundary_states_what_it_protects_and_what_crossing_requires(self):
        boundary_ids = set(re.findall(r"\bB(\d+)\b", self.doc))
        self.assertGreaterEqual(
            len(boundary_ids), 5, "too few trust boundaries to be a decomposition"
        )
        for column in (
            "Protects",
            "Crossing requires today",
            "Crossing must require remotely",
        ):
            self.assertIn(column, self.doc, f"boundary table is missing '{column}'")

    def test_declares_itself_documentation_only(self):
        self.assertIn("documentation only", self.doc.lower())


class ThreatModelConsistencyTests(unittest.TestCase):
    """Counts stated in prose must match the rows actually present."""

    def setUp(self):
        self.doc = _read(DOC_PATH)

    def _declared_ids(self, prefix):
        # Table rows begin '| CR1 |' / '| B3 |' / '| A2 |'.
        return sorted(
            {
                int(m)
                for m in re.findall(
                    r"^\|\s*%s(\d+)\s*\|" % prefix, self.doc, re.MULTILINE
                )
            }
        )

    def test_identifier_sequences_have_no_gaps(self):
        for prefix, label in (
            ("A", "assets"),
            ("B", "boundaries"),
            ("CR", "credentials"),
        ):
            ids = self._declared_ids(prefix)
            self.assertTrue(ids, f"no {label} declared")
            self.assertEqual(
                list(range(1, len(ids) + 1)),
                ids,
                f"{label} identifiers must run 1..n with no gaps; got {ids}",
            )

    def test_stated_credential_count_matches_the_rows(self):
        ids = self._declared_ids("CR")
        match = re.search(r"(\d+)\s+credential(?:s)? in total", self.doc)
        self.assertIsNotNone(match, "the credential inventory must state its own total")
        self.assertEqual(
            len(ids),
            int(match.group(1)),
            "stated credential total disagrees with the number of rows",
        )

    def test_every_boundary_is_owned_by_at_least_one_child(self):
        """Each boundary must be owned by a child *in the mapping table*.

        Scanning the whole section would let a prose summary line ("Boundary
        coverage: ... B5 (#936)") satisfy the assertion while the table row
        that actually assigns the owner had been emptied — verified by
        deliberately blanking a row and watching a whole-section check still
        pass. Only table rows count.
        """
        mapping_section = _section_body(self.doc, "Child-to-boundary mapping")
        self.assertIsNotNone(
            mapping_section, "child-to-boundary mapping section is missing"
        )
        rows = [
            line
            for line in mapping_section.splitlines()
            if line.lstrip().startswith("|") and re.search(r"#93\d", line)
        ]
        self.assertGreaterEqual(
            len(rows), len(REQUIRED_CHILDREN), "mapping table has too few child rows"
        )
        mapped = set(re.findall(r"\bB(\d+)\b", "\n".join(rows)))
        declared = {str(i) for i in self._declared_ids("B")}
        unmapped = sorted(declared - mapped, key=int)
        self.assertEqual(
            [],
            unmapped,
            "boundaries with no owning child: " + ", ".join("B" + u for u in unmapped),
        )


if __name__ == "__main__":
    unittest.main()