stack/packages/harness/src/gate.mjs

// The tool gate both harnesses share. decide(policy, tool, input) answers
// one tool call: the policy's built-in tools and its typed tools pass,
// everything else is blocked, and every path argument of a file tool must
// resolve inside the workspace.
//
// This is a gate on the harness's own tool calls, not a sandbox. `bash`
// isn't confined (the role's tool limit decides whether it exists at all),
// and anything bash can reach, the session can reach. See the README's
// limits.

import { lstatSync, realpathSync } from "node:fs";
import { homedir } from "node:os";
import { basename, dirname, isAbsolute, join, resolve, sep } from "node:path";
import { fileURLToPath } from "node:url";

// pi built-in tool -> Claude Code tool. find and ls both map to Glob.
export const CLAUDE_TOOLS = Object.freeze({ read: "Read", write: "Write", edit: "Edit", bash: "Bash", grep: "Grep", find: "Glob", ls: "Glob" });
export const MCP_PREFIX = "mcp__mosaic__";

// The input field that holds a path, per tool, per harness. A missing
// optional path means the working directory, which is the workspace.
const PATH_FIELDS = {
  pi: { read: "path", write: "path", edit: "path", grep: "path", find: "path", ls: "path" },
  "claude-code": { Read: "file_path", Write: "file_path", Edit: "file_path", Grep: "path", Glob: "path" },
};

// Claude Code tool names the session may call, for --tools and the gate.
export function claudeBuiltins(policy) {
  return [...new Set(policy.tools.map((t) => CLAUDE_TOOLS[t]))];
}

