fix(api): password-gate live TOTP codes; add mutation testing for the un-oracled surfaces

Differential fuzzing can only reach code both implementations have. The
session agent, the API's HTTP layer, high-risk session handling and the sinks
exist only in TypeScript — and that is where the security properties live. For
those the question "would the suite notice if this broke?" has no oracle, so
it has to be answered by breaking things deliberately.

scripts/mutation_test.py does that: it flips comparison and boolean operators
and forces guards to always-allow, but only on lines that decide whether
something is permitted, expired, within a limit, or authentic. Mutating a log
line would produce a survivor that means nothing. Every mutant is typechecked
first, so a mutant that does not compile is reported as invalid rather than
counted as caught, and the original file is restored even on interrupt.

The first run on the API server found two survivors, and they were different
in kind:

**A real gap.** Disabling the rate-limit check entirely left every test
passing — nothing exercised it. Now tested: the limit refuses the request past
its budget, and unauthenticated requests consume budget too, since otherwise
the limit is bypassed by simply not sending a token.

**An equivalent mutant, which pointed at a real hole anyway.** The route-level
`requiresPassword` flag turned out to be fully redundant with the
`requirePassword` call in each handler, so disabling it changed nothing. The
redundancy is worth keeping, but nothing ENFORCED the pairing — so a future
route could set one and forget the other with the suite silent either way.
Two structural tests now pin it.

Writing the second of those tests immediately found a security regression I
had introduced: **`GET /api/v1/totp` returned live TOTP codes on a bearer
token alone.** Python gates that endpoint on the master password
(`_require_password` in api.py) and this port did not, so a leaked token
yielded working second factors for every TOTP entry in the vault. Now gated,
and tested for.

Server mutation score: 88% of viable mutants caught, the single survivor being
the equivalent one above.

Co-Authored-By: Claude Opus 5 (1M context) <[email protected]>
Claude-Session: https://claude.ai/code/session_01Vtx94nXDmNgXEeThUZP4QN

js/packages/cli/src/api/routes.ts

