williamloo-scale · Nov 3, 2026
@@ -20,8 +20,10 @@ them. The page shows the universe's time.
   `edit_note`, `toggle_note` and `delete_note`. Errors are raised as `NotFound`, `BadRequest` and `Unauthorized`.
 - **Roles.** `all_notes` (`GET /api/admin/notes`) is declared with `roles=["admin"]`. It answers only a caller that
   sends `AgentEnv-Role: admin` with `AgentEnv-Role-Key` equal to the container's `WEBSITE_ROLE_KEY`; any other caller
-  gets 403. The gateway passes on whatever role a browser sends, so without the key a page could claim `admin`; with no
-  `WEBSITE_ROLE_KEY` set, every caller is `default` and the route is closed.
+  gets 403. A deploy with `keyed_roles` gives the site its own role key as `WEBSITE_ROLE_KEY` and its gateway sends it
+  on only beside a role it verified with the same key, so a page cannot claim `admin` through it. Any other deploy sets
+  no key, and its gateway passes on whatever role a browser sends; with no `WEBSITE_ROLE_KEY` set, every caller is
+  `default` and the route is closed.
 - **Users.** `ApiContext.user` is the user in the signed session cookie. With no cookie, it is the acting user (the
   store's `current_user`), and `GET /api/session` signs that user in. `POST /api/sign-in` switches to another seeded
   user. A user's notes are invisible to every other user, and another user's note answers 404.

@@ -16,12 +16,14 @@
 import re
 from pathlib import Path
 from typing import Any, ClassVar, NamedTuple, Optional
+from urllib.parse import unquote
 
 from starlette.applications import Starlette
 from starlette.datastructures import MutableHeaders
 from starlette.middleware import Middleware
 from starlette.requests import Request
 from starlette.responses import FileResponse, JSONResponse, Response
+from starlette._utils import get_route_path
 from starlette.routing import Route
 
 from ..agent_env_environment import (
@@ -39,7 +41,8 @@
 from .clock import CLOCK_EXTENSION_URI, UNIVERSE_NOW_HEADER, UniverseClock, clock_patch_script, iso
 from .errors import NotFound
 from .readiness import WEBSITE_EXTENSION_URI, WEBSITE_READY_OP, WebsiteReadiness
-from .routes import ApiContext, ApiRoute, endpoint, error_response, openapi, openapi_path, request_role, reserved_path, routes_of
+from .routes import (ApiContext, ApiRoute, endpoint, error_response, forwarded_prefix, openapi, openapi_path, request_role,
+                     reserved_path, routes_of)
 from .session import SESSION_COOKIE, SessionSigner, expires_at
 from .store import Collections
 
@@ -284,9 +287,11 @@ def _check_health(cls: type) -> None:
 
 
 class _Forwarded:
-    """Puts the address the browser reached the site at in the scope, from ``X-Forwarded-Host`` and
-    ``X-Forwarded-Proto``, before anything routes it, so redirects, ``request.url`` and a mounted app's ``url_for``
-    name that address and not the internal one the proxy reached the server at."""
+    """Puts the address the browser reached the site at in the scope, from ``X-Forwarded-Host``,
+    ``X-Forwarded-Proto`` and ``X-Forwarded-Prefix``, before anything routes it, so redirects, ``request.url`` and a
+    mounted app's ``url_for`` name that address and not the internal one the proxy reached the server at. A prefix
+    becomes the scope's ``root_path`` and leads its ``path``, as ASGI has a mounted app's, so routing still sees the
+    site's own path."""
 
     def __init__(self, app: Any) -> None:
         self.app = app
@@ -301,6 +306,12 @@ async def __call__(self, scope: dict, receive: Any, send: Any) -> None:
                 scope["headers"] = [(k, v) for k, v in scope["headers"] if k != b"host"] + [(b"host", host.encode())]
             if proto in ("http", "https"):
                 scope["scheme"] = proto if scope["type"] == "http" else {"http": "ws", "https": "wss"}[proto]
+            prefix = forwarded_prefix(_first(headers.get(b"x-forwarded-prefix")))
+            if prefix:
+                scope["root_path"] = unquote(prefix) + scope.get("root_path", "")
+                scope["path"] = unquote(prefix) + scope["path"]
+                if scope.get("raw_path") is not None:
+                    scope["raw_path"] = prefix.encode("latin-1") + scope["raw_path"]
         await self.app(scope, receive, send)
 
 
@@ -381,7 +392,7 @@ async def __call__(self, scope: dict, receive: Any, send: Any) -> None:
             if self.app is not None:
                 await self.app(scope, receive, send)
             return
-        path, reading = scope["path"], scope["method"] in ("GET", "HEAD")
+        path, reading = get_route_path(scope) or "/", scope["method"] in ("GET", "HEAD")
         if reserved_path(path):
             return await error_response(NotFound("not found"))(scope, receive, send)
         if reading and (file := self._file(path)) is not None:

@@ -53,8 +53,9 @@ class ApiRoute(NamedTuple):
     fn: Callable
 
     def advertised(self) -> dict:
-        return {"method": self.method, "path": openapi_path(self.path), "name": self.name,
-                "roles": list(self.roles) or None}
+        """The route as the card names it: its path is the Starlette path it is served at, converters and all
+        (``/files/{rest:path}``), so a gateway matches the requests it answers, nested paths included."""
+        return {"method": self.method, "path": self.path, "name": self.name, "roles": list(self.roles) or None}
 
 
 class _Api:
@@ -159,11 +160,23 @@ def apply(self, response: Response) -> Response:
 
 def base_url(request: Request) -> str:
     """The site's address as the browser reached it: ``X-Forwarded-Proto`` and ``X-Forwarded-Host`` when a proxy
-    sent them, the request's own scheme and ``Host`` otherwise."""
+    sent them, the request's own scheme and ``Host`` otherwise, under the path ``X-Forwarded-Prefix`` names when the
+    proxy serves the site below one (the gateway's ``/website/<name>``)."""
     proto = request.headers.get("x-forwarded-proto", "").split(",")[0].strip() or request.url.scheme
     host = (request.headers.get("x-forwarded-host", "").split(",")[0].strip() or request.headers.get("host")
             or request.url.netloc)
-    return f"{proto}://{host}"
+    return f"{proto}://{host}{forwarded_prefix(request.headers.get('x-forwarded-prefix', ''))}"
+
+
+def forwarded_prefix(value: str) -> str:
+    """The path an ``X-Forwarded-Prefix`` value names, without its trailing slash; empty for none, and for one that is
+    not an absolute path or would make the site's URLs name another host (``//host``).
+
+    >>> [forwarded_prefix(v) for v in ("/website/notes/", "/a, /b", "", "/", "website", "//evil.example", "/a//b")]
+    ['/website/notes', '/a', '', '', '', '', '']
+    """
+    prefix = value.split(",")[0].strip().rstrip("/")
+    return prefix if prefix.startswith("/") and "//" not in prefix and prefix.isprintable() else ""
 
 
 ROLE_KEY_ENV = "WEBSITE_ROLE_KEY"
@@ -172,7 +185,8 @@ def base_url(request: Request) -> str:
 
 def request_role(request: Request) -> str:
     """The role ``AgentEnv-Role`` names, believed only from a caller that also sends ``AgentEnv-Role-Key`` equal to
-    ``WEBSITE_ROLE_KEY``; every other caller, and every caller of a site with no key set, is ``default``. The gateway
+    ``WEBSITE_ROLE_KEY``; every other caller, and every caller of a site with no key set, is ``default``. A deploy with
+    ``keyed_roles`` sets the key and its gateway sends it on beside a role it verified the same way; any other gateway
     forwards whatever role a browser sends, so without the key a page could claim any role."""
     claimed = caller_from_request(None, request.headers).role or DEFAULT_ROLE
     key = os.environ.get(ROLE_KEY_ENV, "")

@@ -320,6 +320,19 @@ def test_a_redirect_names_the_forwarded_address(client: TestClient) -> None:
     assert client.get("/api/items/", follow_redirects=False).headers["location"] == "http://testserver/api/items"
 
 
+def test_a_site_behind_a_forwarded_prefix_names_it_in_its_address_and_redirects_and_routes_its_own_paths(
+        client: TestClient) -> None:
+    behind = {"Host": "internal-x.invalid", "X-Forwarded-Host": "gw.example", "X-Forwarded-Proto": "https",
+              "X-Forwarded-Prefix": "/website/shop/"}
+    assert client.get("/api/whoami", headers=behind).json()["base_url"] == "https://gw.example/website/shop"
+    assert client.get("/api/items/1", headers=behind).json()["title"] == "mug"
+    redirected = client.get("/api/items/", follow_redirects=False, headers=behind)
+    assert (redirected.status_code, redirected.headers["location"]) == (307, "https://gw.example/website/shop/api/items")
+    for unusable in ("//evil.example", "website", ""):
+        seen = client.get("/api/whoami", headers={**behind, "X-Forwarded-Prefix": unusable}).json()
+        assert seen["base_url"] == "https://gw.example", unusable
+
+
 def test_parameterized_routes_never_take_the_data_plane_or_card() -> None:
     @website(domains=["wiki.example"])
     class Wiki(WebsiteEnvironment):

@@ -102,6 +102,18 @@ def test_static_files_and_index_fallback_answer_for_any_host(client: TestClient)
     assert client.post("/orders").status_code == 404
 
 
+def test_behind_a_forwarded_prefix_the_site_serves_its_own_paths_and_the_fallback_still_guards_reserved_ones(
+        client: TestClient) -> None:
+    behind = {"X-Forwarded-Prefix": "/website/shop"}
+    assert "home" in client.get("/", headers=behind).text
+    assert client.get("/assets/app.js", headers=behind).text == "console.log(1)"
+    assert "home" in client.get("/orders/42/edit", headers=behind).text
+    assert client.get("/api/orders", headers=behind).json() == []
+    assert client.get("/.well-known/agent-env.json", headers=behind).json()["name"] == "Shop"
+    assert client.get("/website/shop/assets/app.js", headers=behind).status_code == 404
+    assert client.post("/agentenv/x", headers=behind).status_code in (404, 405)
+
+
 def test_a_link_out_of_the_static_folder_is_never_served(dist: Path) -> None:
     secret = dist.parent / "secret.txt"
     (dist / "leak.txt").symlink_to(secret)

@@ -275,29 +275,33 @@ def _skill_extension(deployed: "DeployedA2AAgent") -> dict:
     @staticmethod
     async def install_cli(
         deployed: "DeployedA2AAgent", cli_artifact: "CliArtifact", gateway_url: str, container: str | None = None,
-        on_host: bool = False,
+        on_host: bool = False, role_key: str | None = None,
     ) -> str:
         """Install a CliArtifact bundle into the agent's filesystem at /opt/cli/<command>/, in ``container`` when
         install_agent put the agent in one, or on the VM host, ``on_host``, when it put it there. Writes a sidecar
-        bin/.env with the gateway URL and chmod +x's the entrypoint. Idempotent. Returns the path to the entrypoint."""
+        bin/.env with the gateway URL, and with ``role_key``, the key of that gateway's deploy, when it has one, so
+        each installed CLI sends its own deploy's key beside its role; then chmod +x's the entrypoint. Idempotent.
+        Returns the path to the entrypoint."""
         from agent_env.providers.sandbox_providers.sandbox import VmSandbox, into_container, stage_files_into_container
+        from agent_env.task_step.task_steps.mcp_cli_builder.codegen import GATEWAY_URL_ENV_VAR, ROLE_KEY_ENV_VAR
         from agent_env.providers.sandbox_providers.sandbox_provider import build_sandbox_provider, get_agent_sandbox_provider
         provider = build_sandbox_provider(deployed.sandbox_type) if deployed.sandbox_type else get_agent_sandbox_provider()
         sandbox = await provider.get_sandbox(deployed.sandbox_id)
         target_root = f"/opt/cli/{cli_artifact.command_name}"
         entrypoint_path = f"{target_root}/{cli_artifact.entrypoint}"
         env_dir = posixpath.dirname(entrypoint_path)
+        sidecar = f"{GATEWAY_URL_ENV_VAR}={gateway_url}\n" + (f"{ROLE_KEY_ENV_VAR}={role_key}\n" if role_key else "")
 
         if on_host:
-            await sandbox.write_host_file(f"AGENT_ENV_GATEWAY_URL={gateway_url}\n".encode(), f"{env_dir}/{GATEWAY_ENV_FILE}")
+            await sandbox.write_host_file(sidecar.encode(), f"{env_dir}/{GATEWAY_ENV_FILE}")
             await stage_files_into_container(
                 sandbox, cli_artifact.get_cli_files().get_file_artifacts(), target_root, on_host=True)
             await sandbox.exec_script(f"chmod +x {shlex.quote(entrypoint_path)}")
             logger.info(f"Installed CliArtifact '{cli_artifact.id}' v{cli_artifact.version} on the host at {entrypoint_path}")
             return entrypoint_path
 
         # write .env first — write_file_from_text creates the parent dir; write_file_from_object does not.
-        await sandbox.write_file_from_text(f"AGENT_ENV_GATEWAY_URL={gateway_url}\n", f"{env_dir}/{GATEWAY_ENV_FILE}",
+        await sandbox.write_file_from_text(sidecar, f"{env_dir}/{GATEWAY_ENV_FILE}",
                                            **into_container(container))
 
         for rel_path, file_artifact in cli_artifact.get_cli_files().get_file_artifacts().items():

@@ -48,7 +48,10 @@
 @click.option("--env-state-instance-id", default=None,
               help="Attach against an EXISTING env state instance instead of creating a fresh store "
                    "(its own type selects the provider).")
-def deploy(env_id: str, env_version: int | None, ttl_seconds: int, gateway_mode: str, sandbox: str, service_db_env_id: str | None, gateway_env_id: str | None, website_browser_env_id: str | None, disk_size_gb: float, cpu: float | None, memory_mb: int | None, env_state_type: str | None, env_state_instance_id: str | None):
+@click.option("--keyed-roles", is_flag=True, default=False,
+              help="Draw a role key for the deploy: its gateway then believes a role other than default only from a "
+                   "caller that sends the key beside it.")
+def deploy(env_id: str, env_version: int | None, ttl_seconds: int, gateway_mode: str, sandbox: str, service_db_env_id: str | None, gateway_env_id: str | None, website_browser_env_id: str | None, disk_size_gb: float, cpu: float | None, memory_mb: int | None, env_state_type: str | None, env_state_instance_id: str | None, keyed_roles: bool):
     """Deploy an MCP server environment."""
     from agent_env.providers import build_sandbox_provider
     from agent_env.config import get_config
@@ -89,6 +92,8 @@ def deploy(env_id: str, env_version: int | None, ttl_seconds: int, gateway_mode:
     if env_state_instance_id:
         deploy_kwargs["env_state_instance_id"] = env_state_instance_id
         click.echo(f"Env state instance id: {env_state_instance_id}")
+    if keyed_roles:
+        deploy_kwargs["keyed_roles"] = True
     try:
         with run_scope(env.id):
             deployed_env = asyncio.run(env.deploy(**deploy_kwargs))

@@ -17,7 +17,7 @@
 from agent_env.store.routing import namespace_routing_enabled
 from agent_env.utils.paths import validate_relative_filename
 
-from .gateway import GatewayMode
+from .gateway import AGENT_ENV_ROLE_HEADER, AGENT_ENV_ROLE_KEY_HEADER, GatewayMode
 from .gateway.constants import WELL_KNOWN_PATH
 from .store import get_env_store
 
@@ -83,6 +83,7 @@ class DeployedEnv:
     # The provider that made the record, so the class it loads as (see from_dict); capabilities still come from the card.
     env_provider_type: Optional[str] = None
     environment_card_url: Optional[str] = None
+    role_key: Optional[str] = None
     # The env card as last read, and when (ISO 8601, UTC); written at deploy from the readiness probe.
     environment_card: Optional[dict] = None
     environment_card_read_at_utc: Optional[str] = None
@@ -122,6 +123,11 @@ def control_headers_for(self, url: str) -> dict[str, str]:
         """What a call to ``url`` sends of the env's control token: ``control_headers`` when ``url`` is the env's
         address or under it, as a step's grant gives them, and none anywhere else, an override's address say."""
         return protocol_v1.granted_headers(url, {self.environment_url: self.control_headers})
+    def role_headers(self, role: str) -> dict[str, str]:
+        """The headers a request to the env sends to act as `role`: the role, and the deploy's `role_key` beside it,
+        without which a gateway that has one takes the request for `default`'s. A record from before the key, of a deploy without
+        `keyed_roles`, or of one with no gateway, has none, and its role header alone is believed."""
+        return {AGENT_ENV_ROLE_HEADER: role, **({AGENT_ENV_ROLE_KEY_HEADER: self.role_key} if self.role_key else {})}
 
     def supports(self, uri: str, method: str) -> bool:
         """Whether the stored env card offers `method` on the extension at `uri`; no network."""
@@ -200,6 +206,7 @@ def _fields_from(cls, data: dict) -> dict:
             created_at_utc=data.get("created_at_utc"),
             expires_at_utc=data.get("expires_at_utc"),
             control_token=data.get("control_token"),
+            role_key=data.get("role_key"),
         )
 
 

@@ -107,13 +107,13 @@ def accept_toml(cls, data: dict, ctx: AuthoringContext) -> dict:
         @environment_card when it isn't set. Raises BundleError listing every problem."""
         return ctx.accept_env(data, cls, named_by="image")
 
-    async def deploy(self, ttl_seconds: int = 10800, disk_size_gb: float = 10, gateway_mode: GatewayMode = GatewayMode.PERFORMANCE, cpu: float | None = None, memory_mb: int | None = None, sandbox_type: str | None = None, env_state_type: str | None = None, env_state_instance_id: str | None = None, artifact_id: str | None = None, artifact_version: int | None = None, *, attribution: Optional[Attribution] = None) -> DeployedEnv:
+    async def deploy(self, ttl_seconds: int = 10800, disk_size_gb: float = 10, gateway_mode: GatewayMode = GatewayMode.PERFORMANCE, cpu: float | None = None, memory_mb: int | None = None, sandbox_type: str | None = None, env_state_type: str | None = None, env_state_instance_id: str | None = None, artifact_id: str | None = None, artifact_version: int | None = None, *, attribution: Optional[Attribution] = None, keyed_roles: bool = False) -> DeployedEnv:
         # In container mode the server runs in its own container, so loads stage there, as for a MultiEnv child.
         return await deploy_through_provider(
             self, environment_name=self.environment_name, ttl_seconds=ttl_seconds, sandbox_type=sandbox_type,
             disk_size_gb=disk_size_gb, gateway_mode=gateway_mode, cpu=cpu, memory_mb=memory_mb,
             env_state_type=env_state_type, env_state_instance_id=env_state_instance_id, attribution=attribution,
-            artifact_id=artifact_id, artifact_version=artifact_version,
+            artifact_id=artifact_id, artifact_version=artifact_version, **({"keyed_roles": True} if keyed_roles else {}),
         )
 
     def deploy_refusal(self, *, sandbox_type: str | None = None, **options) -> str | None: