fix(env): the gateway's control routes take the deploy's control token

src/agent_env/env/gateway/constants.py

@@ -68,6 +68,7 @@ def data_plane_load_timeout_s(size_bytes: int | None) -> int:
 METHOD_ADD = "data/add"
 METHOD_GET = "data/get"
 MCP_TRANSPORT = "mcp"
+CONTROL_TOKEN_ENV = "AGENTENV_CONTROL_TOKEN"
 
 # Env/gateway-level extensions the gateway itself owns (tool gating + triggers).
 EXT_DISABLE_TOOL_URI = "urn:agentenv:disable-tool/v1"
@@ -159,6 +160,14 @@ def sources(names: list[str]) -> str:
 WILDCARD = "*"
 TOOL_DISABLE_ACTION = "disable"
 TOOL_ENABLE_ACTION = "enable"
+# The harness's routes that take the deploy's control token: the ones that change tool access, triggers, the clock
+# and notifications. Every other route stays open for now.
+CONTROL_ROUTES = frozenset({
+    f"/tools/{TOOL_DISABLE_ACTION}", f"/tools/{TOOL_ENABLE_ACTION}",
+    "/triggers/register", "/triggers/remove", "/triggers/clear",
+    "/clock/set-time", "/clock/advance", "/clock/clear",
+    "/notifications/config", "/notifications/take", "/notifications/clear",
+})
 
 
 class GatewayMode(str, Enum):

src/agent_env/env/gateway/entrypoint.py

@@ -7,7 +7,7 @@
 import httpx
 
 from . import AGENT_ENV_GATEWAY_MCP_PORT, GatewayMode, InternalMCPServer
-from .constants import DEFAULT_MCP_SERVER_NAME, GATEWAY_PROVIDER_HEADERS_ENV
+from .constants import CONTROL_TOKEN_ENV, DEFAULT_MCP_SERVER_NAME, GATEWAY_PROVIDER_HEADERS_ENV
 from .gateway import Gateway
 
 logger = logging.getLogger(__name__)
@@ -152,6 +152,7 @@ def parse_name_url_map(env_str: str) -> dict[str, str]:
         rest_proxy_urls=rest_proxy_urls,
         gateway_mode=gateway_mode,
         service_db_url=service_db_url,
+        control_token=os.environ.get(CONTROL_TOKEN_ENV),
         provider_headers=os.environ.get(GATEWAY_PROVIDER_HEADERS_ENV, "").split(","),
     )
     gateway.run()

src/agent_env/env/gateway/gateway.py

@@ -4,6 +4,7 @@
 import asyncio
 import atexit
 import contextlib
+import hmac
 import json
 import logging
 import os
@@ -50,6 +51,7 @@
     CHANGES_PAGE_DEFAULT,
     CHANGES_PAGE_MAX,
     CHILD_DISCOVERY_TIMEOUT_S,
+    CONTROL_ROUTES,
     DATA_PLANE_LOAD_TIMEOUT_MAX_S,
     DEFAULT_ROLE,
     EXT_WEBSITE_URI,
@@ -143,6 +145,7 @@ def __init__(
         rest_proxy_urls: dict[str, str] | None = None,
         gateway_mode: GatewayMode = GatewayMode.PERFORMANCE,
         service_db_url: str | None = None,
+        control_token: str | None = None,
         provider_headers: Iterable[str] = (),
     ):
         """`provider_headers` names the headers the sandbox provider in front of the gateway is sent, its credentials,
@@ -163,6 +166,7 @@ def __init__(
                 log.addFilter(TokenRedaction())
         self.gateway_mode = gateway_mode
         self._service_db_url = service_db_url
+        self._control_bearer = f"Bearer {control_token}".encode() if control_token else None
         self._website_routes: dict[str, list[WebsiteRoute] | None] = {}
         self._db_conn: psycopg2.extensions.connection | None = None
         if service_db_url:
@@ -462,8 +466,8 @@ def run(self) -> None:
     # inherited). Process: child sessions and the time-trigger driver, outer to the MCP session manager's lifespan.
 
     def _asgi_app(self):
-        """The served app: FastMCP's Starlette app with the request's caller bound around it, and the process
-        lifespan wrapped around the MCP session manager's."""
+        """The served app: FastMCP's Starlette app with the request's caller bound around it, the control routes behind
+        the deploy's token when it has one, and the process lifespan wrapped around the MCP session manager's."""
         inner = self._mcp.streamable_http_app()
         inner.router.lifespan_context = self._process_lifespan(inner.router.lifespan_context)
         if self.rest_proxy_urls:
@@ -473,6 +477,11 @@ async def app(scope, receive, send):
             if scope["type"] != "http":
                 await inner(scope, receive, send)
                 return
+            if self._control_bearer and scope["path"] in CONTROL_ROUTES and not hmac.compare_digest(
+                    dict(scope.get("headers", ())).get(b"authorization", b""), self._control_bearer):
+                response = self._json_response({"ok": False, "error": "this route takes the env's control token"}, 401)
+                await response(scope, receive, send)
+                return
             role = DEFAULT_ROLE
             for name, value in scope.get("headers", ()):
                 if name == ROLE_HEADER_BYTES:

src/agent_env/providers/env_providers/env_gateway_provider.py

@@ -45,6 +45,8 @@
     DB_WEB_PORT,
     PGWEB_SERVICE_NAME,
 )
+from agent_env.env.envs.computer.utils.boot import control_environment
+from agent_env.env.envs.computer.utils.boot import control_token as new_control_token
 from agent_env.env.envs.playwright_mcp import PLAYWRIGHT_MCP_RECORDER_PORT
 from agent_env.providers.env_providers.constants import (
     AGENT_ENV_WEBSITE_BACKEND_PORT,
@@ -128,6 +130,8 @@ class DeployedGateway:
     # The env card the readiness probe read, and when (ISO 8601, UTC).
     environment_card: dict | None = None
     environment_card_read_at_utc: str | None = None
+    # The token the gateway's control routes take, which the record keeps; None for a gateway started without one.
+    control_token: str | None = None
 
 
 @dataclass
@@ -292,6 +296,7 @@ def create_docker_compose(
         host_ips: tuple[str, ...] = (),
         extra_hosts: tuple[str, ...] = (),
         mcp_server_name: str | None = None,
+        control_token: str | None = None,
     ) -> str:
         """Generate docker-compose.yml content for a gateway deployment onto a VM/laptop/arbitrary machine.
 
@@ -500,6 +505,7 @@ def create_docker_compose(
             gateway_env.append(f"      - REST_PROXY_URLS={','.join(rest_proxy_urls)}")
         gateway_env.append(f"      - GATEWAY_MODE={gateway_mode.value}")
         gateway_env.append(f"      - SERVICE_DB_URL={state_provider.base_url(state_instance)}")
+        gateway_env.extend(f"      - {key}={value}" for key, value in control_environment(control_token).items())
         if provider_headers := sandbox_request_header_names():
             gateway_env.append(f"      - {GATEWAY_PROVIDER_HEADERS_ENV}={','.join(provider_headers)}")
         publish = host_port or (lambda port: port)
@@ -637,6 +643,7 @@ async def deploy(
             mcp_server_images=topology.mcp_server_images, website_configs=topology.website_configs,
             website_images=topology.website_images, mcp_server_name=topology.mcp_server_name, gateway_mode=gateway_mode,
             ttl_seconds=ttl_seconds, disk_size_gb=disk_size_gb, cpu=cpu, memory_mb=memory_mb, attribution=attribution,
+            control_token=new_control_token(),
         )
         return DeployedGatewayEnv(
             env_id=env.id,
@@ -653,6 +660,7 @@ async def deploy(
             website_frontend_urls=deployed.website_frontend_urls,
             gateway_mode=gateway_mode.value,
             env_state_instance_ids=deployed.env_state_instance_ids,
+            control_token=deployed.control_token,
         )
 
     async def create_gateway(
@@ -675,6 +683,7 @@ async def create_gateway(
         mcp_server_name: str | None = None,
         *,
         attribution: Optional[Attribution] = None,
+        control_token: str | None = None,
     ) -> DeployedGateway:
         """Create and start a gateway from explicit topology, for a caller that builds its own record rather than calling ``deploy()``.
 
@@ -688,13 +697,16 @@ async def create_gateway(
 
         ``mcp_server_name`` is the name the gateway presents on its card and as its MCP server; drawn as ``env`` +
         4 random digits when the env declares none; the record takes it from the gateway's card.
+
+        ``control_token`` is the token the gateway's control routes take, which the caller's record must keep; without one
+        they stay open.
         """
         return await self._launch(
             sandbox_provider, env_id, state_instance=state_instance, mcp_servers=mcp_servers, mcp_server_images=mcp_server_images,
             gateway_port=gateway_port, website_configs=website_configs, website_images=website_images,
             gateway_mode=gateway_mode, ttl_seconds=ttl_seconds, disk_size_gb=disk_size_gb, cpu=cpu, memory_mb=memory_mb,
             existing_sandbox=existing_sandbox, sidecars=sidecars, mcp_server_name=mcp_server_name,
-            attribution=attribution,
+            attribution=attribution, control_token=control_token,
         )
 
     async def _launch(self, sandbox_provider: SandboxProvider, env_id: str | None, *, mcp_server_name: str | None = None, **gateway) -> DeployedGateway:
@@ -753,6 +765,7 @@ async def _deploy_via_vm(
         existing_sandbox: Sandbox | None = None,
         sidecars: list[SidecarConfig] | None = None,
         mcp_server_name: str | None = None,
+        control_token: str | None = None,
     ) -> DeployedGateway:
         from agent_env.env.env import Env
         from agent_env.config import gateway_env_id, get_config, website_browser_env_id
@@ -844,6 +857,7 @@ async def _stand_up(_spec):
             host_ips=sandbox.host_ips,
             extra_hosts=sandbox.extra_hosts,
             mcp_server_name=mcp_server_name,
+            control_token=control_token,
         )
         logger.info(f"Generated docker-compose.yml:\n{_redact_compose_secrets(compose_content)}")
 
@@ -939,7 +953,7 @@ async def _stand_up(_spec):
             gateway_url=gateway_url, mcp_url=mcp_url, db_web_url=db_web_url, db_mcp_url=db_mcp_url,
             website_frontend_urls=website_frontend_urls, mcp_server_name=mcp_server_name,
             env_state_instance_ids=[state_instance.instance_id],
-            environment_card=card, environment_card_read_at_utc=card_read_at_utc,
+            environment_card=card, environment_card_read_at_utc=card_read_at_utc, control_token=control_token,
         )
 
     async def _deploy_via_containers(
@@ -956,6 +970,7 @@ async def _deploy_via_containers(
         memory_mb: int | None = None,
         attribution: Optional[Attribution] = None,
         mcp_server_name: str | None = None,
+        control_token: str | None = None,
     ) -> DeployedGateway:
         from agent_env.env.env import Env
         from agent_env.config import gateway_env_id, get_config
@@ -1052,6 +1067,7 @@ async def _deploy_via_containers(
                 "GATEWAY_MODE": gateway_mode.value,
                 "SERVICE_DB_URL": self._state_provider.base_url(state_instance),
                 "MCP_SERVER_NAME": mcp_server_name or DEFAULT_MCP_SERVER_NAME,
+                **control_environment(control_token),
                 **({GATEWAY_PROVIDER_HEADERS_ENV: ",".join(names)} if (names := sandbox_request_header_names()) else {}),
             }
             gateway_sb = await sandbox_provider.create_container(
@@ -1086,6 +1102,7 @@ async def _deploy_via_containers(
                 env_state_instance_ids=[state_instance.instance_id],
                 environment_card=card,
                 environment_card_read_at_utc=card_read_at_utc,
+                control_token=control_token,
             )
         except Exception:
             await self.close()
@@ -1386,6 +1403,7 @@ async def _deploy_gateway(
         existing_sandbox: Sandbox | None = None,
         sidecars: list[SidecarConfig] | None = None,
         mcp_server_name: str | None = None,
+        control_token: str | None = None,
         attribution: Optional[Attribution] = None,
     ) -> DeployedGateway:
         """One gateway deploy on one (unchained) sandbox provider: a VM on one that creates VMs, else a container per
@@ -1411,7 +1429,8 @@ async def _deploy_gateway(
                 sandbox_provider, mcp_servers, mcp_server_images,
                 gateway_port=gateway_port, website_configs=website_configs, gateway_mode=gateway_mode,
                 ttl_seconds=ttl_seconds, disk_size_gb=disk_size_gb,
-                cpu=cpu, memory_mb=memory_mb, mcp_server_name=mcp_server_name, **deploy_kwargs,
+                cpu=cpu, memory_mb=memory_mb, mcp_server_name=mcp_server_name, control_token=control_token,
+                **deploy_kwargs,
             )
         earlier_sandbox = self._sandbox
         try:
@@ -1420,7 +1439,7 @@ async def _deploy_gateway(
                 gateway_port=gateway_port, website_configs=website_configs, website_images=website_images,
                 gateway_mode=gateway_mode, ttl_seconds=ttl_seconds, disk_size_gb=disk_size_gb,
                 cpu=cpu, memory_mb=memory_mb, existing_sandbox=existing_sandbox, sidecars=sidecars,
-                mcp_server_name=mcp_server_name, **deploy_kwargs,
+                mcp_server_name=mcp_server_name, control_token=control_token, **deploy_kwargs,
             )
         except BaseException:
             await self._terminate_new_sandbox(earlier_sandbox=earlier_sandbox, existing_sandbox=existing_sandbox)