fix(mcp-proxy): broaden destructive coverage and harden env passthrough

Close P10.5 security hardening gaps by blocking AVP_* env passthrough, broadening destructive classifier and built-in policy pack coverage, and annotating known-safe evidence SQL construction with B608-specific nosec comments.

Implemented with assistance from Codex.

Author: oleg-bk

agentveil_mcp_proxy/classification.py

@@ -55,7 +55,22 @@
     "open",
     "close",
 )
-_DESTRUCTIVE_PREFIXES = ("delete", "remove", "destroy", "drop", "revoke", "terminate")
+_DESTRUCTIVE_PREFIXES = (
+    "delete",
+    "remove",
+    "destroy",
+    "drop",
+    "revoke",
+    "terminate",
+    "purge",
+    "truncate",
+    "wipe",
+    "format",
+    "rm",
+    "rmdir",
+    "unlink",
+    "clean",
+)
 _PRODUCTION_WORDS = ("prod", "production", "deploy", "release", "rollback", "infra", "cluster")
 _FINANCIAL_WORDS = ("payment", "transfer", "invoice", "billing", "payroll", "purchase", "refund")
 

agentveil_mcp_proxy/evidence/store.py

@@ -238,8 +238,9 @@ def write_pending(self, record: PendingApproval) -> None:
                 record = replace(record, prev_event_hash=prev_event_hash)
                 values = _record_values(record)
                 placeholders = ", ".join("?" for _ in _COLUMNS)
+                # SQL is safe: _COLUMNS is static and values are bound.
                 self._conn.execute(
-                    f"INSERT INTO pending_approvals ({', '.join(_COLUMNS)}) VALUES ({placeholders})",
+                    f"INSERT INTO pending_approvals ({', '.join(_COLUMNS)}) VALUES ({placeholders})",  # nosec B608
                     values,
                 )
                 if not append_only:
@@ -254,8 +255,9 @@ def get_pending(self, request_id: str) -> PendingApproval | None:
         """Return the approval record for a request ID, if present."""
 
         with self._lock:
+            # SQL is safe: _COLUMNS is static and request_id is bound.
             row = self._conn.execute(
-                f"SELECT {', '.join(_COLUMNS)} FROM pending_approvals WHERE request_id = ?",
+                f"SELECT {', '.join(_COLUMNS)} FROM pending_approvals WHERE request_id = ?",  # nosec B608
                 (request_id,),
             ).fetchone()
         return None if row is None else _row_to_record(row)
