SeedPass/scripts/mutation_test.py

#!/usr/bin/env python3
"""Mutation testing for the TypeScript surfaces that have no second implementation.

WHY THIS EXISTS
---------------
Differential fuzzing (scripts/differential_fuzz.py) uses Python as an oracle,
so it can only reach code both implementations have. The session agent, the
API's HTTP layer, high-risk session handling and the secret sinks exist only
in TypeScript — and they are where the security properties live. For those,
the question "would the suite notice if this broke?" has no oracle at all, so
it has to be answered by breaking things on purpose.

A passing suite proves the code works on the cases someone wrote down. It says
nothing about whether a REGRESSION would be caught. This deliberately breaks
one security predicate at a time and checks that some test goes red. A mutant
that survives is a place where the code could silently stop enforcing
something and every test would still pass.

WHAT IT MUTATES
---------------
Security predicates specifically, not arbitrary syntax: comparison and
equality operators, boolean connectives, and guard conditions forced to always
allow. Mutating a log message would produce a survivor that means nothing;
mutating `if (!tokenValid)` means something exact.

Three outcomes per mutant:
  killed     the suite failed. Good: a regression here would be caught.
  SURVIVED   the suite passed. A gap, or an equivalent mutant. Needs triage.
  invalid    the mutant does not typecheck. Not a real mutant; skipped.

Usage:
    .venv/bin/python scripts/mutation_test.py --target agent
    .venv/bin/python scripts/mutation_test.py --target all --limit 40

Survivors are reported with file, line, and the exact edit, so each can be
judged by hand. The script never leaves a mutation behind: the original file
is restored after every run, including on interrupt.
"""

from __future__ import annotations

import argparse
import re
import signal
import subprocess
import sys
import time
from dataclasses import dataclass
from pathlib import Path

REPO = Path(__file__).resolve().parent.parent
CLI = REPO / "js" / "packages" / "cli"

# This script deliberately writes broken security code into the working tree,
# one file at a time, and restores it after each mutant. That is safe on its
# own and NOT safe alongside anything else touching the repository: a `git
# add -A` while a mutant is live commits a deliberately inverted security
# check. That happened once during development -- an inverted fingerprint
# check reached a commit -- so the guard below exists to make it structurally
# impossible rather than a thing to remember.
LOCK = REPO / ".mutation-test-running"


def acquire_lock() -> None:
    if LOCK.exists():
        raise SystemExit(
            f"{LOCK.name} exists: another mutation run is in progress, or a "
            f"previous one died with a mutant still applied. Check "
            f"`git diff` before deleting it."
        )
    LOCK.write_text(
        "A mutation test is running. The working tree contains deliberately\n"
        "broken code. Do not commit. Do not run `git add -A`.\n"
    )


def release_lock() -> None:
    LOCK.unlink(missing_ok=True)

# Each target names the source file and the tests that should defend it.
# Running only the relevant suites keeps a full pass tractable; a mutant that
# survives its own suite but would be caught by another is still a finding,
# because the defending test is not where anyone would look for it.
TARGETS: dict[str, tuple[str, list[str]]] = {
    # agent.ts also holds the high-risk session, whose tests live in
    # highRisk.test.ts. Leaving it out scored the agent at 38% and blamed the
    # code for gaps that were really a gap in this mapping.
    "agent": (
        "src/agent.ts",
        ["test/agentSecurity.test.ts", "test/tokens.test.ts", "test/highRisk.test.ts"],
    ),
    "server": ("src/api/server.ts", ["test/api.test.ts"]),
    "routes": ("src/api/routes.ts", ["test/api.test.ts"]),
    "highrisk": ("src/highRisk.ts", ["test/highRisk.test.ts"]),
    "sinks": ("src/sinks.ts", ["test/sinks.test.ts", "test/tokens.test.ts"]),
    "approvals": ("src/approvals.ts", ["test/highRisk.test.ts"]),
}

# Operator swaps that change a decision rather than a value's shape.
OPERATORS: list[tuple[str, str]] = [
    ("!==", "==="),
    ("===", "!=="),
    (" >= ", " > "),
    (" <= ", " < "),
    (" > ", " >= "),
    (" < ", " <= "),
    (" && ", " || "),
    (" || ", " && "),
]

# Lines worth mutating: they decide whether something is permitted, expired,
# within a limit, or authentic. A mutation anywhere else mostly produces noise.
SECURITY_HINTS = re.compile(
    r"\b("
    r"auth|token|cap|capab|scope|permit|allow|deny|refus|valid|verif|"
    r"expire|expires|ttl|limit|rate|uses|remaining|owner|secret|password|"
    r"factor|unlock|lock|held|fingerprint|hash|equal|match|forbidden"
    r")",
    re.IGNORECASE,
)

SKIP_LINE = re.compile(r"^\s*(//|\*|/\*)")

# A line preceded by this marker has been triaged as an EQUIVALENT mutant:
# the mutation provably cannot change behaviour, so no test can kill it and
# chasing it is wasted effort. Marking it keeps future survivor lists
# meaningful -- a survivor should mean "look at this", not "this again".
# The marker must say WHY, so the judgement can be re-checked rather than
# trusted.
EQUIVALENT_MARKER = "mutation-equivalent:"


@dataclass
class Mutant:
    line_no: int
    original: str
    mutated: str
    description: str


def _marked_equivalent(lines: list[str], line_no: int) -> bool:
    """Is the code at `line_no` (1-based) preceded by an equivalence marker?"""
    idx = line_no - 2  # the line directly above, 0-based
    while idx >= 0 and SKIP_LINE.match(lines[idx]):
        if EQUIVALENT_MARKER in lines[idx]:
            return True
        idx -= 1
    return False


