diff --git a/verifiers/v1/env.py b/verifiers/v1/env.py
index 29c22ab85..824b102e4 100644
--- a/verifiers/v1/env.py
+++ b/verifiers/v1/env.py
@@ -23,10 +23,13 @@
 from verifiers.v1.retries import RetryConfig
 from verifiers.v1.rollout import Rollout
 from verifiers.v1.runtimes import (
+    DockerConfig,
     RuntimeConfig,
     SubprocessConfig,
     runtime_is_local,
+    runtime_reaches_host_locally,
 )
+from verifiers.v1.runtimes.docker import docker_interception_host
 from verifiers.v1.task import Task, resolve_server_config
 from verifiers.v1.taskset import Taskset, TasksetConfig
 from verifiers.v1.utils.generic import generic_type
@@ -371,8 +374,19 @@ async def serving(self):
         what keeps both eval runners (in-process and env-server) on one serving path. Build
         episodes inside this context; the resources are torn down on exit."""
         async with self.shared_tools() as shared:
+            tunneled = self._requires_tunnel(shared)
+            runtime = self.harness.config.runtime
+            extra_host = (
+                await docker_interception_host()
+                if isinstance(runtime, DockerConfig)
+                and not runtime.network_access
+                and not tunneled
+                else None
+            )
             interception = make_interception(
-                self.config.interception, requires_tunnel=self._requires_tunnel(shared)
+                self.config.interception,
+                requires_tunnel=tunneled,
+                extra_host=extra_host,
             )
             async with interception:
                 self._shared_tools = shared
@@ -402,7 +416,9 @@ def _requires_tunnel(self, shared: dict[str, SharedToolServer]) -> bool:
             for server_cls in server_classes
         ]
         return requires_tunnel(
-            runtime_is_local(self.harness.config.runtime), configs, shared.values()
+            runtime_is_local(self.harness.config.runtime),
+            configs,
+            shared.values(),
         )
 
     @contextlib.asynccontextmanager
@@ -411,6 +427,8 @@ async def shared_tools(self):
         if not servers:
             yield {}
             return
-        harness_is_local = runtime_is_local(self.harness.config.runtime)
-        async with serve_shared(servers, harness_is_local=harness_is_local) as shared:
+        harness_reaches_host = runtime_reaches_host_locally(self.harness.config.runtime)
+        async with serve_shared(
+            servers, harness_reaches_host=harness_reaches_host
+        ) as shared:
             yield shared
diff --git a/verifiers/v1/interception/__init__.py b/verifiers/v1/interception/__init__.py
index eb698c675..e56f6e6ac 100644
--- a/verifiers/v1/interception/__init__.py
+++ b/verifiers/v1/interception/__init__.py
@@ -25,7 +25,7 @@
     InterceptionServer,
     InterceptionServerConfig,
 )
-from verifiers.v1.runtimes import runtime_is_local
+from verifiers.v1.runtimes import runtime_reaches_host_locally
 
 if TYPE_CHECKING:
     from verifiers.v1.mcp import SharedToolServer
@@ -44,38 +44,34 @@ def requires_tunnel(
     server_configs: Iterable[BaseConfig] = (),
     shared: "Iterable[SharedToolServer]" = (),
 ) -> bool:
-    """Whether the interception must be exposed via a tunnel — some consumer is off the
-    host network: the harness itself, a live `shared` server in a remote runtime, or a
-    tool/user server config placing one there (each reaches the `/state` channel from
-    its own runtime). Skipped as non-consumers: a `colocated` server (shares the
-    harness's runtime, covered by `harness_is_local`), a config-`url` server (external —
-    it connects out), and an `external` shared server (outside the state machinery
-    entirely). False means every consumer reaches the server at localhost."""
+    """Whether interception needs a public tunnel because any consumer is remote.
+
+    Local runtimes map framework routes through their network policy. Colocated servers
+    share that route; configured URLs and external shared servers do not consume state."""
     if not harness_is_local:
         return True
-    if any(not s.external and not s.local for s in shared):
+    if any(not s.external and not s.reaches_host for s in shared):
         return True
     for config in server_configs:
         if getattr(config, "url", None) or config.colocated:
             continue
-        if not runtime_is_local(config.runtime):
+        if not runtime_reaches_host_locally(config.runtime):
             return True
     return False
 
 
 def make_interception(
-    config: InterceptionConfig, *, requires_tunnel: bool
+    config: InterceptionConfig,
+    *,
+    requires_tunnel: bool,
+    extra_host: str | None = None,
 ) -> Interception:
-    """The interception for a config, picked by type (the host-side counterpart to
-    `make_runtime`). With `requires_tunnel` (some consumer is off the host network — see
-    the `requires_tunnel` util) each server is exposed via its configured tunnel; without
-    it they get none and are reached at localhost. The caller computes it (see
-    `Environment._requires_tunnel`)."""
+    """Build the configured interception shape for the resolved network topology."""
     if isinstance(config, InterceptionServerConfig):
-        return InterceptionServer(config, requires_tunnel)
+        return InterceptionServer(config, requires_tunnel, extra_host=extra_host)
     if isinstance(config, StaticInterceptionPoolConfig):
-        return StaticInterceptionPool(config, requires_tunnel)
-    return ElasticInterceptionPool(config, requires_tunnel)
+        return StaticInterceptionPool(config, requires_tunnel, extra_host=extra_host)
+    return ElasticInterceptionPool(config, requires_tunnel, extra_host=extra_host)
 
 
 __all__ = [
diff --git a/verifiers/v1/interception/pool.py b/verifiers/v1/interception/pool.py
index 324260635..1088aa408 100644
--- a/verifiers/v1/interception/pool.py
+++ b/verifiers/v1/interception/pool.py
@@ -47,12 +47,17 @@ class StaticInterceptionPool(Interception):
     operator's call (it's the shape for pre-provisioned/bring-your-own endpoints)."""
 
     def __init__(
-        self, config: StaticInterceptionPoolConfig, requires_tunnel: bool = False
+        self,
+        config: StaticInterceptionPoolConfig,
+        requires_tunnel: bool = False,
+        *,
+        extra_host: str | None = None,
     ) -> None:
         super().__init__()
         self.config = config
         self.servers = [
-            InterceptionServer(server, requires_tunnel) for server in config.servers
+            InterceptionServer(server, requires_tunnel, extra_host=extra_host)
+            for server in config.servers
         ]
 
     async def start(self) -> None:
@@ -91,10 +96,13 @@ def __init__(
         self,
         config: ElasticInterceptionPoolConfig | None = None,
         requires_tunnel: bool = False,
+        *,
+        extra_host: str | None = None,
     ) -> None:
         super().__init__()
         self.config = config or ElasticInterceptionPoolConfig()
         self.requires_tunnel = requires_tunnel
+        self.extra_host = extra_host
         self.servers: list[InterceptionServer] = []
         self._lock = asyncio.Lock()
 
@@ -110,7 +118,9 @@ async def _server(self) -> InterceptionServer:
                 return server
         # Pin prime explicitly — the only tunnel kind that can be minted on demand.
         server = InterceptionServer(
-            InterceptionServerConfig(tunnel=PrimeTunnelConfig()), self.requires_tunnel
+            InterceptionServerConfig(tunnel=PrimeTunnelConfig()),
+            self.requires_tunnel,
+            extra_host=self.extra_host,
         )
         await self.stack.enter_async_context(server)
         self.servers.append(server)
diff --git a/verifiers/v1/interception/server.py b/verifiers/v1/interception/server.py
index a232266a0..4ad438ad2 100644
--- a/verifiers/v1/interception/server.py
+++ b/verifiers/v1/interception/server.py
@@ -98,17 +98,14 @@ class InterceptionServerConfig(BaseInterceptionConfig):
 
 
 class InterceptionServer(Interception):
-    """A server that proxies model calls for one or more rollouts — and is itself the
-    single-server `Interception` (the pools compose several of these). With
-    `requires_tunnel` (some consumer is off the host network) it mints its configured
-    tunnel; on `start` it then binds where the tunnel says (`bind_host`/`bind_port`) and
-    sets `base_url` — the one URL every consumer reaches it at — to the tunnel's public
-    URL. Without, every consumer is on the host network: it binds loopback, tunnel-free."""
+    """A model proxy for one or more rollouts, optionally exposed through a tunnel."""
 
     def __init__(
         self,
         config: InterceptionServerConfig | None = None,
         requires_tunnel: bool = False,
+        *,
+        extra_host: str | None = None,
     ) -> None:
         super().__init__()
         self.sessions: dict[str, RolloutSession] = {}