// Pi's own path normalisation (dist/utils/paths.js normalizePath with the
// options resolveToCwd passes): unicode spaces, a leading @, ~ and file://.
const UNICODE_SPACES = /[  -   　]/g;
function normalise(p) {
  let s = p.replace(UNICODE_SPACES, " ");
  if (s.startsWith("@")) s = s.slice(1);
  if (s === "~") return homedir();
  if (s.startsWith("~/")) return join(homedir(), s.slice(2));
  if (/^file:\/\//.test(s)) return fileURLToPath(s);
  return s;
}

// Realpath of the nearest ancestor that exists as a name, with the missing
// tail kept. lstat, not exists: a dangling symlink exists as a name, and a
// write through it would create its target wherever that is. So a path that
// reaches a dangling symlink, at any depth, can't be checked and is refused.
function real(p) {
  let head = p;
  const tail = [];
  while (!lstatSync(head, { throwIfNoEntry: false })) {
    const up = dirname(head);
    if (up === head) break;
    tail.unshift(basename(head));
    head = up;
  }
  try {
    return join(realpathSync(head), ...tail);
  } catch (error) {
    if (error.code === "ENOENT") throw Object.assign(new Error("dangling symlink"), { code: "dangling-symlink" });
    throw error;
  }
}

function within(workspace, path) {
  const root = realpathSync(workspace);
  const target = real(path);
  return target === root || target.startsWith(root + sep);
}

// A relative path is resolved the way the tool resolves it: against its
// cwd. Both adapters cd into the workspace, and a process's cwd is the real
// path, so `..` climbs the real path's parents, not those of a workspace or
// dataRoot given through a symlink. It must be inside against the path as
// given too, so the check doesn't rest on how the harness was started.
export function insideWorkspace(workspace, p) {
  const s = normalise(p);
  if (isAbsolute(s)) return within(workspace, resolve(s));
  return within(workspace, resolve(realpathSync(workspace), s)) && within(workspace, resolve(workspace, s));
}

// Pi's read (dist/core/tools/path-utils.js resolveReadPathAsync) opens
// another spelling when the resolved path doesn't exist: U+202F before
// AM/PM, then NFD, then U+2019 for ', then both. Each applies to the whole
// resolved path, directory names included, and Pi resolves against its cwd,
// the workspace's real path. So every spelling that exists as a name is
// checked, not only the one Pi would pick, and the check doesn't depend on
// which of them exists when Pi opens it. Claude Code's Read gets the same
// check; its own retries aren't documented.
function spellings(workspace, p) {
  const s = normalise(p);
  const bases = isAbsolute(s) ? [resolve(s)] : [resolve(workspace, s), resolve(realpathSync(workspace), s)];
  const curly = (v) => v.replace(/'/g, "\u2019");
  const out = new Set();
  for (const r of bases) {
    const nfd = r.normalize("NFD");
    for (const v of [r.replace(/ (AM|PM)\./gi, "\u202F$1."), nfd, curly(r), curly(nfd)]) if (v !== r) out.add(v);
  }
  return out;
}

// The first spelling that exists and resolves outside, or through a
// dangling link, as a refusal reason; null if there is none.
function otherSpelling(workspace, tool, p) {
  for (const v of spellings(workspace, p)) {
    let found;
    try {
      found = lstatSync(v, { throwIfNoEntry: false });
    } catch (error) {
      // A file used as a directory: Pi's access() fails too.
      if (error.code === "ENOTDIR") continue;
      return `${tool} path can't be checked under another spelling: ${error.code ?? error.message}`;
    }
    if (!found) continue;
    try {
      if (!within(workspace, v)) return `${tool} path is outside the workspace under another spelling: ${p} -> ${JSON.stringify(v)}`;
    } catch (error) {
      if (error.code === "dangling-symlink") return `${tool} path goes through a dangling symlink under another spelling: ${p} -> ${JSON.stringify(v)}`;
      return `${tool} path can't be checked under another spelling: ${error.code ?? error.message}`;
    }
  }
  return null;
}

// decide(policy, tool, input) -> { allow: true } | { allow: false, reason }
//   policy  { harness: "pi" | "claude-code", workspace, tools: [pi names], typed: [typed tool names] }
export function decide(policy, tool, input) {
  const no = (reason) => ({ allow: false, reason: `mosaic gate: ${reason}` });
  if (typeof tool !== "string") return no("tool call without a name");
  const claude = policy.harness === "claude-code";
  if (claude ? tool.startsWith(MCP_PREFIX) && policy.typed.includes(tool.slice(MCP_PREFIX.length)) : policy.typed.includes(tool)) {
    return { allow: true };
  }
  const builtins = claude ? claudeBuiltins(policy) : policy.tools;
  if (!builtins.includes(tool)) return no(`${tool} isn't in this session's policy`);
  // A glob pattern is a second way to name a path: Glob and find take it
  // relative to their root, so an absolute pattern, ~ or a .. segment
  // could leave the workspace.
  const pattern = (claude ? tool === "Glob" : tool === "find") ? input?.pattern : undefined;
  if (pattern !== undefined && (typeof pattern !== "string" || /^[~/@]/.test(normalise(pattern).trim()) || /(^|[/\\])\.\.([/\\]|$)/.test(pattern))) {
    return no(`${tool} pattern must stay inside the workspace: ${pattern}`);
  }
  const field = PATH_FIELDS[policy.harness]?.[tool];
  if (!field) return { allow: true };
  const value = input?.[field];
  if (value === undefined || value === null || value === "") return { allow: true };
  if (typeof value !== "string") return no(`${tool}.${field} isn't a string`);
  try {
    if (!insideWorkspace(policy.workspace, value)) return no(`${tool} path is outside the workspace: ${value}`);
  } catch (error) {
    if (error.code === "dangling-symlink") return no(`${tool} path goes through a dangling symlink: ${value}`);
    return no(`${tool} path can't be checked: ${error.code ?? error.message}`);
  }
  if (tool === (claude ? "Read" : "read")) {
    let other;
    try {
      other = otherSpelling(policy.workspace, tool, value);
    } catch (error) {
      return no(`${tool} path can't be checked: ${error.code ?? error.message}`);
    }
    if (other) return no(other);
  }
  return { allow: true };
}