#!/usr/bin/env python3 """Check a NetSuite MCP Standard Tools call log against a finance policy. Uses only the Python standard library. Matrix and fixtures ship with this folder. Risk labels (read / attach / write / execute) are Agents Finance classifications based on Oracle's official tool list; they are not Oracle labels. Exit codes: 0 = all calls allowed, 1 = at least one blocked, 2 = usage / IO error. """ from __future__ import annotations import argparse import json import sys from pathlib import Path HERE = Path(__file__).resolve().parent DEFAULT_MATRIX = HERE / "mcp-standard-tools-matrix.json" def load_json(path: Path) -> dict: try: return json.loads(path.read_text(encoding="utf-8")) except FileNotFoundError: raise SystemExit(f"File not found: {path}") except json.JSONDecodeError as exc: raise SystemExit(f"Invalid JSON in {path}: {exc}") def index_tools(matrix: dict) -> dict[str, dict]: out = {} for tool in matrix.get("tools", []): name = tool.get("name") if not name: raise SystemExit("Matrix tool missing name") out[name] = tool return out def resolve_allowed(matrix: dict, policy: dict) -> tuple[set[str], set[str]]: """Return (allowed tool names, blocked tool names).""" tools = index_tools(matrix) profile_name = policy.get("profile") profiles = matrix.get("profiles", {}) if profile_name not in profiles: raise SystemExit( f"Unknown profile {profile_name!r}. Known: {', '.join(sorted(profiles))}" ) allow_risk = set(profiles[profile_name].get("allow_risk", [])) blocked_always = set(matrix.get("blocked_always", [])) deny = set(policy.get("deny_tools", [])) | blocked_always allowed: set[str] = set() for name, tool in tools.items(): if name in deny: continue if tool.get("risk") in allow_risk: allowed.add(name) return allowed, deny def check_calls(matrix: dict, policy: dict, log: dict) -> list[dict]: allowed, _deny = resolve_allowed(matrix, policy) tools = index_tools(matrix) findings: list[dict] = [] session_ok = True if policy.get("require_session_check"): first = (log.get("calls") or [None])[0] if not first or first.get("tool") != "ns_getSessionInfo": session_ok = False findings.append( { "index": 0, "tool": (first or {}).get("tool"), "status": "blocked", "reason": "policy.require_session_check: first call must be ns_getSessionInfo", } ) for i, call in enumerate(log.get("calls") or []): if not session_ok and i == 0: continue name = call.get("tool") if not name: findings.append( {"index": i, "tool": None, "status": "blocked", "reason": "call missing tool"} ) continue meta = tools.get(name) if meta is None: findings.append( { "index": i, "tool": name, "status": "blocked", "reason": "tool not in official MCP Standard Tools matrix (unknown / custom)", } ) continue if name not in allowed: findings.append( { "index": i, "tool": name, "status": "blocked", "risk": meta.get("risk"), "reason": ( f"tool risk {meta.get('risk')!r} not allowed for profile " f"{policy.get('profile')!r} (or listed in deny_tools / blocked_always)" ), } ) continue findings.append( { "index": i, "tool": name, "status": "allowed", "risk": meta.get("risk"), "reason": "ok", } ) return findings def self_test(matrix_path: Path) -> int: matrix = load_json(matrix_path) assert len(matrix["tools"]) == 25, len(matrix["tools"]) risks = {t["name"]: t["risk"] for t in matrix["tools"]} assert risks["ns_createRecord"] == "write" assert risks["ns_updateRecord"] == "write" assert risks["ns_transformRecord"] == "write" assert risks["ns_executeRecordAction"] == "execute" assert risks["ns_attachFile"] == "attach" assert risks["ns_runCustomSuiteQL"] == "read" policy_ro = load_json(HERE / "policy.example.json") ok = check_calls(matrix, policy_ro, load_json(HERE / "tool-calls-ok.json")) assert all(f["status"] == "allowed" for f in ok), ok bad_write = check_calls(matrix, policy_ro, load_json(HERE / "tool-calls-bad-write.json")) blocked = [f for f in bad_write if f["status"] == "blocked"] assert {f["tool"] for f in blocked} >= {"ns_createRecord", "ns_updateRecord"}, bad_write # Even a sandbox write profile must block execute policy_sandbox = { "profile": "close-operator-sandbox", "deny_tools": [], "require_session_check": False, } bad_exec = check_calls( matrix, policy_sandbox, load_json(HERE / "tool-calls-bad-execute.json") ) assert any( f["tool"] == "ns_executeRecordAction" and f["status"] == "blocked" for f in bad_exec ), bad_exec # Unknown tool fails closed unknown = check_calls( matrix, policy_ro, {"calls": [{"tool": "ns_getSessionInfo"}, {"tool": "ns_deleteEverything"}]}, ) assert any(f["tool"] == "ns_deleteEverything" and f["status"] == "blocked" for f in unknown) print("self-test: 6 checks passed (25 tools, ok log, write block, execute block, unknown tool)") return 0 def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--matrix", type=Path, default=DEFAULT_MATRIX) parser.add_argument("--policy", type=Path, help="policy JSON (see policy.example.json)") parser.add_argument("--calls", type=Path, help="tool-call log JSON") parser.add_argument("--self-test", action="store_true", help="run fixture tests and exit") args = parser.parse_args(argv) if args.self_test: return self_test(args.matrix) if not args.policy or not args.calls: parser.error("--policy and --calls are required unless --self-test") matrix = load_json(args.matrix) policy = load_json(args.policy) log = load_json(args.calls) findings = check_calls(matrix, policy, log) blocked = [f for f in findings if f["status"] == "blocked"] print(json.dumps({"label": log.get("label"), "profile": policy.get("profile"), "findings": findings}, indent=2)) if blocked: print(f"\nBLOCKED: {len(blocked)} call(s)", file=sys.stderr) return 1 print(f"\nOK: {len(findings)} call(s) allowed", file=sys.stderr) return 0 if __name__ == "__main__": raise SystemExit(main())