def generate(source: str) -> list[Mutant]:
    mutants: list[Mutant] = []
    lines = source.splitlines()
    for i, line in enumerate(lines, start=1):
        if SKIP_LINE.match(line) or not SECURITY_HINTS.search(line):
            continue
        # Skip lines a human has already judged unkillable, with a reason.
        # Scans the whole contiguous comment block above, not just the line
        # immediately before: the marker belongs at the START of an
        # explanation, and requiring it on the last line would mean writing
        # the reason upside down.
        if _marked_equivalent(lines, i):
            continue
        for old, new in OPERATORS:
            if old not in line:
                continue
            mutated = line.replace(old, new, 1)
            if mutated == line:
                continue
            mutants.append(
                Mutant(i, line, mutated, f"{old.strip()} -> {new.strip()}")
            )
        # Force a guard to always allow. This is the single most informative
        # mutation for security code: if no test notices a check being removed
        # entirely, that check is unverified.
        guard = re.match(r"^(\s*)if \((.+)\) \{\s*$", line)
        if guard and SECURITY_HINTS.search(guard.group(2)):
            mutants.append(
                Mutant(
                    i,
                    line,
                    f"{guard.group(1)}if (false) {{",
                    "guard disabled (always allow)",
                )
            )
    return mutants


def run(cmd: list[str], cwd: Path, timeout: int) -> tuple[int, str]:
    try:
        proc = subprocess.run(
            cmd, cwd=cwd, capture_output=True, text=True, timeout=timeout
        )
        return proc.returncode, proc.stdout + proc.stderr
    except subprocess.TimeoutExpired:
        # A hang counts as killed: the suite noticed, however unhappily.
        return 1, "TIMEOUT"


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--target", default="all", choices=[*sorted(TARGETS), "all"]
    )
    parser.add_argument(
        "--limit", type=int, default=0, help="cap mutants per target (0 = all)"
    )
    parser.add_argument(
        "--offset",
        type=int,
        default=0,
        help=(
            "skip the first N mutants. With --limit this runs a slice, so a "
            "long target can be measured in foreground-sized pieces rather "
            "than backgrounded -- this script writes broken security code "
            "into the tree, and must not run unattended alongside commits."
        ),
    )
    parser.add_argument("--timeout", type=int, default=420)
    args = parser.parse_args()

    targets = sorted(TARGETS) if args.target == "all" else [args.target]
    overall_survivors: list[tuple[str, Mutant]] = []
    overall_counts = {"killed": 0, "survived": 0, "invalid": 0}

    acquire_lock()
    print(
        "NOTE: the working tree will contain deliberately broken code until "
        "this finishes.\n      Do not commit from another shell while it runs.\n"
    )

    for name in targets:
        rel, tests = TARGETS[name]
        path = CLI / rel
        original = path.read_text()
        all_mutants = generate(original)
        mutants = all_mutants[args.offset :]
        if args.limit:
            mutants = mutants[: args.limit]
        span = f"{args.offset + 1}-{args.offset + len(mutants)} of {len(all_mutants)}"
        print(f"\n=== {name} ({rel}): mutants {span} ===")

        # Restore the file whatever happens, including Ctrl-C. Leaving a
        # deliberate security hole in the tree would be a spectacular own goal.
        def restore(*_: object) -> None:
            path.write_text(original)

        signal.signal(
            signal.SIGINT, lambda *a: (restore(), release_lock(), sys.exit(130))
        )

        try:
            for idx, mutant in enumerate(mutants, start=args.offset + 1):
                lines = original.splitlines(keepends=True)
                ending = "\n" if lines[mutant.line_no - 1].endswith("\n") else ""
                lines[mutant.line_no - 1] = mutant.mutated + ending
                path.write_text("".join(lines))

                code, _ = run(
                    ["npx", "tsc", "--noEmit"], CLI, timeout=args.timeout
                )
                if code != 0:
                    overall_counts["invalid"] += 1
                    print(f"  [{idx}/{len(all_mutants)}] line {mutant.line_no}: invalid (does not typecheck)")
                    continue

                code, _ = run(
                    ["npx", "vitest", "run", *tests], CLI, timeout=args.timeout
                )
                if code != 0:
                    overall_counts["killed"] += 1
                    print(f"  [{idx}/{len(all_mutants)}] line {mutant.line_no}: killed  ({mutant.description})")
                else:
                    overall_counts["survived"] += 1
                    overall_survivors.append((f"{rel}:{mutant.line_no}", mutant))
                    print(f"  [{idx}/{len(all_mutants)}] line {mutant.line_no}: SURVIVED ({mutant.description})")
        finally:
            restore()
            # Prove the restore worked rather than assuming it: a mutant left
            # behind is a deliberately broken security check sitting in the
            # tree waiting to be committed.
            if path.read_text() != original:
                release_lock()
                raise SystemExit(f"FAILED TO RESTORE {rel} -- fix this before committing")

    release_lock()

    total = sum(overall_counts.values())
    viable = overall_counts["killed"] + overall_counts["survived"]
    print(f"\n{'=' * 70}")
    print(
        f"{overall_counts['killed']} killed, {overall_counts['survived']} survived, "
        f"{overall_counts['invalid']} invalid (of {total})"
    )
    if viable:
        print(f"mutation score: {overall_counts['killed'] / viable:.0%} of viable mutants caught")

    if overall_survivors:
        print(f"\nSURVIVORS — each is either a gap or an equivalent mutant, and needs judging:\n")
        for where, mutant in overall_survivors:
            print(f"  {where}  [{mutant.description}]")
            print(f"    was: {mutant.original.strip()[:110]}")
            print(f"    now: {mutant.mutated.strip()[:110]}")
            print()
    return 0


if __name__ == "__main__":
    raise SystemExit(main())