@@ -265,14 +267,16 @@ def list_pending(self, *, since_timestamp: int | None = None) -> list[PendingApp
 
         with self._lock:
             if since_timestamp is None:
+                # SQL is safe: _COLUMNS is static and status is bound.
                 rows = self._conn.execute(
-                    f"SELECT {', '.join(_COLUMNS)} FROM pending_approvals WHERE status = ? "
+                    f"SELECT {', '.join(_COLUMNS)} FROM pending_approvals WHERE status = ? "  # nosec B608
                     "ORDER BY created_at, request_id",
                     (ApprovalStatus.PENDING.value,),
                 ).fetchall()
             else:
+                # SQL is safe: _COLUMNS is static and filters are bound.
                 rows = self._conn.execute(
-                    f"SELECT {', '.join(_COLUMNS)} FROM pending_approvals WHERE status = ? "
+                    f"SELECT {', '.join(_COLUMNS)} FROM pending_approvals WHERE status = ? "  # nosec B608
                     "AND created_at >= ? ORDER BY created_at, request_id",
                     (ApprovalStatus.PENDING.value, since_timestamp),
                 ).fetchall()
@@ -290,8 +294,9 @@ def transition(self, request_id: str, new_status: str, **fields: Any) -> Pending
         with self._lock:
             self._begin()
             try:
+                # SQL is safe: _COLUMNS is static and request_id is bound.
                 row = self._conn.execute(
-                    f"SELECT {', '.join(_COLUMNS)} FROM pending_approvals WHERE request_id = ?",
+                    f"SELECT {', '.join(_COLUMNS)} FROM pending_approvals WHERE request_id = ?",  # nosec B608
                     (request_id,),
                 ).fetchone()
                 if row is None:
@@ -317,8 +322,9 @@ def transition(self, request_id: str, new_status: str, **fields: Any) -> Pending
                 updates["status"] = normalized
                 self._validate_transition_fields(updates)
                 assignments = ", ".join(f"{column} = ?" for column in updates)
+                # SQL is safe: assignment columns are restricted and values are bound.
                 self._conn.execute(
-                    f"UPDATE pending_approvals SET {assignments} WHERE request_id = ?",
+                    f"UPDATE pending_approvals SET {assignments} WHERE request_id = ?",  # nosec B608
                     (*updates.values(), request_id),
                 )
                 self._rebuild_chain_locked()
@@ -438,8 +444,9 @@ def list_records(
             params.extend(ids)
         where = f"WHERE {' AND '.join(clauses)}" if clauses else ""
         with self._lock:
+            # SQL is safe: _COLUMNS is static and where clauses are validated.
             rows = self._conn.execute(
-                f"SELECT {', '.join(_COLUMNS)} FROM pending_approvals "
+                f"SELECT {', '.join(_COLUMNS)} FROM pending_approvals "  # nosec B608
                 f"{where} ORDER BY created_at, request_id",
                 params,
             ).fetchall()
@@ -465,8 +472,9 @@ def find_active_similar_grant(
         )
         placeholders = ", ".join("?" for _ in reusable_statuses)
         with self._lock:
+            # SQL is safe: _COLUMNS is static and filters are bound.
             row = self._conn.execute(
-                f"SELECT {', '.join(_COLUMNS)} FROM pending_approvals "
+                f"SELECT {', '.join(_COLUMNS)} FROM pending_approvals "  # nosec B608
                 f"WHERE status IN ({placeholders}) "
                 "AND approval_scope = ? AND granted_scope_expires_at > ? "
                 "AND downstream_server = ? AND tool_name = ? AND risk_class = ? "
@@ -494,16 +502,18 @@ def vacuum_terminal_records(self, *, before_timestamp: int) -> int:
         with self._lock:
             self._begin()
             try:
+                # SQL is safe: terminal status placeholders and cutoff are controlled.
                 row = self._conn.execute(
                     "SELECT COUNT(*) FROM pending_approvals "
-                    f"WHERE status IN ({placeholders}) AND created_at < ?",
+                    f"WHERE status IN ({placeholders}) AND created_at < ?",  # nosec B608
                     (*terminal, int(before_timestamp)),
                 ).fetchone()
                 deleted = int(row[0])
                 if deleted:
+                    # SQL is safe: terminal status placeholders and cutoff are controlled.
                     self._conn.execute(
                         "DELETE FROM pending_approvals "
-                        f"WHERE status IN ({placeholders}) AND created_at < ?",
+                        f"WHERE status IN ({placeholders}) AND created_at < ?",  # nosec B608
                         (*terminal, int(before_timestamp)),
                     )
                     self._rebuild_chain_locked()
@@ -607,8 +617,9 @@ def _migrate_v3_to_v4_locked(self) -> None:
         # SQLite cannot relax a NOT NULL column constraint in place; rebuild the table.
         columns = ", ".join(_COLUMNS)
         self._conn.execute(_CREATE_PENDING_APPROVALS_MIGRATION_SQL)
+        # SQL is safe: migration columns are restricted to static _COLUMNS.
         self._conn.execute(
-            f"INSERT INTO pending_approvals_new ({columns}) "
+            f"INSERT INTO pending_approvals_new ({columns}) "  # nosec B608
             f"SELECT {columns} FROM pending_approvals"
         )
         self._conn.execute("DROP TABLE pending_approvals")
@@ -624,8 +635,9 @@ def _set_schema_version_locked(self, version: int) -> None:
         )
 
     def _compute_chain_link_for_insert_locked(self, record: PendingApproval) -> tuple[str, bool]:
+        # SQL is safe: _COLUMNS is static.
         row = self._conn.execute(
-            f"SELECT {', '.join(_COLUMNS)} FROM pending_approvals "
+            f"SELECT {', '.join(_COLUMNS)} FROM pending_approvals "  # nosec B608
             "ORDER BY created_at DESC, request_id DESC LIMIT 1"
         ).fetchone()
         if row is None:
@@ -639,8 +651,9 @@ def _compute_chain_link_for_insert_locked(self, record: PendingApproval) -> tupl
         return GENESIS_PREV_EVENT_HASH, False
 
     def _rebuild_chain_locked(self) -> None:
+        # SQL is safe: _COLUMNS is static.
         rows = self._conn.execute(
-            f"SELECT {', '.join(_COLUMNS)} FROM pending_approvals ORDER BY created_at, request_id"
+            f"SELECT {', '.join(_COLUMNS)} FROM pending_approvals ORDER BY created_at, request_id"  # nosec B608
         ).fetchall()
         prev_hash = GENESIS_PREV_EVENT_HASH
         for row in rows:
@@ -654,8 +667,9 @@ def _rebuild_chain_locked(self) -> None:
             prev_hash = record_hash(data)
 
     def _validate_chain_locked(self) -> None:
+        # SQL is safe: _COLUMNS is static.
         rows = self._conn.execute(
-            f"SELECT {', '.join(_COLUMNS)} FROM pending_approvals ORDER BY created_at, request_id"
+            f"SELECT {', '.join(_COLUMNS)} FROM pending_approvals ORDER BY created_at, request_id"  # nosec B608
         ).fetchall()
         prev_hash = GENESIS_PREV_EVENT_HASH
         for row in rows:

agentveil_mcp_proxy/passthrough.py

@@ -63,6 +63,7 @@
 MAX_CLIENT_MESSAGE_BYTES = 1 * 1024 * 1024
 MAX_PENDING_RESPONSES = 1000
 DEFAULT_TIMED_OUT_ID_RETENTION_SECONDS = 600.0
+_ENV_PASSTHROUGH_BLOCKED_PREFIXES = ("AVP_",)
 SAFE_ENV_KEYS = (
     "PATH",
     "HOME",
@@ -233,6 +234,12 @@ def from_proxy_config(cls, config: ProxyConfig) -> "DownstreamConfig":
             for item in env_passthrough
         ):
             raise PassthroughError("downstream.env_passthrough must be a list of strings")
+        for item in env_passthrough:
+            if any(item.startswith(prefix) for prefix in _ENV_PASSTHROUGH_BLOCKED_PREFIXES):
+                raise PassthroughError(
+                    f"downstream.env_passthrough cannot forward {item!r}: "
+                    "AVP_* prefix is reserved for proxy-internal secrets"
+                )
 
         response_timeout = data.get(
             "response_timeout_seconds",

agentveil_mcp_proxy/policy.py

@@ -752,7 +752,14 @@ def builtin_policy_pack(name: str) -> PolicyConfig:
                 "risk_class": "destructive",
                 "match": {
                     "server": ["github", "github-*", "github_*", "*github*"],
-                    "tool": ["delete_*", "remove_*"],
+                    "tool": [
+                        "delete_*",
+                        "remove_*",
+                        "purge_*",
+                        "drop_*",
+                        "destroy_*",
+                        "revoke_*",
+                    ],
                 },
             },
         ],
@@ -784,7 +791,19 @@ def builtin_policy_pack(name: str) -> PolicyConfig:
                 "risk_class": "destructive",
                 "match": {
                     "server": ["filesystem", "fs", "*filesystem*"],
-                    "tool": ["delete_*", "remove_*"],
+                    "tool": [
+                        "delete_*",
+                        "remove_*",
+                        "purge_*",
+                        "truncate_*",
+                        "wipe_*",
+                        "format_*",
+                        "rm",
+                        "rm_*",
+                        "rmdir_*",
+                        "unlink_*",
+                        "clean_*",
+                    ],
                 },
             },
         ],

