fix(mcp): make VQL mutations opt-in

.github/workflows/ci.yml

@@ -0,0 +1,24 @@
+name: CI
+
+on:
+  pull_request:
+  push:
+    branches: [main]
+
+permissions:
+  contents: read
+
+jobs:
+  mcp2vql:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.13"
+      - name: Install test dependencies
+        run: python -m pip install pytest "mcp>=1,<3"
+      - name: Test mutation safety contract
+        env:
+          PYTHONPATH: src:packages/dsl2vql/src:packages/mcp2vql/src
+        run: python -m pytest -q packages/mcp2vql/tests/test_mcp2vql.py

packages/mcp2vql/README.md

@@ -0,0 +1,8 @@
+# mcp2vql
+
+MCP server exposing VQL query, conversion, image diagnosis and mutation tools.
+
+Queries, UI detection, comparison and NL-to-DSL conversion remain read-only.
+Patching, DSL execution, metadata writes and `vql_apply_nl(..., execute=true)`
+are disabled unless the server operator sets `VQL_MCP_ALLOW_MUTATION=1` for
+trusted MCP clients. Natural-language application defaults to `execute=false`.

packages/mcp2vql/src/mcp2vql/server.py

@@ -2,10 +2,21 @@
 
 from __future__ import annotations
 
-import json
+import os
 from dataclasses import dataclass
+from pathlib import Path
 from typing import Any
 
+_MUTATION_ENV = "VQL_MCP_ALLOW_MUTATION"
+
+
+def _require_mutation(action: str) -> None:
+    enabled = os.getenv(_MUTATION_ENV, "").strip().lower() in {"1", "true", "yes", "on"}
+    if not enabled:
+        raise PermissionError(
+            f"MCP mutation '{action}' is disabled; start the server with {_MUTATION_ENV}=1"
+        )
+
 
 def _require_fastmcp():
     try:
@@ -41,16 +52,19 @@ def vql_query(uri: str, file: str = "", fmt: str = "json") -> dict[str, Any]:
 
         @self.app.tool()
         def vql_run_dsl(script: str, default_file: str = "") -> list[dict[str, Any]]:
+            _require_mutation("vql_run_dsl")
             results = execute_dsl(script, default_file=default_file or None)
             return [r.to_dict() for r in results]
 
         @self.app.tool()
         def vql_run_command(command: str, default_file: str = "") -> dict[str, Any]:
+            _require_mutation("vql_run_command")
             result = execute_dsl_line(command, default_file=default_file or None)
             return result.to_dict()
 
         @self.app.tool()
         def vql_run_command_pb(envelope_bytes: bytes, default_file: str = "") -> bytes:
+            _require_mutation("vql_run_command_pb")
             result = dispatch(envelope_bytes, default_file=default_file or None)
             return encode_result_protobuf(result)
 
@@ -59,12 +73,19 @@ def vql_to_dsl(prompt: str) -> str:
             return to_dsl(prompt)
 
         @self.app.tool()
-        def vql_apply_nl(prompt: str, default_file: str = "") -> dict[str, Any]:
-            return apply_nl(prompt, file=default_file or None).to_dict()
+        def vql_apply_nl(
+            prompt: str,
+            default_file: str = "",
+            execute: bool = False,
+        ) -> dict[str, Any]:
+            if execute:
+                _require_mutation("vql_apply_nl")
+            return apply_nl(prompt, file=default_file or None, execute=execute).to_dict()
 
         @self.app.tool()
         def vql_patch(uri: str, with_path: str, file: str = "") -> dict[str, Any]:
-            content = open(with_path, encoding="utf-8").read()
+            _require_mutation("vql_patch")
+            content = Path(with_path).read_text(encoding="utf-8")
             return patch_uri(uri, content=content, file=file or None).to_dict()
 
         @self.app.tool()
@@ -98,6 +119,8 @@ def vql_diagnose_window(
             """img2nl diagnose for screenshot; optionally persist to VQL program metadata."""
             from img2vql.diagnose import diagnose_for_vql
 
+            if save:
+                _require_mutation("vql_diagnose_window")
             return diagnose_for_vql(
                 image,
                 vql_program=vql_program or None,
@@ -116,6 +139,7 @@ def vql_refresh_window_metadata(
             from img2vql.metadata import refresh_program_metadata
             from uri2vql.window import _resolve_window_image
 
+            _require_mutation("vql_refresh_window_metadata")
             img = _resolve_window_image(vql_program, image or None)
             if not img:
                 return {"ok": False, "error": "image missing; pass image or set metadata.image"}

packages/mcp2vql/tests/test_mcp2vql.py

@@ -0,0 +1,14 @@
+"""Safety tests for mcp2vql."""
+
+import pytest
+
+from mcp2vql.server import _require_mutation
+
+
+def test_mcp_mutations_require_operator_capability(monkeypatch) -> None:
+    monkeypatch.delenv("VQL_MCP_ALLOW_MUTATION", raising=False)
+    with pytest.raises(PermissionError, match="VQL_MCP_ALLOW_MUTATION"):
+        _require_mutation("vql_patch")
+
+    monkeypatch.setenv("VQL_MCP_ALLOW_MUTATION", "1")
+    _require_mutation("vql_patch")