stack/packages/harness/tests/gate.test.mjs

import { test } from "node:test";
import assert from "node:assert/strict";
import { mkdirSync, realpathSync, symlinkSync, writeFileSync } from "node:fs";
import { homedir } from "node:os";
import { join, resolve } from "node:path";
import { pathToFileURL } from "node:url";
import { claudeBuiltins, decide, insideWorkspace } from "../src/gate.mjs";
import { scratch } from "./helpers.mjs";

function setup(t, harness, tools = ["read", "write", "edit", "grep", "find", "ls"]) {
  const dir = scratch(t);
  const workspace = join(dir, "ws");
  mkdirSync(join(workspace, "sub"), { recursive: true });
  writeFileSync(join(workspace, "a.txt"), "a");
  return { dir, workspace, policy: { harness, workspace, tools, typed: ["send_message", "list_agents"] } };
}
const allowed = (v) => assert.deepEqual(v, { allow: true });
const blocked = (v, pattern) => {
  assert.equal(v.allow, false);
  assert.match(v.reason, /^mosaic gate: /);
  if (pattern) assert.match(v.reason, pattern);
};

test("pi: policy tools and typed tools pass, anything else is blocked", (t) => {
  const { policy } = setup(t, "pi", ["read", "write"]);
  allowed(decide(policy, "send_message", { to: "pm", body: "x" }));
  allowed(decide(policy, "read", { path: "a.txt" }));
  blocked(decide(policy, "bash", { command: "true" }), /bash isn't in this session's policy/);
  blocked(decide(policy, "mcp__mosaic__send_message", {}), /isn't in this session's policy/);
  blocked(decide(policy, "resolve_decision", {}));
  blocked(decide(policy, undefined, {}), /without a name/);
});

test("claude: builtins map from pi names, typed tools need the mcp prefix", (t) => {
  const { policy } = setup(t, "claude-code", ["read", "find", "ls", "bash"]);
  assert.deepEqual(claudeBuiltins(policy), ["Read", "Glob", "Bash"]);
  allowed(decide(policy, "mcp__mosaic__list_agents", {}));
  allowed(decide(policy, "Bash", { command: "ls /" }));
  blocked(decide(policy, "list_agents", {}));
  blocked(decide(policy, "mcp__mosaic__launch", {}));
  blocked(decide(policy, "mcp__other__list_agents", {}));
  blocked(decide(policy, "Write", { file_path: join(policy.workspace, "x") }), /Write isn't in this session's policy/);
  blocked(decide(policy, "WebFetch", { url: "https://example.test" }));
  blocked(decide(policy, "Task", {}));
});