tests/test_mcp_proxy_classification.py

@@ -229,6 +229,54 @@ def test_risk_inference_destructive_wins_over_production_compounds():
     assert infer_risk_class("auth.revoke_prod_access", tool="revoke_prod_access") is RiskClass.DESTRUCTIVE
 
 
+def test_infer_risk_class_recognizes_purge_as_destructive():
+    assert infer_risk_class("database.purge_database", tool="purge_database") is (
+        RiskClass.DESTRUCTIVE
+    )
+
+
+def test_infer_risk_class_recognizes_truncate_as_destructive():
+    assert infer_risk_class("database.truncate_table", tool="truncate_table") is (
+        RiskClass.DESTRUCTIVE
+    )
+
+
+def test_infer_risk_class_recognizes_wipe_as_destructive():
+    assert infer_risk_class("storage.wipe_disk", tool="wipe_disk") is RiskClass.DESTRUCTIVE
+
+
+def test_infer_risk_class_recognizes_format_as_destructive():
+    assert infer_risk_class("storage.format_volume", tool="format_volume") is (
+        RiskClass.DESTRUCTIVE
+    )
+
+
+def test_infer_risk_class_recognizes_rm_as_destructive():
+    assert infer_risk_class("filesystem.rm", tool="rm") is RiskClass.DESTRUCTIVE
+
+
+def test_infer_risk_class_recognizes_rmdir_as_destructive():
+    assert infer_risk_class("filesystem.rmdir_tree", tool="rmdir_tree") is (
+        RiskClass.DESTRUCTIVE
+    )
+
+
+def test_infer_risk_class_recognizes_unlink_as_destructive():
+    assert infer_risk_class("filesystem.unlink_file", tool="unlink_file") is (
+        RiskClass.DESTRUCTIVE
+    )
+
+
+def test_infer_risk_class_recognizes_clean_as_destructive():
+    assert infer_risk_class("filesystem.clean_temp", tool="clean_temp") is RiskClass.DESTRUCTIVE
+
+
+def test_infer_risk_class_destructive_wins_over_read_on_compound():
+    assert infer_risk_class("filesystem.purge_files", tool="purge_files") is (
+        RiskClass.DESTRUCTIVE
+    )
+
+
 def test_risk_inference_does_not_over_classify_substring_collisions():
     assert infer_risk_class("github.get_infrastructure", tool="get_infrastructure") is RiskClass.READ
     assert infer_risk_class("github.list_endpoints", tool="list_endpoints") is RiskClass.READ

