diff --git a/packages/agentveil-mcp-proxy/README.md b/packages/agentveil-mcp-proxy/README.md index 3a97ea5..83274cb 100644 --- a/packages/agentveil-mcp-proxy/README.md +++ b/packages/agentveil-mcp-proxy/README.md @@ -23,6 +23,13 @@ Use credential custody, egress boundaries, or API gates when an action must be controlled below the agent process. Those boundary patterns are preview and design-partner work, not general public release paths in this package. +## Claude Code connector + +For the project-local Claude Code connector — the hook that denies native +mutations with a redirect, the routed MCP approval/evidence path, setup path, +and its limits — see +[Claude Code Connector — Scope and Quickstart](docs/CLAUDE_CODE_SCOPE.md). + ## Install ```bash diff --git a/packages/agentveil-mcp-proxy/agentveil_mcp_proxy/approval/persistent.py b/packages/agentveil-mcp-proxy/agentveil_mcp_proxy/approval/persistent.py index 0aeec92..f82ee07 100644 --- a/packages/agentveil-mcp-proxy/agentveil_mcp_proxy/approval/persistent.py +++ b/packages/agentveil-mcp-proxy/agentveil_mcp_proxy/approval/persistent.py @@ -145,9 +145,14 @@ def _windows_process_alive(pid: int) -> bool: def _health_check(manifest: ApprovalCenterManifest) -> bool: - url = f"{manifest.approval_center_url()}/api/approvals" try: - return loopback_get_status(url, timeout=HEALTH_TIMEOUT_SECONDS) == 200 + return ( + loopback_get_status( + manifest.approval_center_url(), + timeout=HEALTH_TIMEOUT_SECONDS, + ) + == 200 + ) except (OSError, TimeoutError, ValueError): return False @@ -265,8 +270,6 @@ def _parse_http_response(response: bytes) -> tuple[int, bytes]: def manifest_is_reachable(manifest: ApprovalCenterManifest) -> bool: - if not is_process_alive(manifest.pid): - return False return _health_check(manifest) diff --git a/packages/agentveil-mcp-proxy/agentveil_mcp_proxy/claude_center_lifecycle.py b/packages/agentveil-mcp-proxy/agentveil_mcp_proxy/claude_center_lifecycle.py new file mode 100644 index 0000000..26f1856 --- /dev/null +++ b/packages/agentveil-mcp-proxy/agentveil_mcp_proxy/claude_center_lifecycle.py @@ -0,0 +1,262 @@ +"""Approval Center lifecycle for the one-command Claude Code connector. + +S5 added `setup claude-code --yes` to install the connector in one command, but +without an active local Approval Center the controlled MCP write path produces +URLs that hit a dead loopback port (``ERR_CONNECTION_REFUSED``) — the user sees +"empty windows with no approve". This module gives the public setup command +ownership of the Approval Center process lifecycle so the one-command promise +holds end-to-end: + +- ``ensure_running``: idempotent start of a project-local Approval Center using + the SAME ``--home`` / ``--config`` as the setup-managed proxy. If a manifest + exists and points at a live, healthy process, reuse it. If the manifest is + stale (no PID or PID is dead), restart. Returns a bounded status. +- ``check_status``: read-only health probe over the manifest + PID + loopback + HTTP. Returns ``running`` / ``down`` / ``stale``. +- ``stop_if_managed``: terminate ONLY a process whose PID is in the manifest + this module wrote. We will not kill someone else's center. + +Public/private boundary: this module manages the generic local Approval Center +process. It does not add hosted auth, license activation, policy packs, or +private playbook logic. +""" + +from __future__ import annotations + +import os +import signal +import subprocess +import sys +import time +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +from agentveil_mcp_proxy.approval.persistent import ( + ApprovalCenterManifest, + is_process_alive, + load_manifest, + loopback_get_status, + manifest_path, +) + + +# How long to wait for a freshly-spawned center to publish its manifest and +# answer a loopback health probe. Bounded so a stuck spawn cannot hang setup. +_START_TIMEOUT_SECONDS = 12.0 +_POLL_INTERVAL_SECONDS = 0.2 +_HEALTH_TIMEOUT_SECONDS = 1.5 + + +def _proxy_dir(home: Path) -> Path: + """Directory where the proxy + center share the manifest and config.""" + return Path(home) / "mcp-proxy" + + +def _center_health(manifest: ApprovalCenterManifest) -> bool: + try: + return ( + loopback_get_status( + manifest.approval_center_url(), + timeout=_HEALTH_TIMEOUT_SECONDS, + ) + == 200 + ) + except (OSError, TimeoutError, ValueError): + return False + + +@dataclass(frozen=True) +class CenterStatus: + """Bounded Approval Center status for setup reporting. + + ``state`` semantics: + - ``running``: manifest present, PID alive, health-check passes. + - ``down``: no manifest, or the manifest is structurally unreadable. + - ``stale``: manifest present but PID missing/dead OR health-check fails. + """ + + state: str # "running" | "down" | "stale" + pid: int | None + port: int | None + url: str | None # full approval URL when running, else None + + +def check_status(home: Path) -> CenterStatus: + """Read-only Approval Center health probe.""" + manifest = load_manifest(_proxy_dir(home)) + if manifest is None: + return CenterStatus(state="down", pid=None, port=None, url=None) + healthy = _center_health(manifest) + if healthy: + return CenterStatus( + state="running", + pid=manifest.pid, + port=manifest.port, + url=manifest.approval_center_url(), + ) + return CenterStatus(state="stale", pid=manifest.pid, port=manifest.port, url=None) + + +@dataclass(frozen=True) +class EnsureRunningResult: + """Outcome of an idempotent center start.""" + + status: CenterStatus + started: bool # True if this call spawned a new center + reused: bool # True if an existing healthy center was reused + restarted: bool # True if a stale center was replaced + reason: str # short bounded reason string + + +def _spawn_center( + *, + proxy_command: str, + home: Path, + passphrase_file: Path | None = None, +) -> subprocess.Popen[bytes]: + """Spawn `approval-center serve` as a detached background process. + + The center inherits the same ``--home`` and ``--config`` that the + setup-managed proxy uses, so they share manifest + sqlite store. We detach + stdio so the parent setup command can return without keeping a pipe open. + """ + config_path = _proxy_dir(home) / "config.json" + devnull = subprocess.DEVNULL + # On POSIX, start_new_session detaches from the parent so the setup command + # may exit while the center keeps running. On Windows, no-op kwargs. + kwargs: dict[str, Any] = { + "stdin": devnull, + "stdout": devnull, + "stderr": devnull, + "close_fds": True, + } + if os.name == "posix": + kwargs["start_new_session"] = True + args = [ + proxy_command, + "approval-center", + "serve", + "--home", str(home), + "--config", str(config_path), + "--port", "0", + ] + if passphrase_file is not None: + args.extend(["--passphrase-file", str(passphrase_file)]) + return subprocess.Popen( # noqa: S603 - args constructed from validated paths + args, + **kwargs, + ) + + +def _wait_for_running(home: Path, *, deadline: float) -> CenterStatus: + """Poll manifest + health until running or deadline.""" + last = CenterStatus(state="down", pid=None, port=None, url=None) + while time.monotonic() < deadline: + last = check_status(home) + if last.state == "running": + return last + time.sleep(_POLL_INTERVAL_SECONDS) + return last + + +def ensure_running( + *, + home: Path, + proxy_command: str, + passphrase_file: Path | None = None, +) -> EnsureRunningResult: + """Idempotently ensure a healthy project-local Approval Center is running. + + - If manifest, PID, and health check pass: reuse, do not respawn. + - If manifest is stale (PID gone or unhealthy): respawn. + - If no manifest: spawn. + + Returns a bounded status. If start fails the result's status.state stays + ``down``/``stale`` and ``started`` is False, so the caller (setup) can + refuse to claim ready/protected. + """ + initial = check_status(home) + if initial.state == "running": + return EnsureRunningResult( + status=initial, started=False, reused=True, restarted=False, + reason="center already running and healthy", + ) + + restarted = initial.state == "stale" + try: + _spawn_center( + proxy_command=proxy_command, + home=home, + passphrase_file=passphrase_file, + ) + except (OSError, ValueError) as exc: + return EnsureRunningResult( + status=initial, started=False, reused=False, restarted=False, + reason=f"could not spawn approval-center: {exc.__class__.__name__}", + ) + + deadline = time.monotonic() + _START_TIMEOUT_SECONDS + final = _wait_for_running(home, deadline=deadline) + if final.state == "running": + return EnsureRunningResult( + status=final, started=True, reused=False, restarted=restarted, + reason="center restarted" if restarted else "center started", + ) + return EnsureRunningResult( + status=final, started=False, reused=False, restarted=False, + reason="approval-center did not become healthy within the start timeout", + ) + + +def stop_if_managed(home: Path) -> dict[str, Any]: + """Stop ONLY a center whose PID is in the project manifest. + + Bounded result; no raw paths leaked. Idempotent: if no manifest or PID is + not ours, the call is a no-op. + """ + proxy_dir = _proxy_dir(home) + manifest = load_manifest(proxy_dir) + if manifest is None or manifest.pid is None: + return {"stopped": False, "reason": "no managed approval-center manifest"} + + if not _center_health(manifest): + if is_process_alive(manifest.pid): + return { + "stopped": False, + "reason": "manifest pid is not a healthy AgentVeil Approval Center; not stopped", + } + # The center is not reachable and the manifest PID is not alive. Remove + # it so a later setup does not see a stale running-looking record. + try: + manifest_path(proxy_dir).unlink(missing_ok=True) + except OSError: + pass + return {"stopped": False, "reason": "managed pid not alive; manifest cleared"} + + try: + os.kill(manifest.pid, signal.SIGTERM) + except ProcessLookupError: + return {"stopped": False, "reason": "managed pid already exited"} + except PermissionError: + return {"stopped": False, "reason": "no permission to stop managed pid"} + + # Best-effort wait for exit, then clear manifest. + for _ in range(20): + if not is_process_alive(manifest.pid): + break + time.sleep(0.1) + try: + manifest_path(proxy_dir).unlink(missing_ok=True) + except OSError: + pass + return {"stopped": True, "reason": "managed approval-center stopped"} + + +__all__ = [ + "CenterStatus", + "EnsureRunningResult", + "check_status", + "ensure_running", + "stop_if_managed", +] diff --git a/packages/agentveil-mcp-proxy/agentveil_mcp_proxy/claude_hook.py b/packages/agentveil-mcp-proxy/agentveil_mcp_proxy/claude_hook.py new file mode 100644 index 0000000..bf1e16d --- /dev/null +++ b/packages/agentveil-mcp-proxy/agentveil_mcp_proxy/claude_hook.py @@ -0,0 +1,677 @@ +"""Claude Code PreToolUse hook adapter for AgentVeil MCP Proxy. + +Productizes the P10D.11-C/C2 probe surface into a reusable module that maps +Claude Code `PreToolUse` hook payloads onto the existing AgentVeil policy +engine, returns Claude-compatible decision JSON, and writes bounded JSONL +evidence records. + +Scope for S1: +- Decision plumbing only (deny/allow); no approval round-trip. +- No installer / `.claude/settings.json` writer; no CLI subcommand wiring. +- Reuses ``PolicyEngine`` / ``ToolCallContext`` / ``RiskClass`` / + ``PolicyDecision`` from ``agentveil_mcp_proxy.policy``. There is no parallel + decision engine here. +- Evidence rows are privacy-bounded: ``tool_input`` is reduced to a SHA-256 + digest of its JCS-canonicalized form plus a sorted list of its top-level + key names. Raw prompt, file content, shell command bodies, tokens, and tool + output are not copied into the evidence record. + +Conservative fallback contract (spec clarification #2): +- Tool names that map to ``RiskClass.UNKNOWN`` are routed through the built-in + policy's ``default_decision`` which is ``ASK_BACKEND``. With no backend + available in S1, ``ASK_BACKEND`` maps to deny. This + keeps unknown mutation-shaped tools from silently allowing. +""" + +from __future__ import annotations + +import datetime as _dt +import hashlib +import json +import sys +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Mapping, MutableMapping + +from agentveil_mcp_proxy.classification import infer_action_family, infer_risk_class +from agentveil_mcp_proxy.policy import ( + PolicyConfig, + PolicyDecision, + PolicyEngine, + PolicyEvaluation, + ProxyConfig, + RiskClass, + ToolCallContext, +) + + +CLAUDE_SERVER_LABEL = "claude_code" +HOOK_EVENT_DEFAULT = "PreToolUse" + +# MCP server name written by `agentveil-mcp-proxy connect `. Tool calls +# routed to this server (``mcp__agentveil-mcp-proxy__*``) are the AgentVeil +# controlled route: they already pass through the proxy's own approval boundary. +# The PreToolUse hook must NOT double-block them — otherwise the redirect that +# tells the agent to "use the controlled MCP tool" dead-ends on the hook itself. +AGENTVEIL_CONTROLLED_MCP_SERVER = "agentveil-mcp-proxy" + +# Generic agent-facing redirect appended to NATIVE-tool deny reasons (S2 +# corrective). This is static instruction text only: no approval round-trip +# (S3), no auto-transformation of Write into an MCP call, and no private +# playbook content. It tells the agent to re-route the same intent through a +# controlled AgentVeil MCP tool when one is available. +# claim-check: allow "blocked" is literal hook-deny user-facing text; tested in +# tests/test_mcp_proxy_claude_hook.py native redirect assertions. +NATIVE_REDIRECT_INSTRUCTION = ( + "Direct native tool use was blocked before mutation. " # claim-check: allow literal hook-deny text tested below. + "Use an AgentVeil controlled MCP tool for the same operation when available, " + "preserving the same path, content, and intent. " + "If approval is required, ask the user to approve and then retry the controlled tool call." +) + + +# Claude Code built-in tool names and their natural risk class. Bash is +# special-cased at runtime because its input determines mutation vs read. +_CLAUDE_BUILTIN_RISK: Mapping[str, RiskClass] = { + "Write": RiskClass.WRITE, + "Edit": RiskClass.WRITE, + "MultiEdit": RiskClass.WRITE, + "NotebookEdit": RiskClass.WRITE, + "Read": RiskClass.READ, + "LS": RiskClass.READ, + "Glob": RiskClass.READ, + "Grep": RiskClass.READ, + "WebSearch": RiskClass.READ, + "WebFetch": RiskClass.READ, +} + +# Shell classifier uses a deny fallback: default-deny with a small allowlist of +# unambiguously read-only commands. Denylist heuristics catch obvious +# mutation shapes for telemetry (WRITE/DESTRUCTIVE risk_class), but ANYTHING +# not on the allowlist falls through to UNKNOWN -> ASK_BACKEND -> deny. +# +# Rationale: arbitrary interpreters (python3 -c, node -e, perl -e) and +# arbitrary subprocesses can write to the filesystem without matching any +# specific token pattern. Token-based denylists fail open on these; an +# allowlist of read-only commands denies unknown commands. The corrective fix for +# P10D.14 S1 reverses the original token-denylist design. + +_SHELL_DESTRUCTIVE_TOKENS: tuple[str, ...] = ( + "rm ", + "rmdir ", + "unlink ", + "shred ", + "wipe ", + " -delete", # `find ... -delete` +) + +_SHELL_MUTATION_TOKENS: tuple[str, ...] = ( + " > ", + " >> ", + " >|", + " tee ", + "mv ", + "cp ", + "mkdir ", + "touch ", + "chmod ", + "chown ", + " ln ", + "curl -o", + "wget -O", + " dd ", + " -exec", # `find -exec`, `xargs -I {} -exec` + " -i ", # `sed -i`, `perl -i` + " -pi", # `perl -pi` +) + +# Single-token executables that are unambiguously read-only when invoked +# without mutation operators (caught above by destructive/mutation tokens). +_BASH_READONLY_FIRST_TOKEN: frozenset[str] = frozenset({ + "pwd", + "ls", + "cat", + "head", + "tail", + "grep", + "find", + "wc", + "which", + "whoami", + "date", + "echo", + "true", + "false", +}) + +# Git subcommands that are read-only. Anything else (checkout, reset, clean, +# push, pull, commit, add, merge, rebase, ...) falls through to UNKNOWN. +_BASH_GIT_READONLY_SUBCOMMANDS: frozenset[str] = frozenset({ + "status", + "diff", +}) + +# Shell composition / metacharacters. Presence of ANY of these means the +# command is not a simple single read-only invocation: it can command- +# substitute, pipe, chain, background, or redirect into a mutation while the +# first token still looks read-only (e.g. ``echo $(python3 -c "...")``). +# +# Corrective-2 finding: a first-token allowlist is insufficient; an attacker +# can hide a write inside a substitution/pipe/chain. Any composition token => +# deny fallback (UNKNOWN -> deny), regardless of the first token. +_SHELL_COMPOSITION_PATTERNS: tuple[str, ...] = ( + "$(", # command substitution + "`", # backtick command substitution + "|", # pipe (also covers ||) + ";", # command separator + "&", # background and && chaining + ">", # any output redirect (>, >>, >|, >( ) + "<(", # process substitution (executes the inner command) + "\n", # embedded newline => multiple commands + "\r", +) + + +def _has_shell_composition(command: str) -> bool: + """Return True if the command contains shell composition metacharacters.""" + return any(pattern in command for pattern in _SHELL_COMPOSITION_PATTERNS) + + +def _classify_bash(command: str) -> RiskClass: + """Classify a Bash command. Unknown by default. + + Order of evaluation: + 1. Destructive tokens (rm, rmdir, find -delete) -> DESTRUCTIVE. + 2. Mutation tokens (redirects with spaces, sed -i, find -exec, ...) -> WRITE. + 3. Shell composition metacharacters ($(), backtick, |, ;, &, >, <(, newline) + -> UNKNOWN. A composed command is not simple enough to allowlist. + 4. First-token allowlist (ls, cat, grep, ...) on a simple command -> READ. + 5. `git ` with subcommand in the read-only subset -> READ. + 6. Anything else -> UNKNOWN (the policy then routes to ASK_BACKEND, which + the hook maps to deny in S1). + + Composition (step 3) is checked BEFORE the allowlist (steps 4-5) so a + read-looking first token does not carry a hidden mutation through. + """ + lowered = command.lower().strip() + if not lowered: + return RiskClass.UNKNOWN + + if any(tok in lowered for tok in _SHELL_DESTRUCTIVE_TOKENS): + return RiskClass.DESTRUCTIVE + if any(tok in lowered for tok in _SHELL_MUTATION_TOKENS): + return RiskClass.WRITE + + # Composition guard: nothing past this point may reach the READ allowlist + # unless it is a single, simple command. + if _has_shell_composition(lowered): + return RiskClass.UNKNOWN + + tokens = lowered.split() + if not tokens: + return RiskClass.UNKNOWN + + first = tokens[0] + if first == "git": + if len(tokens) >= 2 and tokens[1] in _BASH_GIT_READONLY_SUBCOMMANDS: + return RiskClass.READ + return RiskClass.UNKNOWN + + if first in _BASH_READONLY_FIRST_TOKEN: + return RiskClass.READ + + return RiskClass.UNKNOWN + + +def _split_mcp_tool_name(tool_name: str) -> tuple[str, str] | None: + """Return ``(server, tool_suffix)`` for ``mcp__server__tool`` or ``None``.""" + if not tool_name.startswith("mcp__"): + return None + parts = tool_name.split("__", 2) + if len(parts) < 3 or not parts[1] or not parts[2]: + return None + return parts[1], parts[2] + + +def classify_claude_tool( + tool_name: str, + tool_input: Mapping[str, Any] | None = None, +) -> RiskClass: + """Classify a Claude tool call into a ``RiskClass``. + + The classifier maps tool metadata into the engine's risk vocabulary. It + does NOT make the deny/allow decision; that responsibility belongs to the + policy engine (spec clarification: no parallel decision engine here). + """ + + if tool_name in _CLAUDE_BUILTIN_RISK: + return _CLAUDE_BUILTIN_RISK[tool_name] + if tool_name == "Bash": + command = "" + if isinstance(tool_input, Mapping): + command = str(tool_input.get("command") or "") + return _classify_bash(command) + mcp_split = _split_mcp_tool_name(tool_name) + if mcp_split is not None: + _server, tool_suffix = mcp_split + arguments = tool_input if isinstance(tool_input, Mapping) else None + return infer_risk_class( + action=tool_name, + tool=tool_suffix, + resource=None, + arguments=arguments, + ) + return RiskClass.UNKNOWN + + +def build_tool_call_context(payload: Mapping[str, Any]) -> ToolCallContext: + """Build a ``ToolCallContext`` from a Claude PreToolUse payload.""" + tool_name = str(payload.get("tool_name") or "") + tool_input = payload.get("tool_input") or {} + if not isinstance(tool_input, Mapping): + tool_input = {} + + risk = classify_claude_tool(tool_name, tool_input) + + mcp_split = _split_mcp_tool_name(tool_name) + if mcp_split is not None: + server, tool_suffix = mcp_split + action_family = infer_action_family(tool_suffix) + else: + server = CLAUDE_SERVER_LABEL + tool_suffix = tool_name or "unknown" + action_family = infer_action_family(tool_suffix) + + return ToolCallContext( + server=server, + tool=tool_suffix, + action=f"{server}.{tool_suffix}", + risk_class=risk, + action_family=action_family, + ) + + +def default_hook_policy() -> PolicyConfig: + """Built-in protect-mode policy for S1. + + Rules are minimal and risk-class driven: + - read -> allow + - write -> approval (treated as deny in S1; no approval surface) + - production -> approval # claim-check: allow "production" is RiskClass vocabulary. + - destructive -> block + - financial -> block + + Anything else (UNKNOWN) falls through to ``default_decision=ASK_BACKEND`` + which the hook maps to deny in S1 (spec clarification #2). + """ + return PolicyConfig.from_dict({ + "id": "claude_hook_s1_default", + "policy_schema_version": 1, + "default_decision": "ask_backend", + "default_risk_class": "unknown", + "rules": [ + { + "id": "claude-hook-read-allow", + "source": "builtin", + "decision": "allow", + "risk_class": "read", + "match": {"risk_class": ["read"]}, + }, + { + "id": "claude-hook-write-approval", + "source": "builtin", + "decision": "approval", + "risk_class": "write", + "match": {"risk_class": ["write"]}, + }, + { + "id": "claude-hook-prod-risk-approval", + "source": "builtin", + "decision": "approval", + "risk_class": "production", # claim-check: allow "production" is RiskClass vocabulary. + "match": {"risk_class": ["production"]}, # claim-check: allow "production" is RiskClass vocabulary. + }, + { + "id": "claude-hook-destructive-block", + "source": "builtin", + "decision": "block", + "risk_class": "destructive", + "match": {"risk_class": ["destructive"]}, + }, + { + "id": "claude-hook-financial-block", + "source": "builtin", + "decision": "block", + "risk_class": "financial", + "match": {"risk_class": ["financial"]}, + }, + ], + }) + + +def default_proxy_config_for_hook() -> ProxyConfig: + """Minimal ProxyConfig wrapping the hook default policy.""" + return ProxyConfig.from_dict({ + "proxy_config_schema_version": 1, + "avp": { + "base_url": "https://agentveil.dev", + "agent_name": "claude-hook", + "trusted_signer_dids": ["did:key:z6MktrustedSigner"], + }, + "mode": "protect", + "privacy": { + "action": "redacted", + "resource": "hash", + "payload": "hash_only", + "evidence_upload": False, + }, + "approval": { + "approval_timeout_seconds": 300, + "on_timeout": "deny", + "ui_open_mode": "none", + }, + "policy": { + "id": "claude_hook_s1_default", + "policy_schema_version": 1, + "default_decision": "ask_backend", + "default_risk_class": "unknown", + "rules": [ + { + "id": "claude-hook-read-allow", + "source": "builtin", + "decision": "allow", + "risk_class": "read", + "match": {"risk_class": ["read"]}, + }, + { + "id": "claude-hook-write-approval", + "source": "builtin", + "decision": "approval", + "risk_class": "write", + "match": {"risk_class": ["write"]}, + }, + { + "id": "claude-hook-prod-risk-approval", + "source": "builtin", + "decision": "approval", + "risk_class": "production", # claim-check: allow "production" is RiskClass vocabulary. + "match": {"risk_class": ["production"]}, # claim-check: allow "production" is RiskClass vocabulary. + }, + { + "id": "claude-hook-destructive-block", + "source": "builtin", + "decision": "block", + "risk_class": "destructive", + "match": {"risk_class": ["destructive"]}, + }, + { + "id": "claude-hook-financial-block", + "source": "builtin", + "decision": "block", + "risk_class": "financial", + "match": {"risk_class": ["financial"]}, + }, + ], + }, + }) + + +@dataclass(frozen=True) +class HookDecision: + """Result of one PreToolUse evaluation.""" + + hook_action: str # "allow" or "deny" + reason_code: str + context: ToolCallContext + evaluation: PolicyEvaluation + + +_FAIL_CLOSED_DECISIONS = frozenset({ + PolicyDecision.BLOCK, + PolicyDecision.APPROVAL, + PolicyDecision.ASK_BACKEND, +}) + + +def _reason_code(evaluation: PolicyEvaluation, hook_action: str) -> str: + if hook_action == "deny": + if evaluation.decision is PolicyDecision.BLOCK: + return "risky_blocked" + if evaluation.decision is PolicyDecision.APPROVAL: + # S1 has no approval surface; approval-required becomes deny. + return "risky_blocked" + if evaluation.decision is PolicyDecision.ASK_BACKEND: + # Conservative fallback when no backend exists. + return "risky_blocked" + return "risky_blocked" + return "allowed" + + +def decide(payload: Mapping[str, Any], engine: PolicyEngine) -> HookDecision: + """Evaluate one PreToolUse payload and return the hook decision.""" + context = build_tool_call_context(payload) + evaluation = engine.evaluate(context) + + # Controlled-route pass-through: calls to the AgentVeil proxy's own MCP + # tools self-govern at the proxy's approval boundary. The hook allows them + # through (instead of denying write-shaped MCP tools on this controlled route) so the redirect + # to "use the controlled MCP tool" is reachable. The proxy, not the hook, + # then applies approval/redirect/evidence to these calls. + if context.server == AGENTVEIL_CONTROLLED_MCP_SERVER: + return HookDecision( + hook_action="allow", + reason_code="controlled_route_passthrough", + context=context, + evaluation=evaluation, + ) + + if evaluation.decision in (PolicyDecision.ALLOW, PolicyDecision.OBSERVE): + hook_action = "allow" + elif evaluation.decision in _FAIL_CLOSED_DECISIONS: + hook_action = "deny" + else: + # Defensive: any decision we don't recognize fails closed. + hook_action = "deny" + return HookDecision( + hook_action=hook_action, + reason_code=_reason_code(evaluation, hook_action), + context=context, + evaluation=evaluation, + ) + + +def format_hook_output(decision: HookDecision) -> str | None: + """Format Claude-compatible PreToolUse JSON, or ``None`` to allow silently. + + For native Claude tools (Write/Edit/MultiEdit/NotebookEdit/Bash), the deny + reason carries a generic redirect instruction so the message is actionable + for the agent, not just "denied". MCP tool denies keep the bounded base + reason (the redirect-to-MCP instruction does not apply to an MCP call). + """ + if decision.hook_action == "allow": + return None + reason = ( + f"agentveil: denied {decision.context.tool} " + f"(risk_class={decision.evaluation.risk_class.value}, " + f"policy_decision={decision.evaluation.decision.value}, " + f"reason_code={decision.reason_code}); target_reached=false" + ) + if decision.context.server == CLAUDE_SERVER_LABEL: + reason = f"{reason}. {NATIVE_REDIRECT_INSTRUCTION}" + return json.dumps({ + "hookSpecificOutput": { + "hookEventName": HOOK_EVENT_DEFAULT, + "permissionDecision": "deny", + "permissionDecisionReason": reason, + } + }) + + +def _bounded_input_ref(tool_input: Any) -> dict[str, Any]: + """Privacy-bounded reference: SHA-256 prefix of canonical input + key names. + + The full raw tool_input is not returned from this function; only the hash prefix + and the sorted list of top-level keys are returned. This is the explicit + contract that test_evidence_bounded asserts via sentinel value injection. + """ + if not isinstance(tool_input, Mapping): + keys: list[str] = [] + canonical = json.dumps({"_non_dict": True}, separators=(",", ":")) + else: + keys = sorted(str(k) for k in tool_input.keys()) + canonical = json.dumps(tool_input, sort_keys=True, separators=(",", ":"), default=str) + digest = hashlib.sha256(canonical.encode("utf-8")).hexdigest() + return { + "input_hash": f"sha256:{digest[:16]}", + "input_keys": keys, + } + + +def _bounded_cwd_ref(cwd: str) -> str: + """Privacy-bounded reference for the workspace cwd. + + Raw cwd is a customer workspace path and must not appear verbatim in + evidence (corrective finding: P10D.14 S1 review). Returns a SHA-256 + digest prefix so audit trails can still compare same-cwd sessions + without leaking the path. + """ + if not cwd: + return "sha256:empty" + digest = hashlib.sha256(cwd.encode("utf-8")).hexdigest() + return f"sha256:{digest[:16]}" + + +def build_evidence_record( + payload: Mapping[str, Any], + decision: HookDecision, + *, + now: _dt.datetime | None = None, +) -> dict[str, Any]: + """Build a bounded JSONL evidence record. + + Privacy contract (corrective S1 review, sentinel-tested): + - No raw ``tool_input`` values copied (only ``input_hash`` + ``input_keys``). + - No raw prompt, file content, shell command body, tokens, or tool output. + - No raw workspace ``cwd``; recorded as ``cwd_digest`` (SHA-256 prefix). + - ``tool_name``, ``session_id``, and derived fields (server/tool/risk_class) + are recorded as standard policy metadata. + """ + timestamp = (now or _dt.datetime.now(_dt.timezone.utc)).isoformat() + tool_input = payload.get("tool_input") + record: dict[str, Any] = { + "ts": timestamp, + "session_id": str(payload.get("session_id") or ""), + "cwd_digest": _bounded_cwd_ref(str(payload.get("cwd") or "")), + "hook_event_name": str(payload.get("hook_event_name") or HOOK_EVENT_DEFAULT), + "tool_name": str(payload.get("tool_name") or ""), + "server": decision.context.server, + "tool": decision.context.tool, + "action_family": decision.context.action_family or "", + "risk_class": decision.evaluation.risk_class.value, + "policy_decision": decision.evaluation.decision.value, + "hook_action": decision.hook_action, + "reason_code": decision.reason_code, + "policy_id": decision.evaluation.policy_id, + "policy_rule_id": decision.evaluation.policy_rule_id, + "matched_rule_ids": list(decision.evaluation.matched_rule_ids), + "target_reached": False if decision.hook_action == "deny" else None, + "input_ref": _bounded_input_ref(tool_input), + } + return record + + +def write_evidence(record: Mapping[str, Any], evidence_path: Path) -> None: + """Append one bounded record as a JSONL line.""" + evidence_path.parent.mkdir(parents=True, exist_ok=True) + with evidence_path.open("a", encoding="utf-8") as fh: + fh.write(json.dumps(record, separators=(",", ":"), default=str) + "\n") + + +def process_hook( + payload: Mapping[str, Any], + *, + config: ProxyConfig | None = None, + evidence_path: Path | None = None, + out: Any = None, +) -> HookDecision: + """Process one PreToolUse payload end-to-end. + + Returns the ``HookDecision`` so callers (and tests) can inspect the + structured outcome. Side effects: + - Appends a bounded evidence record to ``evidence_path`` when provided. + - Writes Claude-compatible deny JSON to ``out`` when the decision is deny. + """ + config = config or default_proxy_config_for_hook() + engine = PolicyEngine(config) + decision = decide(payload, engine) + if evidence_path is not None: + record = build_evidence_record(payload, decision) + write_evidence(record, evidence_path) + output = format_hook_output(decision) + if output is not None: + if out is None: + out = sys.stdout + out.write(output + "\n") + return decision + + +def main(argv: list[str] | None = None, *, stdin: Any = None, stdout: Any = None) -> int: + """Hook runtime entrypoint, invoked by Claude Code as the PreToolUse command. + + Reads a PreToolUse payload from stdin, processes it, writes the + hookSpecificOutput JSON (or nothing for allow) to stdout, and returns 0. + + Evidence path resolution order: + 1. ``--evidence-path `` argument (set by the installed hook command). + 2. ``AGENTVEIL_HOOK_EVIDENCE_PATH`` environment variable. + 3. None (no evidence written). + + The ``--evidence-path`` argument is preferred because it is cross-platform + (no shell ``VAR=x cmd`` prefix needed) and is what S2's installer writes + into the project ``.claude/settings.json`` hook command. + """ + import argparse + import os + + parser = argparse.ArgumentParser(prog="agentveil-claude-hook", add_help=True) + parser.add_argument("--evidence-path", default=None) + args = parser.parse_args(argv if argv is not None else []) + + in_stream = stdin if stdin is not None else sys.stdin + out_stream = stdout if stdout is not None else sys.stdout + try: + payload = json.load(in_stream) + except Exception as exc: # noqa: BLE001 - external input + sys.stderr.write(f"claude_hook: invalid PreToolUse JSON: {exc}\n") + return 1 + if not isinstance(payload, Mapping): + sys.stderr.write("claude_hook: PreToolUse payload must be a JSON object\n") + return 1 + + evidence_arg = args.evidence_path or os.environ.get("AGENTVEIL_HOOK_EVIDENCE_PATH") + evidence_path = Path(evidence_arg) if evidence_arg else None + process_hook(payload, evidence_path=evidence_path, out=out_stream) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main(sys.argv[1:])) + + +__all__ = [ + "AGENTVEIL_CONTROLLED_MCP_SERVER", + "CLAUDE_SERVER_LABEL", + "HOOK_EVENT_DEFAULT", + "NATIVE_REDIRECT_INSTRUCTION", + "HookDecision", + "build_evidence_record", + "build_tool_call_context", + "classify_claude_tool", + "decide", + "default_hook_policy", + "default_proxy_config_for_hook", + "format_hook_output", + "main", + "process_hook", + "write_evidence", +] diff --git a/packages/agentveil-mcp-proxy/agentveil_mcp_proxy/claude_hook_setup.py b/packages/agentveil-mcp-proxy/agentveil_mcp_proxy/claude_hook_setup.py new file mode 100644 index 0000000..9cc551c --- /dev/null +++ b/packages/agentveil-mcp-proxy/agentveil_mcp_proxy/claude_hook_setup.py @@ -0,0 +1,596 @@ +"""Project-local Claude Code hook install / status / uninstall (P10D.14 S2). + +Turns the S1 ``claude_hook`` adapter into a usable project-local install +surface. Writes a merge-preserving AgentVeil-managed ``PreToolUse`` entry into a +project ``.claude/settings.json`` without disturbing unrelated settings or +hooks, reports a bounded status, and removes only the managed entry. + +Scope for S2: +- Project scope only (``--project``). No user/global Claude config. +- No approval round-trip (S3), no docs/release (S4/S5). +- Status language is scoped strictly to Claude Code project hooks; it makes no + host-wide protection claim. + +The managed hook entry is identified by ``AGENTVEIL_HOOK_MARKER`` appearing in +its command string. We do not add non-schema marker keys to the settings JSON, +so an unrelated tool parsing the file will not choke on AgentVeil fields. +""" + +from __future__ import annotations + +import json +import shlex +import sys +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + + +# Substring that identifies an AgentVeil-managed hook command. The installed +# command generated by this module invokes this module path, so its presence in +# a command string marks the entry as ours for idempotent upsert and removal. +AGENTVEIL_HOOK_MARKER = "agentveil_mcp_proxy.claude_hook" + +# Tool classes the hook intercepts: Claude built-in mutators + any MCP tool. +HOOK_MATCHER = "Write|Edit|MultiEdit|NotebookEdit|Bash|mcp__.*" + +# Tool-class labels surfaced in bounded status (no regex internals leaked). +MATCHED_TOOL_CLASSES = ("Write", "Edit", "MultiEdit", "NotebookEdit", "Bash", "mcp__*") + + +class HookSetupError(RuntimeError): + """Raised when install/uninstall cannot proceed safely.""" + + +# ----- path helpers --------------------------------------------------------- + + +def project_claude_dir(project_dir: Path) -> Path: + return Path(project_dir) / ".claude" + + +def project_settings_path(project_dir: Path) -> Path: + return project_claude_dir(project_dir) / "settings.json" + + +def project_evidence_path(project_dir: Path) -> Path: + """Project-local, AgentVeil-scoped evidence file.""" + return project_claude_dir(project_dir) / "agentveil" / "evidence.jsonl" + + +# ----- managed hook entry --------------------------------------------------- + + +def build_hook_command(*, python: str, evidence_path: Path) -> str: + """Build the PreToolUse command string that runs the installed module. + + The interpreter path and the evidence path are shell-quoted with + ``shlex.quote`` so paths containing spaces or quotes do not break the + command Claude Code runs via the shell. The ``-m `` token stays + literal so status detection (`-m agentveil_mcp_proxy.claude_hook`) and the + ``--evidence-path`` marker remain matchable. + """ + return ( + f"{shlex.quote(python)} -m {AGENTVEIL_HOOK_MARKER} " + f"--evidence-path {shlex.quote(str(evidence_path))}" + ) + + +def build_managed_hook_entry(*, python: str, evidence_path: Path) -> dict[str, Any]: + """Build the AgentVeil-managed PreToolUse group entry.""" + return { + "matcher": HOOK_MATCHER, + "hooks": [ + { + "type": "command", + "command": build_hook_command(python=python, evidence_path=evidence_path), + } + ], + } + + +def _entry_is_agentveil_managed(entry: Any) -> bool: + """True if a PreToolUse group entry is an AgentVeil-managed one.""" + if not isinstance(entry, dict): + return False + hooks = entry.get("hooks") + if not isinstance(hooks, list): + return False + for hook in hooks: + if isinstance(hook, dict) and AGENTVEIL_HOOK_MARKER in str(hook.get("command", "")): + return True + return False + + +def _command_points_to_module(entry: Any) -> bool: + """True if the managed entry's command invokes this module via ``-m``.""" + if not isinstance(entry, dict): + return False + for hook in entry.get("hooks", []) or []: + if isinstance(hook, dict): + command = str(hook.get("command", "")) + if f"-m {AGENTVEIL_HOOK_MARKER}" in command: + return True + return False + + +def _hook_is_agentveil(hook: Any) -> bool: + """True for a single inner hook object whose command is AgentVeil-managed.""" + return isinstance(hook, dict) and AGENTVEIL_HOOK_MARKER in str(hook.get("command", "")) + + +def _strip_managed_from_group(group: Any) -> tuple[Any, int]: + """Remove only AgentVeil inner hooks from one PreToolUse group. + + Returns ``(group_or_None, removed_count)``. Operates at the inner-hook + object level so a group that mixes a user hook and an AgentVeil hook keeps + the user hook. If stripping empties the group's ``hooks`` list, the group + is dropped (returns ``None``); otherwise the group is returned with only + the non-AgentVeil hooks preserved. + """ + if not isinstance(group, dict): + return group, 0 + hooks = group.get("hooks") + if not isinstance(hooks, list): + return group, 0 + kept_hooks = [h for h in hooks if not _hook_is_agentveil(h)] + removed = len(hooks) - len(kept_hooks) + if removed == 0: + return group, 0 + if not kept_hooks: + return None, removed + new_group = dict(group) + new_group["hooks"] = kept_hooks + return new_group, removed + + +# ----- settings IO (invalid JSON stops writes) ------------------------------- + + +def load_settings(settings_path: Path) -> dict[str, Any]: + """Load settings JSON. Returns {} if absent. + + Raises HookSetupError on invalid JSON so callers stop WITHOUT rewriting a + file they could not parse. + """ + if not settings_path.exists(): + return {} + try: + raw = settings_path.read_text(encoding="utf-8") + except OSError as exc: + raise HookSetupError(f"cannot read {settings_path}: {exc}") from exc + if not raw.strip(): + return {} + try: + data = json.loads(raw) + except json.JSONDecodeError as exc: + raise HookSetupError( + f"existing {settings_path} is not valid JSON; refusing to overwrite ({exc})" + ) from exc + if not isinstance(data, dict): + raise HookSetupError(f"existing {settings_path} must be a JSON object") + return data + + +def _write_settings(settings_path: Path, data: dict[str, Any]) -> None: + settings_path.parent.mkdir(parents=True, exist_ok=True) + settings_path.write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8") + + +def _pre_tool_use_list(settings: dict[str, Any]) -> list[Any]: + hooks = settings.get("hooks") + if not isinstance(hooks, dict): + return [] + pre = hooks.get("PreToolUse") + return pre if isinstance(pre, list) else [] + + +# ----- results -------------------------------------------------------------- + + +@dataclass(frozen=True) +class InstallResult: + settings_path: Path + evidence_path: Path + created_settings: bool + replaced_existing_managed: bool + reload_required: bool = True + + +@dataclass(frozen=True) +class UninstallResult: + settings_path: Path + removed_entries: int + settings_existed: bool + reload_required: bool + + +@dataclass +class StatusResult: + scope: str = "project" + status: str = "unsafe" # protected | advisory | unsafe + state: str = "missing" # installed | missing | stale | invalid-json + settings_present: bool = False + managed_hook_present: bool = False + hook_command_points_to_module: bool = False + evidence_path_configured: bool = False + reload_required: bool = False + matched_tool_classes: tuple[str, ...] = field(default_factory=lambda: MATCHED_TOOL_CLASSES) + notes: tuple[str, ...] = () + + def to_bounded_dict(self) -> dict[str, Any]: + """Bounded JSON. No absolute paths, prompts, payloads, or commands.""" + return { + "scope": self.scope, + "status": self.status, + "state": self.state, + "settings_present": self.settings_present, + "managed_hook_present": self.managed_hook_present, + "hook_command_points_to_module": self.hook_command_points_to_module, + "evidence_path_configured": self.evidence_path_configured, + "reload_required": self.reload_required, + "matched_tool_classes": list(self.matched_tool_classes), + "notes": list(self.notes), + } + + +# ----- install / uninstall / status ----------------------------------------- + + +def install_hook( + project_dir: Path, + *, + python: str | None = None, + evidence_path: Path | None = None, +) -> InstallResult: + """Install/upsert the AgentVeil PreToolUse hook in project settings. + + Idempotent and merge-preserving: preserves unrelated settings and hooks, removes + any prior AgentVeil-managed entries, and appends exactly one fresh managed + entry. Fails closed (raises) on invalid existing JSON without rewriting. + """ + project_dir = Path(project_dir) + settings_path = project_settings_path(project_dir) + resolved_python = python or sys.executable + resolved_evidence = evidence_path or project_evidence_path(project_dir) + + created = not settings_path.exists() + settings = load_settings(settings_path) # raises on invalid JSON + + hooks = settings.get("hooks") + if not isinstance(hooks, dict): + hooks = {} + pre = hooks.get("PreToolUse") + if not isinstance(pre, list): + pre = [] + + # Strip any prior AgentVeil inner hooks (idempotent upsert) at the + # hook-object level, preserving user hooks that share a group. + cleaned: list[Any] = [] + removed_total = 0 + for group in pre: + new_group, removed = _strip_managed_from_group(group) + removed_total += removed + if new_group is not None: + cleaned.append(new_group) + replaced = removed_total > 0 + + cleaned.append(build_managed_hook_entry(python=resolved_python, evidence_path=resolved_evidence)) + hooks["PreToolUse"] = cleaned + settings["hooks"] = hooks + + _write_settings(settings_path, settings) + + return InstallResult( + settings_path=settings_path, + evidence_path=resolved_evidence, + created_settings=created, + replaced_existing_managed=replaced, + ) + + +def uninstall_hook(project_dir: Path) -> UninstallResult: + """Remove only AgentVeil-managed PreToolUse entries. Idempotent.""" + project_dir = Path(project_dir) + settings_path = project_settings_path(project_dir) + + if not settings_path.exists(): + return UninstallResult( + settings_path=settings_path, + removed_entries=0, + settings_existed=False, + reload_required=False, + ) + + settings = load_settings(settings_path) # raises on invalid JSON + hooks = settings.get("hooks") + if not isinstance(hooks, dict): + return UninstallResult(settings_path, 0, True, reload_required=False) + + pre = hooks.get("PreToolUse") + if not isinstance(pre, list): + return UninstallResult(settings_path, 0, True, reload_required=False) + + # Remove only AgentVeil inner hooks; preserve user hooks in mixed groups. + cleaned: list[Any] = [] + removed = 0 + for group in pre: + new_group, removed_here = _strip_managed_from_group(group) + removed += removed_here + if new_group is not None: + cleaned.append(new_group) + + if removed == 0: + return UninstallResult(settings_path, 0, True, reload_required=False) + + # Clean only containers we may have generated, and only when empty. + if cleaned: + hooks["PreToolUse"] = cleaned + else: + hooks.pop("PreToolUse", None) + if not hooks: + settings.pop("hooks", None) + + _write_settings(settings_path, settings) + return UninstallResult( + settings_path=settings_path, + removed_entries=removed, + settings_existed=True, + reload_required=True, + ) + + +def status_hook(project_dir: Path, *, evidence_path: Path | None = None) -> StatusResult: + """Return bounded project-hook status. + + Status semantics (scoped to Claude Code project hooks only): + - ``protected``: managed hook present, command points to the installed + module, AND local evidence proves the hook has actually fired + (project evidence file exists and is non-empty). Config correctness + alone is NOT enough to claim protected — we cannot observe whether the + running Claude Code process reloaded the config, so protected requires + real run evidence. + - ``advisory``: managed hook present and command points to the module, but + no firing evidence yet (reload/restart likely required) OR the command + is a stale form that does not invoke the installed module. + - ``unsafe``: no settings, no managed hook, or invalid JSON. + + This deliberately does not return ``protected`` for a freshly installed + hook with ``reload_required=true`` — that would overclaim protection before + the hook has run. + """ + project_dir = Path(project_dir) + settings_path = project_settings_path(project_dir) + result = StatusResult() + + if not settings_path.exists(): + result.settings_present = False + result.state = "missing" + result.status = "unsafe" + result.notes = ("no project .claude/settings.json; hook not installed",) + return result + + result.settings_present = True + try: + settings = load_settings(settings_path) + except HookSetupError: + result.state = "invalid-json" + result.status = "unsafe" + result.notes = ("project settings.json is not valid JSON",) + return result + + pre = _pre_tool_use_list(settings) + managed = [e for e in pre if _entry_is_agentveil_managed(e)] + if not managed: + result.managed_hook_present = False + result.state = "missing" + result.status = "unsafe" + result.notes = ("no AgentVeil-managed PreToolUse hook found",) + return result + + result.managed_hook_present = True + points_to_module = any(_command_points_to_module(e) for e in managed) + result.hook_command_points_to_module = points_to_module + result.evidence_path_configured = any( + "--evidence-path" in str(h.get("command", "")) + for e in managed + for h in (e.get("hooks") or []) + if isinstance(h, dict) + ) + + if not points_to_module: + result.state = "stale" + result.status = "advisory" + result.reload_required = True + result.notes = ( + "managed hook present but command does not invoke the installed module; reinstall", + ) + return result + + # Config is correct. Only claim protected with real firing evidence that + # post-dates the current install. A leftover evidence file from a previous + # install (settings.json is rewritten on each install) must NOT count, or + # a reinstall would falsely report protected before the new hook has run. + result.state = "installed" + ev_path = Path(evidence_path) if evidence_path is not None else project_evidence_path(project_dir) + try: + settings_mtime = settings_path.stat().st_mtime + fired = ( + ev_path.exists() + and ev_path.stat().st_size > 0 + and ev_path.stat().st_mtime > settings_mtime + ) + except OSError: + fired = False + + if fired: + result.status = "protected" + result.reload_required = False + result.notes = ( + "managed hook installed and local evidence shows it has fired", + ) + else: + result.status = "advisory" + result.reload_required = True + result.notes = ( + "managed hook installed but not yet proven to fire; restart Claude Code " + "and run a tool to confirm (no evidence recorded yet)", + ) + return result + + +# ----- project .mcp.json route (written by `connect claude_code`) ------------ + +# mcpServers key that `connect claude_code` writes into the project .mcp.json. +AGENTVEIL_MCP_SERVER_NAME = "agentveil-mcp-proxy" + + +def project_mcp_config_path(project_dir: Path) -> Path: + return Path(project_dir) / ".mcp.json" + + +def load_mcp_config(path: Path) -> dict[str, Any]: + """Load a project .mcp.json. Returns {} if absent. Refuses bad JSON.""" + if not path.exists(): + return {} + try: + raw = path.read_text(encoding="utf-8") + except OSError as exc: + raise HookSetupError(f"cannot read {path}: {exc}") from exc + if not raw.strip(): + return {} + try: + data = json.loads(raw) + except json.JSONDecodeError as exc: + raise HookSetupError( + f"existing {path} is not valid JSON; refusing to overwrite ({exc})" + ) from exc + if not isinstance(data, dict): + raise HookSetupError(f"existing {path} must be a JSON object") + return data + + +def mcp_route_present(project_dir: Path) -> bool: + """True if the AgentVeil MCP route is configured in the project .mcp.json.""" + try: + data = load_mcp_config(project_mcp_config_path(project_dir)) + except HookSetupError: + return False + servers = data.get("mcpServers") + return isinstance(servers, dict) and AGENTVEIL_MCP_SERVER_NAME in servers + + +@dataclass(frozen=True) +class RemoveMcpRouteResult: + config_path: Path + removed: bool + config_existed: bool + + +def remove_mcp_route(project_dir: Path) -> RemoveMcpRouteResult: + """Remove only the AgentVeil mcpServers entry from project .mcp.json. + + Preserves unrelated MCP servers and unrelated keys. Idempotent. Invalid + JSON raises without rewriting. + """ + path = project_mcp_config_path(project_dir) + if not path.exists(): + return RemoveMcpRouteResult(config_path=path, removed=False, config_existed=False) + + data = load_mcp_config(path) # raises on invalid JSON + servers = data.get("mcpServers") + if not isinstance(servers, dict) or AGENTVEIL_MCP_SERVER_NAME not in servers: + return RemoveMcpRouteResult(config_path=path, removed=False, config_existed=True) + + servers.pop(AGENTVEIL_MCP_SERVER_NAME, None) + if servers: + data["mcpServers"] = servers + else: + data.pop("mcpServers", None) + path.write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8") + return RemoveMcpRouteResult(config_path=path, removed=True, config_existed=True) + + +# ----- aggregated connector status ------------------------------------------ + + +def connector_status( + project_dir: Path, + *, + evidence_path: Path | None = None, + proxy_route_present: bool = False, +) -> dict[str, Any]: + """Bounded one-command connector status. + + Aggregates the project-local hook status, the .mcp.json route, and a + caller-supplied proxy-route flag into a single product-facing summary. No + absolute paths, prompts, secrets, or payloads. + """ + hook = status_hook(project_dir, evidence_path=evidence_path) + route_present = mcp_route_present(project_dir) + + if hook.state == "invalid-json": + hook_state = "stale" + elif not hook.managed_hook_present: + hook_state = "missing" + elif hook.state == "stale": + hook_state = "stale" + elif hook.status == "protected": + hook_state = "fired" + else: + hook_state = "installed" + + configured = hook.managed_hook_present and route_present and proxy_route_present + if not configured or hook.state == "invalid-json": + status = "unsafe" + elif hook.status == "protected": + status = "protected" + else: + status = "advisory" + + if not hook.managed_hook_present or not route_present or not proxy_route_present: + restart_required: bool | None = None # setup incomplete; reload not the issue yet + next_step = "run `agentveil-mcp-proxy setup claude-code --yes`" + elif hook.status == "protected": + restart_required = False + next_step = "connector active; nothing to do" + else: + restart_required = True + next_step = "restart Claude Code, then run `agentveil-mcp-proxy setup status`" + + return { + "scope": "project", + "status": status, + "hook": hook_state, + "mcp_route": "configured" if route_present else "missing", + "proxy_route": "configured" if proxy_route_present else "missing", + "restart_required": restart_required, + "matched_tool_classes": list(MATCHED_TOOL_CLASSES), + "next_step": next_step, + } + + +__all__ = [ + "AGENTVEIL_HOOK_MARKER", + "AGENTVEIL_MCP_SERVER_NAME", + "HOOK_MATCHER", + "MATCHED_TOOL_CLASSES", + "HookSetupError", + "InstallResult", + "RemoveMcpRouteResult", + "StatusResult", + "UninstallResult", + "build_hook_command", + "build_managed_hook_entry", + "connector_status", + "install_hook", + "load_mcp_config", + "load_settings", + "mcp_route_present", + "project_evidence_path", + "project_mcp_config_path", + "project_settings_path", + "remove_mcp_route", + "status_hook", + "uninstall_hook", +] diff --git a/packages/agentveil-mcp-proxy/agentveil_mcp_proxy/cli.py b/packages/agentveil-mcp-proxy/agentveil_mcp_proxy/cli.py index 2926023..98798b4 100644 --- a/packages/agentveil-mcp-proxy/agentveil_mcp_proxy/cli.py +++ b/packages/agentveil-mcp-proxy/agentveil_mcp_proxy/cli.py @@ -3659,9 +3659,21 @@ def build_parser() -> argparse.ArgumentParser: setup_status = setup_subparsers.add_parser( "status", - help="Report setup status from actual proxy and client config files", + help=( + "Without --home: project-local Claude connector status. " + "With --home: adaptive setup wizard status." + ), + ) + # --home is optional (option A): bare `setup status` reports the project + # connector status; `setup status --home ` keeps the adaptive wizard + # behavior, back-compatible with existing callers. + setup_status.add_argument("--home", type=Path, default=None, help="AVP home to inspect (wizard mode)") + setup_status.add_argument( + "--project-dir", + type=Path, + default=None, + help="Project directory for connector status (default: current directory)", ) - setup_status.add_argument("--home", type=Path, required=True, help="AVP home to inspect") setup_status.add_argument( "--client", default="cursor", @@ -3702,6 +3714,109 @@ def build_parser() -> argparse.ArgumentParser: ) _add_json_arg(setup_restore) + setup_claude_code = setup_subparsers.add_parser( + "claude-code", + help="One-command Claude Code connector setup (proxy route + MCP route + hook)", + ) + setup_claude_code.add_argument( + "--project-dir", type=Path, default=None, + help="Project directory to set up (default: current directory)", + ) + setup_claude_code.add_argument( + "--choose-folder", action="store_true", + help="Open the macOS folder picker to choose the project directory", + ) + setup_claude_code.add_argument( + "--sandbox", type=Path, default=None, + help="Quickstart filesystem sandbox path (default: selected project directory)", + ) + setup_claude_code.add_argument( + "--yes", action="store_true", + help="Apply the setup; without it the command only previews", + ) + setup_claude_code.add_argument( + "--passphrase-file", type=Path, default=None, + help="Encrypt the proxy identity with this passphrase file (default: plaintext quickstart)", + ) + setup_claude_code.add_argument( + "--force", action="store_true", + help="Re-initialize the proxy route even if its config already exists", + ) + _add_json_arg(setup_claude_code) + + setup_remove = setup_subparsers.add_parser( + "remove", + help="Remove a one-command connector (AgentVeil-managed entries only)", + ) + setup_remove.add_argument("target", choices=["claude-code"], help="Connector to remove") + setup_remove.add_argument( + "--project-dir", type=Path, default=None, + help="Project directory to remove from (default: current directory)", + ) + setup_remove.add_argument( + "--yes", action="store_true", + help="Apply the removal; without it the command only previews", + ) + _add_json_arg(setup_remove) + + install_claude_hook = subparsers.add_parser( + "install-claude-hook", + help="Install the AgentVeil PreToolUse hook into a project's .claude/settings.json", + ) + install_claude_hook.add_argument( + "--project", + action="store_true", + required=True, + help="Project scope (required; the only supported scope in this release)", + ) + install_claude_hook.add_argument( + "--project-dir", + type=Path, + default=None, + help="Project directory to install into (default: current directory)", + ) + install_claude_hook.add_argument( + "--yes", + action="store_true", + help="Proceed with the write; without it the command only previews", + ) + _add_json_arg(install_claude_hook) + + status_claude_hook = subparsers.add_parser( + "status-claude-hook", + help="Show AgentVeil Claude Code project hook status (bounded)", + ) + status_claude_hook.add_argument( + "--project", action="store_true", required=True, help="Project scope (required)" + ) + status_claude_hook.add_argument( + "--project-dir", + type=Path, + default=None, + help="Project directory to inspect (default: current directory)", + ) + _add_json_arg(status_claude_hook) + + uninstall_claude_hook = subparsers.add_parser( + "uninstall-claude-hook", + help="Remove the AgentVeil PreToolUse hook from a project's .claude/settings.json", + ) + uninstall_claude_hook.add_argument( + "--project", action="store_true", required=True, help="Project scope (required)" + ) + uninstall_claude_hook.add_argument( + "--project-dir", + type=Path, + default=None, + help="Project directory to uninstall from (default: current directory)", + ) + uninstall_claude_hook.add_argument( + "--yes", + action="store_true", + help="Proceed with the removal; without it the command only previews", + ) + _add_json_arg(uninstall_claude_hook) + return parser @@ -3733,6 +3848,452 @@ def _normalize_downstream_arg_values(argv: list[str]) -> list[str]: return normalized +def run_install_claude_hook_cli( + *, project_dir: Path | None, assume_yes: bool, output_json: bool +) -> int: + """Install/upsert the AgentVeil PreToolUse hook into a project.""" + from agentveil_mcp_proxy import claude_hook_setup + + target = Path(project_dir) if project_dir is not None else Path.cwd() + settings_path = claude_hook_setup.project_settings_path(target) + rel_settings = ".claude/settings.json" + + if not assume_yes: + message = ( + f"Would install the AgentVeil PreToolUse hook into {settings_path}. " + "Re-run with --yes to write it." + ) + if output_json: + _print_json({ + "ok": False, + "action": "install-claude-hook", + "applied": False, + "errors": ["confirmation required: pass --yes"], + "warnings": [message], + "settings_relpath": rel_settings, + }) + else: + print(message) + return 0 + + try: + result = claude_hook_setup.install_hook(target) + except claude_hook_setup.HookSetupError as exc: + raise ProxyCliError(str(exc), exit_code=1) from exc + + if output_json: + _print_operator_json({ + "ok": True, + "action": "install-claude-hook", + "applied": True, + "scope": "project", + "settings_relpath": rel_settings, + "evidence_relpath": ".claude/agentveil/evidence.jsonl", + "created_settings": result.created_settings, + "replaced_existing_managed": result.replaced_existing_managed, + "reload_required": result.reload_required, + "matched_tool_classes": list(claude_hook_setup.MATCHED_TOOL_CLASSES), + }) + else: + print(f"Installed AgentVeil PreToolUse hook: {settings_path}") + print(f"Evidence: {result.evidence_path}") + print("Restart Claude Code to load the hook.") + return 0 + + +def run_status_claude_hook_cli(*, project_dir: Path | None, output_json: bool) -> int: + """Print bounded project hook status.""" + from agentveil_mcp_proxy import claude_hook_setup + + target = Path(project_dir) if project_dir is not None else Path.cwd() + status = claude_hook_setup.status_hook(target) + bounded = status.to_bounded_dict() + if output_json: + _print_operator_json(bounded) + else: + print(f"status: {bounded['status']} ({bounded['state']})") + print(f"managed hook present: {bounded['managed_hook_present']}") + print(f"command points to module: {bounded['hook_command_points_to_module']}") + print(f"reload required: {bounded['reload_required']}") + for note in bounded["notes"]: + print(f"- {note}") + return 0 + + +def run_uninstall_claude_hook_cli( + *, project_dir: Path | None, assume_yes: bool, output_json: bool +) -> int: + """Remove only AgentVeil-managed hook entries from a project.""" + from agentveil_mcp_proxy import claude_hook_setup + + target = Path(project_dir) if project_dir is not None else Path.cwd() + settings_path = claude_hook_setup.project_settings_path(target) + + if not assume_yes: + message = ( + f"Would remove the AgentVeil PreToolUse hook from {settings_path}. " + "Re-run with --yes to apply." + ) + if output_json: + _print_json({ + "ok": False, + "action": "uninstall-claude-hook", + "applied": False, + "errors": ["confirmation required: pass --yes"], + "warnings": [message], + "settings_relpath": ".claude/settings.json", + }) + else: + print(message) + return 0 + + try: + result = claude_hook_setup.uninstall_hook(target) + except claude_hook_setup.HookSetupError as exc: + raise ProxyCliError(str(exc), exit_code=1) from exc + + if output_json: + _print_operator_json({ + "ok": True, + "action": "uninstall-claude-hook", + "applied": True, + "scope": "project", + "settings_existed": result.settings_existed, + "removed_entries": result.removed_entries, + "reload_required": result.reload_required, + }) + else: + if result.removed_entries: + print(f"Removed {result.removed_entries} AgentVeil hook entry(ies): {settings_path}") + print("Restart Claude Code to drop the hook.") + else: + print("No AgentVeil-managed hook entry found; nothing to remove.") + return 0 + + +def _setup_claude_code_home(project_dir: Path) -> Path: + """Project-local proxy home for the one-command connector.""" + return project_dir / ".avp" + + +def _choose_setup_project_folder() -> Path: + """Open the native macOS folder picker and return the selected directory.""" + if sys.platform != "darwin": + raise ProxyCliError("--choose-folder is currently supported on macOS only", exit_code=2) + import subprocess + + script = ( + 'POSIX path of (choose folder with prompt ' + '"Choose the project folder to protect with AgentVeil")' + ) + try: + result = subprocess.run( # noqa: S603 - fixed osascript invocation + ["osascript", "-e", script], + check=False, + capture_output=True, + text=True, + timeout=120, + ) + except (OSError, subprocess.TimeoutExpired) as exc: + raise ProxyCliError( + f"could not open folder picker: {exc.__class__.__name__}", + exit_code=2, + ) from exc + if result.returncode != 0: + raise ProxyCliError("folder selection cancelled", exit_code=2) + selected = result.stdout.strip() + if not selected: + raise ProxyCliError("folder selection returned an empty path", exit_code=2) + return Path(selected) + + +def _resolve_setup_proxy_command() -> str | None: + """Resolve the console script used by generated Claude setup config.""" + import shutil + + invoked = Path(sys.argv[0]) + if invoked.exists() and invoked.name == "agentveil-mcp-proxy": + return str(invoked.resolve()) + resolved = shutil.which("agentveil-mcp-proxy") + if resolved: + return resolved + return None + + +def _configure_claude_setup_approval_ui(config_path: Path) -> None: + """Disable browser auto-open for Claude setup-managed approvals. + + Claude Code already displays the exact pending approval URL in chat. Opening + the bare Approval Center dashboard from the proxy creates a confusing empty + browser tab during first-run setup and approval retries. + """ + config_payload = _read_json(config_path, "proxy config") + approval = config_payload.get("approval") + if not isinstance(approval, dict): + approval = {} + config_payload["approval"] = approval + approval["ui_open_mode"] = ApprovalUiOpenMode.TERMINAL.value + _secure_write_json(config_path, config_payload, force=True) + + +def run_setup_claude_code_cli( + *, + project_dir: Path | None, + choose_folder: bool, + sandbox: Path | None, + assume_yes: bool, + passphrase_file: Path | None, + force: bool, + output_json: bool, +) -> int: + """One-command Claude Code connector setup. + + Orchestrates the existing primitives end-to-end: proxy quickstart route + (`init_proxy`), Claude MCP route (`connect claude_code`), and the + project-local hook (`install_hook`). It does not duplicate policy/approval + logic. It does not claim Claude Code has reloaded; it states restart is + required. + """ + from agentveil_mcp_proxy import claude_hook_setup + + if choose_folder and project_dir is not None: + raise ProxyCliError("--choose-folder cannot be combined with --project-dir", exit_code=2) + selected_project = _choose_setup_project_folder() if choose_folder else project_dir + target = (Path(selected_project) if selected_project is not None else Path.cwd()).resolve() + sandbox_path = Path(sandbox).resolve() if sandbox is not None else target + home = _setup_claude_code_home(target) + config_path = home / "mcp-proxy" / "config.json" + + if not assume_yes: + message = ( + f"Would set up the Claude Code connector in {target}: proxy quickstart " + f"route, project .mcp.json, and the project hook. Re-run with --yes." + ) + if output_json: + _print_json({ + "ok": False, "action": "setup-claude-code", "applied": False, + "errors": ["confirmation required: pass --yes"], "warnings": [message], + }) + else: + print(message) + return 0 + + quiet = io.StringIO() + # 1. Proxy quickstart route (idempotent unless --force). + proxy_present = config_path.exists() + proxy_initialized = False + if force or not proxy_present: + downstream = quickstart_filesystem_downstream(sandbox_path) + try: + init_proxy( + home=home, + config_path=None, + downstream_config=downstream, + plaintext=(passphrase_file is None), + passphrase_file=passphrase_file, + force=force, + err=quiet, + ) + except ProxyCliError: + raise + proxy_initialized = True + if config_path.exists(): + _configure_claude_setup_approval_ui(config_path) + + # 2. Claude Code MCP route (.mcp.json) — reuse the connect command. Resolve + # the installed console script explicitly (matches the supported manual + # `connect --proxy-command "$(which agentveil-mcp-proxy)"` path). The route + # is considered configured once .mcp.json carries the AgentVeil entry; a + # non-zero launch-proof is not fatal to setup if the route was written. + resolved_proxy_command = _resolve_setup_proxy_command() + run_connect_cli( + client_id="claude_code", + home=home, + project_root=target, + write=True, + proxy_command=resolved_proxy_command, + out=quiet, + ) + if not claude_hook_setup.mcp_route_present(target): + raise ProxyCliError( + "could not write the Claude MCP route; ensure agentveil-mcp-proxy is " + "installed and on PATH", + exit_code=1, + ) + + # 3. Project-local hook. + try: + install_result = claude_hook_setup.install_hook(target) + except claude_hook_setup.HookSetupError as exc: + raise ProxyCliError(str(exc), exit_code=1) from exc + + # 4. Approval Center: take ownership of lifecycle. Without a live center, + # controlled MCP write approvals would surface URLs that the user cannot + # open (ERR_CONNECTION_REFUSED). Setup must not claim ready in that case. + from agentveil_mcp_proxy import claude_center_lifecycle + + proxy_cmd_for_center = resolved_proxy_command + if not proxy_cmd_for_center: + raise ProxyCliError( + "agentveil-mcp-proxy console script not on PATH; cannot start Approval Center", + exit_code=1, + ) + center_result = claude_center_lifecycle.ensure_running( + home=home, + proxy_command=proxy_cmd_for_center, + passphrase_file=passphrase_file, + ) + center_running = center_result.status.state == "running" + + # 5. Bounded status summary. + status = claude_hook_setup.connector_status(target, proxy_route_present=config_path.exists()) + sandbox_label = "project" if sandbox is None else "custom" + + if not center_running: + if output_json: + _print_json({ + "ok": False, + "action": "setup-claude-code", + "applied": True, + "scope": "project", + "proxy_route_initialized": proxy_initialized, + "mcp_route_relpath": ".mcp.json", + "hook_relpath": ".claude/settings.json", + "sandbox_relpath": sandbox_label, + "identity_encrypted": passphrase_file is not None, + "approval_center": {"state": center_result.status.state, "reason": center_result.reason}, + "errors": [ + "approval_center not running; controlled MCP writes would surface " + "unreachable approval URLs. Setup is not ready/protected." + ], + "warnings": [], + }) + else: + print("Claude Code connector partially set up:") + print(f" proxy route: {'initialized' if proxy_initialized else 'already present'}") + print(" MCP route: .mcp.json written") + print(" hook: .claude/settings.json present") + print(f" approval_center: {center_result.status.state} ({center_result.reason})") + print("ERROR: setup is NOT ready — Approval Center could not start.", file=sys.stderr) + return 1 + + if output_json: + _print_operator_json({ + "ok": True, + "action": "setup-claude-code", + "applied": True, + "scope": "project", + "proxy_route_initialized": proxy_initialized, + "mcp_route_relpath": ".mcp.json", + "hook_relpath": ".claude/settings.json", + "sandbox_relpath": sandbox_label, + "identity_encrypted": passphrase_file is not None, + "approval_center": { + "state": center_result.status.state, + "started": center_result.started, + "reused": center_result.reused, + "restarted": center_result.restarted, + }, + "reload_required": True, + "status": status, + }) + else: + print("Claude Code connector set up for this project:") + print(f" proxy route: {'initialized' if proxy_initialized else 'already present'}") + print(" MCP route: .mcp.json written") + print(f" hook: .claude/settings.json ({'created' if install_result.created_settings else 'updated'})") + if passphrase_file is None: + print(" identity: stored unencrypted (quickstart); use --passphrase-file to encrypt") + action = "reused" if center_result.reused else ("restarted" if center_result.restarted else "started") + print(f" approval_center: running ({action})") + print(f" status: {status['status']}") + print("Restart Claude Code for this project, then run `agentveil-mcp-proxy setup status`.") + return 0 + + +def run_setup_connector_status_cli(*, project_dir: Path | None, output_json: bool) -> int: + """Project-local Claude connector status (bare `setup status`).""" + from agentveil_mcp_proxy import claude_center_lifecycle, claude_hook_setup + + target = (Path(project_dir) if project_dir is not None else Path.cwd()).resolve() + home = _setup_claude_code_home(target) + config_path = home / "mcp-proxy" / "config.json" + status = claude_hook_setup.connector_status(target, proxy_route_present=config_path.exists()) + center = claude_center_lifecycle.check_status(home) + status["approval_center"] = center.state + # Setup is not "ready" unless the center is running; downgrade product + # status accordingly so we never say protected with a dead approval path. + if center.state != "running" and status["status"] != "unsafe": + status["status"] = "unsafe" if status["mcp_route"] == "missing" else "advisory" + if output_json: + _print_operator_json(status) + else: + print(f"status: {status['status']}") + print(f" hook: {status['hook']}") + print(f" mcp route: {status['mcp_route']}") + print(f" proxy route:{status['proxy_route']}") + print(f" approval_center: {center.state}") + print(f" restart required: {status['restart_required']}") + print(f" next: {status['next_step']}") + return 0 + + +def run_setup_remove_claude_code_cli( + *, project_dir: Path | None, assume_yes: bool, output_json: bool +) -> int: + """Remove the one-command Claude connector — AgentVeil-managed entries only.""" + from agentveil_mcp_proxy import claude_hook_setup + + target = (Path(project_dir) if project_dir is not None else Path.cwd()).resolve() + + if not assume_yes: + message = ( + f"Would remove AgentVeil-managed Claude hook and MCP route entries from " + f"{target} (unrelated settings and MCP servers preserved). Re-run with --yes." + ) + if output_json: + _print_json({ + "ok": False, "action": "setup-remove-claude-code", "applied": False, + "errors": ["confirmation required: pass --yes"], "warnings": [message], + }) + else: + print(message) + return 0 + + from agentveil_mcp_proxy import claude_center_lifecycle + + try: + hook_result = claude_hook_setup.uninstall_hook(target) + route_result = claude_hook_setup.remove_mcp_route(target) + except claude_hook_setup.HookSetupError as exc: + raise ProxyCliError(str(exc), exit_code=1) from exc + + center_stop = claude_center_lifecycle.stop_if_managed(_setup_claude_code_home(target)) + removed_any = ( + hook_result.removed_entries > 0 + or route_result.removed + or center_stop["stopped"] + ) + if output_json: + _print_operator_json({ + "ok": True, + "action": "setup-remove-claude-code", + "applied": True, + "scope": "project", + "hook_entries_removed": hook_result.removed_entries, + "mcp_route_removed": route_result.removed, + "approval_center_stopped": center_stop["stopped"], + "reload_required": removed_any, + }) + else: + print(f"Removed: hook entries={hook_result.removed_entries}, mcp route={route_result.removed}, " + f"approval_center={'stopped' if center_stop['stopped'] else 'not running'}") + print("Unrelated Claude settings and MCP servers preserved.") + if removed_any: + print("Restart Claude Code for this project to drop the connector.") + return 0 + + def main(argv: list[str] | None = None) -> int: parser = build_parser() parse_argv = sys.argv[1:] if argv is None else argv @@ -4129,6 +4690,13 @@ def main(argv: list[str] | None = None) -> int: output_json=args.json_output, ) if args.setup_action == "status": + # Option A: bare `setup status` = project connector status; + # `setup status --home ` = existing adaptive wizard status. + if args.home is None: + return run_setup_connector_status_cli( + project_dir=args.project_dir, + output_json=args.json_output, + ) return print_setup_status_cli( home=args.home, client_id=args.client, @@ -4143,7 +4711,44 @@ def main(argv: list[str] | None = None) -> int: client_id=args.client, output_json=args.json_output, ) - raise ProxyCliError("setup action must be run, status, or restore") + if args.setup_action == "claude-code": + return run_setup_claude_code_cli( + project_dir=args.project_dir, + choose_folder=args.choose_folder, + sandbox=args.sandbox, + assume_yes=args.yes, + passphrase_file=args.passphrase_file, + force=args.force, + output_json=args.json_output, + ) + if args.setup_action == "remove": + if args.target != "claude-code": + raise ProxyCliError("setup remove target must be claude-code") + return run_setup_remove_claude_code_cli( + project_dir=args.project_dir, + assume_yes=args.yes, + output_json=args.json_output, + ) + raise ProxyCliError( + "setup action must be claude-code, status, remove, run, or restore" + ) + if args.command == "install-claude-hook": + return run_install_claude_hook_cli( + project_dir=args.project_dir, + assume_yes=args.yes, + output_json=args.json_output, + ) + if args.command == "status-claude-hook": + return run_status_claude_hook_cli( + project_dir=args.project_dir, + output_json=args.json_output, + ) + if args.command == "uninstall-claude-hook": + return run_uninstall_claude_hook_cli( + project_dir=args.project_dir, + assume_yes=args.yes, + output_json=args.json_output, + ) except ( ProxyCliError, ApprovalEvidenceError, diff --git a/packages/agentveil-mcp-proxy/docs/CLAUDE_CODE_SCOPE.md b/packages/agentveil-mcp-proxy/docs/CLAUDE_CODE_SCOPE.md new file mode 100644 index 0000000..fbe53fe --- /dev/null +++ b/packages/agentveil-mcp-proxy/docs/CLAUDE_CODE_SCOPE.md @@ -0,0 +1,147 @@ +# Claude Code Connector — Scope and Quickstart + +The AgentVeil MCP Proxy can be installed as a project-local connector for +Claude Code. The connector makes Claude ask before changing files in the +configured project: direct native mutations are denied before they run, Claude +can retry through the AgentVeil-managed write path, and the MCP Proxy owns +approval plus bounded evidence. + +This document describes only the public connector path and its limits. + +## What the connector controls + +- A project-local Claude Code `PreToolUse` hook installed into + `./.claude/settings.json`. +- Native Claude Code mutation tools (`Write`, `Edit`, `MultiEdit`, + `NotebookEdit`, mutating `Bash`) are denied **before** the mutation, with a + short instruction for Claude to retry the same change through AgentVeil. +- MCP tool calls routed to the `agentveil-mcp-proxy` server pass through the + hook and are governed by the MCP Proxy itself (classification, approval, + evidence). +- Bounded JSONL evidence for hook decisions under + `./.claude/agentveil/evidence.jsonl`. Evidence rows carry hashes and bounded + references only — no raw prompt, file content, shell command body, tokens, or + full tool payload. + +## Setup UX + +The intended product setup is one project command after package install: + +```bash +pip install agentveil-mcp-proxy +agentveil-mcp-proxy setup claude-code --choose-folder --yes +``` + +The folder picker selects the project to protect without typing a path. Terminal +users can also run the setup from inside the project folder: + +```bash +agentveil-mcp-proxy setup claude-code --yes +``` + +That one-command setup is the target public UX. The same connector can still be +configured with the lower-level primitives below for diagnostics. Do not treat +the lower-level sequence as the final product setup shape. + +## Current lower-level setup + +```bash +pip install agentveil-mcp-proxy + +# initialize a local proxy with a sandboxed filesystem downstream +agentveil-mcp-proxy init --quickstart-filesystem ./sandbox + +# install the project-local Claude Code PreToolUse hook +agentveil-mcp-proxy install-claude-hook --project --yes + +# write the project-local Claude Code MCP route (.mcp.json) +agentveil-mcp-proxy connect claude_code --write + +# show bounded connector status +agentveil-mcp-proxy status-claude-hook --project --json +``` + +Both setup shapes require restarting / reloading Claude Code for the project so +it loads the hook and the MCP route. To remove the connector with the current +lower-level command: + +```bash +agentveil-mcp-proxy uninstall-claude-hook --project --yes +``` + +`uninstall-claude-hook` removes only the AgentVeil-managed hook entry and +preserves unrelated Claude Code settings and hooks. Restart Claude Code after +removal. + +## Expected flow + +1. The agent attempts a native mutation (for example `Write`). The project hook + denies it before the mutation and tells Claude to retry the same change + through AgentVeil. +2. The agent calls the AgentVeil MCP write tool + (`mcp__agentveil-mcp-proxy__*`). The hook passes that call through; the MCP + Proxy classifies it and requires approval for a write. +3. An approval is surfaced through the MCP Proxy approval path. The target file + is not written before approval. +4. After approval, a retry of the same controlled MCP call writes the file under + the configured project root or explicit sandbox, and the MCP Proxy records + bounded evidence for the routed decision. + +The MCP Proxy owns approval and evidence for AgentVeil-managed writes. The +hook's role is to stop direct native mutations before they bypass that approval +step. +Claude setup does not auto-open a browser tab for the bare Approval Center +dashboard; Claude Code shows the exact pending approval URL when an action +requires approval. + +## Status meaning (connector-local only) + +`status-claude-hook` reports connector-local truth about the project hook: + +- `unsafe` — no managed hook, missing settings, or unparseable settings. +- `advisory` — the managed hook is installed and points at the installed + module, but it has not been proven to fire yet (restart/reload likely + required). +- `protected` — the managed hook is installed **and** local evidence shows it + has fired after the current install. + +This status is scoped to the project Claude Code hook only. It is not a +host-wide or machine-wide protection signal. + +## Limits + +The connector is deliberately narrow. It does **not** claim host-wide control. + +- **Project-local only.** The hook and MCP route apply to the project where you + installed them. The connector does not modify user/global Claude Code config + and does not protect other projects. +- **Not host-wide.** The connector controls only Claude Code tool calls in the + configured project. It does not monitor or control the machine. +- **`claude --bare` bypasses the hook.** Claude Code's `--bare` mode skips + hooks; actions run under `--bare` are not governed by the connector. +- **Out-of-band actions are outside control.** Manual edits in an IDE, external + terminals, and direct filesystem changes are not Claude Code tool calls and + are not controlled. +- **Only configured AgentVeil calls are controlled.** Native mutation tools are + denied; AgentVeil-managed MCP calls are governed by the proxy. Calls that are + not routed through the connector are not classified or logged by it. +- **Controlled writes stay under the configured root.** The one-command Claude + setup uses the current project folder by default. The lower-level + `--quickstart-filesystem ./sandbox` primitive writes within the configured + sandbox path. +- **Claude Code may prompt first.** Claude Code can show its own MCP tool + permission prompt before the AgentVeil approval step. +- **Exact approval scope.** Approval is bound to the exact action payload, so a + retry whose content differs from what was approved may require a fresh + approval. This protects against approving one action and executing a + different one. + +## Not in this connector + +This connector is the minimal public path. It does not include advanced team +policy packages, hosted custody, or any host-wide/all-terminal capability. + + +See [MCP Proxy Operations](../../../docs/MCP_PROXY_OPERATIONS.md) for downstream +lifecycle and [Data Handling](../../../docs/DATA_HANDLING.md) for the evidence +privacy model. diff --git a/packages/agentveil-mcp-proxy/tests/test_mcp_proxy_approval.py b/packages/agentveil-mcp-proxy/tests/test_mcp_proxy_approval.py index aa44934..106ee44 100644 --- a/packages/agentveil-mcp-proxy/tests/test_mcp_proxy_approval.py +++ b/packages/agentveil-mcp-proxy/tests/test_mcp_proxy_approval.py @@ -3565,3 +3565,62 @@ def test_invalid_approval_token_returns_403_html_without_leaking_pending(): _assert_stale_html_privacy_safe(response.text, session_token=server.session_token) finally: server.stop() + + +# ----- P10D.14 S3 follow-up: G4 approve->retry payload-drift diagnosis ------- +# +# Root cause of the live G4 failure: a controlled write_file retry through a +# *separate* Claude session produced the same path but different content bytes +# (e.g. a trailing newline), so payload_hash drifted while resource_hash stayed +# identical. exact-scope approval is payload-bound, so it correctly did NOT # claim-check: allow approval binding property asserted by tests below. +# cover the changed-content retry. similar_5m is resource-bound and +# payload-AGNOSTIC (store.find_active_similar_grant has no payload_hash filter), +# so enabling it for filesystem writes would approve *changed content* to the +# same path for 5 minutes — which the slice explicitly forbids. These tests pin +# that security property so the "broad similar" fix cannot be introduced by +# accident. + + +def test_g4_changed_content_changes_payload_but_not_resource() -> None: + from agentveil_mcp_proxy.classification import ( + extract_resource, + sha256_jcs, + sha256_text, + ) + + args_a = {"path": "config.py", "content": "FEATURE_X=true"} + args_b = {"path": "config.py", "content": "FEATURE_X=true\n"} # trailing-newline drift + + # Same path -> identical resource label and resource_hash. + assert extract_resource(args_a) == extract_resource(args_b) + assert sha256_text(extract_resource(args_a)) == sha256_text(extract_resource(args_b)) + + # Different content -> different payload_hash. This is why an exact-scope + # approval of args_a does NOT cover an args_b retry (security-correct). + assert sha256_jcs(args_a) != sha256_jcs(args_b) + + +def test_g4_identical_retry_keeps_same_payload_hash() -> None: + """Exact retry stability: a byte-identical retry hashes the same, so an + exact grant would be reused (the in-session interactive retry path).""" + from agentveil_mcp_proxy.classification import sha256_jcs + + args = {"path": "config.py", "content": "FEATURE_X=true"} + assert sha256_jcs(args) == sha256_jcs(dict(args)) + + +def test_g4_similar_grant_matching_is_payload_agnostic_by_design() -> None: + """Guard: find_active_similar_grant does not constrain payload_hash, so + enabling similar_5m for filesystem writes would cover changed content. + This documents WHY similar_5m is rejected for the controlled write route.""" + import inspect + + from agentveil_mcp_proxy.evidence.store import ApprovalEvidenceStore + + src = inspect.getsource(ApprovalEvidenceStore.find_active_similar_grant) + # The SQL match binds resource_hash but must not bind payload_hash. + assert "resource_hash" in src + assert "payload_hash" not in src, ( + "similar grant matching now references payload_hash; re-verify the G4 " + "scope analysis before enabling similar_5m for filesystem writes" + ) diff --git a/packages/agentveil-mcp-proxy/tests/test_mcp_proxy_claude_hook.py b/packages/agentveil-mcp-proxy/tests/test_mcp_proxy_claude_hook.py new file mode 100644 index 0000000..79e6414 --- /dev/null +++ b/packages/agentveil-mcp-proxy/tests/test_mcp_proxy_claude_hook.py @@ -0,0 +1,740 @@ +"""Tests for agentveil_mcp_proxy.claude_hook (P10D.14 S1). + +Covers the seven spec cases (Write/Edit deny, Bash deny, read allow, +MCP write deny, MCP read allow, evidence bounded, unknown maps to deny) +plus three implementer clarifications (sentinel-based privacy, explicit +ASK_BACKEND deny-fallback semantics, JSONL append correctness). +""" + +from __future__ import annotations + +import io +import json +import os +from pathlib import Path + +import pytest + +from agentveil_mcp_proxy import claude_hook +from agentveil_mcp_proxy.claude_hook import ( + CLAUDE_SERVER_LABEL, + HookDecision, + _bounded_input_ref, + build_evidence_record, + build_tool_call_context, + classify_claude_tool, + decide, + default_hook_policy, + default_proxy_config_for_hook, + format_hook_output, + main, + process_hook, +) +from agentveil_mcp_proxy.policy import ( + PolicyConfig, + PolicyDecision, + PolicyEngine, + ProxyConfig, + RiskClass, +) + + +# ----- helpers --------------------------------------------------------------- + + +def _payload(tool_name: str, tool_input: dict | None = None, **extra) -> dict: + return { + "session_id": extra.get("session_id", "test-session"), + "cwd": extra.get("cwd", "/tmp/probe"), + "hook_event_name": "PreToolUse", + "tool_name": tool_name, + "tool_input": tool_input or {}, + } + + +# ----- classifier basics ----------------------------------------------------- + + +@pytest.mark.parametrize( + "tool_name,expected_risk", + [ + ("Write", RiskClass.WRITE), + ("Edit", RiskClass.WRITE), + ("MultiEdit", RiskClass.WRITE), + ("NotebookEdit", RiskClass.WRITE), + ("Read", RiskClass.READ), + ("LS", RiskClass.READ), + ("Glob", RiskClass.READ), + ("Grep", RiskClass.READ), + ("WebSearch", RiskClass.READ), + ("WebFetch", RiskClass.READ), + ], +) +def test_classify_claude_builtin_tools(tool_name: str, expected_risk: RiskClass) -> None: + assert classify_claude_tool(tool_name, {}) is expected_risk + + +@pytest.mark.parametrize( + "command,expected_risk", + [ + # Allowlist read-only commands -> READ + ("ls -la", RiskClass.READ), + ("cat /tmp/x.txt", RiskClass.READ), + ("pwd", RiskClass.READ), + ("grep foo /tmp/x", RiskClass.READ), + ("find . -name '*.py'", RiskClass.READ), + # Mutation operators on otherwise-read-looking commands -> WRITE + ("echo hello > /tmp/out.txt", RiskClass.WRITE), + ("touch /tmp/x", RiskClass.WRITE), + ("mv /tmp/a /tmp/b", RiskClass.WRITE), + # Destructive -> DESTRUCTIVE + ("rm /tmp/x", RiskClass.DESTRUCTIVE), + ("rm -rf /tmp/probe", RiskClass.DESTRUCTIVE), + ("rmdir /tmp/probe", RiskClass.DESTRUCTIVE), + ], +) +def test_classify_bash_by_command(command: str, expected_risk: RiskClass) -> None: + assert classify_claude_tool("Bash", {"command": command}) is expected_risk + + +# ----- corrective: Bash deny fallback for arbitrary interpreters / mutations -- + + +@pytest.mark.parametrize( + "command", + [ + # Arbitrary interpreters can write to filesystem without matching + # any mutation token. They MUST NOT classify as READ. + "python3 -c \"open('owned.txt','w').write('x')\"", + "python -c 'import os; os.remove(\"x\")'", + "node -e 'require(\"fs\").writeFileSync(\"x\",\"y\")'", + "ruby -e 'File.write(\"x\",\"y\")'", + # sed -i / perl -pi in-place edits — caught by mutation tokens + "sed -i 's/foo/bar/' file", + "perl -pi -e 's/foo/bar/' file", + # git mutation subcommands — caught by git allowlist + "git checkout main", + "git reset --hard HEAD~1", + "git clean -fd", + "git push origin main", + "git commit -m 'x'", + "git add .", + # awk, xargs, eval, source — not on allowlist, must not auto-allow + "awk '{print}' /etc/passwd", + "xargs -I {} echo {}", + "eval 'rm x'", + "source ~/.bashrc", + # interactive shells — not on allowlist + "bash -c 'echo x > y'", + "zsh -c 'rm x'", + ], +) +def test_classify_bash_fail_closed_for_non_allowlisted(command: str) -> None: + """Anything not unambiguously read-only must be non-READ.""" + risk = classify_claude_tool("Bash", {"command": command}) + assert risk is not RiskClass.READ, ( + f"Bash command {command!r} should NOT classify as READ; got {risk}" + ) + + +@pytest.mark.parametrize("command", ["git status", "git status -s", "git diff", "git diff --stat"]) +def test_classify_bash_git_readonly_subcommands_allowed(command: str) -> None: + assert classify_claude_tool("Bash", {"command": command}) is RiskClass.READ + + +def test_classify_bash_empty_command_is_unknown() -> None: + assert classify_claude_tool("Bash", {"command": ""}) is RiskClass.UNKNOWN + assert classify_claude_tool("Bash", {"command": " "}) is RiskClass.UNKNOWN + assert classify_claude_tool("Bash", {}) is RiskClass.UNKNOWN + + +# ----- corrective-2: shell composition uses deny fallback ------------------- + +# The four exact repro commands the reviewer found that bypassed the +# first-token allowlist (read-looking first token + hidden mutation). +_COMPOSITION_REPRO = [ + "echo $(python3 -c \"open('owned.txt','w').write('x')\")", + "ls | python3 -c \"open('owned.txt','w').write('x')\"", + "cat `python3 -c \"open('owned.txt','w').write('x')\"`", + "echo hi; python3 -c \"open('owned.txt','w').write('x')\"", +] + + +@pytest.mark.parametrize("command", _COMPOSITION_REPRO) +def test_corrective2_composition_repro_not_read(command: str) -> None: + """Reviewer repro: these read-looking commands hid a mutation. Must NOT + classify as READ.""" + risk = classify_claude_tool("Bash", {"command": command}) + assert risk is not RiskClass.READ, f"{command!r} classified READ ({risk})" + + +@pytest.mark.parametrize("command", _COMPOSITION_REPRO) +def test_corrective2_composition_repro_denied_end_to_end(command: str) -> None: + engine = PolicyEngine(default_proxy_config_for_hook()) + decision = decide(_payload("Bash", {"command": command}), engine) + assert decision.hook_action == "deny", ( + f"{command!r} must deny; got {decision.hook_action} " + f"(risk={decision.evaluation.risk_class.value})" + ) + + +@pytest.mark.parametrize( + "command", + [ + "echo $(whoami)", # command substitution + "echo `whoami`", # backtick substitution + "ls | grep foo", # pipe + "echo a; echo b", # semicolon chain + "true && echo ok", # && chain + "false || echo fallback", # || chain + "cat <(echo x)", # process substitution (read side) + "diff <(ls a) <(ls b)", # process substitution + "echo x > file", # redirect (also caught by mutation token) + "echo x>file", # redirect no spaces (was a gap) + "echo x >> file", # append redirect + "cat foo &", # background + "ls\nrm x", # embedded newline + ], +) +def test_corrective2_each_composition_metachar_denied(command: str) -> None: + """Each individual composition metacharacter maps to deny.""" + risk = classify_claude_tool("Bash", {"command": command}) + assert risk is not RiskClass.READ, f"{command!r} classified READ ({risk})" + engine = PolicyEngine(default_proxy_config_for_hook()) + decision = decide(_payload("Bash", {"command": command}), engine) + assert decision.hook_action == "deny" + + +@pytest.mark.parametrize( + "command", + [ + "ls -la", + "cat file", + "cat /tmp/some/file.txt", + "grep foo file", + "grep -rn pattern src", + "find . -name '*.py'", + "git status", + "git status -s", + "git diff", + "git diff --stat HEAD~1", + "pwd", + "head -n 20 file", + "tail -f file", + "wc -l file", + ], +) +def test_corrective2_simple_reads_still_allowed(command: str) -> None: + """Composition guard must not regress simple single read-only commands.""" + risk = classify_claude_tool("Bash", {"command": command}) + assert risk is RiskClass.READ, f"{command!r} should be READ; got {risk}" + engine = PolicyEngine(default_proxy_config_for_hook()) + decision = decide(_payload("Bash", {"command": command}), engine) + assert decision.hook_action == "allow", f"{command!r} should allow" + + +def test_classify_mcp_tool_via_classification_primitives() -> None: + # mcp__server__list_items -> infer matches the "list" prefix -> READ + assert classify_claude_tool("mcp__probe__list_items", {}) is RiskClass.READ + # mcp__server__write_note -> "write" prefix -> WRITE + assert classify_claude_tool("mcp__probe__write_note", {"content": "x"}) is RiskClass.WRITE + # mcp__server__delete_thing -> "delete" prefix -> DESTRUCTIVE + assert classify_claude_tool("mcp__probe__delete_thing", {}) is RiskClass.DESTRUCTIVE + + +def test_classify_unknown_returns_unknown_risk() -> None: + # No prefix match, not in built-in table, not Bash, not mcp__ + assert classify_claude_tool("ZorblatXyz", {}) is RiskClass.UNKNOWN + + +def test_classify_mcp_with_unrecognized_suffix_returns_unknown() -> None: + # mcp__server__zorblat -> not matching any prefix in classification.py + assert classify_claude_tool("mcp__probe__zorblat", {}) is RiskClass.UNKNOWN + + +# ----- context building ------------------------------------------------------ + + +def test_context_for_claude_builtin() -> None: + ctx = build_tool_call_context(_payload("Write", {"file_path": "/tmp/x", "content": "y"})) + assert ctx.server == CLAUDE_SERVER_LABEL + assert ctx.tool == "Write" + assert ctx.action == f"{CLAUDE_SERVER_LABEL}.Write" + assert ctx.risk_class is RiskClass.WRITE + + +def test_context_for_mcp_tool_splits_server_and_tool() -> None: + ctx = build_tool_call_context(_payload("mcp__probe__write_note", {"content": "x"})) + assert ctx.server == "probe" + assert ctx.tool == "write_note" + assert ctx.action == "probe.write_note" + assert ctx.risk_class is RiskClass.WRITE + + +# ----- spec test 1: Write/Edit/MultiEdit/NotebookEdit deny ------------------ + + +@pytest.mark.parametrize("tool_name", ["Write", "Edit", "MultiEdit", "NotebookEdit"]) +def test_spec1_write_family_denied_under_default_policy(tool_name: str) -> None: + engine = PolicyEngine(default_proxy_config_for_hook()) + decision = decide(_payload(tool_name, {"file_path": "/tmp/x", "content": "y"}), engine) + assert decision.hook_action == "deny" + assert decision.reason_code == "risky_blocked" + assert decision.evaluation.risk_class is RiskClass.WRITE + # write -> approval rule fires; with no approval surface in S1, treated as deny. + assert decision.evaluation.decision is PolicyDecision.APPROVAL + + +# ----- spec test 2: Bash mutation deny -------------------------------------- + + +def test_spec2_bash_mutation_denied() -> None: + engine = PolicyEngine(default_proxy_config_for_hook()) + decision = decide(_payload("Bash", {"command": "echo y > /tmp/out"}), engine) + assert decision.hook_action == "deny" + assert decision.evaluation.risk_class is RiskClass.WRITE + + +def test_spec2_bash_destructive_blocked_by_rule() -> None: + engine = PolicyEngine(default_proxy_config_for_hook()) + decision = decide(_payload("Bash", {"command": "rm -rf /tmp/foo"}), engine) + assert decision.hook_action == "deny" + assert decision.evaluation.risk_class is RiskClass.DESTRUCTIVE + assert decision.evaluation.decision is PolicyDecision.BLOCK + + +# ----- corrective: end-to-end deny for arbitrary interpreters / git mutations + + +@pytest.mark.parametrize( + "command", + [ + "python3 -c \"open('owned.txt','w').write('x')\"", + "sed -i 's/foo/bar/' file", + "perl -pi -e 's/foo/bar/' file", + "git checkout main", + "git reset --hard", + "git clean -fd", + "node -e 'require(\"fs\").writeFileSync(\"x\",\"y\")'", + ], +) +def test_corrective_bash_fail_closed_denies_at_decide(command: str) -> None: + """Reviewer-found blocker: these commands previously classified as READ + and were allowed. After fix they must be denied end-to-end.""" + engine = PolicyEngine(default_proxy_config_for_hook()) + decision = decide(_payload("Bash", {"command": command}), engine) + assert decision.hook_action == "deny", ( + f"Bash {command!r} must be denied but got {decision.hook_action} " + f"(risk={decision.evaluation.risk_class.value}, " + f"policy_decision={decision.evaluation.decision.value})" + ) + + +@pytest.mark.parametrize("command", ["git status", "git diff", "git status -s"]) +def test_corrective_bash_git_readonly_still_allowed_at_decide(command: str) -> None: + """Allowlist preserved end-to-end.""" + engine = PolicyEngine(default_proxy_config_for_hook()) + decision = decide(_payload("Bash", {"command": command}), engine) + assert decision.hook_action == "allow" + + +# ----- spec test 3: read/list allows ---------------------------------------- + + +@pytest.mark.parametrize("tool_name", ["Read", "LS", "Glob", "Grep"]) +def test_spec3_safe_reads_allowed(tool_name: str) -> None: + engine = PolicyEngine(default_proxy_config_for_hook()) + decision = decide(_payload(tool_name, {"path": "/tmp/probe"}), engine) + assert decision.hook_action == "allow" + assert decision.evaluation.risk_class is RiskClass.READ + assert decision.evaluation.decision is PolicyDecision.ALLOW + + +def test_spec3_bash_readonly_command_allowed() -> None: + engine = PolicyEngine(default_proxy_config_for_hook()) + decision = decide(_payload("Bash", {"command": "ls -la"}), engine) + assert decision.hook_action == "allow" + + +# ----- spec test 4: MCP write-like denied through Claude MCP naming -------- + + +def test_spec4_mcp_write_note_denied() -> None: + engine = PolicyEngine(default_proxy_config_for_hook()) + decision = decide( + _payload("mcp__probe__write_note", {"content": "x"}), engine + ) + assert decision.hook_action == "deny" + assert decision.evaluation.risk_class is RiskClass.WRITE + assert decision.context.server == "probe" + assert decision.context.tool == "write_note" + + +def test_spec4_mcp_destructive_blocked() -> None: + engine = PolicyEngine(default_proxy_config_for_hook()) + decision = decide(_payload("mcp__probe__delete_record", {}), engine) + assert decision.hook_action == "deny" + assert decision.evaluation.risk_class is RiskClass.DESTRUCTIVE + assert decision.evaluation.decision is PolicyDecision.BLOCK + + +# ----- spec test 5: MCP read/list allowed ----------------------------------- + + +def test_spec5_mcp_safe_list_allowed() -> None: + engine = PolicyEngine(default_proxy_config_for_hook()) + decision = decide(_payload("mcp__probe__safe_list", {}), engine) + assert decision.hook_action == "allow" + assert decision.evaluation.risk_class is RiskClass.READ + + +# ----- spec test 6: evidence rows are bounded (sentinel privacy proof) ----- + + +SENTINEL_CONTENT = "SENTINEL_RAW_VALUE_xyz_should_never_appear_in_evidence" +SENTINEL_PATH = "SENTINEL_RAW_FILE_PATH_abc_should_never_appear_in_evidence" +SENTINEL_COMMAND = "SENTINEL_RAW_SHELL_COMMAND_def_should_never_appear_in_evidence" + + +def _evidence_to_json_str(record: dict) -> str: + """Serialize record exactly as write_evidence does for grep checks.""" + return json.dumps(record, separators=(",", ":"), default=str) + + +def test_spec6_write_evidence_does_not_contain_raw_content_or_path() -> None: + engine = PolicyEngine(default_proxy_config_for_hook()) + payload = _payload( + "Write", + {"file_path": SENTINEL_PATH, "content": SENTINEL_CONTENT}, + ) + decision = decide(payload, engine) + record = build_evidence_record(payload, decision) + serialized = _evidence_to_json_str(record) + assert SENTINEL_CONTENT not in serialized, "raw write content leaked" + assert SENTINEL_PATH not in serialized, "raw file path leaked" + + +def test_spec6_bash_evidence_does_not_contain_raw_command() -> None: + engine = PolicyEngine(default_proxy_config_for_hook()) + payload = _payload("Bash", {"command": f"echo {SENTINEL_COMMAND} > /tmp/x"}) + decision = decide(payload, engine) + record = build_evidence_record(payload, decision) + serialized = _evidence_to_json_str(record) + assert SENTINEL_COMMAND not in serialized, "raw shell command leaked" + + +def test_spec6_mcp_evidence_does_not_contain_raw_input_values() -> None: + engine = PolicyEngine(default_proxy_config_for_hook()) + payload = _payload( + "mcp__probe__write_note", + {"content": SENTINEL_CONTENT, "filename": SENTINEL_PATH}, + ) + decision = decide(payload, engine) + record = build_evidence_record(payload, decision) + serialized = _evidence_to_json_str(record) + assert SENTINEL_CONTENT not in serialized + assert SENTINEL_PATH not in serialized + + +def test_spec6_input_ref_contains_only_hash_and_keys() -> None: + ref = _bounded_input_ref({"content": "hello", "file_path": "/tmp/x"}) + assert set(ref.keys()) == {"input_hash", "input_keys"} + assert ref["input_hash"].startswith("sha256:") + assert ref["input_keys"] == ["content", "file_path"] + # No raw values + serialized = json.dumps(ref) + assert "hello" not in serialized + assert "/tmp/x" not in serialized + + +# ----- corrective: cwd must not appear raw in evidence --------------------- + + +SENTINEL_CWD = "/Users/olegboiko/SENTINEL_SECRET_WORKSPACE_zlj9k" + + +def test_corrective_evidence_does_not_contain_raw_cwd() -> None: + """Reviewer-found blocker: build_evidence_record was writing raw cwd. + After fix, the workspace path must not appear in serialized evidence.""" + payload = _payload( + "Write", + {"file_path": "/tmp/x", "content": "y"}, + cwd=SENTINEL_CWD, + ) + engine = PolicyEngine(default_proxy_config_for_hook()) + decision = decide(payload, engine) + record = build_evidence_record(payload, decision) + serialized = _evidence_to_json_str(record) + assert SENTINEL_CWD not in serialized, "raw cwd leaked in evidence" + assert "SENTINEL_SECRET_WORKSPACE" not in serialized, "raw cwd path fragment leaked" + # The bounded digest field MUST be present in its place. + assert "cwd_digest" in record + assert record["cwd_digest"].startswith("sha256:") + assert "cwd" not in record, "raw cwd field must not be present" + + +def test_corrective_cwd_digest_is_deterministic_for_same_path() -> None: + """Audit need: same workspace -> same digest, so sessions can be grouped + without leaking the path.""" + payload_a = _payload("Read", {"file_path": "/tmp/x"}, cwd="/Users/x/proj") + payload_b = _payload("Read", {"file_path": "/tmp/y"}, cwd="/Users/x/proj") + payload_c = _payload("Read", {"file_path": "/tmp/z"}, cwd="/Users/x/other") + engine = PolicyEngine(default_proxy_config_for_hook()) + rec_a = build_evidence_record(payload_a, decide(payload_a, engine)) + rec_b = build_evidence_record(payload_b, decide(payload_b, engine)) + rec_c = build_evidence_record(payload_c, decide(payload_c, engine)) + assert rec_a["cwd_digest"] == rec_b["cwd_digest"] + assert rec_a["cwd_digest"] != rec_c["cwd_digest"] + + +def test_corrective_empty_cwd_yields_marker_digest() -> None: + payload = _payload("Read", {"file_path": "/tmp/x"}, cwd="") + engine = PolicyEngine(default_proxy_config_for_hook()) + record = build_evidence_record(payload, decide(payload, engine)) + assert record["cwd_digest"] == "sha256:empty" + + +# ----- spec test 7: unknown mutation-shaped fails closed ------------------- + + +def test_spec7_unknown_tool_fails_closed() -> None: + """Conservative fallback: UNKNOWN risk -> default_decision=ASK_BACKEND + -> hook treats as deny (no backend in S1).""" + engine = PolicyEngine(default_proxy_config_for_hook()) + decision = decide(_payload("ZorblatUnknownTool", {"any": "value"}), engine) + assert decision.evaluation.risk_class is RiskClass.UNKNOWN + assert decision.evaluation.decision is PolicyDecision.ASK_BACKEND + assert decision.hook_action == "deny" + assert decision.reason_code == "risky_blocked" + + +def test_spec7_unknown_mcp_tool_fails_closed() -> None: + engine = PolicyEngine(default_proxy_config_for_hook()) + decision = decide(_payload("mcp__weird__zorblat", {}), engine) + assert decision.evaluation.risk_class is RiskClass.UNKNOWN + assert decision.hook_action == "deny" + + +# ----- output formatting + evidence writer --------------------------------- + + +def test_format_hook_output_allow_returns_none() -> None: + engine = PolicyEngine(default_proxy_config_for_hook()) + decision = decide(_payload("Read", {"file_path": "/tmp/x"}), engine) + assert format_hook_output(decision) is None + + +def test_format_hook_output_deny_returns_claude_compatible_json() -> None: + engine = PolicyEngine(default_proxy_config_for_hook()) + decision = decide(_payload("Write", {"file_path": "/tmp/x", "content": "y"}), engine) + raw = format_hook_output(decision) + assert raw is not None + out = json.loads(raw) + assert out["hookSpecificOutput"]["hookEventName"] == "PreToolUse" + assert out["hookSpecificOutput"]["permissionDecision"] == "deny" + reason = out["hookSpecificOutput"]["permissionDecisionReason"] + assert "target_reached=false" in reason + assert "risk_class=write" in reason + + +# ----- S2 corrective: native deny carries an agent-facing redirect ---------- + + +from agentveil_mcp_proxy.claude_hook import NATIVE_REDIRECT_INSTRUCTION + + +@pytest.mark.parametrize( + "tool_name,tool_input", + [ + ("Write", {"file_path": "/tmp/x", "content": "y"}), + ("Edit", {"file_path": "/tmp/x", "old_string": "a", "new_string": "b"}), + ("MultiEdit", {"file_path": "/tmp/x", "edits": []}), + ("Bash", {"command": "echo y > /tmp/out"}), + ], +) +def test_native_mutation_deny_includes_redirect_instruction(tool_name, tool_input) -> None: + engine = PolicyEngine(default_proxy_config_for_hook()) + decision = decide(_payload(tool_name, tool_input), engine) + assert decision.hook_action == "deny" + raw = format_hook_output(decision) + reason = json.loads(raw)["hookSpecificOutput"]["permissionDecisionReason"] + # Each required instruction element is present. + assert "Direct native tool use was blocked before mutation" in reason # claim-check: allow literal hook-deny text asserted by this test. + assert "controlled MCP tool" in reason + assert "same path, content, and intent" in reason + assert "ask the user to approve" in reason and "retry the controlled tool call" in reason + assert NATIVE_REDIRECT_INSTRUCTION in reason + + +def test_native_deny_redirect_does_not_leak_raw_input() -> None: + """Redirect text must not reintroduce a raw-value leak.""" + engine = PolicyEngine(default_proxy_config_for_hook()) + payload = _payload( + "Write", + {"file_path": SENTINEL_PATH, "content": SENTINEL_CONTENT}, + ) + decision = decide(payload, engine) + raw = format_hook_output(decision) + assert SENTINEL_CONTENT not in raw + assert SENTINEL_PATH not in raw + + +def test_native_bash_deny_redirect_does_not_leak_command() -> None: + engine = PolicyEngine(default_proxy_config_for_hook()) + payload = _payload("Bash", {"command": f"echo {SENTINEL_COMMAND} > /tmp/x"}) + decision = decide(payload, engine) + raw = format_hook_output(decision) + assert SENTINEL_COMMAND not in raw + + +# ----- S3 corrective: controlled AgentVeil MCP route must pass through ------- + + +from agentveil_mcp_proxy.claude_hook import AGENTVEIL_CONTROLLED_MCP_SERVER + + +@pytest.mark.parametrize( + "tool_name,tool_input", + [ + (f"mcp__{AGENTVEIL_CONTROLLED_MCP_SERVER}__write_file", {"path": "/x", "content": "y"}), + (f"mcp__{AGENTVEIL_CONTROLLED_MCP_SERVER}__delete_file", {"path": "/x"}), + (f"mcp__{AGENTVEIL_CONTROLLED_MCP_SERVER}__move_file", {"src": "/a", "dst": "/b"}), + (f"mcp__{AGENTVEIL_CONTROLLED_MCP_SERVER}__list_workspace", {}), + ], +) +def test_controlled_mcp_route_passes_through(tool_name, tool_input) -> None: + """S3 blocker fix: the hook must NOT deny the AgentVeil controlled MCP + tools, or the redirect dead-ends. They self-govern at the proxy.""" + engine = PolicyEngine(default_proxy_config_for_hook()) + decision = decide(_payload(tool_name, tool_input), engine) + assert decision.hook_action == "allow", ( + f"controlled route {tool_name} must pass through; got {decision.hook_action}" + ) + assert decision.reason_code == "controlled_route_passthrough" + # No deny JSON emitted -> the call reaches the proxy. + assert format_hook_output(decision) is None + + +def test_controlled_passthrough_does_not_weaken_other_mcp_servers() -> None: + """A non-AgentVeil MCP write tool is still denied (no blanket allow).""" + engine = PolicyEngine(default_proxy_config_for_hook()) + decision = decide(_payload("mcp__other_server__write_note", {"content": "x"}), engine) + assert decision.hook_action == "deny" + + +def test_controlled_passthrough_does_not_weaken_native_writes() -> None: + """Native Write/Bash mutations are still denied after the passthrough fix.""" + engine = PolicyEngine(default_proxy_config_for_hook()) + assert decide(_payload("Write", {"file_path": "/x", "content": "y"}), engine).hook_action == "deny" + assert decide(_payload("Bash", {"command": "echo y > /x"}), engine).hook_action == "deny" + + +def test_mcp_deny_does_not_carry_native_redirect() -> None: + """The native redirect is scoped to native tools, not MCP tool denies.""" + engine = PolicyEngine(default_proxy_config_for_hook()) + decision = decide(_payload("mcp__probe__write_note", {"content": "x"}), engine) + assert decision.hook_action == "deny" + reason = json.loads(format_hook_output(decision))["hookSpecificOutput"]["permissionDecisionReason"] + assert NATIVE_REDIRECT_INSTRUCTION not in reason + assert "target_reached=false" in reason # base bounded reason still present + + +def test_write_evidence_appends_jsonl_line(tmp_path: Path) -> None: + engine = PolicyEngine(default_proxy_config_for_hook()) + payload = _payload("Write", {"file_path": "/tmp/x", "content": "y"}) + decision = decide(payload, engine) + record = build_evidence_record(payload, decision) + evidence_path = tmp_path / "decisions.jsonl" + claude_hook.write_evidence(record, evidence_path) + claude_hook.write_evidence(record, evidence_path) + lines = evidence_path.read_text(encoding="utf-8").strip().splitlines() + assert len(lines) == 2 + for line in lines: + decoded = json.loads(line) + assert decoded["tool_name"] == "Write" + assert decoded["hook_action"] == "deny" + + +# ----- process_hook end-to-end + main() CLI -------------------------------- + + +def test_process_hook_writes_evidence_and_deny_output(tmp_path: Path) -> None: + payload = _payload("Write", {"file_path": "/tmp/x", "content": "y"}) + out = io.StringIO() + evidence = tmp_path / "decisions.jsonl" + decision = process_hook(payload, evidence_path=evidence, out=out) + assert decision.hook_action == "deny" + assert "permissionDecision" in out.getvalue() + assert evidence.read_text(encoding="utf-8").strip() != "" + + +def test_process_hook_allow_writes_evidence_but_no_output(tmp_path: Path) -> None: + payload = _payload("Read", {"file_path": "/tmp/x"}) + out = io.StringIO() + evidence = tmp_path / "decisions.jsonl" + decision = process_hook(payload, evidence_path=evidence, out=out) + assert decision.hook_action == "allow" + assert out.getvalue() == "" # silent allow + assert evidence.read_text(encoding="utf-8").strip() != "" # but evidence written + + +def test_main_reads_stdin_and_writes_stdout(tmp_path: Path, monkeypatch) -> None: + payload = _payload("Write", {"file_path": "/tmp/x", "content": "y"}) + evidence = tmp_path / "decisions.jsonl" + monkeypatch.setenv("AGENTVEIL_HOOK_EVIDENCE_PATH", str(evidence)) + in_stream = io.StringIO(json.dumps(payload)) + out_stream = io.StringIO() + rc = main(stdin=in_stream, stdout=out_stream) + assert rc == 0 + assert "permissionDecision" in out_stream.getvalue() + assert evidence.read_text(encoding="utf-8").strip() != "" + + +def test_main_rejects_non_object_payload() -> None: + in_stream = io.StringIO('"not an object"') + out_stream = io.StringIO() + rc = main(stdin=in_stream, stdout=out_stream) + assert rc == 1 + assert out_stream.getvalue() == "" + + +def test_main_evidence_path_arg_writes_evidence(tmp_path: Path) -> None: + """S2 wiring: the installed hook command passes --evidence-path.""" + payload = _payload("Write", {"file_path": "/tmp/x", "content": "y"}) + evidence = tmp_path / "agentveil" / "evidence.jsonl" + in_stream = io.StringIO(json.dumps(payload)) + out_stream = io.StringIO() + rc = main(["--evidence-path", str(evidence)], stdin=in_stream, stdout=out_stream) + assert rc == 0 + assert "permissionDecision" in out_stream.getvalue() + assert evidence.read_text(encoding="utf-8").strip() != "" + + +def test_main_evidence_path_arg_overrides_env(tmp_path: Path, monkeypatch) -> None: + arg_path = tmp_path / "arg.jsonl" + env_path = tmp_path / "env.jsonl" + monkeypatch.setenv("AGENTVEIL_HOOK_EVIDENCE_PATH", str(env_path)) + payload = _payload("Read", {"file_path": "/tmp/x"}) + rc = main( + ["--evidence-path", str(arg_path)], + stdin=io.StringIO(json.dumps(payload)), + stdout=io.StringIO(), + ) + assert rc == 0 + assert arg_path.exists() + assert not env_path.exists() # arg wins over env + + +# ----- ASK_BACKEND deny fallback semantics (spec clarification #2) ---------- + + +def test_ask_backend_is_fail_closed_in_s1() -> None: + """Direct construction of an ASK_BACKEND decision proves deny mapping.""" + # Use a context with risk that won't match any rule -> default_decision fires. + config = default_proxy_config_for_hook() + engine = PolicyEngine(config) + # default_decision is ASK_BACKEND; UNKNOWN risk doesn't match any rule. + payload = _payload("ZorblatUnknownTool", {}) + decision = decide(payload, engine) + assert decision.evaluation.decision is PolicyDecision.ASK_BACKEND + assert decision.hook_action == "deny" diff --git a/packages/agentveil-mcp-proxy/tests/test_mcp_proxy_claude_hook_setup.py b/packages/agentveil-mcp-proxy/tests/test_mcp_proxy_claude_hook_setup.py new file mode 100644 index 0000000..e3234f4 --- /dev/null +++ b/packages/agentveil-mcp-proxy/tests/test_mcp_proxy_claude_hook_setup.py @@ -0,0 +1,573 @@ +"""Tests for agentveil_mcp_proxy.claude_hook_setup (P10D.14 S2). + +Covers install / status / uninstall with merge preservation, idempotency, +invalid-JSON no-rewrite handling, bounded status, and project-local evidence +path. +""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from agentveil_mcp_proxy import claude_hook_setup +from agentveil_mcp_proxy.claude_hook_setup import ( + AGENTVEIL_HOOK_MARKER, + HOOK_MATCHER, + HookSetupError, + build_hook_command, + build_managed_hook_entry, + install_hook, + load_settings, + project_evidence_path, + project_settings_path, + status_hook, + uninstall_hook, +) + + +# ----- helpers --------------------------------------------------------------- + + +def _read_settings(project: Path) -> dict: + return json.loads(project_settings_path(project).read_text(encoding="utf-8")) + + +def _managed_entries(settings: dict) -> list: + pre = settings.get("hooks", {}).get("PreToolUse", []) + return [e for e in pre if AGENTVEIL_HOOK_MARKER in json.dumps(e)] + + +# ----- build helpers --------------------------------------------------------- + + +def test_build_hook_command_invokes_module_with_evidence_path() -> None: + cmd = build_hook_command(python="/usr/bin/python3", evidence_path=Path("/proj/.claude/agentveil/evidence.jsonl")) + assert "-m agentveil_mcp_proxy.claude_hook" in cmd + assert "--evidence-path" in cmd + assert "/proj/.claude/agentveil/evidence.jsonl" in cmd + + +def test_managed_entry_uses_combined_matcher() -> None: + entry = build_managed_hook_entry(python="/usr/bin/python3", evidence_path=Path("/x/e.jsonl")) + assert entry["matcher"] == HOOK_MATCHER + assert "Bash" in entry["matcher"] + assert "mcp__" in entry["matcher"] + assert entry["hooks"][0]["type"] == "command" + + +def test_evidence_path_is_project_local_under_dot_claude(tmp_path: Path) -> None: + ev = project_evidence_path(tmp_path) + assert ev.is_relative_to(tmp_path / ".claude") + + +# ----- install --------------------------------------------------------------- + + +def test_install_creates_settings_with_managed_entry(tmp_path: Path) -> None: + result = install_hook(tmp_path) + assert result.created_settings is True + assert result.reload_required is True + settings = _read_settings(tmp_path) + managed = _managed_entries(settings) + assert len(managed) == 1 + command = managed[0]["hooks"][0]["command"] + assert "-m agentveil_mcp_proxy.claude_hook" in command + assert "--evidence-path" in command + + +def test_install_is_idempotent(tmp_path: Path) -> None: + install_hook(tmp_path) + install_hook(tmp_path) + install_hook(tmp_path) + settings = _read_settings(tmp_path) + assert len(_managed_entries(settings)) == 1, "install must not duplicate managed entries" + + +def test_install_replaces_stale_managed_entry_without_duplicate(tmp_path: Path) -> None: + # Pre-seed an older managed entry (marker present, old command form). + settings_path = project_settings_path(tmp_path) + settings_path.parent.mkdir(parents=True, exist_ok=True) + settings_path.write_text(json.dumps({ + "hooks": { + "PreToolUse": [ + { + "matcher": "Bash", + "hooks": [{"type": "command", "command": "python /old/agentveil_mcp_proxy.claude_hook.py"}], + } + ] + } + }), encoding="utf-8") + result = install_hook(tmp_path) + assert result.replaced_existing_managed is True + settings = _read_settings(tmp_path) + assert len(_managed_entries(settings)) == 1 + + +def test_install_preserves_unrelated_top_level_settings(tmp_path: Path) -> None: + settings_path = project_settings_path(tmp_path) + settings_path.parent.mkdir(parents=True, exist_ok=True) + settings_path.write_text(json.dumps({ + "model": "claude-opus-4-8", + "env": {"FOO": "bar"}, + "permissions": {"allow": ["Read"]}, + }), encoding="utf-8") + install_hook(tmp_path) + settings = _read_settings(tmp_path) + assert settings["model"] == "claude-opus-4-8" + assert settings["env"] == {"FOO": "bar"} + assert settings["permissions"] == {"allow": ["Read"]} + assert len(_managed_entries(settings)) == 1 + + +def test_install_preserves_unrelated_hooks(tmp_path: Path) -> None: + settings_path = project_settings_path(tmp_path) + settings_path.parent.mkdir(parents=True, exist_ok=True) + settings_path.write_text(json.dumps({ + "hooks": { + "PostToolUse": [ + {"matcher": "Write", "hooks": [{"type": "command", "command": "echo post"}]} + ], + "PreToolUse": [ + {"matcher": "Read", "hooks": [{"type": "command", "command": "echo user-pre"}]} + ], + } + }), encoding="utf-8") + install_hook(tmp_path) + settings = _read_settings(tmp_path) + # Unrelated PostToolUse preserved + assert settings["hooks"]["PostToolUse"][0]["hooks"][0]["command"] == "echo post" + # Unrelated PreToolUse preserved + pre = settings["hooks"]["PreToolUse"] + user_pre = [e for e in pre if e.get("matcher") == "Read"] + assert len(user_pre) == 1 + assert user_pre[0]["hooks"][0]["command"] == "echo user-pre" + # AgentVeil entry added alongside + assert len(_managed_entries(settings)) == 1 + + +def test_install_fails_closed_on_invalid_json_without_rewrite(tmp_path: Path) -> None: + settings_path = project_settings_path(tmp_path) + settings_path.parent.mkdir(parents=True, exist_ok=True) + garbage = "{ this is not valid json ]" + settings_path.write_text(garbage, encoding="utf-8") + with pytest.raises(HookSetupError): + install_hook(tmp_path) + # File must be untouched. + assert settings_path.read_text(encoding="utf-8") == garbage + + +# ----- uninstall ------------------------------------------------------------- + + +def test_uninstall_removes_only_managed_entry(tmp_path: Path) -> None: + settings_path = project_settings_path(tmp_path) + settings_path.parent.mkdir(parents=True, exist_ok=True) + settings_path.write_text(json.dumps({ + "model": "x", + "hooks": { + "PreToolUse": [ + {"matcher": "Read", "hooks": [{"type": "command", "command": "echo user-pre"}]} + ], + }, + }), encoding="utf-8") + install_hook(tmp_path) + assert len(_managed_entries(_read_settings(tmp_path))) == 1 + + result = uninstall_hook(tmp_path) + assert result.removed_entries == 1 + settings = _read_settings(tmp_path) + assert len(_managed_entries(settings)) == 0 + # unrelated preserved + assert settings["model"] == "x" + pre = settings["hooks"]["PreToolUse"] + assert any(e.get("matcher") == "Read" for e in pre) + + +def test_uninstall_idempotent_no_file(tmp_path: Path) -> None: + result = uninstall_hook(tmp_path) + assert result.removed_entries == 0 + assert result.settings_existed is False + + +def test_uninstall_idempotent_no_managed_entry(tmp_path: Path) -> None: + settings_path = project_settings_path(tmp_path) + settings_path.parent.mkdir(parents=True, exist_ok=True) + settings_path.write_text(json.dumps({"model": "x"}), encoding="utf-8") + result = uninstall_hook(tmp_path) + assert result.removed_entries == 0 + # file preserved + assert _read_settings(tmp_path)["model"] == "x" + + +def test_uninstall_cleans_empty_generated_containers(tmp_path: Path) -> None: + # install into an otherwise-empty project, then uninstall. + install_hook(tmp_path) + uninstall_hook(tmp_path) + settings = _read_settings(tmp_path) + # The generated empty PreToolUse list and hooks dict should be cleaned. + assert "hooks" not in settings or settings.get("hooks") not in ({}, {"PreToolUse": []}) + + +def test_uninstall_fails_closed_on_invalid_json(tmp_path: Path) -> None: + settings_path = project_settings_path(tmp_path) + settings_path.parent.mkdir(parents=True, exist_ok=True) + garbage = "{bad json" + settings_path.write_text(garbage, encoding="utf-8") + with pytest.raises(HookSetupError): + uninstall_hook(tmp_path) + assert settings_path.read_text(encoding="utf-8") == garbage + + +def test_uninstall_preserves_user_hook_in_mixed_group(tmp_path: Path) -> None: + """Corrective: a group containing BOTH a user hook and an AgentVeil hook + must keep the user hook on uninstall (remove only the AgentVeil hook).""" + settings_path = project_settings_path(tmp_path) + settings_path.parent.mkdir(parents=True, exist_ok=True) + settings_path.write_text(json.dumps({ + "hooks": { + "PreToolUse": [ + { + "matcher": "Bash", + "hooks": [ + {"type": "command", "command": "echo user-hook"}, + {"type": "command", "command": "python -m agentveil_mcp_proxy.claude_hook --evidence-path /x"}, + ], + } + ] + } + }), encoding="utf-8") + result = uninstall_hook(tmp_path) + assert result.removed_entries == 1 + settings = _read_settings(tmp_path) + pre = settings["hooks"]["PreToolUse"] + # The group survives with the user hook intact. + assert len(pre) == 1 + commands = [h["command"] for h in pre[0]["hooks"]] + assert "echo user-hook" in commands + assert all(AGENTVEIL_HOOK_MARKER not in c for c in commands) # claim-check: allow Python all() assertion. + + +def test_install_preserves_user_hook_in_mixed_group(tmp_path: Path) -> None: + """Install upsert must not drop a user hook sharing a group with an old + AgentVeil hook.""" + settings_path = project_settings_path(tmp_path) + settings_path.parent.mkdir(parents=True, exist_ok=True) + settings_path.write_text(json.dumps({ + "hooks": { + "PreToolUse": [ + { + "matcher": "Bash", + "hooks": [ + {"type": "command", "command": "echo user-hook"}, + {"type": "command", "command": "python -m agentveil_mcp_proxy.claude_hook --evidence-path /old"}, + ], + } + ] + } + }), encoding="utf-8") + install_hook(tmp_path) + settings = _read_settings(tmp_path) + all_commands = [ + h["command"] + for g in settings["hooks"]["PreToolUse"] + for h in g["hooks"] + ] + assert "echo user-hook" in all_commands, "user hook in mixed group was dropped" + # Exactly one current managed entry (the fresh one; old stale one removed). + assert len(_managed_entries(settings)) == 1 + + +def test_install_command_shell_quotes_paths_with_spaces(tmp_path: Path) -> None: + """Corrective: project/python paths with spaces must be shell-quoted so the + generated command is parseable.""" + import shlex + + spaced = tmp_path / "dir with spaces" + spaced.mkdir() + result = install_hook(spaced, python="/opt/py thon/bin/python3") + settings = json.loads(project_settings_path(spaced).read_text(encoding="utf-8")) + command = _managed_entries(settings)[0]["hooks"][0]["command"] + # The command must split cleanly via shlex (no broken quoting). + tokens = shlex.split(command) + assert "/opt/py thon/bin/python3" in tokens + assert "-m" in tokens + assert "agentveil_mcp_proxy.claude_hook" in tokens + # The evidence path (with spaces) round-trips as a single token. + ev_index = tokens.index("--evidence-path") + 1 + assert "dir with spaces" in tokens[ev_index] + + +def test_install_uninstall_roundtrip_preserves_unrelated(tmp_path: Path) -> None: + settings_path = project_settings_path(tmp_path) + settings_path.parent.mkdir(parents=True, exist_ok=True) + original = {"model": "x", "hooks": {"PostToolUse": [{"matcher": "Write", "hooks": [{"type": "command", "command": "echo post"}]}]}} + settings_path.write_text(json.dumps(original), encoding="utf-8") + install_hook(tmp_path) + uninstall_hook(tmp_path) + settings = _read_settings(tmp_path) + assert settings["model"] == "x" + assert settings["hooks"]["PostToolUse"][0]["hooks"][0]["command"] == "echo post" + assert len(_managed_entries(settings)) == 0 + + +# ----- status ---------------------------------------------------------------- + + +def test_status_missing_is_unsafe(tmp_path: Path) -> None: + status = status_hook(tmp_path).to_bounded_dict() + assert status["status"] == "unsafe" + assert status["state"] == "missing" + assert status["settings_present"] is False + + +def test_status_installed_without_evidence_is_advisory_not_protected(tmp_path: Path) -> None: + """Corrective: a freshly installed hook is advisory (reload pending), NOT + protected. Protected would overclaim before the hook has run.""" + install_hook(tmp_path) + status = status_hook(tmp_path).to_bounded_dict() + assert status["status"] == "advisory" + assert status["state"] == "installed" + assert status["managed_hook_present"] is True + assert status["hook_command_points_to_module"] is True + assert status["evidence_path_configured"] is True + assert status["reload_required"] is True + + +def _set_mtime(path: Path, mtime: float) -> None: + import os + os.utime(path, (mtime, mtime)) + + +def test_status_protected_only_after_firing_evidence(tmp_path: Path) -> None: + """Protected requires firing evidence whose mtime post-dates settings.json.""" + install_hook(tmp_path) + ev = project_evidence_path(tmp_path) + ev.parent.mkdir(parents=True, exist_ok=True) + ev.write_text('{"decision":"deny"}\n', encoding="utf-8") + # Force evidence newer than settings.json so the check is deterministic + # regardless of filesystem mtime granularity. + settings_mtime = project_settings_path(tmp_path).stat().st_mtime + _set_mtime(ev, settings_mtime + 5) + status = status_hook(tmp_path).to_bounded_dict() + assert status["status"] == "protected" + assert status["state"] == "installed" + assert status["reload_required"] is False + + +def test_status_old_evidence_before_install_is_advisory(tmp_path: Path) -> None: + """Corrective: stale evidence from a previous install must NOT yield + protected after a reinstall.""" + # 1. An old evidence file predates the (re)install. + ev = project_evidence_path(tmp_path) + ev.parent.mkdir(parents=True, exist_ok=True) + ev.write_text('{"old":"evidence"}\n', encoding="utf-8") + _set_mtime(ev, 1_000_000.0) # far in the past + + # 2. Install now (settings.json mtime is "now", newer than old evidence). + install_hook(tmp_path) + status = status_hook(tmp_path).to_bounded_dict() + assert status["status"] == "advisory", "stale evidence must not claim protected" + assert status["state"] == "installed" + assert status["reload_required"] is True + + # 3. A fresh firing (evidence mtime newer than settings.json) -> protected. + settings_mtime = project_settings_path(tmp_path).stat().st_mtime + ev.write_text('{"new":"evidence"}\n', encoding="utf-8") + _set_mtime(ev, settings_mtime + 5) + status2 = status_hook(tmp_path).to_bounded_dict() + assert status2["status"] == "protected" + assert status2["reload_required"] is False + + +def test_status_empty_evidence_file_is_still_advisory(tmp_path: Path) -> None: + install_hook(tmp_path) + ev = project_evidence_path(tmp_path) + ev.parent.mkdir(parents=True, exist_ok=True) + ev.write_text("", encoding="utf-8") # exists but empty => not fired + status = status_hook(tmp_path).to_bounded_dict() + assert status["status"] == "advisory" + assert status["reload_required"] is True + + +def test_status_invalid_json_is_unsafe(tmp_path: Path) -> None: + settings_path = project_settings_path(tmp_path) + settings_path.parent.mkdir(parents=True, exist_ok=True) + settings_path.write_text("{not json", encoding="utf-8") + status = status_hook(tmp_path).to_bounded_dict() + assert status["status"] == "unsafe" + assert status["state"] == "invalid-json" + + +def test_status_stale_when_command_does_not_point_to_module(tmp_path: Path) -> None: + settings_path = project_settings_path(tmp_path) + settings_path.parent.mkdir(parents=True, exist_ok=True) + # Managed (marker present) but old invocation form (no `-m module`). + settings_path.write_text(json.dumps({ + "hooks": { + "PreToolUse": [ + {"matcher": "Bash", "hooks": [{"type": "command", "command": "python /old/agentveil_mcp_proxy.claude_hook.py --evidence-path /x"}]} + ] + } + }), encoding="utf-8") + status = status_hook(tmp_path).to_bounded_dict() + assert status["state"] == "stale" + assert status["status"] == "advisory" + assert status["managed_hook_present"] is True + assert status["hook_command_points_to_module"] is False + + +def test_status_no_managed_entry_is_unsafe(tmp_path: Path) -> None: + settings_path = project_settings_path(tmp_path) + settings_path.parent.mkdir(parents=True, exist_ok=True) + settings_path.write_text(json.dumps({ + "hooks": {"PreToolUse": [{"matcher": "Read", "hooks": [{"type": "command", "command": "echo x"}]}]} + }), encoding="utf-8") + status = status_hook(tmp_path).to_bounded_dict() + assert status["status"] == "unsafe" + assert status["state"] == "missing" + assert status["managed_hook_present"] is False + + +def test_status_bounded_has_no_absolute_paths_or_raw_data(tmp_path: Path) -> None: + install_hook(tmp_path) + bounded = status_hook(tmp_path).to_bounded_dict() + serialized = json.dumps(bounded) + # No absolute project path, no python interpreter path, no evidence path. + assert str(tmp_path) not in serialized + assert ".claude/settings.json" not in serialized # no settings path string + assert "/usr" not in serialized and "python" not in serialized + # Only bounded keys present. + assert set(bounded.keys()) == { + "scope", "status", "state", "settings_present", "managed_hook_present", + "hook_command_points_to_module", "evidence_path_configured", + "reload_required", "matched_tool_classes", "notes", + } + + +# ----- load_settings edge cases --------------------------------------------- + + +def test_load_settings_absent_returns_empty(tmp_path: Path) -> None: + assert load_settings(project_settings_path(tmp_path)) == {} + + +def test_load_settings_empty_file_returns_empty(tmp_path: Path) -> None: + p = project_settings_path(tmp_path) + p.parent.mkdir(parents=True, exist_ok=True) + p.write_text(" \n", encoding="utf-8") + assert load_settings(p) == {} + + +def test_load_settings_non_object_raises(tmp_path: Path) -> None: + p = project_settings_path(tmp_path) + p.parent.mkdir(parents=True, exist_ok=True) + p.write_text("[1,2,3]", encoding="utf-8") + with pytest.raises(HookSetupError): + load_settings(p) + + +# ----- P10D.14 S5: one-command connector (.mcp.json route + status) ---------- + +from agentveil_mcp_proxy.claude_hook_setup import ( + AGENTVEIL_MCP_SERVER_NAME, + connector_status, + mcp_route_present, + project_mcp_config_path, + remove_mcp_route, +) + + +def _write_mcp(project: Path, data: dict) -> None: + p = project_mcp_config_path(project) + p.parent.mkdir(parents=True, exist_ok=True) + p.write_text(json.dumps(data), encoding="utf-8") + + +def _seed_agentveil_mcp(project: Path, *, with_other: bool = False) -> None: + servers = {AGENTVEIL_MCP_SERVER_NAME: {"command": "agentveil-mcp-proxy", "args": ["run"]}} + if with_other: + servers["other-server"] = {"command": "other", "args": []} + _write_mcp(project, {"mcpServers": servers}) + + +def test_mcp_route_present_detects_agentveil_entry(tmp_path: Path) -> None: + assert mcp_route_present(tmp_path) is False + _seed_agentveil_mcp(tmp_path) + assert mcp_route_present(tmp_path) is True + + +def test_remove_mcp_route_removes_only_agentveil(tmp_path: Path) -> None: + _seed_agentveil_mcp(tmp_path, with_other=True) + result = remove_mcp_route(tmp_path) + assert result.removed is True + data = json.loads(project_mcp_config_path(tmp_path).read_text(encoding="utf-8")) + assert AGENTVEIL_MCP_SERVER_NAME not in data["mcpServers"] + assert "other-server" in data["mcpServers"], "unrelated MCP server must survive" + + +def test_remove_mcp_route_idempotent_and_cleans_empty(tmp_path: Path) -> None: + # no file + assert remove_mcp_route(tmp_path).removed is False + # only agentveil -> removed + mcpServers key cleaned + _seed_agentveil_mcp(tmp_path) + assert remove_mcp_route(tmp_path).removed is True + data = json.loads(project_mcp_config_path(tmp_path).read_text(encoding="utf-8")) + assert "mcpServers" not in data or data.get("mcpServers") == {} + # second remove is a no-op + assert remove_mcp_route(tmp_path).removed is False + + +def test_remove_mcp_route_fail_closed_invalid_json(tmp_path: Path) -> None: + p = project_mcp_config_path(tmp_path) + p.parent.mkdir(parents=True, exist_ok=True) + garbage = "{bad json" + p.write_text(garbage, encoding="utf-8") + with pytest.raises(HookSetupError): + remove_mcp_route(tmp_path) + assert p.read_text(encoding="utf-8") == garbage # not rewritten + + +def test_remove_mcp_route_preserves_unrelated_top_level(tmp_path: Path) -> None: + _write_mcp(tmp_path, { + "mcpServers": {AGENTVEIL_MCP_SERVER_NAME: {"command": "x"}}, + "otherKey": {"keep": True}, + }) + remove_mcp_route(tmp_path) + data = json.loads(project_mcp_config_path(tmp_path).read_text(encoding="utf-8")) + assert data["otherKey"] == {"keep": True} + + +def test_connector_status_missing_is_unsafe(tmp_path: Path) -> None: + st = connector_status(tmp_path, proxy_route_present=False) + assert st["status"] == "unsafe" + assert st["hook"] == "missing" + assert st["mcp_route"] == "missing" + assert st["proxy_route"] == "missing" + assert st["next_step"].startswith("run ") + + +def test_connector_status_complete_is_advisory(tmp_path: Path) -> None: + install_hook(tmp_path) + _seed_agentveil_mcp(tmp_path) + st = connector_status(tmp_path, proxy_route_present=True) + assert st["status"] == "advisory" + assert st["hook"] == "installed" + assert st["mcp_route"] == "configured" + assert st["proxy_route"] == "configured" + assert st["restart_required"] is True + + +def test_connector_status_bounded_no_absolute_paths(tmp_path: Path) -> None: + install_hook(tmp_path) + _seed_agentveil_mcp(tmp_path) + st = connector_status(tmp_path, proxy_route_present=True) + serialized = json.dumps(st) + assert str(tmp_path) not in serialized + assert "/Users/" not in serialized and "/private/" not in serialized + assert set(st.keys()) == { + "scope", "status", "hook", "mcp_route", "proxy_route", + "restart_required", "matched_tool_classes", "next_step", + } diff --git a/packages/agentveil-mcp-proxy/tests/test_mcp_proxy_cli.py b/packages/agentveil-mcp-proxy/tests/test_mcp_proxy_cli.py index 8fc96a0..1610a71 100644 --- a/packages/agentveil-mcp-proxy/tests/test_mcp_proxy_cli.py +++ b/packages/agentveil-mcp-proxy/tests/test_mcp_proxy_cli.py @@ -9,6 +9,7 @@ from pathlib import Path import stat import sys +from types import SimpleNamespace import pytest @@ -1892,3 +1893,439 @@ def test_setup_cli_run_status_and_restore(tmp_path, capsys): assert "/private/" not in restore_text assert "/var/folders/" not in restore_text assert "/Users/" not in restore_text + + +# ----- P10D.14 S2: Claude hook install/status/uninstall CLI dispatch -------- + + +def test_cli_install_claude_hook_preview_does_not_write(tmp_path, capsys): + assert main([ + "install-claude-hook", "--project", "--project-dir", str(tmp_path), + ]) == 0 + capsys.readouterr() + assert not (tmp_path / ".claude" / "settings.json").exists() + + +def test_cli_install_status_uninstall_dispatch_roundtrip(tmp_path, capsys): + # install --yes --json + assert main([ + "install-claude-hook", "--project", "--project-dir", str(tmp_path), "--yes", "--json", + ]) == 0 + install_payload = json.loads(capsys.readouterr().out) + assert install_payload["ok"] is True + assert install_payload["applied"] is True + settings = tmp_path / ".claude" / "settings.json" + assert settings.exists() + + # status --json -> advisory (installed, no firing evidence yet), bounded + assert main([ + "status-claude-hook", "--project", "--project-dir", str(tmp_path), "--json", + ]) == 0 + status_payload = json.loads(capsys.readouterr().out) + assert status_payload["status"] == "advisory" + assert status_payload["state"] == "installed" + assert status_payload["reload_required"] is True + status_text = json.dumps(status_payload) + assert str(tmp_path) not in status_text + assert "/Users/" not in status_text and "/private/" not in status_text + + # uninstall --yes --json + assert main([ + "uninstall-claude-hook", "--project", "--project-dir", str(tmp_path), "--yes", "--json", + ]) == 0 + uninstall_payload = json.loads(capsys.readouterr().out) + assert uninstall_payload["ok"] is True + assert uninstall_payload["removed_entries"] == 1 + + +def test_cli_status_missing_is_unsafe(tmp_path, capsys): + assert main([ + "status-claude-hook", "--project", "--project-dir", str(tmp_path), "--json", + ]) == 0 + payload = json.loads(capsys.readouterr().out) + assert payload["status"] == "unsafe" + assert payload["state"] == "missing" + + +def test_cli_install_claude_hook_requires_project_flag(tmp_path): + # --project is required; omitting it is an argparse error (SystemExit). + with pytest.raises(SystemExit): + main(["install-claude-hook", "--project-dir", str(tmp_path), "--yes"]) + + +def test_cli_status_claude_hook_requires_project_flag(tmp_path): + with pytest.raises(SystemExit): + main(["status-claude-hook", "--project-dir", str(tmp_path), "--json"]) + + +# ----- P10D.14 S5: one-command Claude connector setup CLI -------------------- + + +def test_cli_setup_status_bare_is_connector(tmp_path, capsys): + assert main(["setup", "status", "--project-dir", str(tmp_path), "--json"]) == 0 + payload = json.loads(capsys.readouterr().out) + # connector-shaped keys + assert payload["status"] == "unsafe" + assert payload["hook"] == "missing" + assert payload["mcp_route"] == "missing" + assert "next_step" in payload + + +def test_cli_setup_status_home_routes_to_wizard(tmp_path, capsys): + # --home routes to the adaptive wizard, NOT the connector status. + home = tmp_path / "avp-home" + home.mkdir() + rc = main(["setup", "status", "--home", str(home), "--json"]) + out = capsys.readouterr().out + payload = json.loads(out) + # wizard output must not carry the connector-only keys + assert "mcp_route" not in payload + assert "hook" not in payload or "next_step" not in payload + + +def test_cli_setup_claude_code_preview_does_not_write(tmp_path, capsys): + assert main(["setup", "claude-code", "--project-dir", str(tmp_path)]) == 0 + capsys.readouterr() + assert not (tmp_path / ".claude" / "settings.json").exists() + assert not (tmp_path / ".mcp.json").exists() + + +def test_choose_setup_project_folder_macos(monkeypatch, tmp_path): + selected = tmp_path / "Selected Project" + selected.mkdir() + + monkeypatch.setattr(sys, "platform", "darwin") + monkeypatch.setattr( + "subprocess.run", + lambda *args, **kwargs: SimpleNamespace( + returncode=0, + stdout=f"{selected}\n", + stderr="", + ), + ) + + assert proxy_cli._choose_setup_project_folder() == selected + + +def test_cli_setup_claude_code_rejects_choose_folder_with_project_dir(tmp_path, capsys): + rc = main([ + "setup", "claude-code", + "--choose-folder", + "--project-dir", str(tmp_path), + "--yes", + ]) + + assert rc == 2 + assert "cannot be combined" in capsys.readouterr().err + + +def test_cli_setup_remove_claude_code_apply(tmp_path, capsys): + from agentveil_mcp_proxy import claude_hook_setup + # seed an installed hook + an .mcp.json with agentveil + an unrelated server + claude_hook_setup.install_hook(tmp_path) + (tmp_path / ".mcp.json").write_text(json.dumps({ + "mcpServers": { + "agentveil-mcp-proxy": {"command": "agentveil-mcp-proxy", "args": ["run"]}, + "other-server": {"command": "other"}, + } + }), encoding="utf-8") + + assert main(["setup", "remove", "claude-code", "--project-dir", str(tmp_path), "--yes", "--json"]) == 0 + payload = json.loads(capsys.readouterr().out) + assert payload["ok"] is True + assert payload["hook_entries_removed"] == 1 + assert payload["mcp_route_removed"] is True + + settings = json.loads((tmp_path / ".claude" / "settings.json").read_text(encoding="utf-8")) + pre = settings.get("hooks", {}).get("PreToolUse", []) + for entry in pre: + assert "agentveil_mcp_proxy.claude_hook" not in json.dumps(entry) + mcp = json.loads((tmp_path / ".mcp.json").read_text(encoding="utf-8")) + assert "agentveil-mcp-proxy" not in mcp.get("mcpServers", {}) + assert "other-server" in mcp.get("mcpServers", {}), "unrelated MCP server must survive" + + +def test_cli_setup_status_bounded_no_paths(tmp_path, capsys): + from agentveil_mcp_proxy import claude_hook_setup + claude_hook_setup.install_hook(tmp_path) + assert main(["setup", "status", "--project-dir", str(tmp_path), "--json"]) == 0 + text = capsys.readouterr().out + assert str(tmp_path) not in text + assert "/Users/" not in text and "/private/" not in text + + +def test_cli_setup_claude_code_starts_center_with_passphrase_without_url_leak( + tmp_path, + monkeypatch, + capsys, +): + from agentveil_mcp_proxy import claude_center_lifecycle + + passphrase_file = tmp_path / "passphrase.txt" + passphrase_file.write_text(TEST_PASSPHRASE, encoding="utf-8") + seen: dict[str, object] = {} + + def fake_init_proxy(**kwargs): + home = kwargs["home"] + (home / "mcp-proxy").mkdir(parents=True, exist_ok=True) + (home / "mcp-proxy" / "config.json").write_text("{}", encoding="utf-8") + seen["init_plaintext"] = kwargs["plaintext"] + seen["init_passphrase_file"] = kwargs["passphrase_file"] + seen["downstream_root"] = kwargs["downstream_config"]["args"][-1] + + def fake_connect(**kwargs): + project = kwargs["project_root"] + (project / ".mcp.json").write_text( + json.dumps({"mcpServers": {"agentveil-mcp-proxy": {"command": "agentveil-mcp-proxy"}}}), + encoding="utf-8", + ) + return 0 + + def fake_ensure_running(**kwargs): + seen["center_passphrase_file"] = kwargs["passphrase_file"] + return SimpleNamespace( + status=SimpleNamespace( + state="running", + url="http://127.0.0.1:12345/approval/SECRET_TOKEN", + ), + started=True, + reused=False, + restarted=False, + ) + + monkeypatch.setattr(proxy_cli, "init_proxy", fake_init_proxy) + monkeypatch.setattr(proxy_cli, "run_connect_cli", fake_connect) + monkeypatch.setattr("shutil.which", lambda _name: "agentveil-mcp-proxy") + monkeypatch.setattr(claude_center_lifecycle, "ensure_running", fake_ensure_running) + + rc = main([ + "setup", "claude-code", + "--project-dir", str(tmp_path), + "--passphrase-file", str(passphrase_file), + "--yes", + "--json", + ]) + assert rc == 0 + payload = json.loads(capsys.readouterr().out) + assert payload["ok"] is True + assert payload["approval_center"]["state"] == "running" + assert "url" not in payload["approval_center"] + assert "SECRET_TOKEN" not in json.dumps(payload) + assert payload["identity_encrypted"] is True + assert seen["init_plaintext"] is False + assert seen["init_passphrase_file"] == passphrase_file + assert seen["center_passphrase_file"] == passphrase_file + assert seen["downstream_root"] == str(tmp_path.resolve()) + config = json.loads((tmp_path / ".avp" / "mcp-proxy" / "config.json").read_text(encoding="utf-8")) + assert config["approval"]["ui_open_mode"] == "terminal" + + +def test_cli_setup_claude_code_choose_folder_uses_selected_project( + tmp_path, + monkeypatch, + capsys, +): + from agentveil_mcp_proxy import claude_center_lifecycle + + selected = tmp_path / "Selected Project" + selected.mkdir() + seen: dict[str, object] = {} + + def fake_init_proxy(**kwargs): + home = kwargs["home"] + (home / "mcp-proxy").mkdir(parents=True, exist_ok=True) + (home / "mcp-proxy" / "config.json").write_text("{}", encoding="utf-8") + seen["home"] = home + seen["downstream_root"] = kwargs["downstream_config"]["args"][-1] + + def fake_connect(**kwargs): + project = kwargs["project_root"] + seen["project_root"] = project + (project / ".mcp.json").write_text( + json.dumps({"mcpServers": {"agentveil-mcp-proxy": {"command": "agentveil-mcp-proxy"}}}), + encoding="utf-8", + ) + return 0 + + monkeypatch.setattr(proxy_cli, "_choose_setup_project_folder", lambda: selected) + monkeypatch.setattr(proxy_cli, "init_proxy", fake_init_proxy) + monkeypatch.setattr(proxy_cli, "run_connect_cli", fake_connect) + monkeypatch.setattr("shutil.which", lambda _name: "agentveil-mcp-proxy") + monkeypatch.setattr( + claude_center_lifecycle, + "ensure_running", + lambda **_kwargs: SimpleNamespace( + status=SimpleNamespace(state="running", url="http://127.0.0.1/approval/SECRET"), + started=True, + reused=False, + restarted=False, + ), + ) + + rc = main(["setup", "claude-code", "--choose-folder", "--yes", "--json"]) + + assert rc == 0 + payload = json.loads(capsys.readouterr().out) + assert payload["ok"] is True + assert seen["project_root"] == selected.resolve() + assert seen["home"] == selected.resolve() / ".avp" + assert seen["downstream_root"] == str(selected.resolve()) + + +def test_cli_setup_claude_code_fails_when_center_not_running( + tmp_path, + monkeypatch, + capsys, +): + from agentveil_mcp_proxy import claude_center_lifecycle + + def fake_init_proxy(**kwargs): + home = kwargs["home"] + (home / "mcp-proxy").mkdir(parents=True, exist_ok=True) + (home / "mcp-proxy" / "config.json").write_text("{}", encoding="utf-8") + + def fake_connect(**kwargs): + project = kwargs["project_root"] + (project / ".mcp.json").write_text( + json.dumps({"mcpServers": {"agentveil-mcp-proxy": {"command": "agentveil-mcp-proxy"}}}), + encoding="utf-8", + ) + return 0 + + def fake_ensure_running(**_kwargs): + return SimpleNamespace( + status=SimpleNamespace(state="stale", url=None), + started=False, + reused=False, + restarted=False, + reason="approval-center did not become healthy", + ) + + monkeypatch.setattr(proxy_cli, "init_proxy", fake_init_proxy) + monkeypatch.setattr(proxy_cli, "run_connect_cli", fake_connect) + monkeypatch.setattr("shutil.which", lambda _name: "agentveil-mcp-proxy") + monkeypatch.setattr(claude_center_lifecycle, "ensure_running", fake_ensure_running) + + rc = main(["setup", "claude-code", "--project-dir", str(tmp_path), "--yes", "--json"]) + assert rc == 1 + payload = json.loads(capsys.readouterr().out) + assert payload["ok"] is False + assert payload["approval_center"]["state"] == "stale" + assert "not ready/protected" in payload["errors"][0] + + +def test_claude_center_stop_does_not_kill_unhealthy_manifest(tmp_path, monkeypatch): + from agentveil_mcp_proxy import claude_center_lifecycle + from agentveil_mcp_proxy.approval.persistent import ( + ApprovalCenterManifest, + save_manifest, + token_hash_for, + ) + + proxy_dir = tmp_path / ".avp" / "mcp-proxy" + token = "session-token" + save_manifest( + proxy_dir, + ApprovalCenterManifest( + schema_version=2, + host="127.0.0.1", + port=43210, + session_token=token, + token_hash=token_hash_for(token), + internal_register_token="internal", + pid=12345, + started_at=1, + ), + ) + monkeypatch.setattr(claude_center_lifecycle, "is_process_alive", lambda _pid: True) + monkeypatch.setattr(claude_center_lifecycle, "_center_health", lambda _manifest: False) + + def fail_kill(_pid, _signal): + raise AssertionError("must not kill an unhealthy/non-AgentVeil manifest pid") + + monkeypatch.setattr(os, "kill", fail_kill) + result = claude_center_lifecycle.stop_if_managed(tmp_path / ".avp") + assert result["stopped"] is False + assert "not a healthy AgentVeil Approval Center" in result["reason"] + + +def test_claude_center_health_uses_approval_center_page(tmp_path, monkeypatch): + from agentveil_mcp_proxy import claude_center_lifecycle + from agentveil_mcp_proxy.approval.persistent import ( + ApprovalCenterManifest, + token_hash_for, + ) + + token = "session-token" + manifest = ApprovalCenterManifest( + schema_version=2, + host="127.0.0.1", + port=43210, + session_token=token, + token_hash=token_hash_for(token), + internal_register_token="internal", + pid=12345, + started_at=1, + ) + seen_urls: list[str] = [] + + def fake_status(url: str, *, timeout: float) -> int: + del timeout + seen_urls.append(url) + return 200 if url == manifest.approval_center_url() else 403 + + monkeypatch.setattr(claude_center_lifecycle, "loopback_get_status", fake_status) + + assert claude_center_lifecycle._center_health(manifest) + assert seen_urls == [manifest.approval_center_url()] + + +def test_claude_center_status_uses_loopback_not_pid_probe(tmp_path, monkeypatch): + from agentveil_mcp_proxy import claude_center_lifecycle + from agentveil_mcp_proxy.approval.persistent import ( + ApprovalCenterManifest, + save_manifest, + token_hash_for, + ) + + token = "session-token" + manifest = ApprovalCenterManifest( + schema_version=2, + host="127.0.0.1", + port=43210, + session_token=token, + token_hash=token_hash_for(token), + internal_register_token="internal", + pid=12345, + started_at=1, + ) + save_manifest(tmp_path / ".avp" / "mcp-proxy", manifest) + monkeypatch.setattr(claude_center_lifecycle, "is_process_alive", lambda _pid: False) + monkeypatch.setattr(claude_center_lifecycle, "_center_health", lambda _manifest: True) + + status = claude_center_lifecycle.check_status(tmp_path / ".avp") + + assert status.state == "running" + assert status.port == 43210 + + +def test_setup_proxy_command_falls_back_to_invoked_console_script(tmp_path, monkeypatch): + script = tmp_path / "agentveil-mcp-proxy" + script.write_text("#!/bin/sh\n", encoding="utf-8") + monkeypatch.setattr("shutil.which", lambda _name: None) + monkeypatch.setattr(sys, "argv", [str(script), "setup", "claude-code"]) + + assert proxy_cli._resolve_setup_proxy_command() == str(script.resolve()) + + +def test_setup_proxy_command_prefers_invoked_script_over_path(tmp_path, monkeypatch): + invoked = tmp_path / "venv" / "bin" / "agentveil-mcp-proxy" + invoked.parent.mkdir(parents=True) + invoked.write_text("#!/bin/sh\n", encoding="utf-8") + global_script = tmp_path / "global" / "agentveil-mcp-proxy" + global_script.parent.mkdir() + global_script.write_text("#!/bin/sh\n", encoding="utf-8") + monkeypatch.setattr("shutil.which", lambda _name: str(global_script)) + monkeypatch.setattr(sys, "argv", [str(invoked), "setup", "claude-code"]) + + assert proxy_cli._resolve_setup_proxy_command() == str(invoked.resolve()) diff --git a/packages/agentveil-mcp-proxy/tests/test_mcp_proxy_persistent_approval_center.py b/packages/agentveil-mcp-proxy/tests/test_mcp_proxy_persistent_approval_center.py index bc85f79..0d8ecac 100644 --- a/packages/agentveil-mcp-proxy/tests/test_mcp_proxy_persistent_approval_center.py +++ b/packages/agentveil-mcp-proxy/tests/test_mcp_proxy_persistent_approval_center.py @@ -171,6 +171,47 @@ def direct_loopback_connection(address, *args, **kwargs): store.close() +def test_manifest_health_check_uses_approval_center_page(tmp_path, monkeypatch): + store, server, _manager, proxy_dir = _start_persistent_center(tmp_path) + try: + manifest = load_manifest(proxy_dir) + assert manifest is not None + seen_urls: list[str] = [] + + def fake_status(url: str, *, timeout: float) -> int: + del timeout + seen_urls.append(url) + return 200 if url == manifest.approval_center_url() else 403 + + monkeypatch.setattr(persistent_module, "loopback_get_status", fake_status) + + assert manifest_is_reachable(manifest) + assert seen_urls == [manifest.approval_center_url()] + finally: + server.stop() + store.close() + + +def test_manifest_reachability_uses_loopback_not_pid_probe(tmp_path, monkeypatch): + store, server, _manager, proxy_dir = _start_persistent_center(tmp_path) + try: + manifest = load_manifest(proxy_dir) + assert manifest is not None + monkeypatch.setattr(persistent_module, "is_process_alive", lambda _pid: False) + monkeypatch.setattr( + persistent_module, + "loopback_get_status", + lambda url, *, timeout: ( + 200 if url == manifest.approval_center_url() else 403 + ), + ) + + assert manifest_is_reachable(manifest) + finally: + server.stop() + store.close() + + def test_public_session_token_cannot_register_prompts(tmp_path): store, server, _manager, proxy_dir = _start_persistent_center(tmp_path) manifest = load_manifest(proxy_dir)