@@ -116,7 +113,7 @@ def __init__(
         self.tunnel: Tunnel | None = (
             make_tunnel(self.config.tunnel) if requires_tunnel else None
         )
-        self.host = "127.0.0.1"
+        self.hosts = ("127.0.0.1",) + ((extra_host,) if extra_host else ())
         self.port = 0
         self.base_url = ""  # set by `start`
         self.runner: web.AppRunner | None = None
@@ -176,19 +173,19 @@ async def start(self) -> None:
         self.runner = web.AppRunner(app)
         await self.runner.setup()
         self.stack.push_async_callback(self.runner.cleanup)
-        # No tunnel → every consumer shares the host network: bind loopback on any ephemeral
-        # port. Otherwise the tunnel says where to bind for it to reach the port, and
-        # `expose` publishes it.
+        # Local Docker also listens on its bridge gateway while localhost stays canonical.
         if self.tunnel is None:
-            self.host, bind_port = "127.0.0.1", 0
+            hosts, bind_port = self.hosts, 0
         else:
-            self.host, bind_port = self.tunnel.bind_host, self.tunnel.bind_port
-        site = web.TCPSite(self.runner, self.host, bind_port)
+            hosts, bind_port = (self.tunnel.bind_host,), self.tunnel.bind_port
+        site = web.TCPSite(self.runner, hosts[0], bind_port)
         await site.start()
         self.port = site._server.sockets[0].getsockname()[1]  # actual bound port
-        logger.info("interception up: url=http://%s:%d", self.host, self.port)
+        for host in hosts[1:]:
+            await web.TCPSite(self.runner, host, self.port).start()
+        logger.info("interception up: hosts=%s port=%d", hosts, self.port)
         self.stack.callback(
-            logger.info, "interception down: url=http://%s:%d", self.host, self.port
+            logger.info, "interception down: hosts=%s port=%d", hosts, self.port
         )
         if self.tunnel is None:
             self.base_url = f"http://127.0.0.1:{self.port}"
diff --git a/verifiers/v1/mcp/launch.py b/verifiers/v1/mcp/launch.py
index d28315d55..1643a461b 100644
--- a/verifiers/v1/mcp/launch.py
+++ b/verifiers/v1/mcp/launch.py
@@ -17,13 +17,13 @@
 from typing import TYPE_CHECKING
 from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit
 
-from verifiers.v1.errors import RolloutError, ToolsetError, UserError
+from verifiers.v1.errors import RolloutError, SandboxError, ToolsetError, UserError
 from verifiers.v1.interception.tunnel import PrimeTunnel
 from verifiers.v1.mcp.server import STATE_SECRET_PARAM, STATE_URL_PARAM, ServerBase
 from verifiers.v1.runtimes import (
     Runtime,
     make_runtime,
-    runtime_is_local,
+    runtime_reaches_host_locally,
 )
 from verifiers.v1.runtimes.base import _ENSURE_UV
 from verifiers.v1.types import Messages
@@ -217,21 +217,24 @@ async def serve_in_runtime(
 
 @contextlib.asynccontextmanager
 async def reachable_url(
-    service: Runtime, port: int, *, colocated: bool, consumer_is_local: bool
+    service: Runtime, port: int, *, colocated: bool, consumer_reaches_host: bool
 ) -> AsyncIterator[str]:
     """Yield the URL a consumer uses to reach the server at (`service`, `port`), over two
     primitives: `Runtime.expose` (publish a port out of a sandbox) and a host `Tunnel` (reach
     into the host from a remote runtime). `colocated` = the server shares the consumer's
-    runtime; `consumer_is_local` = the consumer is on the host network.
+    runtime; `consumer_reaches_host` = the consumer reaches the host network directly.
 
     - `colocated` -> localhost (same runtime, in-sandbox or host loopback);
-    - the server runs in a remote sandbox -> its own published URL (`expose`), reachable anywhere;
+    - the server is not host-reachable -> its published URL (`expose`), reachable anywhere;
     - else it's on the host network -> localhost to a local consumer, a host tunnel to a remote one."""
     if colocated:
         yield f"http://127.0.0.1:{port}"
-    elif not service.is_local:  # in a remote sandbox → it publishes its own port
-        yield await service.expose(port)
-    elif consumer_is_local:  # host network, local consumer → localhost, no tunnel
+    elif not service.reachable_from_host_locally:
+        url = await service.expose(port)
+        if url is None:
+            raise SandboxError(f"{service.type} runtime cannot expose port {port}")
+        yield url
+    elif consumer_reaches_host:  # host network, direct consumer → localhost, no tunnel
         yield f"http://127.0.0.1:{port}"
     else:  # host network, remote consumer → a host tunnel publishes the port outward
         async with PrimeTunnel().expose(port) as url:
@@ -243,7 +246,7 @@ async def serve(
     server: ServerBase,
     harness_runtime: Runtime | None = None,
     for_host: bool = False,
-    harness_is_local: bool = True,
+    harness_reaches_host: bool = True,
     *,
     state_secret: str = "",
     state_base: str | None = None,
@@ -254,6 +257,9 @@ async def serve(
             runtime = harness_runtime
         else:
             runtime = make_runtime(cfg.runtime)
+            if runtime.network_policy.restricted:
+                error = UserError if for_host else ToolsetError
+                raise error("network policies are only applied to harness runtimes")
             await runtime.start()
             stack.push_async_callback(runtime.stop)
         # Only consumers outside the server runtime need its fixed published port. Colocated tools
@@ -276,19 +282,22 @@ async def serve(
         # Who consumes the server decides reachability: a user sim is reached by the HOST
         # (`for_host`, always local, never colocated with it); a tool by the harness — colocated
         # when it shares the harness's runtime, reached with the harness's locality (read off the
-        # harness runtime when there is one, else `harness_is_local` for an eval-level shared tool).
+        # harness runtime when there is one, else `harness_reaches_host` for a shared tool).
         if for_host:
-            colocated, consumer_is_local = False, True
+            colocated, consumer_reaches_host = False, True
         else:
             colocated = runtime is harness_runtime
-            consumer_is_local = (
-                harness_runtime.is_local
+            consumer_reaches_host = (
+                harness_runtime.reaches_host_locally
                 if harness_runtime is not None
-                else harness_is_local
+                else harness_reaches_host
             )
         base = await stack.enter_async_context(
             reachable_url(
-                runtime, port, colocated=colocated, consumer_is_local=consumer_is_local
+                runtime,
+                port,
+                colocated=colocated,
+                consumer_reaches_host=consumer_reaches_host,
             )
         )
         if not for_host and not colocated and harness_runtime is not None:
@@ -299,7 +308,7 @@ async def serve(
 @dataclass(frozen=True)
 class SharedToolServer:
     """One live taskset-scoped (shared) server, as the rollouts see it: its eval-level
-    `url` plus whether its runtime is `local` (host-reachable) — a remote one is an
+    `url` plus whether its runtime reaches the host directly — otherwise it is an
     interception consumer, so the interception must be exposed for it to reach the
     `/state` channel (see `Environment._requires_tunnel`). An `external` server (a
     config-`url` endpoint) was not launched by the framework and sits outside its state
@@ -307,17 +316,17 @@ class SharedToolServer:
     secret sent to a third party)."""
 
     url: str
-    local: bool
+    reaches_host: bool
     external: bool = False
 
 
 @contextlib.asynccontextmanager
-async def serve_shared(toolsets: list[Toolset], harness_is_local: bool = True):
+async def serve_shared(toolsets: list[Toolset], harness_reaches_host: bool = True):
     """Start the taskset-scoped (shared) tool servers ONCE for a whole eval, each in its OWN
     `runtime`, and yield `{name: SharedToolServer}` reachable by every rollout's harness.
     Reachability mirrors a per-rollout tool, but there's no single harness runtime to read
     locality off — the caller (`Environment.shared_tools`) passes the harness runtime's
-    `harness_is_local`, so a host tool gets one host bridge (tunnel) when the harness runs
+    `harness_reaches_host`, so a host tool gets one host bridge when the harness runs
     remotely, and a remote tool runtime publishes its own URL. Torn down when the eval ends.
     A shared server is task-agnostic — the taskset carries no per-row data — so its `setup`
     gets no task (its `setup_task` is never called; the per-rollout servers fetch
@@ -342,14 +351,15 @@ async def serve_shared(toolsets: list[Toolset], harness_is_local: bool = True):
                 )
             if cfg.url:  # already running remotely; nothing launched, nothing to bridge
                 servers[name] = SharedToolServer(
-                    url=cfg.url, local=False, external=True
+                    url=cfg.url, reaches_host=False, external=True
                 )
             else:
                 url = await stack.enter_async_context(
-                    serve(toolset, harness_is_local=harness_is_local)
+                    serve(toolset, harness_reaches_host=harness_reaches_host)
                 )
                 servers[name] = SharedToolServer(
-                    url=url, local=runtime_is_local(cfg.runtime)
+                    url=url,
+                    reaches_host=runtime_reaches_host_locally(cfg.runtime),
                 )
             logger.info("shared tool server '%s': %s", name, servers[name].url)
         yield servers
@@ -393,7 +403,11 @@ async def serve_tools(
                 urls[name] = server.url
                 logger.info("tool server '%s' (shared, external): %s", name, server.url)
                 continue
-            url = harness_runtime.host_url(server.url) if server.local else server.url
+            url = (
+                harness_runtime.host_url(server.url)
+                if server.reaches_host
+                else server.url
+            )
             urls[name] = _shared_url_for_rollout(url, state_base, state_secret)
             # The tagged URL contains the bearer secret; log only the untagged base URL.
             logger.info("tool server '%s' (shared): %s", name, server.url)
diff --git a/verifiers/v1/rollout.py b/verifiers/v1/rollout.py
index 8a04ef785..eff8132f1 100644
--- a/verifiers/v1/rollout.py
+++ b/verifiers/v1/rollout.py
@@ -13,6 +13,7 @@
     RolloutError,
     TaskError,
     ToolsetError,
+    UserError,
     boundary,
 )
 from verifiers.v1.interception import (
@@ -94,7 +95,9 @@ async def _serve_interception(
             [server.config for server in servers],
             self.shared_tools.values(),
         )
-        server = InterceptionServer(requires_tunnel=tunneled)
+        server = InterceptionServer(
+            requires_tunnel=tunneled, extra_host=runtime.interception_host
+        )
         async with server:
             async with server.acquire(session) as slot:
                 yield slot
@@ -150,7 +153,22 @@ async def run(self) -> Trace:
                 await self.harness.setup(runtime)
             async with boundary(ToolsetError, "building tool servers"):
                 tool_servers = self.task.tool_servers()
+                if not runtime.supports_colocated_tools and any(
+                    server.config.colocated and not server.config.url
+                    for server in tool_servers
+                ):
+                    raise ToolsetError(
+                        "this harness network policy does not support colocated MCP servers"
+                    )
             user = self.task.user_server()
+            if (
+                not runtime.supports_colocated_user
+                and user is not None
+                and user.config.colocated
+            ):
+                raise UserError(
+                    "this harness network policy does not support a colocated user server"
+                )
             # `base_url` is the interception server's reachable URL for this rollout. The
             # harness reaches the model at `{base_url}/v1`; tool/user servers reach this
             # rollout's `/state` + `/task` at `base_url` — it's universally reachable (the
@@ -180,6 +198,14 @@ async def run(self) -> Trace:
                             "conversation; set task.prompt or declare a simulator "
                             "class on Task.user"
                         )
+                    routes = await runtime.apply_network_policy(
+                        {
+                            "model": endpoint,
+                            **{f"mcp:{name}": url for name, url in urls.items()},
+                        }
+                    )
+                    endpoint = routes["model"]
+                    urls = {name: routes[f"mcp:{name}"] for name in urls}
                     now = time.time()
                     trace.timing.setup.end = now
                     trace.timing.generation.start = now
diff --git a/verifiers/v1/runtimes/__init__.py b/verifiers/v1/runtimes/__init__.py
index bd5341806..fe6a7ac13 100644
--- a/verifiers/v1/runtimes/__init__.py
+++ b/verifiers/v1/runtimes/__init__.py
@@ -47,12 +47,15 @@ def make_runtime(config: RuntimeConfig, name: str | None = None) -> Runtime:
 
 
 def runtime_is_local(config: RuntimeConfig) -> bool:
-    """Whether a runtime of this config shares the host network (so a program inside it reaches a
-    host service at localhost, no tunnel) — read off the runtime class, without provisioning one.
-    The interception / tool serving use it to decide whether to tunnel their host port."""
+    """Whether this runtime runs locally rather than in a provider sandbox."""
     return _runtime_cls(config).is_local
 
 
+def runtime_reaches_host_locally(config: RuntimeConfig) -> bool:
+    """Whether this config reaches host services at localhost rather than through a tunnel."""
+    return _runtime_cls(config).config_reaches_host_locally(config)
+
+
 __all__ = [
     "ProgramResult",
     "Runtime",
@@ -61,6 +64,7 @@ def runtime_is_local(config: RuntimeConfig) -> bool:
     "BaseRuntimeInfo",
     "make_runtime",
     "runtime_is_local",
+    "runtime_reaches_host_locally",
     "SubprocessConfig",
     "SubprocessRuntime",
     "SubprocessRuntimeInfo",
diff --git a/verifiers/v1/runtimes/base.py b/verifiers/v1/runtimes/base.py
index a6fdf08f1..303073ee7 100644
--- a/verifiers/v1/runtimes/base.py
+++ b/verifiers/v1/runtimes/base.py
@@ -15,6 +15,8 @@
 
 from pydantic_config import BaseConfig
 
+from verifiers.v1.errors import SandboxError
+from verifiers.v1.runtimes.network import NetworkPolicy
 from verifiers.v1.utils.aio import run_shielded
 
 logger = logging.getLogger(__name__)
@@ -100,13 +102,15 @@ class BaseRuntimeInfo(BaseConfig):
 
 class Runtime(ABC):
     is_local: ClassVar[bool] = True
-    """Whether this runtime shares the host network — a program inside it reaches a host service
-    at localhost (no tunnel) and a service inside it is reachable at localhost. True for
-    subprocess / docker(--network host); remote runtimes (modal/prime) override to False (they
-    need a tunnel each way: a host `Tunnel` (interception.tunnel) inward, `expose` outward)."""
+    """Whether this runtime is provisioned on the local machine rather than by a provider."""
 
     info: BaseRuntimeInfo
 
+    @classmethod
+    def config_reaches_host_locally(cls, _config: BaseConfig) -> bool:
+        """Whether this runtime config reaches host services without a tunnel."""
+        return cls.is_local
+
     def __init__(self, name: str | None = None) -> None:
         self.name = name or f"vf-{uuid.uuid4().hex[:12]}"
         self._uv_interpreters: dict[str, str] = {}
@@ -120,6 +124,36 @@ def type(self) -> str:
     def descriptor(self) -> str | None:
         return self.info.id
 
+    @property
+    def reaches_host_locally(self) -> bool:
+        """Whether programs here reach host services at localhost, without a tunnel."""
+        return type(self).config_reaches_host_locally(self.config)
+
+    @property
+    def reachable_from_host_locally(self) -> bool:
+        """Whether the host reaches services here at localhost, without exposure."""
+        return self.is_local
+
+    @property
+    def interception_host(self) -> str | None:
+        """Additional host address where local interception must listen."""
+        return None
+
+    @property
+    def network_policy(self) -> NetworkPolicy:
+        """Egress policy applied at the trusted setup-to-execution boundary."""
+        return NetworkPolicy()
+
+    @property
+    def supports_colocated_tools(self) -> bool:
+        """Whether colocated MCP servers work under this runtime policy."""
+        return True
+
+    @property
+    def supports_colocated_user(self) -> bool:
+        """Whether a colocated user server is host-reachable under this policy."""
+        return True
+
     @abstractmethod
     async def start(self) -> None:
         pass
@@ -157,6 +191,15 @@ async def run_program(self, argv: list[str], env: dict[str, str]) -> ProgramResu
         still retry individual safe transport operations underneath `run`."""
         return await self.run(argv, env)
 
+    async def apply_network_policy(self, routes: dict[str, str]) -> dict[str, str]:
+        """Apply the policy once after setup and map required framework route URLs.
+
+        The policy remains active until `stop()` tears down the runtime.
+        """
+        if self.network_policy.restricted:
+            raise SandboxError(f"{self.type} runtime does not support network policies")
+        return routes
+
     async def run_background(
         self, argv: list[str], env: dict[str, str], log: str
     ) -> None:
@@ -244,14 +287,13 @@ def published_port(self) -> int | None:
         """A fixed port this runtime exposes to the outside at startup, declared up front to the
         provider (Modal forwards only ports named at `Sandbox.create`). When set, a server placed
         here binds it instead of a host-chosen free port, and `expose` returns its public URL.
-        `None` for host-networked runtimes (subprocess/docker), which pick a free port and are
-        reached over the shared host network."""
+        `None` for runtimes reached directly from the host, which pick a free port."""
         return None
 
     async def expose(self, port: int) -> str | None:
         """Publish a port running *inside this runtime* to a URL reachable from the host/outside,
-        or None when local (it's on the host network — reach it at localhost). A remote runtime
-        overrides this with the provider's native port exposure (modal `tunnels()`, prime
-        `client.expose`), torn down with the sandbox in `stop()`. The reverse of a host `Tunnel`
+        or None when `reachable_from_host_locally` is true. A remote runtime overrides this
+        with the provider's native port exposure (modal `tunnels()`, prime `client.expose`),
+        torn down with the sandbox in `stop()`. The reverse of a host `Tunnel`
         (interception.tunnel, which reaches a host port from inside a runtime)."""
         return None
diff --git a/verifiers/v1/runtimes/docker.py b/verifiers/v1/runtimes/docker.py
index 3fc689c8a..bea76b6e4 100644
--- a/verifiers/v1/runtimes/docker.py
+++ b/verifiers/v1/runtimes/docker.py
@@ -1,14 +1,15 @@
-"""Local Docker runtime using host networking."""
+"""Local Docker runtime with policy-controlled agent networking."""
 
 import asyncio
-import contextlib
+import json
 import logging
+import re
 import shlex
 import subprocess
 import sys
 from pathlib import PurePosixPath
-from typing import Literal
-from urllib.parse import urlsplit
+from typing import Literal, cast
+from urllib.parse import urlsplit, urlunsplit
 
 from pydantic_config import BaseConfig
 
@@ -19,14 +20,21 @@
     Runtime,
     parse_gpu,
 )
+from verifiers.v1.runtimes.network import NetworkPolicy
 
 logger = logging.getLogger(__name__)
 
+_EGRESS_IMAGE = "nginx:1.28.3-alpine"
+_EGRESS_HOST = "egress"
+_EGRESS_PORT = 8080
+
 
 class DockerConfig(BaseConfig):
     type: Literal["docker"] = "docker"
     image: str = "python:3.11-slim"
     workdir: str = "/app"
+    network_access: bool = True
+    """Allow arbitrary agent egress. Setup has access; false then allows framework routes only."""
     # TaskData.resources uses these units; non-default runtime config values take precedence.
     cpu: float | None = None
     """Pin the container to this many CPU cores (docker `--cpus`). None = unlimited."""
@@ -59,14 +67,66 @@ async def docker(*args: str) -> ProgramResult:
     )
 
 
+async def docker_checked(*args: str, error: str) -> ProgramResult:
+    result = await docker(*args)
+    if result.exit_code == 0:
+        return result
+    detail = (result.stderr or result.stdout).strip()
+    raise SandboxError(f"{error}: {detail}")
+
+
+async def docker_interception_host() -> str | None:
+    """Return the host address used by native Linux containers."""
+    if sys.platform != "linux":
+        return None
+    result = await docker(
+        "network", "inspect", "--format", "{{(index .IPAM.Config 0).Gateway}}", "bridge"
+    )
+    gateway = result.stdout.strip()
+    if result.exit_code != 0 or not gateway:
+        raise SandboxError(
+            "could not resolve the Docker host gateway: "
+            f"{(result.stderr or result.stdout).strip()}"
+        )
+    return gateway
+
+
 class DockerRuntime(Runtime):
     def __init__(self, config: DockerConfig, name: str | None = None) -> None:
         super().__init__(name)
         self.config = config
         self.info = DockerRuntimeInfo(**config.model_dump())
         self._container: str | None = None  # our `--name` (used for exec/rm)
+        self._egress: str | None = None
+        self._setup_network: str | None = None
+        self._execution_network: str | None = None
+        self._interception_host: str | None = None
         self._stopped = False
 
+    @classmethod
+    def config_reaches_host_locally(cls, config: BaseConfig) -> bool:
+        return cast(DockerConfig, config).network_access
+
+    @property
+    def reachable_from_host_locally(self) -> bool:
+        return self.config.network_access
+
+    @property
+    def interception_host(self) -> str | None:
+        return self._interception_host
+
+    @property
+    def network_policy(self) -> NetworkPolicy:
+        return NetworkPolicy(allow=None if self.config.network_access else [])
+
+    @property
+    def supports_colocated_tools(self) -> bool:
+        return self.config.network_access
+
+    @property
+    def supports_colocated_user(self) -> bool:
+        return self.config.network_access
+
     async def start(self) -> None:
         try:
             version = await docker("version", "--format", "{{.Server.Version}}")
@@ -95,11 +155,36 @@ async def start(self) -> None:
         _, gpu_count = parse_gpu(self.config.gpu)
         if gpu_count:
             limits += ["--gpus", str(gpu_count)]
-        run = await docker(
+        if self.network_policy.restricted:
+            self._interception_host = await docker_interception_host()
+            self._setup_network = f"{self.name}-setup"
+            await docker_checked(
+                "network",
+                "create",
+                "--label",
+                f"verifiers.runtime={self.name}",
+                self._setup_network,
+                error="docker setup network creation failed",
+            )
+        network = self._setup_network or "host"
+        restrictions = (
+            []
+            if self.config.network_access
+            else [
+                "--cap-drop",
+                "NET_ADMIN",
+                "--cap-drop",
+                "NET_RAW",
+                "--sysctl",
+                "net.ipv6.conf.all.disable_ipv6=1",
+            ]
+        )
+        run = await docker_checked(
             "run",
             "--detach",
             "--network",
-            "host",
+            network,
+            *restrictions,
             *limits,
             "--workdir",
             self.config.workdir,
@@ -108,26 +193,215 @@ async def start(self) -> None:
             self.config.image,
             "sleep",
             "infinity",
+            error="docker run failed",
         )
-        if run.exit_code != 0:
-            raise SandboxError(f"docker run failed: {run.stderr.strip()}")
         self.info.id = run.stdout.strip()[
             :12
         ]  # `docker run -d` prints the container id
         logger.info(
-            "docker: started container %s (image=%s)",
+            "docker: started container %s (image=%s, network=%s)",
             self._container,
             self.config.image,
+            network,
         )
 
     def host_url(self, url: str) -> str:
-        # Docker Desktop (macOS/Windows) runs containers in a VM, so `--network host`
-        # doesn't reach the host's loopback; `host.docker.internal` does.
+        # Docker Desktop runs containers in a VM, so host networking does not reach
+        # the host's loopback; host.docker.internal does.
         host = urlsplit(url).hostname
         if sys.platform != "linux" and host in ("127.0.0.1", "localhost"):
             return url.replace(host, "host.docker.internal", 1)
         return url
 
+    async def apply_network_policy(self, routes: dict[str, str]) -> dict[str, str]:
+        if not self.network_policy.restricted:
+            return routes
+        if self._container is None or self._setup_network is None:
+            raise SandboxError("docker container is not running")
+
+        parsed_routes = {}
+        servers = []
+        for offset, (name, url) in enumerate(routes.items()):
+            parsed = urlsplit(url)
+            host = parsed.hostname
+            try:
+                port = parsed.port
+            except ValueError as e:
+                raise SandboxError(
+                    f"invalid required network route {url!r}: {e}"
+                ) from e
+            if (
+                parsed.scheme not in {"http", "https"}
+                or host is None
+                or parsed.username is not None
+                or parsed.password is not None
+                or re.fullmatch(r"[A-Za-z0-9.-]+", host) is None
+            ):
+                raise SandboxError(f"invalid required network route {url!r}")
+            path = parsed.path or "/"
+            if (
+                not path.isascii()
+                or (path != "/" and not path.strip("/"))
+                or any(
+                    not char.isprintable() or char.isspace() or char == "$"
+                    for char in path
+                )
+            ):
+                raise SandboxError(f"invalid required network route {url!r}")
+            authority = f"{host}:{port}" if port is not None else host
+            upstream_host = (
+                "host.docker.internal" if host in {"127.0.0.1", "localhost"} else host
+            )
+            upstream_authority = (
+                f"{upstream_host}:{port}" if port is not None else upstream_host
+            )
+            relay_port = _EGRESS_PORT + offset
+            parsed_routes[name] = parsed, relay_port
+            locations = (
+                f"location / {{ proxy_pass {parsed.scheme}://{upstream_authority}; }}"
+                if path == "/"
+                else f"""
+        location = {json.dumps(path)} {{ proxy_pass {parsed.scheme}://{upstream_authority}; }}
+        location ^~ {json.dumps(f"{path.rstrip('/')}/")} {{
+            proxy_pass {parsed.scheme}://{upstream_authority};
+        }}
+        location / {{ return 403; }}"""
+            )
+            servers.append(
+                f"""
+    server {{
+        listen {relay_port};
+        client_max_body_size 0;
+        proxy_set_header Host {authority};
+        proxy_ssl_server_name on;
+        proxy_ssl_name {host};
+        proxy_ssl_verify on;
+        proxy_ssl_trusted_certificate /etc/ssl/certs/ca-certificates.crt;
+        proxy_http_version 1.1;
+        proxy_set_header Connection "";
+        proxy_request_buffering off;
+        proxy_buffering off;
+        proxy_next_upstream off;
+        proxy_read_timeout 86400s;
+        proxy_send_timeout 86400s;
+        {locations}
+    }}
+"""
+            )
+        config = f"""
+worker_processes 1;
+pid /tmp/nginx.pid;
+error_log stderr warn;
+events {{ worker_connections 128; }}
+http {{
+    access_log off;
+    client_body_temp_path /tmp/client_body;
+    proxy_temp_path /tmp/proxy;
+    fastcgi_temp_path /tmp/fastcgi;
+    uwsgi_temp_path /tmp/uwsgi;
+    scgi_temp_path /tmp/scgi;
+{"".join(servers)}}}
+"""
+
+        self._egress = f"{self.name}-egress"
+        self._execution_network = f"{self.name}-execution"
+        await docker_checked(
+            "network",
+            "create",
+            "--internal",
+            "--opt",
+            "com.docker.network.bridge.gateway_mode_ipv4=isolated",
+            "--label",
+            f"verifiers.runtime={self.name}",
+            self._execution_network,
+            error="docker execution network creation failed",
+        )
+        await docker_checked(
+            "create",
+            "--name",
+            self._egress,
+            "--network",
+            self._setup_network,
+            "--add-host",
+            f"host.docker.internal:{self._interception_host or 'host-gateway'}",
+            "--label",
+            f"verifiers.runtime={self.name}",
+            "--user",
+            "101:101",
+            "--cap-drop",
+            "ALL",
+            "--sysctl",
+            "net.ipv4.ip_forward=0",
+            "--security-opt",
+            "no-new-privileges",
+            "--read-only",
+            "--tmpfs",
+            "/tmp:rw,noexec,nosuid,size=16m,uid=101,gid=101",
+            "--entrypoint",
+            "sh",
+            _EGRESS_IMAGE,
+            "-c",
+            'printf "%s" "$1" > /tmp/nginx.conf && '
+            'exec nginx -c /tmp/nginx.conf -g "daemon off;"',
+            "vf-egress",
+            config,
+            error="docker egress creation failed",
+        )
+        await docker_checked(
+            "network",
+            "connect",
+            "--alias",
+            _EGRESS_HOST,
+            self._execution_network,
+            self._egress,
+            error="docker egress network connection failed",
+        )
+        await docker_checked("start", self._egress, error="docker egress start failed")
+        for _ in range(20):
+            ready = await docker(
+                "exec", self._egress, "nginx", "-t", "-c", "/tmp/nginx.conf"
+            )
+            if ready.exit_code == 0:
+                break
+            await asyncio.sleep(0.05)
+        else:
+            logs = await docker("logs", self._egress)
+            raise SandboxError(
+                "docker egress did not become ready: "
+                f"{(logs.stderr or logs.stdout).strip()[-2000:]}"
+            )
+        await docker_checked(
+            "network",
+            "connect",
+            self._execution_network,
+            self._container,
+            error="docker execution network connection failed",
+        )
+        await docker_checked(
+            "network",
+            "disconnect",
+            self._setup_network,
+            self._container,
+            error="docker setup network disconnection failed",
+        )
+        logger.info(
+            "docker: applied network policy to %s via %s",
+            self._container,
+            self._egress,
+        )
+        return {
+            name: urlunsplit(
+                (
+                    "http",
+                    f"{_EGRESS_HOST}:{port}",
+                    parsed.path,
+                    parsed.query,
+                    "",
+                )
+            )
+            for name, (parsed, port) in parsed_routes.items()
+        }
+
     async def run(self, argv: list[str], env: dict[str, str]) -> ProgramResult:
         env_args = [arg for k, v in env.items() for arg in ("--env", f"{k}={v}")]
         return await docker(
@@ -139,7 +413,7 @@ async def run_background(
     ) -> None:
         env_args = [arg for k, v in env.items() for arg in ("--env", f"{k}={v}")]
         inner = f"{' '.join(shlex.quote(a) for a in argv)} > {shlex.quote(log)} 2>&1"
-        run = await docker(
+        await docker_checked(
             "exec",
             "--detach",
             *env_args,
@@ -149,9 +423,8 @@ async def run_background(
             "sh",
             "-c",
             inner,
+            error="docker exec -d failed",
         )  # detached → lives in the container until it's removed in stop()
-        if run.exit_code != 0:
-            raise SandboxError(f"docker exec -d failed: {run.stderr.strip()}")
 
     async def read(self, path: str) -> bytes:
         proc = await asyncio.create_subprocess_exec(
@@ -195,16 +468,51 @@ async def write(self, path: str, data: bytes) -> None:
             )
 
     def cleanup(self) -> None:
-        if self._container is None or self._stopped:
-            return
-        self._stopped = (
-            True  # idempotency guard; keep `_container` so the name still shows
+        resources = (
+            self._container,
+            self._egress,
+            self._setup_network,
+            self._execution_network,
         )
-        logger.debug("docker: removing container %s", self._container)
-        with contextlib.suppress(Exception):
-            subprocess.run(
-                ["docker", "rm", "--force", self._container],
-                stdout=subprocess.DEVNULL,
-                stderr=subprocess.DEVNULL,
-                timeout=30,
+        if self._stopped or not any(resources):
+            return
+        for attr in ("_egress", "_container"):
+            container = getattr(self, attr)
+            if container is None:
+                continue
+            logger.debug("docker: removing container %s", container)
+            try:
+                result = subprocess.run(
+                    ["docker", "rm", "--force", container],
+                    stdout=subprocess.DEVNULL,
+                    stderr=subprocess.DEVNULL,
+                    timeout=30,
+                )
+            except Exception:
+                continue
+            if result.returncode == 0:
+                setattr(self, attr, None)
+        for attr in ("_execution_network", "_setup_network"):
+            network = getattr(self, attr)
+            if network is None:
+                continue
+            logger.debug("docker: removing network %s", network)
+            try:
+                result = subprocess.run(
+                    ["docker", "network", "rm", network],
+                    stdout=subprocess.DEVNULL,
+                    stderr=subprocess.DEVNULL,
+                    timeout=30,
+                )
+            except Exception:
+                continue
+            if result.returncode == 0:
+                setattr(self, attr, None)
+        self._stopped = not any(
+            (
+                self._container,
+                self._egress,
+                self._setup_network,
+                self._execution_network,
             )
+        )
diff --git a/verifiers/v1/runtimes/network.py b/verifiers/v1/runtimes/network.py
new file mode 100644
index 000000000..58d214f1d
--- /dev/null
+++ b/verifiers/v1/runtimes/network.py
@@ -0,0 +1,23 @@
+"""Internal execution-network policy shared by runtime backends."""
+
+from pydantic_config import BaseConfig
+
+
+class NetworkPolicy(BaseConfig):
+    """Agent egress rules applied at the trusted setup-to-execution boundary.
+
+    Framework routes such as model and MCP endpoints are passed separately to the
+    runtime. They remain reachable regardless of these rules and may be rewritten
+    through backend-specific relays.
+
+    This stays internal until network target syntax and backend support are defined.
+    """
+
+    allow: list[str] | None = None
+    """Allowed network targets. None leaves arbitrary egress enabled; [] denies it."""
+    block: list[str] = []
+    """Blocked arbitrary-egress targets. These take precedence over allow rules."""
+
+    @property
+    def restricted(self) -> bool:
+        return self.allow is not None or bool(self.block)