tests/test_mcp_proxy_passthrough.py

@@ -58,6 +58,39 @@ def _write_json(path: Path, data: dict) -> None:
     os.chmod(path, 0o600)
 
 
+def _config_with_env_passthrough(env_passthrough: list[str]) -> ProxyConfig:
+    return ProxyConfig.from_dict({
+        "proxy_config_schema_version": 1,
+        "avp": {
+            "base_url": "https://agentveil.dev",
+            "agent_name": "agentveil-mcp-proxy",
+            "trusted_signer_dids": ["did:key:z6MktrustedSigner"],
+        },
+        "mode": "protect",
+        "privacy": {
+            "action": "redacted",
+            "resource": "hash",
+            "payload": "hash_only",
+            "evidence_upload": False,
+        },
+        "fallback": {},
+        "approval": {},
+        "policy": {
+            "id": "test-policy",
+            "policy_schema_version": 1,
+            "default_decision": "ask_backend",
+            "default_risk_class": "unknown",
+            "rules": [],
+        },
+        "downstream": {
+            "name": "env-test",
+            "command": sys.executable,
+            "args": [],
+            "env_passthrough": env_passthrough,
+        },
+    })
+
+
 def _set_downstream(config_path: Path, script: Path, *, log_path: Path | None = None) -> None:
     config = json.loads(config_path.read_text(encoding="utf-8"))
     env = {}