test("file tool paths must resolve inside the workspace", (t) => {
  const { dir, workspace, policy } = setup(t, "pi");
  allowed(decide(policy, "read", { path: "a.txt" }));
  allowed(decide(policy, "write", { path: "sub/new/deep.txt" }));
  allowed(decide(policy, "read", { path: join(workspace, "a.txt") }));
  allowed(decide(policy, "ls", {}));
  allowed(decide(policy, "grep", { pattern: "x", path: "" }));
  blocked(decide(policy, "read", { path: "../outside.txt" }), /outside the workspace/);
  blocked(decide(policy, "read", { path: "/etc/passwd" }), /outside the workspace/);
  blocked(decide(policy, "write", { path: join(dir, "ws-evil", "x") }), /outside the workspace/);
  blocked(decide(policy, "read", { path: 42 }), /isn't a string/);
});

test("pi's own path normalisation can't be used to step out", (t) => {
  const { workspace, policy } = setup(t, "pi");
  blocked(decide(policy, "read", { path: "~/.ssh/id_ed25519" }));
  blocked(decide(policy, "read", { path: "~" }));
  blocked(decide(policy, "read", { path: "@/etc/passwd" }));
  blocked(decide(policy, "read", { path: pathToFileURL("/etc/passwd").href }));
  blocked(decide(policy, "read", { path: "/etc/passwd" }));
  allowed(decide(policy, "read", { path: `@${join(workspace, "a.txt")}` }));
  allowed(decide(policy, "read", { path: pathToFileURL(join(workspace, "a.txt")).href }));
  // A workspace under home is still judged by its real path.
  assert.equal(insideWorkspace(workspace, "~/"), homedir() === workspace);
});

test("a symlink inside the workspace that points out is outside", (t) => {
  const { dir, workspace, policy } = setup(t, "pi");
  mkdirSync(join(dir, "secret"));
  writeFileSync(join(dir, "secret", "s.txt"), "s");
  symlinkSync(join(dir, "secret"), join(workspace, "link"));
  blocked(decide(policy, "read", { path: "link/s.txt" }), /outside the workspace/);
  blocked(decide(policy, "write", { path: "link/new.txt" }), /outside the workspace/);
});

test("a dangling symlink is refused at any depth, in both harnesses", (t) => {
  for (const [harness, tool, field] of [["pi", "write", "path"], ["claude-code", "Write", "file_path"]]) {
    const { dir, workspace, policy } = setup(t, harness, ["read", "write"]);
    mkdirSync(join(dir, "outside"));
    // notes.md names a file that doesn't exist yet, outside; dangling names a
    // directory that doesn't; inner points inside but at nothing.
    symlinkSync(join(dir, "outside", "planted.txt"), join(workspace, "notes.md"));
    symlinkSync(join(dir, "nowhere"), join(workspace, "dangling"));
    symlinkSync(join(workspace, "later.txt"), join(workspace, "inner"));
    for (const rel of ["notes.md", "dangling/x", "dangling/deep/x", "inner"]) {
      blocked(decide(policy, tool, { [field]: rel }), /goes through a dangling symlink/);
      blocked(decide(policy, tool, { [field]: join(workspace, rel) }), /goes through a dangling symlink/);
    }
    // Once the target exists, the usual realpath rule decides.
    writeFileSync(join(dir, "outside", "planted.txt"), "p");
    blocked(decide(policy, tool, { [field]: "notes.md" }), /outside the workspace/);
    writeFileSync(join(workspace, "later.txt"), "l");
    allowed(decide(policy, tool, { [field]: "inner" }));
    // A file used as a directory can't be checked.
    blocked(decide(policy, tool, { [field]: "a.txt/x" }), /can't be checked: ENOTDIR/);
  }
});

// Pi's read opens another spelling when the name it's given doesn't exist
// (path-utils.js resolveReadPathAsync). Asked name -> name on disk.
const SPELLINGS = [
  ["quote", "notes's.txt", "notes\u2019s.txt"],
  ["ampm", "shot 9.41 AM.png", "shot 9.41\u202FAM.png"],
  ["nfd", "r\u00E9sum\u00E9.txt", "r\u00E9sum\u00E9.txt".normalize("NFD")],
  ["nfd and quote", "d'\u00E9cran.png", "d\u2019\u00E9cran.png".normalize("NFD")],
  // Each family alone, on a name with both an apostrophe and an accent.
  ["nfd, straight quote", "l'\u00E9t\u00E9.txt", "l'\u00E9t\u00E9.txt".normalize("NFD")],
  ["quote, composed", "l'\u00E9t\u00E9.md", "l\u2019\u00E9t\u00E9.md"],
];

test("read is checked under every spelling pi's read would open, in both harnesses", (t) => {
  for (const [harness, read, write, field] of [["pi", "read", "write", "path"], ["claude-code", "Read", "Write", "file_path"]]) {
    for (const [, asked, disk] of SPELLINGS) {
      const { dir, workspace, policy } = setup(t, harness, ["read", "write"]);
      mkdirSync(join(dir, "outside"));
      writeFileSync(join(dir, "outside", "secret.txt"), "s");
      symlinkSync(join(dir, "outside", "secret.txt"), join(workspace, disk));
      const why = new RegExp(`outside the workspace under another spelling: (.*/)?${asked.replace(/[.']/g, "\\$&")}`);
      blocked(decide(policy, read, { [field]: asked }), why);
      blocked(decide(policy, read, { [field]: join(workspace, asked) }), why);
      // Write takes the name as given, so writing the asked name stays inside.
      allowed(decide(policy, write, { [field]: asked }));
      // The same spelling pointing inside is fine.
      const inner = setup(t, harness, ["read"]);
      symlinkSync(join(inner.workspace, "a.txt"), join(inner.workspace, disk));
      allowed(decide(inner.policy, read, { [field]: asked }));
    }
  }
});

test("other spellings cover directories, dangling links and pi's cwd", (t) => {
  const { dir, workspace, policy } = setup(t, "pi", ["read"]);
  mkdirSync(join(dir, "outside"));
  writeFileSync(join(dir, "outside", "s.txt"), "s");
  // A directory name is respelled too.
  symlinkSync(join(dir, "outside"), join(workspace, "it\u2019s"));
  blocked(decide(policy, "read", { path: "it's/s.txt" }), /outside the workspace under another spelling/);
  // A dangling link under another spelling is refused, like the name itself.
  symlinkSync(join(dir, "nowhere"), join(workspace, "gone\u2019s"));
  blocked(decide(policy, "read", { path: "gone's" }), /dangling symlink under another spelling/);
  // A spelling that doesn't exist, or uses a file as a directory, isn't opened.
  allowed(decide(policy, "read", { path: "nobody's.txt" }));
  writeFileSync(join(workspace, "f\u2019s"), "f");
  allowed(decide(policy, "read", { path: "f's/x" }));
  // Pi resolves against its cwd, the workspace's real path, and respells
  // that too: here the real path has an apostrophe the given path lacks.
  mkdirSync(join(dir, "jo's", "ws"), { recursive: true });
  mkdirSync(join(dir, "jo\u2019s", "ws"), { recursive: true });
  writeFileSync(join(dir, "jo\u2019s", "ws", "x.txt"), "x");
  symlinkSync(join(dir, "jo's"), join(dir, "jo"));
  const viaLink = { ...policy, workspace: join(dir, "jo", "ws") };
  blocked(decide(viaLink, "read", { path: "x.txt" }), /outside the workspace under another spelling/);
});

// Both adapters cd into the workspace, and the tool's cwd is its real path,
// so `..` climbs the real path's parents. Through a symlinked workspace or
// dataRoot those differ from the given path's.
const CLIMBS = {
  // <dir>/a/ws -> <dir>/deep/store/ws: up two is <dir> as given, <dir>/deep for pi.
  workspace(dir) {
    mkdirSync(join(dir, "deep", "store", "ws"), { recursive: true });
    mkdirSync(join(dir, "a"));
    symlinkSync(join(dir, "deep", "store", "ws"), join(dir, "a", "ws"));
    return { workspace: join(dir, "a", "ws"), outside: join(dir, "deep", "a", "ws"), escape: "../../a/ws", stay: "../ws", strict: "../../store/ws" };
  },
  // dataRoot <dir>/data -> <dir>/deep/store, the workspace under it as the
  // launcher builds it: up four is <dir> as given, <dir>/deep for pi.
  dataRoot(dir) {
    mkdirSync(join(dir, "deep", "store", "workspaces", "b", "i"), { recursive: true });
    symlinkSync(join(dir, "deep", "store"), join(dir, "data"));
    return { workspace: join(dir, "data", "workspaces", "b", "i"), outside: join(dir, "deep", "data", "workspaces", "b", "i"), escape: "../../../../data/workspaces/b/i", stay: "../i", strict: "../../../../store/workspaces/b/i" };
  },
};

test("a relative path climbs from the workspace's real path, in both harnesses", (t) => {
  for (const [name, build] of Object.entries(CLIMBS)) {
    for (const harness of ["pi", "claude-code"]) {
      const dir = scratch(t);
      const { workspace, outside, escape, stay, strict } = build(dir);
      writeFileSync(join(workspace, "a.txt"), "a");
      mkdirSync(outside, { recursive: true });
      writeFileSync(join(outside, "secret.txt"), "s");
      const policy = { harness, workspace, tools: ["read", "write"], typed: [] };
      const [read, write, field] = harness === "pi" ? ["read", "write", "path"] : ["Read", "Write", "file_path"];
      const at = `${name}, ${harness}`;
      // As given, the escape is inside; from the real path it is outside.
      assert.equal(resolve(workspace, escape), workspace, at);
      assert.equal(resolve(realpathSync(workspace), escape), outside, at);
      blocked(decide(policy, read, { [field]: `${escape}/secret.txt` }), /outside the workspace/);
      blocked(decide(policy, write, { [field]: `${escape}/planted.txt` }), /outside the workspace/);
      // Climbing out and back in by the same name stays inside on both paths.
      allowed(decide(policy, read, { [field]: `${stay}/a.txt` }));
      allowed(decide(policy, write, { [field]: `${stay}/new.txt` }));
      // The other way round, inside for pi but outside as given, is refused
      // too: the gate doesn't rest on the cwd the harness started in.
      assert.equal(resolve(realpathSync(workspace), strict), realpathSync(workspace), at);
      blocked(decide(policy, read, { [field]: `${strict}/a.txt` }), /outside the workspace/);
    }
  }
});

test("claude path fields per tool", (t) => {
  const { workspace, policy } = setup(t, "claude-code", ["read", "write", "edit", "grep", "find"]);
  allowed(decide(policy, "Read", { file_path: join(workspace, "a.txt") }));
  allowed(decide(policy, "Edit", { file_path: join(workspace, "a.txt"), old_string: "a", new_string: "b" }));
  allowed(decide(policy, "Grep", { pattern: "x" }));
  blocked(decide(policy, "Write", { file_path: "/tmp/x", content: "" }), /outside the workspace/);
  blocked(decide(policy, "Grep", { pattern: "x", path: "/etc" }), /outside the workspace/);
  blocked(decide(policy, "Glob", { pattern: "*", path: "/" }), /outside the workspace/);
});

test("glob patterns stay inside the workspace", (t) => {
  const pi = setup(t, "pi").policy;
  const claude = setup(t, "claude-code", ["find"]).policy;
  allowed(decide(pi, "find", { pattern: "**/*.mjs" }));
  allowed(decide(claude, "Glob", { pattern: "src/**/*.mjs" }));
  for (const pattern of ["/etc/*", "~/.ssh/*", "@/etc/*", "../*", "a/../../*", "..", " /etc/*"]) {
    blocked(decide(pi, "find", { pattern }), /pattern must stay inside/);
    blocked(decide(claude, "Glob", { pattern }), /pattern must stay inside/);
  }
  blocked(decide(claude, "Glob", { pattern: ["*"] }), /pattern must stay inside/);
});

test("a path that can't be checked is blocked", (t) => {
  const { policy } = setup(t, "pi");
  blocked(decide({ ...policy, workspace: join(policy.workspace, "missing") }, "read", { path: "a.txt" }), /can't be checked: ENOENT/);
});