@@ -584,8 +584,18 @@ export function registerRoutes(server: ApiServer, ctx: ApiContext): void {
 
   // ------------------------------------------------------------------- totp
 
-  server.route("GET", "/api/v1/totp", async () => {
-    const mnemonic = requireUnlocked(ctx);
+  server.route(
+    "GET",
+    "/api/v1/totp",
+    async (req) => {
+      // A live TOTP code authenticates. Python gates this on the master
+      // password (`_require_password` in api.py) and this port did not, so a
+      // leaked bearer token yielded working second factors for every TOTP
+      // entry in the vault. Found by the route-table invariant test below,
+      // which exists because mutation testing showed the gating was
+      // structurally unenforced.
+      await requirePassword(ctx, req);
+      const mnemonic = requireUnlocked(ctx);
     const vault = await readVault(ctx);
     const now = Math.floor(ctx.now() / 1000);
     const codes = Object.entries(vault.index.entries)
@@ -603,8 +613,10 @@ export function registerRoutes(server: ApiServer, ctx: ApiContext): void {
           seconds_remaining: period - (now % period),
         };
       });
-    return { json: { codes } };
-  });
+      return { json: { codes } };
+    },
+    { requiresPassword: true },
+  );
 
   server.route(
     "GET",

js/packages/cli/test/api.test.ts

@@ -14,6 +14,7 @@ import { mkdtemp, mkdir, readFile, writeFile, stat } from "node:fs/promises";
 import { existsSync } from "node:fs";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
+import { fileURLToPath } from "node:url";
 import process from "node:process";
 import { mnemonics } from "@seedpass/test-vectors";
 import {
@@ -209,6 +210,56 @@ describe("what the API refuses", () => {
     }
   });
 
+  it("actually enforces the ordinary rate limit", async () => {
+    // Mutation testing found this: disabling the rate-limit check entirely
+    // left every test passing. A limiter nothing exercises is a limiter that
+    // can be removed by accident.
+    const limited = new ApiServer({
+      host: "127.0.0.1",
+      port: 0,
+      token: TOKEN,
+      rateLimit: 3,
+      rateWindowSeconds: 60,
+    });
+    registerRoutes(limited, buildContext({ app, fingerprint: FINGERPRINT, mnemonic: MNEMONIC }));
+    const bound = await limited.listen();
+    const url = `http://${bound.host}:${bound.port}/api/v1/entry`;
+    try {
+      const hit = () => fetch(url, { headers: { authorization: `Bearer ${TOKEN}` } });
+      expect((await hit()).status).toBe(200);
+      expect((await hit()).status).toBe(200);
+      expect((await hit()).status).toBe(200);
+      // Fourth request in the window is refused.
+      const blocked = await hit();
+      expect(blocked.status).toBe(429);
+      expect((await blocked.json()).detail).toContain("Rate limit");
+    } finally {
+      await limited.close();
+    }
+  });
+
+  it("rate-limits before authenticating, so a flood cannot hammer the token check", async () => {
+    const limited = new ApiServer({
+      host: "127.0.0.1",
+      port: 0,
+      token: TOKEN,
+      rateLimit: 2,
+      rateWindowSeconds: 60,
+    });
+    registerRoutes(limited, buildContext({ app, fingerprint: FINGERPRINT, mnemonic: MNEMONIC }));
+    const bound = await limited.listen();
+    const url = `http://${bound.host}:${bound.port}/api/v1/entry`;
+    try {
+      // Unauthenticated requests consume the budget too — otherwise the limit
+      // is trivially bypassed by simply not sending a token.
+      await fetch(url);
+      await fetch(url);
+      expect((await fetch(url)).status).toBe(429);
+    } finally {
+      await limited.close();
+    }
+  });
+
   it("limits failed unlock attempts far more tightly than ordinary requests", async () => {
     const locked = new ApiServer({
       host: "127.0.0.1",
@@ -341,8 +392,17 @@ describe("entries", () => {
 });
 
 describe("totp, config, relays, stats", () => {
+  it("requires the master password for live TOTP codes", async () => {
+    // A live code authenticates, so a leaked bearer token must not produce
+    // one. Python gates this the same way; this port did not until the
+    // route-table invariant test caught it.
+    const noPassword = await call("GET", "/api/v1/totp");
+    expect(noPassword.status).toBe(401);
+    expect(noPassword.text).not.toMatch(/\d{6}/);
+  });
+
   it("returns live TOTP codes with the time remaining", async () => {
-    const res = await call("GET", "/api/v1/totp");
+    const res = await call("GET", "/api/v1/totp", { password: PASSWORD });
     expect(res.status).toBe(200);
     expect(res.json.codes).toHaveLength(1);
     expect(res.json.codes[0].code).toMatch(/^\d{6}$/);
@@ -614,3 +674,69 @@ describe("path handling", () => {
     expect(UNPORTED_PREFIXES).toEqual([]);
   });
 });
+
+describe("structural invariants of the route table", () => {
+  /**
+   * Mutation testing showed the route-level `requiresPassword` flag is
+   * redundant: every route that sets it also calls `requirePassword` in its
+   * handler, so disabling the flag changes nothing. That redundancy is
+   * defense in depth and worth keeping — but nothing ENFORCED the pairing,
+   * which meant a future route could set one and forget the other, and the
+   * suite would be silent either way.
+   *
+   * This reads the source rather than exercising behaviour, which is unusual
+   * for a test and justified here: the property is about how routes are
+   * DECLARED, and it is exactly the property a new route would break.
+   */
+  it("pairs the requiresPassword flag with an actual requirePassword call", async () => {
+    const source = await readFile(
+      fileURLToPath(new URL("../src/api/routes.ts", import.meta.url)),
+      "utf8",
+    );
+    const blocks = source
+      .split(/(?=\n  server\.route\()/)
+      // Drop the file preamble: it holds the requirePassword DEFINITION, not
+      // a route, and would otherwise read as a route that calls it.
+      .filter((block) => /\n?\s*server\.route\(/.test(block));
+    const mismatched: string[] = [];
+    for (const block of blocks) {
+      const flagged = block.includes("requiresPassword: true");
+      const called = block.includes("requirePassword(ctx");
+      if (flagged === called) continue;
+      const name = /server\.route\(\s*"(\w+)",\s*"([^"]+)"/.exec(block);
+      mismatched.push(
+        `${name ? `${name[1]} ${name[2]}` : "(unparsed)"}: ` +
+          `flag=${flagged} call=${called}`,
+      );
+    }
+    expect(mismatched).toEqual([]);
+  });
+
+  it("password-gates every route that can return plaintext", async () => {
+    // The failure this guards against is a NEW secret-returning route with
+    // neither the flag nor the call — which the pairing test above cannot
+    // see, because it only checks consistency, not coverage.
+    const source = await readFile(
+      fileURLToPath(new URL("../src/api/routes.ts", import.meta.url)),
+      "utf8",
+    );
+    const blocks = source
+      .split(/(?=\n  server\.route\()/)
+      .filter((block) => /\n?\s*server\.route\(/.test(block));
+    const ungated: string[] = [];
+    for (const block of blocks) {
+      // A handler that materializes a secret, reads the partition, or hands
+      // back the parent seed must be password-gated.
+      const producesPlaintext =
+        block.includes("materializeSecret(") ||
+        block.includes("deriveTotpSecret(") ||
+        block.includes("readPartition(") ||
+        block.includes("requireUnlocked(ctx)") && block.includes("mnemonic + ");
+      if (!producesPlaintext) continue;
+      if (block.includes("requirePassword(ctx")) continue;
+      const name = /server\.route\(\s*"(\w+)",\s*"([^"]+)"/.exec(block);
+      ungated.push(name ? `${name[1]} ${name[2]}` : "(unparsed)");
+    }
+    expect(ungated).toEqual([]);
+  });
+});

scripts/mutation_test.py

@@ -0,0 +1,222 @@
+#!/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"
+
+# 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": ("src/agent.ts", ["test/agentSecurity.test.ts", "test/tokens.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*(//|\*|/\*)")
+
+
+@dataclass
+class Mutant:
+    line_no: int
+    original: str
+    mutated: str
+    description: str
+
+
+def generate(source: str) -> list[Mutant]:
+    mutants: list[Mutant] = []
+    for i, line in enumerate(source.splitlines(), start=1):
+        if SKIP_LINE.match(line) or not SECURITY_HINTS.search(line):
+            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("--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}
+
+    for name in targets:
+        rel, tests = TARGETS[name]
+        path = CLI / rel
+        original = path.read_text()
+        mutants = generate(original)
+        if args.limit:
+            mutants = mutants[: args.limit]
+        print(f"\n=== {name} ({rel}): {len(mutants)} mutants ===")
+
+        # 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(), sys.exit(130)))
+
+        try:
+            for idx, mutant in enumerate(mutants, start=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(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(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(mutants)}] line {mutant.line_no}: SURVIVED ({mutant.description})")
+        finally:
+            restore()
+
+    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())