@@ -167,7 +200,7 @@ def _env_downstream(tmp_path: Path) -> Path:
     result = {
         "secret": os.environ.get("AWS_SECRET_ACCESS_KEY"),
         "explicit": os.environ.get("EXPLICIT_DOWNSTREAM_ENV"),
-        "passthrough": os.environ.get("AVP_TEST_ALLOWED_ENV"),
+        "passthrough": os.environ.get("MY_TOOL_ALLOWED_ENV"),
     }
     print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": result}), flush=True)
 """.lstrip(),
@@ -547,7 +580,7 @@ def __init__(self, *args, **kwargs):
 
 def test_downstream_env_is_minimal_by_default_and_explicit_only(tmp_path, monkeypatch):
     monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", SECRET)
-    monkeypatch.setenv("AVP_TEST_ALLOWED_ENV", "allowed")
+    monkeypatch.setenv("MY_TOOL_ALLOWED_ENV", "allowed")
     home = tmp_path / "avp-home"
     init = init_proxy(home=home, agent_name="proxy", plaintext=True)
     config = json.loads(init.config_path.read_text(encoding="utf-8"))
@@ -556,7 +589,7 @@ def test_downstream_env_is_minimal_by_default_and_explicit_only(tmp_path, monkey
         "command": sys.executable,
         "args": ["-u", str(_env_downstream(tmp_path))],
         "env": {"EXPLICIT_DOWNSTREAM_ENV": "explicit"},
-        "env_passthrough": ["AVP_TEST_ALLOWED_ENV"],
+        "env_passthrough": ["MY_TOOL_ALLOWED_ENV"],
     }
     _write_json(init.config_path, config)
 
@@ -574,6 +607,43 @@ def test_downstream_env_is_minimal_by_default_and_explicit_only(tmp_path, monkey
     }
 
 
+def test_downstream_config_rejects_avp_passphrase_in_env_passthrough():
+    config = _config_with_env_passthrough(["AVP_PROXY_PASSPHRASE"])
+
+    with pytest.raises(passthrough_module.PassthroughError, match="AVP_\\* prefix"):
+        DownstreamConfig.from_proxy_config(config)
+
+
+def test_downstream_config_rejects_other_avp_var_in_env_passthrough():
+    config = _config_with_env_passthrough(["AVP_HOME"])
+
+    with pytest.raises(passthrough_module.PassthroughError, match="AVP_\\* prefix"):
+        DownstreamConfig.from_proxy_config(config)
+
+
+def test_downstream_config_accepts_non_avp_env_passthrough():
+    config = _config_with_env_passthrough(["MY_TOOL_VAR", "HOME"])
+
+    parsed = DownstreamConfig.from_proxy_config(config)
+
+    assert parsed.env_passthrough == ("MY_TOOL_VAR", "HOME")
+
+
+def test_downstream_config_accepts_lowercase_avp_var():
+    config = _config_with_env_passthrough(["avp_internal"])
+
+    parsed = DownstreamConfig.from_proxy_config(config)
+
+    assert parsed.env_passthrough == ("avp_internal",)
+
+
+def test_downstream_config_rejects_avp_var_among_safe_vars():
+    config = _config_with_env_passthrough(["HOME", "AVP_PROXY_PASSPHRASE", "PATH"])
+
+    with pytest.raises(passthrough_module.PassthroughError, match="AVP_PROXY_PASSPHRASE"):
+        DownstreamConfig.from_proxy_config(config)
+
+
 def test_downstream_notifications_are_forwarded_before_matching_response(tmp_path):
     home = tmp_path / "avp-home"
     init = init_proxy(home=home, agent_name="proxy", plaintext=True)