mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-09-09 13:39:26 +00:00
feat(sandbox): add controlled egress with approvals (#5152)
* feat(sandbox): add controlled egress approvals * Apply batched suggestions from code review * fix(sandbox): harden restricted network policy * fix(sandbox): harden denied egress handling * fix(sandbox): isolate network proxy sidecar * chore: retry sandbox image smoke * fix(sandbox): close remaining network policy gaps * fix(sandbox): harden relay token rejection * fix(sandbox): fence incompatible policy replacement * fix(sandbox): replace containers across network modes * fix(sandbox): close remaining lifecycle gaps --------- Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
This commit is contained in:
Aari
GitHub
parent
eebe909ebd
commit
0f7d8709d3
78
.github/workflows/sandbox-network-proxy-image.yaml
vendored
Normal file
78
.github/workflows/sandbox-network-proxy-image.yaml
vendored
Normal file
@ -0,0 +1,78 @@
|
||||
name: Publish Sandbox Network Proxy
|
||||
|
||||
on:
|
||||
pull_request:
|
||||
paths:
|
||||
- "docker/sandbox-network-proxy/**"
|
||||
- "backend/packages/harness/deerflow/community/aio_sandbox/network_proxy.py"
|
||||
- ".github/workflows/sandbox-network-proxy-image.yaml"
|
||||
push:
|
||||
branches:
|
||||
- main
|
||||
paths:
|
||||
- "docker/sandbox-network-proxy/**"
|
||||
- "backend/packages/harness/deerflow/community/aio_sandbox/network_proxy.py"
|
||||
- ".github/workflows/sandbox-network-proxy-image.yaml"
|
||||
env:
|
||||
REGISTRY: ghcr.io
|
||||
IMAGE_NAME: bytedance/deer-flow-sandbox-network-proxy
|
||||
|
||||
jobs:
|
||||
validate:
|
||||
if: github.event_name == 'pull_request'
|
||||
runs-on: ubuntu-latest
|
||||
permissions:
|
||||
contents: read
|
||||
steps:
|
||||
- name: Checkout repository
|
||||
uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3
|
||||
- name: Set up QEMU
|
||||
uses: docker/setup-qemu-action@29109295f81e9208d7d86ff1c6c12d2833863392 # v3.6.0
|
||||
- name: Set up Docker Buildx
|
||||
uses: docker/setup-buildx-action@e468171a9de216ec08956ac3ada2f0791b6bd435 # v3.11.1
|
||||
- name: Build image
|
||||
uses: docker/build-push-action@263435318d21b8e681c14492fe198d362a7d2c83 # v6.18.0
|
||||
with:
|
||||
context: .
|
||||
file: docker/sandbox-network-proxy/Dockerfile
|
||||
platforms: linux/amd64,linux/arm64
|
||||
push: false
|
||||
|
||||
publish:
|
||||
if: github.event_name == 'push' && github.ref == 'refs/heads/main' && github.repository == 'bytedance/deer-flow'
|
||||
runs-on: ubuntu-latest
|
||||
permissions:
|
||||
contents: read
|
||||
packages: write
|
||||
attestations: write
|
||||
id-token: write
|
||||
steps:
|
||||
- name: Checkout repository
|
||||
uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3
|
||||
- name: Set up QEMU
|
||||
uses: docker/setup-qemu-action@29109295f81e9208d7d86ff1c6c12d2833863392 # v3.6.0
|
||||
- name: Set up Docker Buildx
|
||||
uses: docker/setup-buildx-action@e468171a9de216ec08956ac3ada2f0791b6bd435 # v3.11.1
|
||||
- name: Log in to the container registry
|
||||
uses: docker/login-action@74a5d142397b4f367a81961eba4e8cd7edddf772 # v3.4.0
|
||||
with:
|
||||
registry: ${{ env.REGISTRY }}
|
||||
username: ${{ github.actor }}
|
||||
password: ${{ secrets.GITHUB_TOKEN }}
|
||||
- name: Build and publish image
|
||||
id: build
|
||||
uses: docker/build-push-action@263435318d21b8e681c14492fe198d362a7d2c83 # v6.18.0
|
||||
with:
|
||||
context: .
|
||||
file: docker/sandbox-network-proxy/Dockerfile
|
||||
platforms: linux/amd64,linux/arm64
|
||||
push: true
|
||||
tags: |
|
||||
${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}:latest
|
||||
${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}:sha-${{ github.sha }}
|
||||
- name: Generate artifact attestation
|
||||
uses: actions/attest-build-provenance@e8998f949152b193b063cb0ec769d69d929409be # v2.4.0
|
||||
with:
|
||||
subject-name: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}
|
||||
subject-digest: ${{ steps.build.outputs.digest }}
|
||||
push-to-registry: true
|
||||
13
README.md
13
README.md
@ -1292,6 +1292,19 @@ downloads at the Gateway boundary.
|
||||
|
||||
With `AioSandboxProvider`, shell execution runs inside isolated containers. With `LocalSandboxProvider`, file tools still map to per-thread directories on the host, but host `bash` is disabled by default because it is not a secure isolation boundary. Re-enable host bash only for fully trusted local workflows. Host bash commands have a wall-clock timeout, and long-lived processes should be started in the background with output redirected to a workspace log. On Windows, Git Bash/MSYS argument-conversion exclusions are limited to safe non-root virtual path prefixes, so host-native CLI launchers retain their normal MSYS compatibility.
|
||||
|
||||
Docker AIO sandboxes default to their existing open egress behavior for
|
||||
compatibility. Operators can set `sandbox.network.mode` to `isolated` or
|
||||
`allowlist`; allowlist mode supports operator-defined domains and an interactive
|
||||
Human Input card for temporary or sandbox-lifetime HTTP(S) approval. Private,
|
||||
loopback, link-local, multicast, and cloud metadata addresses remain
|
||||
unapprovable. Denied hostnames are rejected before DNS resolution, and
|
||||
scheduled or otherwise non-interactive runs auto-deny without opening a card.
|
||||
The trusted sidecar uses a dedicated per-sandbox egress bridge rather than
|
||||
Docker's shared default bridge, and rejects ambiguous HTTP field names before
|
||||
forwarding. See
|
||||
[Sandbox configuration](backend/docs/CONFIGURATION.md#sandbox-network-policy)
|
||||
for runtime requirements and the complete policy model.
|
||||
|
||||
`AioSandboxProvider` normally detects thread-data mounts from its backend: local
|
||||
containers use the mounted gateway directories, while remote/provisioner
|
||||
sandboxes receive uploaded files through explicit synchronization. Deployments
|
||||
|
||||
@ -664,6 +664,91 @@ sandbox:
|
||||
|
||||
When you configure `sandbox.mounts`, DeerFlow exposes those `container_path` values in the agent prompt so the agent can discover and operate on mounted directories directly instead of assuming everything must live under `/mnt/user-data`.
|
||||
|
||||
#### Sandbox network policy
|
||||
|
||||
Local Docker AIO sandboxes can opt into an outbound policy:
|
||||
|
||||
```yaml
|
||||
sandbox:
|
||||
use: deerflow.community.aio_sandbox:AioSandboxProvider
|
||||
network:
|
||||
mode: allowlist
|
||||
allow_domains:
|
||||
- pypi.org
|
||||
- files.pythonhosted.org
|
||||
- registry.npmjs.org
|
||||
approval: prompt
|
||||
temporary_grant_ttl: 300
|
||||
```
|
||||
|
||||
`open` (the compatibility default) keeps normal Docker egress. `isolated`
|
||||
places each sandbox on a per-sandbox internal bridge and denies all outbound
|
||||
traffic. `allowlist` uses the same bridge and a trusted sidecar that supports
|
||||
HTTP and HTTPS CONNECT only. Exact domains and leading wildcards such as
|
||||
`*.pythonhosted.org` are accepted; URLs, ports, and a catch-all `*` are
|
||||
rejected. Traffic that ignores proxy environment variables still has no route
|
||||
out of the internal bridge. DeerFlow also sets the upstream AIO image's
|
||||
`PROXY_SERVER`/`PROXY_EXCLUDE` variables so its Chromium service uses the same
|
||||
policy sidecar; standard upper/lower-case HTTP, HTTPS, and ALL proxy variables
|
||||
cover shell and package-manager clients.
|
||||
|
||||
The sidecar is dual-homed between the sandbox's internal bridge and a separate
|
||||
per-sandbox egress bridge with inter-container communication disabled. It is
|
||||
never attached to Docker's shared default `bridge`, so unrelated containers
|
||||
cannot reach its container address directly. Its published sandbox-API relay
|
||||
also requires a cryptographically random per-sandbox token, so containers on
|
||||
other bridge networks cannot use Docker's host-port mapping to bypass that
|
||||
separation. Only the sidecar is attached to the egress bridge; outbound traffic
|
||||
still goes through Docker NAT.
|
||||
|
||||
Plain HTTP connections carry exactly one fully framed, policy-checked request;
|
||||
the sidecar closes the upstream connection afterward so a pipelined request
|
||||
cannot reuse the first request's decision. HTTP field names are parsed once,
|
||||
must use the RFC token grammar with the colon immediately following the name,
|
||||
and malformed or ambiguous fields are rejected before any policy lookup or
|
||||
forwarding. HTTPS CONNECT validates both the
|
||||
CONNECT authority and the TLS ClientHello SNI without intercepting TLS. Because
|
||||
the encrypted HTTP `Host`/`:authority` remains invisible, a deliberately
|
||||
malicious client may still reach another virtual host co-located on an allowed
|
||||
endpoint if that server accepts a mismatched inner authority. Deployments that
|
||||
require strict origin-level HTTPS isolation should use `isolated` mode or an
|
||||
operator-managed TLS-inspecting egress gateway.
|
||||
|
||||
With `approval: prompt`, a denied public domain becomes a Human Input card with
|
||||
**Deny**, **Allow temporarily**, and **Allow for this sandbox** choices. DeerFlow
|
||||
does not replay the failed command after approval because it may already have
|
||||
performed local side effects; the agent must retry it explicitly. Non-interactive
|
||||
runs auto-deny without opening a card or waiting for input. The sidecar rejects
|
||||
hostnames that policy does not allow before DNS resolution; allowed hostnames
|
||||
that resolve to loopback, private, link-local, multicast, IPv6 ULA/site-local,
|
||||
or cloud metadata destinations are rejected and can never be approved. Raw
|
||||
TCP/UDP, Git-over-SSH, and other non-HTTP protocols remain unavailable in
|
||||
restricted modes.
|
||||
|
||||
Restricted modes currently require the local Docker backend and Docker Engine
|
||||
28 or newer. They fail closed on Apple Container, provisioner mode, and older
|
||||
engines. DeerFlow applies Engine 28's isolated bridge gateway mode to both IPv4
|
||||
and IPv6 so the sandbox cannot reach services bound to either host-side bridge
|
||||
address. The sandbox, sidecar, internal network, and egress network carry a
|
||||
digest of the effective policy, proxy source, and image reference; startup and
|
||||
reconciliation destroy and recreate a persisted resource set when that identity
|
||||
or its required network properties no longer match. Docker sandboxes in every
|
||||
mode also carry stable identity and mode labels. After a Gateway restart,
|
||||
changing between `open` and a restricted mode is therefore reported as an
|
||||
incompatible persisted sandbox and replaced only after the normal ownership,
|
||||
orphan-grace, and teardown fences. Unlabelled open containers from older
|
||||
DeerFlow versions are recognized when they use the configured image and retain
|
||||
their published API port. Docker Desktop is detected
|
||||
from the daemon, not the Gateway process, so Docker-outside-of-Docker deployments
|
||||
handle its synthetic DNS range correctly. The policy sidecar publishes only its
|
||||
fixed sandbox-API relay back to the Gateway; the sandbox API itself is not
|
||||
published. DeerFlow generates a separate relay token for each sandbox, requires
|
||||
it on every new relay connection, reconstructs it from Docker during discovery,
|
||||
and injects it only into Gateway control-plane clients. The token is excluded
|
||||
from `SandboxInfo` serialization, representations, and command logs.
|
||||
Mirror or digest-pin `network.proxy_image` in production environments that
|
||||
require supply-chain pinning.
|
||||
|
||||
#### Sandbox container network exposure and hardening
|
||||
|
||||
The sandbox HTTP API (`/v1/shell/*` and friends) has no authentication: anyone who can reach a published sandbox port can execute arbitrary commands in that sandbox. For bare-metal Docker sandbox runs that use localhost, DeerFlow binds the sandbox port to `127.0.0.1` so it is not exposed on other host interfaces. For Docker-outside-of-Docker deployments that connect through `host.docker.internal`, the port is bound to the address that hostname actually resolves to — the daemon's `host-gateway-ip` mapping (customizable, possibly IPv6) — so the published port and the address the gateway connects to always match, and the port is no longer published on external network interfaces (previously it was bound to `0.0.0.0`). If resolution fails, the Docker default bridge gateway (via `docker network inspect bridge`, falling back to `172.17.0.1`) is used as a best-effort bind and a warning is logged. Set `DEER_FLOW_SANDBOX_BIND_HOST` explicitly if your deployment needs a different bind address; setting it to `0.0.0.0` restores the legacy broad bind, which re-exposes the unauthenticated exec API on every interface and should be paired with an external firewall.
|
||||
@ -682,9 +767,9 @@ A custom image that is already fully initialized as a non-root user and needs no
|
||||
| `DEER_FLOW_SANDBOX_CPUS` | `2` | `--cpus` limit per sandbox container. `0`/`none` disables the limit. |
|
||||
| `DEER_FLOW_SANDBOX_PIDS_LIMIT` | `512` | `--pids-limit` per sandbox container (fork-bomb guard). `0`/`none` disables the limit. |
|
||||
| `DEER_FLOW_SANDBOX_CONTAINER_USER` | unset (image default) | Passed through as `--user` (e.g. `1000:1000`). The default AIO image's user is upstream-controlled, so DeerFlow does not force one; set this only if you know your image's runtime user. |
|
||||
| `DEER_FLOW_SANDBOX_NETWORK` | unset (daemon default network) | Passed through as `--network`. Point it at a dedicated, egress-controlled Docker network so sandbox egress can be filtered by that network's policy; by default sandbox code can otherwise reach internal networks and cloud metadata endpoints directly. `host`, `container:<name>`, and `none` are rejected at startup (including through Docker's extended `name=<network>` syntax, whose effective target is validated): Docker drops `-p/--publish` in host mode (and shares the namespace for `container:<name>`), which would void the hardened port bind and re-expose the unauthenticated exec API; `none` leaves the container loopback-only, so the published sandbox API port cannot receive traffic and every acquisition would time out. |
|
||||
| `DEER_FLOW_SANDBOX_NETWORK` | unset (daemon default network) | Legacy `open`-mode escape hatch passed through as `--network`. Prefer `sandbox.network` for managed isolation. `host`, `container:<name>`, and `none` are rejected at startup. Restricted modes ignore this variable and use their own per-sandbox internal network. |
|
||||
|
||||
These hardening flags are Docker-only; Apple Container (`container` runtime) keeps its previous, unhardened invocation.
|
||||
These hardening flags are Docker-only; Apple Container (`container` runtime) keeps its previous, unhardened invocation and therefore supports only `network.mode: open`. On macOS, an `open` Gateway normally prefers Apple Container, but it keeps using Docker while the configured sandbox prefix has managed Docker sandboxes so startup reconciliation can safely replace resources left by a restricted-mode deployment before the runtime changes.
|
||||
|
||||
Sandbox control-plane HTTP calls to loopback/private IPs, single-label cluster
|
||||
hosts, and Docker/Podman internal hostnames bypass `HTTP_PROXY`/`HTTPS_PROXY`
|
||||
|
||||
@ -89,6 +89,18 @@ CI.
|
||||
|
||||
**Gateway Conformance Tests** (`TestGatewayConformance`): Validate that every dict-returning client method conforms to the corresponding Gateway Pydantic response model. Each test parses the client output through the Gateway model — if Gateway adds a required field that the client doesn't provide, Pydantic raises `ValidationError` and CI catches the drift. Covers: `ModelsListResponse`, `ModelResponse`, `SkillsListResponse`, `SkillResponse`, `SkillInstallResponse`, `McpConfigResponse`, `UploadResponse`, `MemoryConfigResponse`, `MemoryStatusResponse`.
|
||||
|
||||
### AIO Sandbox Network Policy
|
||||
|
||||
Restricted AIO keeps sandboxes internal; a per-sandbox, ICC-disabled sidecar
|
||||
handles egress and its token-authenticated API relay. Parse headers strictly;
|
||||
reject policy-denied names before DNS and try all validated answers. Claim the
|
||||
oldest unsurfaced denial; subagent/non-interactive runs drain and deny. Approvals
|
||||
never replay tools; policy labels fence reuse. CONNECT/SNI cannot inspect
|
||||
encrypted authority. Discovery and enumeration are read-only, including on a
|
||||
policy or network-mode mismatch; only the provider may replace it after the
|
||||
orphan grace, local teardown reservation, and cross-instance teardown lease.
|
||||
Destroy the sandbox, sidecar, and both networks together.
|
||||
|
||||
### E2B Mount Uploads
|
||||
|
||||
The E2B provider uploads host mounts during sandbox creation. It passes binary file objects to the E2B SDK.
|
||||
|
||||
@ -59,25 +59,35 @@ class AioSandbox(Sandbox):
|
||||
#: the next — recorded bash evidence cannot prove a clean environment.
|
||||
persistent_shell_sessions = True
|
||||
|
||||
def __init__(self, id: str, base_url: str, home_dir: str | None = None):
|
||||
def __init__(
|
||||
self,
|
||||
id: str,
|
||||
base_url: str,
|
||||
home_dir: str | None = None,
|
||||
request_headers: dict[str, str] | None = None,
|
||||
):
|
||||
"""Initialize the AIO sandbox.
|
||||
|
||||
Args:
|
||||
id: Unique identifier for this sandbox instance.
|
||||
base_url: URL of the sandbox API (e.g., http://localhost:8080).
|
||||
home_dir: Home directory inside the sandbox. If None, will be fetched from the sandbox.
|
||||
request_headers: Trusted control-plane headers required by a local
|
||||
relay. These are never injected into sandbox commands.
|
||||
"""
|
||||
super().__init__(id)
|
||||
self._base_url = base_url
|
||||
client_kwargs = {
|
||||
"base_url": base_url,
|
||||
"timeout": 600,
|
||||
}
|
||||
if request_headers:
|
||||
client_kwargs["headers"] = dict(request_headers)
|
||||
if sandbox_http_trust_env(base_url):
|
||||
self._client = AioSandboxClient(base_url=base_url, timeout=600)
|
||||
self._client = AioSandboxClient(**client_kwargs)
|
||||
else:
|
||||
direct_client = httpx.Client(timeout=600, follow_redirects=True, trust_env=False)
|
||||
self._client = AioSandboxClient(
|
||||
base_url=base_url,
|
||||
timeout=600,
|
||||
httpx_client=direct_client,
|
||||
)
|
||||
self._client = AioSandboxClient(**client_kwargs, httpx_client=direct_client)
|
||||
self._home_dir = home_dir
|
||||
self._lock = threading.Lock()
|
||||
self._scope_registry_lock = threading.Lock()
|
||||
|
||||
@ -86,6 +86,14 @@ class SandboxBeingDestroyedError(RuntimeError):
|
||||
self.sandbox_id = sandbox_id
|
||||
|
||||
|
||||
class SandboxPolicyReplacementDeferredError(RuntimeError):
|
||||
"""An incompatible sandbox cannot be replaced until it is a true orphan."""
|
||||
|
||||
def __init__(self, sandbox_id: str) -> None:
|
||||
super().__init__(f"sandbox {sandbox_id} has an incompatible provisioning policy; replacement is deferred until its current owner releases it")
|
||||
self.sandbox_id = sandbox_id
|
||||
|
||||
|
||||
class SandboxIdentityCollisionError(RuntimeError):
|
||||
"""A deterministic ID is already tracked for a different user/thread."""
|
||||
|
||||
@ -248,6 +256,8 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
|
||||
"""
|
||||
provisioner_url = self._config.get("provisioner_url")
|
||||
if provisioner_url:
|
||||
if self.sandbox_network_mode() != "open":
|
||||
raise RuntimeError("sandbox.network restricted modes are currently supported only by the local Docker AIO backend")
|
||||
logger.info(f"Using remote sandbox backend with provisioner at {provisioner_url}")
|
||||
api_key = self._config.get("provisioner_api_key", "")
|
||||
return RemoteSandboxBackend(provisioner_url=provisioner_url, api_key=api_key)
|
||||
@ -259,6 +269,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
|
||||
container_prefix=self._config["container_prefix"],
|
||||
config_mounts=self._config["mounts"],
|
||||
environment=self._config["environment"],
|
||||
network_config=self._config["network"],
|
||||
)
|
||||
|
||||
# ── Configuration ────────────────────────────────────────────────────
|
||||
@ -287,6 +298,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
|
||||
"mounts": sandbox_config.mounts or [],
|
||||
"thread_data_mounts": getattr(sandbox_config, "thread_data_mounts", None),
|
||||
"environment": self._resolve_env_vars(sandbox_config.environment or {}),
|
||||
"network": sandbox_config.network.model_dump(),
|
||||
"ownership": getattr(sandbox_config, "ownership", None),
|
||||
# A redis stream bridge means the deployment is multi-instance, which
|
||||
# is what the ownership store must default to. Read the same source
|
||||
@ -300,6 +312,27 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
|
||||
),
|
||||
}
|
||||
|
||||
def sandbox_network_mode(self) -> str:
|
||||
return str(self._config.get("network", {}).get("mode", "open"))
|
||||
|
||||
def sandbox_network_temporary_grant_ttl(self) -> int:
|
||||
return int(self._config.get("network", {}).get("temporary_grant_ttl", 300))
|
||||
|
||||
def consume_network_policy_events(self, sandbox_id: str) -> list[dict[str, object]]:
|
||||
if not isinstance(self._backend, LocalContainerBackend):
|
||||
return []
|
||||
return self._backend.consume_network_policy_events(sandbox_id)
|
||||
|
||||
def deny_pending_network_policy_events(self, sandbox_id: str) -> bool:
|
||||
if not isinstance(self._backend, LocalContainerBackend):
|
||||
return True
|
||||
return self._backend.deny_pending_network_policy_events(sandbox_id)
|
||||
|
||||
def decide_network_policy_request(self, sandbox_id: str, request_id: str, decision: str) -> bool:
|
||||
if not isinstance(self._backend, LocalContainerBackend):
|
||||
return False
|
||||
return self._backend.decide_network_policy_request(sandbox_id, request_id, decision)
|
||||
|
||||
@staticmethod
|
||||
def _resolve_env_vars(env_config: dict[str, str]) -> dict[str, str]:
|
||||
"""Resolve environment variable references (values starting with $)."""
|
||||
@ -628,6 +661,42 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
|
||||
first_seen = self._unowned_since.setdefault(sandbox_id, now)
|
||||
return now - first_seen >= compute_lease_ttl(self._ownership_config)
|
||||
|
||||
def _replace_incompatible_sandbox(self, info: SandboxInfo, now: float) -> bool:
|
||||
"""Destroy an incompatible sandbox only after both ownership fences.
|
||||
|
||||
Backends report policy mismatches through ``SandboxInfo`` without
|
||||
mutating Docker state. That is essential during rolling upgrades: an
|
||||
older Gateway may still be serving the container under a live lease.
|
||||
Replacement is therefore an orphan-reconciliation operation, not a
|
||||
discovery side effect. The recovery grace protects against ownership
|
||||
store state loss, the teardown lease excludes peers, and the local
|
||||
reservation excludes this provider's own acquire/reaper paths.
|
||||
"""
|
||||
if not info.requires_replacement:
|
||||
return False
|
||||
if not self._adoptable_after_grace(info.sandbox_id, now):
|
||||
return False
|
||||
if not self._reserve_local_teardown(
|
||||
info.sandbox_id,
|
||||
lambda: info.sandbox_id not in self._sandboxes and info.sandbox_id not in self._sandbox_infos and info.sandbox_id not in self._warm_pool,
|
||||
):
|
||||
return False
|
||||
|
||||
try:
|
||||
if not self._claim_ownership(info.sandbox_id, for_destroy=True):
|
||||
return False
|
||||
try:
|
||||
with self._held_teardown_lease(info.sandbox_id):
|
||||
self._backend.destroy(info)
|
||||
except Exception as e:
|
||||
logger.warning("Failed to replace sandbox %s with incompatible provisioning policy: %s", info.sandbox_id, e)
|
||||
return False
|
||||
self._unowned_since.pop(info.sandbox_id, None)
|
||||
logger.info("Removed orphaned sandbox %s with incompatible provisioning policy", info.sandbox_id)
|
||||
return True
|
||||
finally:
|
||||
self._finish_local_teardown(info.sandbox_id)
|
||||
|
||||
def _reconcile_orphans(self) -> None:
|
||||
"""Reconcile orphaned containers left by previous process lifecycles.
|
||||
|
||||
@ -662,11 +731,23 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
|
||||
|
||||
current_time = time.time()
|
||||
adopted = 0
|
||||
replaced = 0
|
||||
skipped_live = 0
|
||||
deferred = 0
|
||||
|
||||
for info in running:
|
||||
age = current_time - info.created_at if info.created_at > 0 else float("inf")
|
||||
if info.requires_replacement:
|
||||
if self._replace_incompatible_sandbox(info, current_time):
|
||||
replaced += 1
|
||||
else:
|
||||
deferred += 1
|
||||
logger.debug(
|
||||
"Deferring replacement of container %s during reconciliation: owned, locally tracked, or not yet past the recovery grace",
|
||||
info.sandbox_id,
|
||||
)
|
||||
continue
|
||||
|
||||
if not self._adoptable_after_grace(info.sandbox_id, current_time):
|
||||
deferred += 1
|
||||
logger.debug("Deferring container %s during reconciliation: owned, or not yet past the recovery grace", info.sandbox_id)
|
||||
@ -708,8 +789,9 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
|
||||
logger.info(f"Adopted container {info.sandbox_id} into warm pool (age: {age:.0f}s)")
|
||||
|
||||
logger.info(
|
||||
"Startup reconciliation complete: %s adopted into warm pool, %s skipped (live peer ownership), %s deferred (owned or within recovery grace), %s total found",
|
||||
"Startup reconciliation complete: %s adopted into warm pool, %s incompatible orphan(s) replaced, %s skipped (live peer ownership), %s deferred (owned, locally tracked, or within recovery grace), %s total found",
|
||||
adopted,
|
||||
replaced,
|
||||
skipped_live,
|
||||
deferred,
|
||||
len(running),
|
||||
@ -1575,7 +1657,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
|
||||
return None
|
||||
self._warm_pool_identity.pop(sandbox_id, None)
|
||||
info, _ = warm_item
|
||||
sandbox = AioSandbox(id=sandbox_id, base_url=info.sandbox_url)
|
||||
sandbox = AioSandbox(id=sandbox_id, base_url=info.sandbox_url, request_headers=info.request_headers)
|
||||
self._sandboxes[sandbox_id] = sandbox
|
||||
self._sandbox_infos[sandbox_id] = info
|
||||
self._active_sandbox_identity[sandbox_id] = key
|
||||
@ -1607,6 +1689,8 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
|
||||
prevent. The window is a peer's in-flight container stop, so the
|
||||
thread's next turn discovers nothing and cold-starts cleanly.
|
||||
"""
|
||||
if info.requires_replacement:
|
||||
raise SandboxPolicyReplacementDeferredError(info.sandbox_id)
|
||||
key = self._thread_key(thread_id, user_id)
|
||||
with self._lock:
|
||||
if self._being_torn_down_locally(info.sandbox_id):
|
||||
@ -1618,7 +1702,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
|
||||
self._assert_active_identity_available_locked(info.sandbox_id, key)
|
||||
self._assert_warm_identity_available_locked(info.sandbox_id, key)
|
||||
|
||||
sandbox = AioSandbox(id=info.sandbox_id, base_url=info.sandbox_url)
|
||||
sandbox = AioSandbox(id=info.sandbox_id, base_url=info.sandbox_url, request_headers=info.request_headers)
|
||||
# Ownership first, so a failure cannot leave a tracked-but-unowned sandbox.
|
||||
# There is no container to roll back (we did not create it), but the
|
||||
# host-side HTTP client constructed above is ours and must not leak —
|
||||
@ -1664,7 +1748,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
|
||||
|
||||
def _register_created_sandbox(self, thread_id: str | None, sandbox_id: str, info: SandboxInfo, *, user_id: str | None = None) -> str:
|
||||
"""Track a newly-created sandbox in the active maps."""
|
||||
sandbox = AioSandbox(id=sandbox_id, base_url=info.sandbox_url)
|
||||
sandbox = AioSandbox(id=sandbox_id, base_url=info.sandbox_url, request_headers=info.request_headers)
|
||||
key = (
|
||||
self._thread_key(
|
||||
thread_id,
|
||||
@ -1706,21 +1790,14 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
|
||||
SandboxIdentityCollisionError,
|
||||
):
|
||||
logger.error(
|
||||
"Could not register new sandbox %s; destroying it rather than leaking an untracked container",
|
||||
"Could not register new sandbox %s; attempting ownership-fenced cleanup",
|
||||
sandbox_id,
|
||||
)
|
||||
try:
|
||||
sandbox.close()
|
||||
except Exception as e:
|
||||
logger.warning(f"Error closing sandbox {sandbox_id} during ownership rollback: {e}")
|
||||
try:
|
||||
self._backend.destroy(info)
|
||||
except Exception as e:
|
||||
logger.error(
|
||||
"Failed to destroy sandbox %s after registration failure: %s",
|
||||
sandbox_id,
|
||||
e,
|
||||
)
|
||||
self._destroy_unready_sandbox(sandbox_id, info)
|
||||
raise
|
||||
|
||||
logger.info(f"Created sandbox {sandbox_id} for thread {thread_id} at {info.sandbox_url}")
|
||||
@ -2020,7 +2097,11 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
|
||||
# Backend discovery: another process may have created the container.
|
||||
discovered = self._backend.discover(sandbox_id)
|
||||
if discovered is not None:
|
||||
return self._register_discovered_sandbox(thread_id, discovered, user_id=effective_user_id)
|
||||
if discovered.requires_replacement:
|
||||
if not self._replace_incompatible_sandbox(discovered, time.time()):
|
||||
raise SandboxPolicyReplacementDeferredError(sandbox_id)
|
||||
else:
|
||||
return self._register_discovered_sandbox(thread_id, discovered, user_id=effective_user_id)
|
||||
|
||||
return self._create_sandbox(thread_id, sandbox_id, user_id=effective_user_id)
|
||||
finally:
|
||||
@ -2049,10 +2130,19 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
|
||||
# Docker and perform a health check; keep it off the event loop.
|
||||
discovered = await asyncio.to_thread(self._backend.discover, sandbox_id)
|
||||
if discovered is not None:
|
||||
# Registration publishes ownership, which is blocking store IO
|
||||
# (filesystem or network depending on the backend) — same reason
|
||||
# every other step in this coroutine is offloaded.
|
||||
return await asyncio.to_thread(self._register_discovered_sandbox, thread_id, discovered, user_id=effective_user_id)
|
||||
if discovered.requires_replacement:
|
||||
replaced = await asyncio.to_thread(
|
||||
self._replace_incompatible_sandbox,
|
||||
discovered,
|
||||
time.time(),
|
||||
)
|
||||
if not replaced:
|
||||
raise SandboxPolicyReplacementDeferredError(sandbox_id)
|
||||
else:
|
||||
# Registration publishes ownership, which is blocking store
|
||||
# IO (filesystem or network depending on the backend) — same
|
||||
# reason every other step in this coroutine is offloaded.
|
||||
return await asyncio.to_thread(self._register_discovered_sandbox, thread_id, discovered, user_id=effective_user_id)
|
||||
|
||||
return await self._create_sandbox_async(thread_id, sandbox_id, user_id=effective_user_id)
|
||||
finally:
|
||||
@ -2160,7 +2250,8 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
|
||||
)
|
||||
|
||||
# Wait for sandbox to be ready
|
||||
if not wait_for_sandbox_ready(info.sandbox_url, timeout=SANDBOX_LOCAL_PROVIDER_READY_TIMEOUT):
|
||||
readiness_kwargs = {"headers": info.request_headers} if info.request_headers else {}
|
||||
if not wait_for_sandbox_ready(info.sandbox_url, timeout=SANDBOX_LOCAL_PROVIDER_READY_TIMEOUT, **readiness_kwargs):
|
||||
# The container is running but unowned: ownership is published by
|
||||
# ``_register_created_sandbox`` after this gate. Claim the teardown
|
||||
# lease before stopping it so a peer cannot adopt the not-yet-ready
|
||||
@ -2206,7 +2297,12 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
|
||||
)
|
||||
|
||||
# Wait for sandbox to be ready without blocking the event loop.
|
||||
if not await wait_for_sandbox_ready_async(info.sandbox_url, timeout=SANDBOX_LOCAL_PROVIDER_READY_TIMEOUT):
|
||||
readiness_kwargs = {"headers": info.request_headers} if info.request_headers else {}
|
||||
if not await wait_for_sandbox_ready_async(
|
||||
info.sandbox_url,
|
||||
timeout=SANDBOX_LOCAL_PROVIDER_READY_TIMEOUT,
|
||||
**readiness_kwargs,
|
||||
):
|
||||
# The container is running but unowned: ownership is published by
|
||||
# ``_register_created_sandbox`` after this gate. Claim the teardown
|
||||
# lease before stopping it so a peer cannot adopt the not-yet-ready
|
||||
|
||||
@ -7,6 +7,7 @@ import ipaddress
|
||||
import logging
|
||||
import time
|
||||
from abc import ABC, abstractmethod
|
||||
from collections.abc import Mapping
|
||||
from urllib.parse import urlparse
|
||||
|
||||
import httpx
|
||||
@ -48,7 +49,12 @@ def sandbox_http_trust_env(sandbox_url: str) -> bool:
|
||||
SANDBOX_LOCAL_PROVIDER_READY_TIMEOUT = 60
|
||||
|
||||
|
||||
def wait_for_sandbox_ready(sandbox_url: str, timeout: int = 30) -> bool:
|
||||
def wait_for_sandbox_ready(
|
||||
sandbox_url: str,
|
||||
timeout: int = 30,
|
||||
*,
|
||||
headers: Mapping[str, str] | None = None,
|
||||
) -> bool:
|
||||
"""Poll sandbox health endpoint until ready or timeout.
|
||||
|
||||
Args:
|
||||
@ -61,6 +67,8 @@ def wait_for_sandbox_ready(sandbox_url: str, timeout: int = 30) -> bool:
|
||||
start_time = time.time()
|
||||
with requests.Session() as session:
|
||||
session.trust_env = sandbox_http_trust_env(sandbox_url)
|
||||
if headers:
|
||||
session.headers.update(headers)
|
||||
while time.time() - start_time < timeout:
|
||||
try:
|
||||
response = session.get(f"{sandbox_url}/v1/sandbox", timeout=5)
|
||||
@ -72,7 +80,13 @@ def wait_for_sandbox_ready(sandbox_url: str, timeout: int = 30) -> bool:
|
||||
return False
|
||||
|
||||
|
||||
async def wait_for_sandbox_ready_async(sandbox_url: str, timeout: int = 30, poll_interval: float = 1.0) -> bool:
|
||||
async def wait_for_sandbox_ready_async(
|
||||
sandbox_url: str,
|
||||
timeout: int = 30,
|
||||
poll_interval: float = 1.0,
|
||||
*,
|
||||
headers: Mapping[str, str] | None = None,
|
||||
) -> bool:
|
||||
"""Async variant of sandbox readiness polling.
|
||||
|
||||
Use this from async runtime paths so sandbox startup waits do not block the
|
||||
@ -82,7 +96,13 @@ async def wait_for_sandbox_ready_async(sandbox_url: str, timeout: int = 30, poll
|
||||
loop = asyncio.get_running_loop()
|
||||
deadline = loop.time() + timeout
|
||||
|
||||
async with httpx.AsyncClient(timeout=5, trust_env=sandbox_http_trust_env(sandbox_url)) as client:
|
||||
client_kwargs: dict[str, object] = {
|
||||
"timeout": 5,
|
||||
"trust_env": sandbox_http_trust_env(sandbox_url),
|
||||
}
|
||||
if headers:
|
||||
client_kwargs["headers"] = dict(headers)
|
||||
async with httpx.AsyncClient(**client_kwargs) as client:
|
||||
while True:
|
||||
remaining = deadline - loop.time()
|
||||
if remaining <= 0:
|
||||
@ -175,7 +195,10 @@ class SandboxBackend(ABC):
|
||||
sandbox_id: The deterministic sandbox ID to look for.
|
||||
|
||||
Returns:
|
||||
SandboxInfo if found and healthy, None otherwise.
|
||||
SandboxInfo if found, including ``requires_replacement=True`` when
|
||||
the backend can identify an incompatible persisted provisioning
|
||||
policy without safely adopting it. Enumeration must not destroy
|
||||
resources; the provider owns replacement fencing. None otherwise.
|
||||
"""
|
||||
...
|
||||
|
||||
@ -189,6 +212,9 @@ class SandboxBackend(ABC):
|
||||
The default implementation returns an empty list, which is correct
|
||||
for backends that don't manage local containers (e.g., RemoteSandboxBackend
|
||||
delegates lifecycle to the provisioner which handles its own cleanup).
|
||||
Enumeration must be read-only. Backends report resources that need
|
||||
replacement through ``SandboxInfo.requires_replacement`` so the
|
||||
provider can apply ownership and local teardown fencing first.
|
||||
|
||||
Returns:
|
||||
A list of SandboxInfo for all currently running sandboxes.
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@ -0,0 +1,654 @@
|
||||
"""Trusted HTTP(S) policy proxy used by restricted local AIO sandboxes.
|
||||
|
||||
The module is copied into a small sidecar container and executed as a script.
|
||||
It deliberately supports only HTTP absolute-form requests and HTTPS CONNECT;
|
||||
all other protocols remain unavailable on the sandbox's internal-only network.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import asyncio
|
||||
import contextlib
|
||||
import hmac
|
||||
import ipaddress
|
||||
import json
|
||||
import os
|
||||
import socket
|
||||
import sqlite3
|
||||
import time
|
||||
import uuid
|
||||
from pathlib import Path
|
||||
from urllib.parse import urlsplit
|
||||
|
||||
MAX_HEADER_BYTES = 65_536
|
||||
POLICY_DB = Path(os.environ.get("DEERFLOW_POLICY_DB", "/tmp/deerflow-network-policy.sqlite3"))
|
||||
RELAY_AUTH_HEADER = "X-DeerFlow-Relay-Token"
|
||||
RELAY_TOKEN_ENV = "DEERFLOW_RELAY_TOKEN"
|
||||
|
||||
|
||||
class _InvalidHttpRequest(ValueError):
|
||||
pass
|
||||
|
||||
|
||||
class _InvalidHttpBody(_InvalidHttpRequest):
|
||||
pass
|
||||
|
||||
|
||||
class _InvalidHttpHeader(_InvalidHttpRequest):
|
||||
pass
|
||||
|
||||
|
||||
_HTTP_TOKEN_CHARS = frozenset("!#$%&'*+-.^_`|~0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz")
|
||||
|
||||
|
||||
def _connect_db() -> sqlite3.Connection:
|
||||
db = sqlite3.connect(POLICY_DB, timeout=5)
|
||||
db.execute("PRAGMA journal_mode=WAL")
|
||||
db.execute(
|
||||
"""
|
||||
CREATE TABLE IF NOT EXISTS events (
|
||||
request_id TEXT PRIMARY KEY,
|
||||
host TEXT NOT NULL,
|
||||
port INTEGER NOT NULL,
|
||||
method TEXT NOT NULL,
|
||||
created_at REAL NOT NULL,
|
||||
surfaced INTEGER NOT NULL DEFAULT 0,
|
||||
decision TEXT
|
||||
)
|
||||
"""
|
||||
)
|
||||
db.execute(
|
||||
"""
|
||||
CREATE TABLE IF NOT EXISTS grants (
|
||||
host TEXT NOT NULL,
|
||||
port INTEGER NOT NULL,
|
||||
expires_at REAL,
|
||||
PRIMARY KEY (host, port)
|
||||
)
|
||||
"""
|
||||
)
|
||||
db.commit()
|
||||
return db
|
||||
|
||||
|
||||
def normalize_host(raw: str) -> str | None:
|
||||
host = raw.strip().lower().rstrip(".")
|
||||
if not host or len(host) > 253 or any(char in host for char in "/\\\x00\r\n"):
|
||||
return None
|
||||
try:
|
||||
return host.encode("idna").decode("ascii")
|
||||
except UnicodeError:
|
||||
return None
|
||||
|
||||
|
||||
def domain_matches(host: str, rule: str) -> bool:
|
||||
if rule.startswith("*."):
|
||||
suffix = rule[2:]
|
||||
return host.endswith("." + suffix) and host != suffix
|
||||
return host == rule
|
||||
|
||||
|
||||
def address_is_public(address: str, *, allow_synthetic_dns: bool = False) -> bool:
|
||||
try:
|
||||
parsed = ipaddress.ip_address(address.split("%", 1)[0])
|
||||
except ValueError:
|
||||
return False
|
||||
if parsed.is_global and not parsed.is_multicast and not getattr(parsed, "is_site_local", False):
|
||||
return True
|
||||
# Docker Desktop's DNS inhibition layer maps public names into the
|
||||
# benchmarking-only 198.18.0.0/15 range. Accept it only when the host
|
||||
# backend explicitly identifies Desktop; native Linux keeps rejecting it.
|
||||
return allow_synthetic_dns and parsed in ipaddress.ip_network("198.18.0.0/15")
|
||||
|
||||
|
||||
def _static_rules() -> tuple[str, ...]:
|
||||
try:
|
||||
raw = json.loads(os.environ.get("DEERFLOW_ALLOW_DOMAINS_JSON", "[]"))
|
||||
except json.JSONDecodeError:
|
||||
return ()
|
||||
return tuple(value for value in raw if isinstance(value, str))
|
||||
|
||||
|
||||
def _policy_mode() -> str:
|
||||
value = os.environ.get("DEERFLOW_NETWORK_MODE", "isolated")
|
||||
return value if value in {"isolated", "allowlist"} else "isolated"
|
||||
|
||||
|
||||
def _is_granted(host: str, port: int, now: float) -> bool:
|
||||
with _connect_db() as db:
|
||||
row = db.execute("SELECT expires_at FROM grants WHERE host = ? AND port = ?", (host, port)).fetchone()
|
||||
if row is None:
|
||||
return False
|
||||
expires_at = row[0]
|
||||
if expires_at is not None and float(expires_at) <= now:
|
||||
db.execute("DELETE FROM grants WHERE host = ? AND port = ?", (host, port))
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def policy_allows(host: str, port: int, now: float | None = None) -> bool:
|
||||
now = time.time() if now is None else now
|
||||
if _policy_mode() != "allowlist":
|
||||
return False
|
||||
if any(domain_matches(host, rule) for rule in _static_rules()):
|
||||
return True
|
||||
return _is_granted(host, port, now)
|
||||
|
||||
|
||||
def record_denial(host: str, port: int, method: str) -> str:
|
||||
now = time.time()
|
||||
with _connect_db() as db:
|
||||
# Serialize the read-before-insert deduplication so simultaneous proxy
|
||||
# requests cannot create multiple approval cards for one destination.
|
||||
db.execute("BEGIN IMMEDIATE")
|
||||
recent = db.execute(
|
||||
"SELECT request_id FROM events WHERE host = ? AND port = ? AND decision IS NULL AND created_at >= ? ORDER BY created_at DESC LIMIT 1",
|
||||
(host, port, now - 30),
|
||||
).fetchone()
|
||||
if recent is not None:
|
||||
return str(recent[0])
|
||||
request_id = str(uuid.uuid4())
|
||||
db.execute(
|
||||
"INSERT INTO events(request_id, host, port, method, created_at) VALUES (?, ?, ?, ?, ?)",
|
||||
(request_id, host, port, method, now),
|
||||
)
|
||||
return request_id
|
||||
|
||||
|
||||
def pending_events() -> list[dict[str, object]]:
|
||||
with _connect_db() as db:
|
||||
# Tool execution timestamps cannot reliably delimit proxy events: a
|
||||
# background process may emit a denial after its launching tool returns,
|
||||
# and subagent/non-interactive paths must drain events without prompting.
|
||||
# Claim the oldest unsurfaced event atomically, independent of age.
|
||||
db.execute("BEGIN IMMEDIATE")
|
||||
row = db.execute(
|
||||
"SELECT request_id, host, port, method, created_at FROM events WHERE surfaced = 0 AND decision IS NULL ORDER BY created_at LIMIT 1",
|
||||
).fetchone()
|
||||
if row is not None:
|
||||
db.execute("UPDATE events SET surfaced = 1 WHERE request_id = ?", (row[0],))
|
||||
# One Human Input card makes one destination decision. Close any
|
||||
# sibling denials from the same tool call so a later retry records
|
||||
# a fresh event instead of leaving an invisible surfaced request.
|
||||
db.execute("UPDATE events SET decision = 'superseded' WHERE surfaced = 0 AND decision IS NULL")
|
||||
if row is None:
|
||||
return []
|
||||
return [{"request_id": row[0], "host": row[1], "port": row[2], "method": row[3], "created_at": row[4]}]
|
||||
|
||||
|
||||
def deny_pending_events() -> int:
|
||||
"""Atomically deny every event that has not been surfaced to a user."""
|
||||
with _connect_db() as db:
|
||||
db.execute("BEGIN IMMEDIATE")
|
||||
result = db.execute("UPDATE events SET surfaced = 1, decision = 'deny' WHERE surfaced = 0 AND decision IS NULL")
|
||||
return max(result.rowcount, 0)
|
||||
|
||||
|
||||
def decide(request_id: str, decision: str, ttl: int) -> bool:
|
||||
with _connect_db() as db:
|
||||
row = db.execute("SELECT host, port, decision FROM events WHERE request_id = ?", (request_id,)).fetchone()
|
||||
if row is None:
|
||||
return False
|
||||
host, port, existing = str(row[0]), int(row[1]), row[2]
|
||||
if existing is not None:
|
||||
return str(existing) == decision
|
||||
if decision == "allow_temporary":
|
||||
db.execute(
|
||||
"INSERT INTO grants(host, port, expires_at) VALUES (?, ?, ?) ON CONFLICT(host, port) DO UPDATE SET expires_at = excluded.expires_at",
|
||||
(host, port, time.time() + ttl),
|
||||
)
|
||||
elif decision == "allow_sandbox":
|
||||
db.execute(
|
||||
"INSERT INTO grants(host, port, expires_at) VALUES (?, ?, NULL) ON CONFLICT(host, port) DO UPDATE SET expires_at = NULL",
|
||||
(host, port),
|
||||
)
|
||||
elif decision != "deny":
|
||||
return False
|
||||
db.execute("UPDATE events SET decision = ? WHERE request_id = ?", (decision, request_id))
|
||||
return True
|
||||
|
||||
|
||||
async def resolve_public(host: str, port: int) -> tuple[tuple[int, tuple], ...] | None:
|
||||
loop = asyncio.get_running_loop()
|
||||
try:
|
||||
infos = await loop.getaddrinfo(host, port, type=socket.SOCK_STREAM)
|
||||
except OSError:
|
||||
return None
|
||||
allow_synthetic_dns = os.environ.get("DEERFLOW_ALLOW_SYNTHETIC_DNS") == "1"
|
||||
public = [(family, sockaddr) for family, _socktype, _proto, _canonname, sockaddr in infos if address_is_public(str(sockaddr[0]), allow_synthetic_dns=allow_synthetic_dns)]
|
||||
if len(public) != len(infos) or not public:
|
||||
return None
|
||||
return tuple(public)
|
||||
|
||||
|
||||
async def _relay(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
|
||||
try:
|
||||
while chunk := await reader.read(65_536):
|
||||
writer.write(chunk)
|
||||
await writer.drain()
|
||||
except (ConnectionError, asyncio.CancelledError):
|
||||
pass
|
||||
finally:
|
||||
with contextlib.suppress(Exception):
|
||||
writer.close()
|
||||
await writer.wait_closed()
|
||||
|
||||
|
||||
async def _open_public(resolved: tuple[tuple[int, tuple], ...], port: int) -> tuple[asyncio.StreamReader, asyncio.StreamWriter] | None:
|
||||
"""Try every pre-validated DNS answer within one shared deadline."""
|
||||
loop = asyncio.get_running_loop()
|
||||
deadline = loop.time() + 15
|
||||
for index, (family, sockaddr) in enumerate(resolved):
|
||||
remaining = deadline - loop.time()
|
||||
if remaining <= 0:
|
||||
break
|
||||
# Reserve an equal share of the remaining budget for every later
|
||||
# address so a black-holed first answer cannot consume the full
|
||||
# deadline and strand otherwise reachable candidates.
|
||||
attempt_timeout = remaining / (len(resolved) - index)
|
||||
try:
|
||||
return await asyncio.wait_for(
|
||||
asyncio.open_connection(sockaddr[0], port, family=family),
|
||||
timeout=attempt_timeout,
|
||||
)
|
||||
except (OSError, TimeoutError):
|
||||
continue
|
||||
return None
|
||||
|
||||
|
||||
async def _reject(writer: asyncio.StreamWriter, status: str, body: str) -> None:
|
||||
encoded = body.encode("utf-8")
|
||||
writer.write(f"HTTP/1.1 {status}\r\nContent-Type: text/plain; charset=utf-8\r\nContent-Length: {len(encoded)}\r\nConnection: close\r\n\r\n".encode() + encoded)
|
||||
await writer.drain()
|
||||
writer.close()
|
||||
await writer.wait_closed()
|
||||
|
||||
|
||||
def _parse_http_header_fields(header_lines: list[str]) -> list[tuple[str, str, str]]:
|
||||
"""Parse one header block using a single strict field-name grammar."""
|
||||
fields: list[tuple[str, str, str]] = []
|
||||
for line in header_lines:
|
||||
if not line:
|
||||
continue
|
||||
if line.startswith((" ", "\t")) or ":" not in line:
|
||||
raise _InvalidHttpHeader("Obsolete or malformed HTTP headers are not supported")
|
||||
raw_name, raw_value = line.split(":", 1)
|
||||
if not raw_name or any(char not in _HTTP_TOKEN_CHARS for char in raw_name):
|
||||
raise _InvalidHttpHeader("HTTP field names must use token characters followed immediately by a colon")
|
||||
if any((ord(char) < 0x20 and char != "\t") or ord(char) == 0x7F for char in raw_value):
|
||||
raise _InvalidHttpHeader("HTTP field values cannot contain control characters")
|
||||
fields.append((raw_name, raw_name.lower(), raw_value.strip(" \t")))
|
||||
return fields
|
||||
|
||||
|
||||
def _http_request_body_framing(header_fields: list[tuple[str, str, str]]) -> tuple[str, int]:
|
||||
"""Return the strictly validated framing for one HTTP proxy request."""
|
||||
content_lengths = [value for _raw_name, name, value in header_fields if name == "content-length"]
|
||||
transfer_encodings = [value.lower() for _raw_name, name, value in header_fields if name == "transfer-encoding"]
|
||||
if content_lengths and transfer_encodings:
|
||||
raise _InvalidHttpBody("Content-Length and Transfer-Encoding cannot be combined")
|
||||
if transfer_encodings:
|
||||
if transfer_encodings != ["chunked"]:
|
||||
raise _InvalidHttpBody("Only a single chunked Transfer-Encoding is supported")
|
||||
return ("chunked", 0)
|
||||
if not content_lengths:
|
||||
return ("fixed", 0)
|
||||
if len(content_lengths) != 1 or not content_lengths[0].isdigit():
|
||||
raise _InvalidHttpBody("Content-Length must be one non-negative decimal integer")
|
||||
return ("fixed", int(content_lengths[0]))
|
||||
|
||||
|
||||
def _build_http_outbound_header(method: str, path: str, version: str, header_fields: list[tuple[str, str, str]]) -> bytes:
|
||||
"""Strip proxy/hop headers and force the one-request upstream connection closed."""
|
||||
connection_tokens = {token.strip().lower() for _raw_name, name, value in header_fields if name == "connection" for token in value.split(",") if token.strip()}
|
||||
if any(any(char not in _HTTP_TOKEN_CHARS for char in token) for token in connection_tokens):
|
||||
raise _InvalidHttpHeader("Connection header options must use HTTP token characters")
|
||||
if connection_tokens & {"host", "content-length", "transfer-encoding"}:
|
||||
raise _InvalidHttpBody("Connection cannot remove request framing headers")
|
||||
hop_headers = {
|
||||
"connection",
|
||||
"keep-alive",
|
||||
"proxy-authenticate",
|
||||
"proxy-authorization",
|
||||
"proxy-connection",
|
||||
"te",
|
||||
"trailer",
|
||||
"upgrade",
|
||||
*connection_tokens,
|
||||
}
|
||||
kept_headers = [f"{raw_name}: {value}" for raw_name, name, value in header_fields if name not in hop_headers]
|
||||
kept_headers.append("Connection: close")
|
||||
return f"{method} {path} {version}\r\n".encode("latin-1") + "\r\n".join(kept_headers).encode("latin-1") + b"\r\n\r\n"
|
||||
|
||||
|
||||
async def _copy_exact_request_bytes(reader: asyncio.StreamReader, writer: asyncio.StreamWriter, length: int) -> None:
|
||||
remaining = length
|
||||
while remaining:
|
||||
chunk = await asyncio.wait_for(reader.readexactly(min(remaining, 65_536)), timeout=15)
|
||||
writer.write(chunk)
|
||||
await writer.drain()
|
||||
remaining -= len(chunk)
|
||||
|
||||
|
||||
async def _copy_chunked_request_body(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
|
||||
"""Relay exactly one strictly framed chunked body, including its trailers."""
|
||||
while True:
|
||||
line = await asyncio.wait_for(reader.readuntil(b"\r\n"), timeout=15)
|
||||
if len(line) > MAX_HEADER_BYTES:
|
||||
raise _InvalidHttpBody("Chunk header is too large")
|
||||
size_token = line[:-2].split(b";", 1)[0].strip()
|
||||
if not size_token or any(value not in b"0123456789abcdefABCDEF" for value in size_token):
|
||||
raise _InvalidHttpBody("Invalid chunk size")
|
||||
size = int(size_token, 16)
|
||||
writer.write(line)
|
||||
await writer.drain()
|
||||
if size:
|
||||
await _copy_exact_request_bytes(reader, writer, size)
|
||||
terminator = await asyncio.wait_for(reader.readexactly(2), timeout=15)
|
||||
if terminator != b"\r\n":
|
||||
raise _InvalidHttpBody("Invalid chunk terminator")
|
||||
writer.write(terminator)
|
||||
await writer.drain()
|
||||
continue
|
||||
|
||||
trailer_bytes = len(line)
|
||||
while True:
|
||||
trailer = await asyncio.wait_for(reader.readuntil(b"\r\n"), timeout=15)
|
||||
trailer_bytes += len(trailer)
|
||||
if trailer_bytes > MAX_HEADER_BYTES:
|
||||
raise _InvalidHttpBody("Chunk trailers are too large")
|
||||
if trailer != b"\r\n" and (trailer.startswith((b" ", b"\t")) or b":" not in trailer):
|
||||
raise _InvalidHttpBody("Invalid chunk trailer")
|
||||
writer.write(trailer)
|
||||
await writer.drain()
|
||||
if trailer == b"\r\n":
|
||||
return
|
||||
|
||||
|
||||
def _parse_authority(authority: str, default_port: int) -> tuple[str, int] | None:
|
||||
parsed = urlsplit("//" + authority)
|
||||
try:
|
||||
port = parsed.port or default_port
|
||||
except ValueError:
|
||||
return None
|
||||
host = normalize_host(parsed.hostname or "")
|
||||
if host is None:
|
||||
return None
|
||||
return host, port
|
||||
|
||||
|
||||
async def _read_tls_client_hello(reader: asyncio.StreamReader) -> tuple[str, bytes] | None:
|
||||
"""Read one TLS ClientHello and return its normalized SNI plus wire bytes."""
|
||||
wire = bytearray()
|
||||
handshake = bytearray()
|
||||
expected_handshake_size: int | None = None
|
||||
while len(wire) <= MAX_HEADER_BYTES:
|
||||
try:
|
||||
record_header = await asyncio.wait_for(reader.readexactly(5), timeout=15)
|
||||
record_size = int.from_bytes(record_header[3:5], "big")
|
||||
if record_header[0] != 22 or record_size <= 0 or len(wire) + 5 + record_size > MAX_HEADER_BYTES:
|
||||
return None
|
||||
record = await asyncio.wait_for(reader.readexactly(record_size), timeout=15)
|
||||
except (asyncio.IncompleteReadError, TimeoutError):
|
||||
return None
|
||||
wire.extend(record_header)
|
||||
wire.extend(record)
|
||||
handshake.extend(record)
|
||||
if expected_handshake_size is None and len(handshake) >= 4:
|
||||
if handshake[0] != 1:
|
||||
return None
|
||||
expected_handshake_size = 4 + int.from_bytes(handshake[1:4], "big")
|
||||
if expected_handshake_size > MAX_HEADER_BYTES:
|
||||
return None
|
||||
if expected_handshake_size is not None and len(handshake) >= expected_handshake_size:
|
||||
break
|
||||
if expected_handshake_size is None or len(handshake) < expected_handshake_size:
|
||||
return None
|
||||
|
||||
hello = memoryview(handshake)[4:expected_handshake_size]
|
||||
try:
|
||||
offset = 2 + 32
|
||||
offset += 1 + hello[offset]
|
||||
cipher_size = int.from_bytes(hello[offset : offset + 2], "big")
|
||||
offset += 2 + cipher_size
|
||||
offset += 1 + hello[offset]
|
||||
extensions_size = int.from_bytes(hello[offset : offset + 2], "big")
|
||||
offset += 2
|
||||
extensions_end = offset + extensions_size
|
||||
if extensions_end > len(hello):
|
||||
return None
|
||||
sni: str | None = None
|
||||
while offset + 4 <= extensions_end:
|
||||
extension_type = int.from_bytes(hello[offset : offset + 2], "big")
|
||||
extension_size = int.from_bytes(hello[offset + 2 : offset + 4], "big")
|
||||
offset += 4
|
||||
extension = hello[offset : offset + extension_size]
|
||||
offset += extension_size
|
||||
if offset > extensions_end:
|
||||
return None
|
||||
# Reject encrypted ClientHello: its hidden SNI cannot be compared
|
||||
# with the approved CONNECT host without TLS interception.
|
||||
if extension_type in {0xFE0D, 0xFFCE}:
|
||||
return None
|
||||
if extension_type != 0 or len(extension) < 5:
|
||||
continue
|
||||
names_size = int.from_bytes(extension[0:2], "big")
|
||||
cursor = 2
|
||||
while cursor + 3 <= 2 + names_size and cursor + 3 <= len(extension):
|
||||
name_type = extension[cursor]
|
||||
name_size = int.from_bytes(extension[cursor + 1 : cursor + 3], "big")
|
||||
cursor += 3
|
||||
if cursor + name_size > len(extension):
|
||||
return None
|
||||
if name_type == 0:
|
||||
sni = normalize_host(bytes(extension[cursor : cursor + name_size]).decode("ascii"))
|
||||
break
|
||||
cursor += name_size
|
||||
return (sni, bytes(wire)) if sni is not None else None
|
||||
except (IndexError, UnicodeDecodeError, ValueError):
|
||||
return None
|
||||
|
||||
|
||||
async def handle_proxy(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
|
||||
try:
|
||||
header = await asyncio.wait_for(reader.readuntil(b"\r\n\r\n"), timeout=15)
|
||||
except (asyncio.IncompleteReadError, asyncio.LimitOverrunError, TimeoutError):
|
||||
await _reject(writer, "400 Bad Request", "Invalid proxy request")
|
||||
return
|
||||
if len(header) > MAX_HEADER_BYTES:
|
||||
await _reject(writer, "431 Request Header Fields Too Large", "Proxy request headers are too large")
|
||||
return
|
||||
try:
|
||||
request_line, *header_lines = header.decode("latin-1").split("\r\n")
|
||||
method, target, version = request_line.split(" ", 2)
|
||||
except ValueError:
|
||||
await _reject(writer, "400 Bad Request", "Invalid proxy request line")
|
||||
return
|
||||
if version not in {"HTTP/1.0", "HTTP/1.1"}:
|
||||
await _reject(writer, "400 Bad Request", "Unsupported HTTP version")
|
||||
return
|
||||
try:
|
||||
header_fields = _parse_http_header_fields(header_lines)
|
||||
except _InvalidHttpHeader as exc:
|
||||
await _reject(writer, "400 Bad Request", str(exc))
|
||||
return
|
||||
|
||||
method = method.upper()
|
||||
if method == "CONNECT":
|
||||
parsed = _parse_authority(target, 443)
|
||||
if parsed is None or parsed[1] != 443:
|
||||
await _reject(writer, "403 Forbidden", "Only HTTPS CONNECT on port 443 is supported")
|
||||
return
|
||||
host, port = parsed
|
||||
outbound_header = None
|
||||
else:
|
||||
parsed_url = urlsplit(target)
|
||||
if parsed_url.scheme.lower() != "http" or not parsed_url.hostname:
|
||||
await _reject(writer, "403 Forbidden", "Only HTTP absolute-form requests and HTTPS CONNECT are supported")
|
||||
return
|
||||
try:
|
||||
port = parsed_url.port or 80
|
||||
except ValueError:
|
||||
await _reject(writer, "400 Bad Request", "Invalid destination port")
|
||||
return
|
||||
host = normalize_host(parsed_url.hostname)
|
||||
if host is None or port != 80:
|
||||
await _reject(writer, "403 Forbidden", "Only HTTP on port 80 is supported")
|
||||
return
|
||||
path = parsed_url.path or "/"
|
||||
if parsed_url.query:
|
||||
path += "?" + parsed_url.query
|
||||
host_headers = [value for _raw_name, name, value in header_fields if name == "host"]
|
||||
host_authority = _parse_authority(host_headers[0], 80) if len(host_headers) == 1 else None
|
||||
if host_authority != (host, port):
|
||||
await _reject(writer, "400 Bad Request", "HTTP Host header must match the approved proxy destination")
|
||||
return
|
||||
if any(name == "expect" for _raw_name, name, _value in header_fields):
|
||||
await _reject(writer, "417 Expectation Failed", "Expect is not supported by the sandbox network proxy")
|
||||
return
|
||||
try:
|
||||
body_mode, body_length = _http_request_body_framing(header_fields)
|
||||
outbound_header = _build_http_outbound_header(method, path, version, header_fields)
|
||||
except _InvalidHttpRequest as exc:
|
||||
await _reject(writer, "400 Bad Request", str(exc))
|
||||
return
|
||||
|
||||
try:
|
||||
ipaddress.ip_address(host)
|
||||
except ValueError:
|
||||
pass
|
||||
else:
|
||||
await _reject(writer, "403 Forbidden", "IP-literal destinations are not allowed by sandbox network policy")
|
||||
return
|
||||
if not policy_allows(host, port):
|
||||
if os.environ.get("DEERFLOW_RECORD_DENIALS") == "1":
|
||||
request_id = record_denial(host, port, method)
|
||||
detail = f" (request {request_id})"
|
||||
else:
|
||||
detail = ""
|
||||
await _reject(writer, "403 Forbidden", f"Sandbox network policy denied {host}:{port}{detail}")
|
||||
return
|
||||
resolved = await resolve_public(host, port)
|
||||
if resolved is None:
|
||||
await _reject(writer, "403 Forbidden", "Destination did not resolve exclusively to public addresses")
|
||||
return
|
||||
upstream = await _open_public(resolved, port)
|
||||
if upstream is None:
|
||||
await _reject(writer, "403 Forbidden", "Destination did not resolve exclusively to public addresses")
|
||||
return
|
||||
upstream_reader, upstream_writer = upstream
|
||||
if method == "CONNECT":
|
||||
writer.write(b"HTTP/1.1 200 Connection Established\r\n\r\n")
|
||||
await writer.drain()
|
||||
client_hello = await _read_tls_client_hello(reader)
|
||||
if client_hello is None or client_hello[0] != host:
|
||||
upstream_writer.close()
|
||||
await upstream_writer.wait_closed()
|
||||
writer.close()
|
||||
await writer.wait_closed()
|
||||
return
|
||||
upstream_writer.write(client_hello[1])
|
||||
await upstream_writer.drain()
|
||||
await asyncio.gather(_relay(reader, upstream_writer), _relay(upstream_reader, writer))
|
||||
return
|
||||
|
||||
upstream_writer.write(outbound_header or b"")
|
||||
await upstream_writer.drain()
|
||||
try:
|
||||
if body_mode == "chunked":
|
||||
await _copy_chunked_request_body(reader, upstream_writer)
|
||||
else:
|
||||
await _copy_exact_request_bytes(reader, upstream_writer, body_length)
|
||||
except (asyncio.IncompleteReadError, asyncio.LimitOverrunError, TimeoutError, _InvalidHttpBody) as exc:
|
||||
upstream_writer.close()
|
||||
await upstream_writer.wait_closed()
|
||||
await _reject(writer, "400 Bad Request", str(exc) or "Invalid HTTP request body")
|
||||
return
|
||||
if upstream_writer.can_write_eof():
|
||||
upstream_writer.write_eof()
|
||||
await upstream_writer.drain()
|
||||
try:
|
||||
# One request per client connection is intentional. Relaying arbitrary
|
||||
# remaining client bytes would let a pipelined request bypass the next
|
||||
# destination/Host policy check.
|
||||
await _relay(upstream_reader, writer)
|
||||
finally:
|
||||
with contextlib.suppress(Exception):
|
||||
upstream_writer.close()
|
||||
await upstream_writer.wait_closed()
|
||||
|
||||
|
||||
async def handle_relay(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
|
||||
try:
|
||||
header = await asyncio.wait_for(reader.readuntil(b"\r\n\r\n"), timeout=15)
|
||||
except (asyncio.IncompleteReadError, asyncio.LimitOverrunError, TimeoutError):
|
||||
await _reject(writer, "400 Bad Request", "Invalid sandbox relay request")
|
||||
return
|
||||
if len(header) > MAX_HEADER_BYTES:
|
||||
await _reject(writer, "431 Request Header Fields Too Large", "Sandbox relay request headers are too large")
|
||||
return
|
||||
try:
|
||||
_request_line, *header_lines = header.decode("latin-1").split("\r\n")
|
||||
header_fields = _parse_http_header_fields(header_lines)
|
||||
except (UnicodeDecodeError, _InvalidHttpHeader):
|
||||
await _reject(writer, "400 Bad Request", "Invalid sandbox relay request headers")
|
||||
return
|
||||
expected_token = os.environ.get(RELAY_TOKEN_ENV, "")
|
||||
presented_tokens = [value for _raw_name, name, value in header_fields if name == RELAY_AUTH_HEADER.lower()]
|
||||
presented_token = presented_tokens[0].encode("latin-1") if len(presented_tokens) == 1 else b""
|
||||
if not expected_token or len(presented_tokens) != 1 or not hmac.compare_digest(presented_token, expected_token.encode()):
|
||||
await _reject(writer, "403 Forbidden", "Sandbox relay authentication failed")
|
||||
return
|
||||
|
||||
target = os.environ.get("DEERFLOW_SANDBOX_TARGET", "")
|
||||
parsed = _parse_authority(target, 8080)
|
||||
if parsed is None:
|
||||
await _reject(writer, "502 Bad Gateway", "Sandbox relay target is invalid")
|
||||
return
|
||||
try:
|
||||
upstream_reader, upstream_writer = await asyncio.wait_for(asyncio.open_connection(parsed[0], parsed[1]), timeout=5)
|
||||
except (OSError, TimeoutError):
|
||||
await _reject(writer, "502 Bad Gateway", "Sandbox is not ready")
|
||||
return
|
||||
upstream_writer.write(header)
|
||||
await upstream_writer.drain()
|
||||
await asyncio.gather(_relay(reader, upstream_writer), _relay(upstream_reader, writer))
|
||||
|
||||
|
||||
async def serve() -> None:
|
||||
_connect_db().close()
|
||||
proxy = await asyncio.start_server(handle_proxy, "0.0.0.0", 3128, limit=MAX_HEADER_BYTES)
|
||||
relay = await asyncio.start_server(handle_relay, "0.0.0.0", 8080, limit=MAX_HEADER_BYTES)
|
||||
async with proxy, relay:
|
||||
await asyncio.gather(proxy.serve_forever(), relay.serve_forever())
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser()
|
||||
subparsers = parser.add_subparsers(dest="command")
|
||||
subparsers.add_parser("serve")
|
||||
subparsers.add_parser("pending")
|
||||
subparsers.add_parser("deny-pending")
|
||||
decide_parser = subparsers.add_parser("decide")
|
||||
decide_parser.add_argument("request_id")
|
||||
decide_parser.add_argument("decision", choices=("deny", "allow_temporary", "allow_sandbox"))
|
||||
decide_parser.add_argument("--ttl", type=int, default=300)
|
||||
args = parser.parse_args()
|
||||
if args.command == "serve":
|
||||
asyncio.run(serve())
|
||||
return 0
|
||||
if args.command == "pending":
|
||||
print(json.dumps(pending_events(), separators=(",", ":")))
|
||||
return 0
|
||||
if args.command == "deny-pending":
|
||||
print(deny_pending_events())
|
||||
return 0
|
||||
if args.command == "decide":
|
||||
return 0 if decide(args.request_id, args.decision, args.ttl) else 2
|
||||
parser.error("a command is required")
|
||||
return 2
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@ -20,6 +20,16 @@ class SandboxInfo:
|
||||
container_name: str | None = None # Only for local container backend
|
||||
container_id: str | None = None # Only for local container backend
|
||||
created_at: float = field(default_factory=time.time)
|
||||
# Ephemeral control-plane credentials reconstructed from local Docker
|
||||
# discovery. Intentionally excluded from to_dict() and repr so they cannot
|
||||
# leak through metadata persistence or routine lifecycle logs.
|
||||
request_headers: dict[str, str] = field(default_factory=dict, repr=False, compare=False)
|
||||
# Discovery-only lifecycle signal. A backend may report a running sandbox
|
||||
# whose persisted provisioning policy is incompatible with this process,
|
||||
# but it must not destroy that sandbox while merely enumerating it. The
|
||||
# provider consumes this flag and performs replacement only after obtaining
|
||||
# its local teardown reservation and cross-instance teardown lease.
|
||||
requires_replacement: bool = field(default=False, repr=False, compare=False)
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
return {
|
||||
|
||||
@ -1,9 +1,74 @@
|
||||
import ipaddress
|
||||
from typing import Literal
|
||||
|
||||
from pydantic import BaseModel, ConfigDict, Field
|
||||
from pydantic import BaseModel, ConfigDict, Field, field_validator
|
||||
|
||||
SandboxOwnershipType = Literal["memory", "redis"]
|
||||
SandboxOverflowPolicy = Literal["wait", "reject", "burst"]
|
||||
SandboxNetworkMode = Literal["open", "isolated", "allowlist"]
|
||||
SandboxNetworkApproval = Literal["deny", "prompt"]
|
||||
|
||||
|
||||
class SandboxNetworkConfig(BaseModel):
|
||||
"""Outbound network policy for locally managed AIO sandboxes."""
|
||||
|
||||
mode: SandboxNetworkMode = Field(
|
||||
default="open",
|
||||
description="open keeps the current Docker networking behavior; isolated denies all egress; allowlist permits configured domains and optional runtime approval.",
|
||||
)
|
||||
allow_domains: list[str] = Field(
|
||||
default_factory=list,
|
||||
description="Exact domains or leading-wildcard domains (for example *.pythonhosted.org) allowed in allowlist mode.",
|
||||
)
|
||||
approval: SandboxNetworkApproval = Field(
|
||||
default="prompt",
|
||||
description="Whether a denied public HTTP(S) destination may ask an interactive user for a temporary or sandbox-lifetime grant.",
|
||||
)
|
||||
temporary_grant_ttl: int = Field(
|
||||
default=300,
|
||||
ge=30,
|
||||
le=3600,
|
||||
description="Lifetime in seconds for the temporary approval choice.",
|
||||
)
|
||||
proxy_image: str = Field(
|
||||
default="ghcr.io/bytedance/deer-flow-sandbox-network-proxy:latest",
|
||||
min_length=1,
|
||||
description="Managed Python runtime image used for the trusted network-policy sidecar.",
|
||||
)
|
||||
|
||||
@field_validator("allow_domains")
|
||||
@classmethod
|
||||
def _normalize_allow_domains(cls, values: list[str]) -> list[str]:
|
||||
normalized: list[str] = []
|
||||
seen: set[str] = set()
|
||||
for raw in values:
|
||||
value = raw.strip().lower().rstrip(".")
|
||||
suffix = value[2:] if value.startswith("*.") else value
|
||||
if not value or value == "*" or not suffix or "://" in value or "/" in value or ":" in value or "*" in suffix or suffix.startswith(".") or suffix.endswith("."):
|
||||
raise ValueError(f"invalid sandbox network allowlist domain: {raw!r}")
|
||||
try:
|
||||
suffix.encode("idna")
|
||||
except UnicodeError as exc:
|
||||
raise ValueError(f"invalid sandbox network allowlist domain: {raw!r}") from exc
|
||||
try:
|
||||
ipaddress.ip_address(suffix)
|
||||
except ValueError:
|
||||
pass
|
||||
else:
|
||||
raise ValueError(f"invalid sandbox network allowlist domain: {raw!r}")
|
||||
ascii_suffix = suffix.encode("idna").decode("ascii")
|
||||
labels = ascii_suffix.split(".")
|
||||
if (
|
||||
len(labels) < 2
|
||||
or len(ascii_suffix) > 253
|
||||
or any(not label or len(label) > 63 or label.startswith("-") or label.endswith("-") or any(not (char.isascii() and (char.isalnum() or char == "-")) for char in label) for label in labels)
|
||||
):
|
||||
raise ValueError(f"invalid sandbox network allowlist domain: {raw!r}")
|
||||
canonical = ("*." if value.startswith("*.") else "") + ascii_suffix
|
||||
if canonical not in seen:
|
||||
seen.add(canonical)
|
||||
normalized.append(canonical)
|
||||
return normalized
|
||||
|
||||
|
||||
class SandboxOwnershipConfig(BaseModel):
|
||||
@ -173,6 +238,10 @@ class SandboxConfig(BaseModel):
|
||||
default_factory=dict,
|
||||
description="Environment variables to inject into the sandbox container. Values starting with $ will be resolved from host environment variables.",
|
||||
)
|
||||
network: SandboxNetworkConfig = Field(
|
||||
default_factory=SandboxNetworkConfig,
|
||||
description="AioSandboxProvider outbound network isolation and approval policy.",
|
||||
)
|
||||
|
||||
bash_output_max_chars: int = Field(
|
||||
default=20000,
|
||||
|
||||
@ -49,7 +49,7 @@ if TYPE_CHECKING:
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_LEGACY_SUMMARY_MESSAGE_NAME = "summary"
|
||||
_PERSISTED_HIDDEN_HUMAN_INPUT_RESPONSE_SOURCES = frozenset({"ask_clarification"})
|
||||
_PERSISTED_HIDDEN_HUMAN_INPUT_RESPONSE_SOURCES = frozenset({"ask_clarification", "sandbox_network"})
|
||||
|
||||
|
||||
def _should_persist_human_input_message(message: BaseMessage) -> bool:
|
||||
|
||||
@ -1,16 +1,18 @@
|
||||
import asyncio
|
||||
import logging
|
||||
from collections.abc import Awaitable, Callable
|
||||
from collections.abc import Awaitable, Callable, Mapping
|
||||
from dataclasses import replace as dc_replace
|
||||
from typing import NotRequired, override
|
||||
|
||||
from langchain.agents import AgentState
|
||||
from langchain.agents.middleware import AgentMiddleware
|
||||
from langchain_core.messages import ToolMessage
|
||||
from langchain_core.messages import HumanMessage, ToolMessage
|
||||
from langgraph.graph import END
|
||||
from langgraph.prebuilt.tool_node import ToolCallRequest
|
||||
from langgraph.runtime import Runtime
|
||||
from langgraph.types import Command, Overwrite
|
||||
|
||||
from deerflow.agents.human_input import read_human_input_response
|
||||
from deerflow.agents.thread_state import SandboxStateField, ThreadDataState
|
||||
from deerflow.authz.sandbox_authz import (
|
||||
authorize_sandbox_execution,
|
||||
@ -27,9 +29,17 @@ from deerflow.sandbox.lease import (
|
||||
sandbox_lease_owner,
|
||||
)
|
||||
from deerflow.sandbox.overwrite import unwrap_sandbox
|
||||
from deerflow.sandbox.sandbox_provider import get_initialized_sandbox_provider
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
NETWORK_POLICY_HUMAN_INPUT_SOURCE = "sandbox_network"
|
||||
_NETWORK_POLICY_DECISIONS = frozenset({"deny", "allow_temporary", "allow_sandbox"})
|
||||
|
||||
|
||||
def _network_approval_is_non_interactive(context: Mapping[str, object]) -> bool:
|
||||
return bool(context.get("disable_clarification") or context.get("non_interactive"))
|
||||
|
||||
|
||||
class SandboxMiddlewareState(AgentState):
|
||||
"""Compatible with the `ThreadState` schema."""
|
||||
@ -218,6 +228,7 @@ class SandboxMiddleware(AgentMiddleware[SandboxMiddlewareState]):
|
||||
thread_id = (runtime.context or {}).get("thread_id")
|
||||
if thread_id is None:
|
||||
return super().before_agent(state, runtime)
|
||||
self._apply_network_policy_response(state, runtime)
|
||||
user_id = resolve_runtime_user_id(runtime)
|
||||
projection = self._prepare_agent_skill_projection(thread_id, user_id=user_id)
|
||||
owner_id = ensure_sandbox_lease_owner(runtime.context)
|
||||
@ -298,11 +309,46 @@ class SandboxMiddleware(AgentMiddleware[SandboxMiddlewareState]):
|
||||
runtime.context["sandbox_id"] = retained_id
|
||||
return super().before_agent(state, runtime)
|
||||
|
||||
def _apply_network_policy_response(self, state: SandboxMiddlewareState, runtime: Runtime) -> None:
|
||||
sandbox_id = self._read_sandbox_id_from_state(state)
|
||||
if sandbox_id is None:
|
||||
return
|
||||
messages = state.get("messages", [])
|
||||
response = None
|
||||
for message in reversed(messages):
|
||||
if not isinstance(message, HumanMessage):
|
||||
continue
|
||||
candidate = read_human_input_response(message.additional_kwargs)
|
||||
# A network decision is actionable only when it is the current
|
||||
# user turn. Stop at the newest HumanMessage so an older persisted
|
||||
# card response cannot be re-applied after ordinary conversation.
|
||||
if candidate is None or candidate["source"] != NETWORK_POLICY_HUMAN_INPUT_SOURCE:
|
||||
return
|
||||
response = candidate
|
||||
break
|
||||
if response is None or response["response_kind"] != "option":
|
||||
return
|
||||
decision = response["option_id"]
|
||||
if decision not in _NETWORK_POLICY_DECISIONS:
|
||||
raise SandboxRuntimeError("Invalid sandbox network approval response")
|
||||
context = runtime.context or {}
|
||||
applied = context.setdefault("sandbox_network_decisions_applied", set())
|
||||
marker = (sandbox_id, response["request_id"], decision)
|
||||
if marker in applied:
|
||||
return
|
||||
provider = get_sandbox_provider()
|
||||
if provider.sandbox_network_mode() != "allowlist":
|
||||
return
|
||||
if not provider.decide_network_policy_request(sandbox_id, response["request_id"], decision):
|
||||
raise SandboxRuntimeError("The sandbox network approval is stale or does not belong to this sandbox")
|
||||
applied.add(marker)
|
||||
|
||||
@override
|
||||
async def abefore_agent(self, state: SandboxMiddlewareState, runtime: Runtime) -> dict | None:
|
||||
thread_id = (runtime.context or {}).get("thread_id")
|
||||
if thread_id is None:
|
||||
return await super().abefore_agent(state, runtime)
|
||||
await asyncio.to_thread(self._apply_network_policy_response, state, runtime)
|
||||
user_id = resolve_runtime_user_id(runtime)
|
||||
projection = await asyncio.to_thread(
|
||||
self._prepare_agent_skill_projection,
|
||||
@ -497,12 +543,10 @@ class SandboxMiddleware(AgentMiddleware[SandboxMiddlewareState]):
|
||||
) -> ToolMessage | Command:
|
||||
prev_sandbox_id = self._read_sandbox_id_from_request(request)
|
||||
result = handler(request)
|
||||
if prev_sandbox_id is not None:
|
||||
return result
|
||||
curr_sandbox_id = self._read_sandbox_id_from_request(request)
|
||||
if curr_sandbox_id is None:
|
||||
return result
|
||||
return self._attach_sandbox_update(result, curr_sandbox_id)
|
||||
if prev_sandbox_id is None and curr_sandbox_id is not None:
|
||||
result = self._attach_sandbox_update(result, curr_sandbox_id)
|
||||
return self._maybe_request_network_approval(request, result, curr_sandbox_id or prev_sandbox_id)
|
||||
|
||||
@override
|
||||
async def awrap_tool_call(
|
||||
@ -512,9 +556,92 @@ class SandboxMiddleware(AgentMiddleware[SandboxMiddlewareState]):
|
||||
) -> ToolMessage | Command:
|
||||
prev_sandbox_id = self._read_sandbox_id_from_request(request)
|
||||
result = await handler(request)
|
||||
if prev_sandbox_id is not None:
|
||||
return result
|
||||
curr_sandbox_id = self._read_sandbox_id_from_request(request)
|
||||
if curr_sandbox_id is None:
|
||||
if prev_sandbox_id is None and curr_sandbox_id is not None:
|
||||
result = self._attach_sandbox_update(result, curr_sandbox_id)
|
||||
sandbox_id = curr_sandbox_id or prev_sandbox_id
|
||||
if sandbox_id is None:
|
||||
return result
|
||||
return self._attach_sandbox_update(result, curr_sandbox_id)
|
||||
context = getattr(request.runtime, "context", None) or {}
|
||||
provider = get_initialized_sandbox_provider()
|
||||
if provider is None:
|
||||
return result
|
||||
if provider.sandbox_network_mode() != "allowlist":
|
||||
return result
|
||||
if _network_approval_is_non_interactive(context) or context.get("is_subagent"):
|
||||
if not await provider.deny_pending_network_policy_events_async(sandbox_id):
|
||||
logger.warning("Failed to drain sandbox network policy events for non-interactive sandbox %s", sandbox_id)
|
||||
return result
|
||||
events = await provider.consume_network_policy_events_async(sandbox_id)
|
||||
return self._network_approval_result(request, result, sandbox_id, events)
|
||||
|
||||
def _maybe_request_network_approval(
|
||||
self,
|
||||
request: ToolCallRequest,
|
||||
result: ToolMessage | Command,
|
||||
sandbox_id: str | None,
|
||||
) -> ToolMessage | Command:
|
||||
if sandbox_id is None:
|
||||
return result
|
||||
context = getattr(request.runtime, "context", None) or {}
|
||||
provider = get_initialized_sandbox_provider()
|
||||
if provider is None:
|
||||
return result
|
||||
if provider.sandbox_network_mode() != "allowlist":
|
||||
return result
|
||||
if _network_approval_is_non_interactive(context) or context.get("is_subagent"):
|
||||
if not provider.deny_pending_network_policy_events(sandbox_id):
|
||||
logger.warning("Failed to drain sandbox network policy events for non-interactive sandbox %s", sandbox_id)
|
||||
return result
|
||||
events = provider.consume_network_policy_events(sandbox_id)
|
||||
return self._network_approval_result(request, result, sandbox_id, events)
|
||||
|
||||
def _network_approval_result(
|
||||
self,
|
||||
request: ToolCallRequest,
|
||||
result: ToolMessage | Command,
|
||||
sandbox_id: str,
|
||||
events: list[dict[str, object]],
|
||||
) -> ToolMessage | Command:
|
||||
if not events:
|
||||
return result
|
||||
event = events[0]
|
||||
request_id = event.get("request_id")
|
||||
host = event.get("host")
|
||||
port = event.get("port")
|
||||
if not isinstance(request_id, str) or not isinstance(host, str) or not isinstance(port, int):
|
||||
logger.warning("Ignoring malformed trusted sandbox network event: %r", event)
|
||||
return result
|
||||
tool_call_id = str(request.tool_call.get("id") or "")
|
||||
tool_name = str(request.tool_call.get("name") or "sandbox")
|
||||
ttl_seconds = get_sandbox_provider().sandbox_network_temporary_grant_ttl()
|
||||
ttl_label = f"{ttl_seconds // 60} minutes" if ttl_seconds % 60 == 0 else f"{ttl_seconds} seconds"
|
||||
message = ToolMessage(
|
||||
id=f"sandbox-network:{request_id}",
|
||||
content=(f"Sandbox network policy blocked {host}:{port}. The command was not retried. Choose whether this destination should be available, then ask the agent to retry if appropriate."),
|
||||
tool_call_id=tool_call_id,
|
||||
name=tool_name,
|
||||
artifact={
|
||||
"human_input": {
|
||||
"version": 1,
|
||||
"kind": "human_input_request",
|
||||
"source": NETWORK_POLICY_HUMAN_INPUT_SOURCE,
|
||||
"request_id": request_id,
|
||||
"tool_call_id": tool_call_id,
|
||||
"clarification_type": "risk_confirmation",
|
||||
"title": "Sandbox network access",
|
||||
"question": f"Allow this sandbox to connect to {host}:{port}?",
|
||||
"context": "Private, loopback, link-local, multicast, and cloud metadata addresses can never be approved.",
|
||||
"input_mode": "single_choice",
|
||||
"options": [
|
||||
{"id": "deny", "label": "Deny", "value": "Deny network access"},
|
||||
{"id": "allow_temporary", "label": f"Allow for {ttl_label}", "value": f"Allow network access for {ttl_label}"},
|
||||
{"id": "allow_sandbox", "label": "Allow for this sandbox", "value": "Allow network access for this sandbox"},
|
||||
],
|
||||
}
|
||||
},
|
||||
)
|
||||
update: dict = {"messages": [message], "sandbox": {"sandbox_id": sandbox_id}}
|
||||
if isinstance(result, Command) and isinstance(result.update, dict):
|
||||
update = {**result.update, **update}
|
||||
return Command(update=update, goto=END)
|
||||
|
||||
@ -96,6 +96,40 @@ class SandboxProvider(ABC):
|
||||
"""
|
||||
pass
|
||||
|
||||
def sandbox_network_mode(self) -> str:
|
||||
"""Return the provider's effective outbound network mode."""
|
||||
return "open"
|
||||
|
||||
def sandbox_network_temporary_grant_ttl(self) -> int:
|
||||
return 300
|
||||
|
||||
def consume_network_policy_events(self, sandbox_id: str) -> list[dict[str, object]]:
|
||||
"""Claim the oldest unsurfaced trusted-proxy event for a sandbox.
|
||||
|
||||
Providers without a managed network policy use the empty default.
|
||||
"""
|
||||
del sandbox_id
|
||||
return []
|
||||
|
||||
async def consume_network_policy_events_async(self, sandbox_id: str) -> list[dict[str, object]]:
|
||||
return await asyncio.to_thread(self.consume_network_policy_events, sandbox_id)
|
||||
|
||||
def deny_pending_network_policy_events(self, sandbox_id: str) -> bool:
|
||||
"""Atomically deny all unsurfaced trusted-proxy events for a sandbox."""
|
||||
del sandbox_id
|
||||
return False
|
||||
|
||||
async def deny_pending_network_policy_events_async(self, sandbox_id: str) -> bool:
|
||||
return await asyncio.to_thread(self.deny_pending_network_policy_events, sandbox_id)
|
||||
|
||||
def decide_network_policy_request(self, sandbox_id: str, request_id: str, decision: str) -> bool:
|
||||
"""Apply a user decision to one trusted-proxy event."""
|
||||
del sandbox_id, request_id, decision
|
||||
return False
|
||||
|
||||
async def decide_network_policy_request_async(self, sandbox_id: str, request_id: str, decision: str) -> bool:
|
||||
return await asyncio.to_thread(self.decide_network_policy_request, sandbox_id, request_id, decision)
|
||||
|
||||
|
||||
_default_sandbox_provider: SandboxProvider | None = None
|
||||
# Guards every read and write of `_default_sandbox_provider`. The singleton is
|
||||
@ -115,6 +149,12 @@ _default_sandbox_provider: SandboxProvider | None = None
|
||||
_provider_lock = threading.Lock()
|
||||
|
||||
|
||||
def get_initialized_sandbox_provider() -> SandboxProvider | None:
|
||||
"""Return the provider only when another lifecycle path initialized it."""
|
||||
with _provider_lock:
|
||||
return _default_sandbox_provider
|
||||
|
||||
|
||||
def get_sandbox_provider(**kwargs) -> SandboxProvider:
|
||||
"""Get the sandbox provider singleton.
|
||||
|
||||
|
||||
@ -49,6 +49,29 @@ def test_local_sandbox_client_bypasses_environment_proxy():
|
||||
)
|
||||
|
||||
|
||||
def test_local_sandbox_client_forwards_trusted_relay_headers():
|
||||
from deerflow.community.aio_sandbox.aio_sandbox import AioSandbox
|
||||
|
||||
sentinel_httpx = MagicMock()
|
||||
headers = {"X-DeerFlow-Relay-Token": "secret-token"}
|
||||
with (
|
||||
patch("deerflow.community.aio_sandbox.aio_sandbox.httpx.Client", return_value=sentinel_httpx),
|
||||
patch("deerflow.community.aio_sandbox.aio_sandbox.AioSandboxClient") as sdk_cls,
|
||||
):
|
||||
AioSandbox(
|
||||
id="test-sandbox",
|
||||
base_url="http://host.docker.internal:8080",
|
||||
request_headers=headers,
|
||||
)
|
||||
|
||||
sdk_cls.assert_called_once_with(
|
||||
base_url="http://host.docker.internal:8080",
|
||||
timeout=600,
|
||||
headers=headers,
|
||||
httpx_client=sentinel_httpx,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"base_url",
|
||||
[
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
395
backend/tests/test_aio_sandbox_network_proxy.py
Normal file
395
backend/tests/test_aio_sandbox_network_proxy.py
Normal file
@ -0,0 +1,395 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import socket
|
||||
import ssl
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
|
||||
from deerflow.community.aio_sandbox import network_proxy
|
||||
|
||||
|
||||
def test_domain_matches_exact_and_leading_wildcard_only() -> None:
|
||||
assert network_proxy.domain_matches("pypi.org", "pypi.org")
|
||||
assert not network_proxy.domain_matches("evilpypi.org", "pypi.org")
|
||||
assert network_proxy.domain_matches("files.pythonhosted.org", "*.pythonhosted.org")
|
||||
assert not network_proxy.domain_matches("pythonhosted.org", "*.pythonhosted.org")
|
||||
|
||||
|
||||
def test_address_is_public_rejects_host_private_link_local_and_metadata() -> None:
|
||||
for address in (
|
||||
"127.0.0.1",
|
||||
"10.0.0.2",
|
||||
"172.16.0.2",
|
||||
"192.168.1.2",
|
||||
"169.254.169.254",
|
||||
"224.0.0.1",
|
||||
"::1",
|
||||
"fc00::1",
|
||||
"fec0::1",
|
||||
"fe80::1",
|
||||
"ff0e::1",
|
||||
):
|
||||
assert not network_proxy.address_is_public(address)
|
||||
assert network_proxy.address_is_public("8.8.8.8")
|
||||
assert not network_proxy.address_is_public("198.18.1.5")
|
||||
assert network_proxy.address_is_public("198.18.1.5", allow_synthetic_dns=True)
|
||||
|
||||
|
||||
def test_policy_denial_and_temporary_or_sandbox_grants(tmp_path, monkeypatch) -> None:
|
||||
monkeypatch.setattr(network_proxy, "POLICY_DB", tmp_path / "policy.sqlite3")
|
||||
monkeypatch.setenv("DEERFLOW_NETWORK_MODE", "allowlist")
|
||||
monkeypatch.setenv("DEERFLOW_ALLOW_DOMAINS_JSON", json.dumps(["pypi.org"]))
|
||||
|
||||
assert network_proxy.policy_allows("pypi.org", 443, now=100)
|
||||
assert not network_proxy.policy_allows("example.com", 443, now=100)
|
||||
|
||||
temporary = network_proxy.record_denial("example.com", 443, "CONNECT")
|
||||
assert network_proxy.decide(temporary, "allow_temporary", ttl=60)
|
||||
assert network_proxy.policy_allows("example.com", 443)
|
||||
|
||||
sandbox = network_proxy.record_denial("files.example.net", 443, "CONNECT")
|
||||
assert network_proxy.decide(sandbox, "allow_sandbox", ttl=60)
|
||||
assert network_proxy.policy_allows("files.example.net", 443, now=10**12)
|
||||
|
||||
|
||||
def test_pending_events_are_consumed_once(tmp_path, monkeypatch) -> None:
|
||||
monkeypatch.setattr(network_proxy, "POLICY_DB", tmp_path / "policy.sqlite3")
|
||||
request_id = network_proxy.record_denial("example.com", 443, "CONNECT")
|
||||
|
||||
events = network_proxy.pending_events()
|
||||
|
||||
assert events == [
|
||||
{
|
||||
"request_id": request_id,
|
||||
"host": "example.com",
|
||||
"port": 443,
|
||||
"method": "CONNECT",
|
||||
"created_at": events[0]["created_at"],
|
||||
}
|
||||
]
|
||||
assert network_proxy.pending_events() == []
|
||||
|
||||
|
||||
def test_pending_events_surface_only_one_destination_per_approval(tmp_path, monkeypatch) -> None:
|
||||
monkeypatch.setattr(network_proxy, "POLICY_DB", tmp_path / "policy.sqlite3")
|
||||
first = network_proxy.record_denial("one.example", 443, "CONNECT")
|
||||
network_proxy.record_denial("two.example", 443, "CONNECT")
|
||||
|
||||
assert [event["request_id"] for event in network_proxy.pending_events()] == [first]
|
||||
# The sibling is superseded so a retry can create a fresh approvable event.
|
||||
assert network_proxy.pending_events() == []
|
||||
fresh = network_proxy.record_denial("two.example", 443, "CONNECT")
|
||||
assert [event["request_id"] for event in network_proxy.pending_events()] == [fresh]
|
||||
|
||||
|
||||
def test_pending_events_supersede_every_sibling_without_a_batch_limit(tmp_path, monkeypatch) -> None:
|
||||
monkeypatch.setattr(network_proxy, "POLICY_DB", tmp_path / "policy.sqlite3")
|
||||
request_ids = [network_proxy.record_denial(f"host-{index}.example", 443, "CONNECT") for index in range(17)]
|
||||
|
||||
assert [event["request_id"] for event in network_proxy.pending_events()] == [request_ids[0]]
|
||||
assert network_proxy.pending_events() == []
|
||||
|
||||
|
||||
def test_deny_pending_events_atomically_denies_every_unsurfaced_event(tmp_path, monkeypatch) -> None:
|
||||
monkeypatch.setattr(network_proxy, "POLICY_DB", tmp_path / "policy.sqlite3")
|
||||
for index in range(17):
|
||||
network_proxy.record_denial(f"host-{index}.example", 443, "CONNECT")
|
||||
|
||||
assert network_proxy.deny_pending_events() == 17
|
||||
assert network_proxy.pending_events() == []
|
||||
|
||||
with network_proxy._connect_db() as db:
|
||||
assert db.execute("SELECT COUNT(*) FROM events WHERE decision = 'deny'").fetchone() == (17,)
|
||||
|
||||
|
||||
def test_deny_pending_events_preserves_an_already_surfaced_user_decision(tmp_path, monkeypatch) -> None:
|
||||
monkeypatch.setattr(network_proxy, "POLICY_DB", tmp_path / "policy.sqlite3")
|
||||
surfaced = network_proxy.record_denial("interactive.example", 443, "CONNECT")
|
||||
assert [event["request_id"] for event in network_proxy.pending_events()] == [surfaced]
|
||||
unsurfaced = network_proxy.record_denial("scheduled.example", 443, "CONNECT")
|
||||
|
||||
assert network_proxy.deny_pending_events() == 1
|
||||
assert network_proxy.decide(surfaced, "allow_temporary", ttl=60)
|
||||
assert network_proxy.decide(unsurfaced, "deny", ttl=60)
|
||||
|
||||
|
||||
def test_pending_events_claims_old_unsurfaced_denial_on_retry(tmp_path, monkeypatch) -> None:
|
||||
monkeypatch.setattr(network_proxy, "POLICY_DB", tmp_path / "policy.sqlite3")
|
||||
now = [100.0]
|
||||
monkeypatch.setattr(network_proxy.time, "time", lambda: now[0])
|
||||
request_id = network_proxy.record_denial("late.example", 443, "CONNECT")
|
||||
|
||||
now[0] = 105.0
|
||||
assert network_proxy.record_denial("late.example", 443, "CONNECT") == request_id
|
||||
assert [event["request_id"] for event in network_proxy.pending_events()] == [request_id]
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_resolve_public_fails_closed_when_dns_contains_private_answer(monkeypatch) -> None:
|
||||
loop = __import__("asyncio").get_running_loop()
|
||||
|
||||
async def fake_getaddrinfo(*_args, **_kwargs):
|
||||
return [
|
||||
(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("8.8.8.8", 443)),
|
||||
(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("10.0.0.5", 443)),
|
||||
]
|
||||
|
||||
monkeypatch.setattr(loop, "getaddrinfo", fake_getaddrinfo)
|
||||
assert await network_proxy.resolve_public("example.com", 443) is None
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_resolve_public_returns_every_validated_answer_and_open_retries(monkeypatch) -> None:
|
||||
loop = asyncio.get_running_loop()
|
||||
answers = [
|
||||
(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("8.8.8.8", 443)),
|
||||
(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("1.1.1.1", 443)),
|
||||
]
|
||||
|
||||
async def fake_getaddrinfo(*_args, **_kwargs):
|
||||
return answers
|
||||
|
||||
attempts: list[str] = []
|
||||
connected = (MagicMock(), MagicMock())
|
||||
|
||||
async def fake_open_connection(host: str, _port: int, *, family: int):
|
||||
attempts.append(host)
|
||||
assert family == socket.AF_INET
|
||||
if host == "8.8.8.8":
|
||||
raise OSError("first address unavailable")
|
||||
return connected
|
||||
|
||||
monkeypatch.setattr(loop, "getaddrinfo", fake_getaddrinfo)
|
||||
monkeypatch.setattr(network_proxy.asyncio, "open_connection", fake_open_connection)
|
||||
|
||||
resolved = await network_proxy.resolve_public("example.com", 443)
|
||||
|
||||
assert resolved == (
|
||||
(socket.AF_INET, ("8.8.8.8", 443)),
|
||||
(socket.AF_INET, ("1.1.1.1", 443)),
|
||||
)
|
||||
assert await network_proxy._open_public(resolved, 443) is connected
|
||||
assert attempts == ["8.8.8.8", "1.1.1.1"]
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
@pytest.mark.parametrize("mode", ["isolated", "allowlist"])
|
||||
async def test_denied_destination_is_rejected_without_dns_resolution(tmp_path, monkeypatch, mode: str) -> None:
|
||||
monkeypatch.setattr(network_proxy, "POLICY_DB", tmp_path / "policy.sqlite3")
|
||||
monkeypatch.setenv("DEERFLOW_NETWORK_MODE", mode)
|
||||
monkeypatch.setenv("DEERFLOW_ALLOW_DOMAINS_JSON", json.dumps(["allowed.example"]))
|
||||
monkeypatch.delenv("DEERFLOW_RECORD_DENIALS", raising=False)
|
||||
resolutions: list[tuple[str, int]] = []
|
||||
|
||||
async def fake_resolve_public(host: str, port: int):
|
||||
resolutions.append((host, port))
|
||||
return None
|
||||
|
||||
monkeypatch.setattr(network_proxy, "resolve_public", fake_resolve_public)
|
||||
|
||||
proxy = await asyncio.start_server(network_proxy.handle_proxy, "127.0.0.1", 0)
|
||||
proxy_port = proxy.sockets[0].getsockname()[1]
|
||||
try:
|
||||
reader, writer = await asyncio.open_connection("127.0.0.1", proxy_port)
|
||||
writer.write(b"CONNECT denied.example:443 HTTP/1.1\r\nHost: denied.example:443\r\n\r\n")
|
||||
await writer.drain()
|
||||
response = await asyncio.wait_for(reader.read(), timeout=2)
|
||||
writer.close()
|
||||
await writer.wait_closed()
|
||||
|
||||
assert b"403 Forbidden" in response
|
||||
assert resolutions == []
|
||||
finally:
|
||||
proxy.close()
|
||||
await proxy.wait_closed()
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_tls_client_hello_sni_is_extracted_for_connect_enforcement() -> None:
|
||||
incoming = ssl.MemoryBIO()
|
||||
outgoing = ssl.MemoryBIO()
|
||||
context = ssl.create_default_context()
|
||||
tls = context.wrap_bio(incoming, outgoing, server_side=False, server_hostname="pypi.org")
|
||||
with pytest.raises(ssl.SSLWantReadError):
|
||||
tls.do_handshake()
|
||||
|
||||
reader = __import__("asyncio").StreamReader()
|
||||
wire = outgoing.read()
|
||||
reader.feed_data(wire)
|
||||
reader.feed_eof()
|
||||
|
||||
parsed = await network_proxy._read_tls_client_hello(reader)
|
||||
|
||||
assert parsed == ("pypi.org", wire)
|
||||
|
||||
|
||||
def test_http_request_framing_rejects_ambiguous_or_duplicate_lengths() -> None:
|
||||
with pytest.raises(ValueError, match="cannot be combined"):
|
||||
fields = network_proxy._parse_http_header_fields(["Content-Length: 4", "Transfer-Encoding: chunked"])
|
||||
network_proxy._http_request_body_framing(fields)
|
||||
with pytest.raises(ValueError, match="one non-negative"):
|
||||
fields = network_proxy._parse_http_header_fields(["Content-Length: 4", "Content-Length: 4"])
|
||||
network_proxy._http_request_body_framing(fields)
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_chunked_request_body_rejects_non_hex_size() -> None:
|
||||
reader = asyncio.StreamReader()
|
||||
reader.feed_data(b"+1\r\na\r\n0\r\n\r\n")
|
||||
reader.feed_eof()
|
||||
writer = MagicMock()
|
||||
|
||||
with pytest.raises(ValueError, match="Invalid chunk size"):
|
||||
await network_proxy._copy_chunked_request_body(reader, writer)
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_http_proxy_relays_exactly_one_request_per_connection(monkeypatch) -> None:
|
||||
received: asyncio.Future[bytes] = asyncio.get_running_loop().create_future()
|
||||
|
||||
async def upstream_handler(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
|
||||
header = await reader.readuntil(b"\r\n\r\n")
|
||||
body = await reader.readexactly(4)
|
||||
received.set_result(header + body)
|
||||
writer.write(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok")
|
||||
await writer.drain()
|
||||
writer.close()
|
||||
await writer.wait_closed()
|
||||
|
||||
upstream = await asyncio.start_server(upstream_handler, "127.0.0.1", 0)
|
||||
upstream_port = upstream.sockets[0].getsockname()[1]
|
||||
|
||||
async def fake_resolve_public(_host: str, _port: int):
|
||||
return (socket.AF_INET, ("127.0.0.1", upstream_port))
|
||||
|
||||
async def fake_open_public(_resolved, _port: int):
|
||||
return await asyncio.open_connection("127.0.0.1", upstream_port)
|
||||
|
||||
monkeypatch.setattr(network_proxy, "resolve_public", fake_resolve_public)
|
||||
monkeypatch.setattr(network_proxy, "_open_public", fake_open_public)
|
||||
monkeypatch.setattr(network_proxy, "policy_allows", lambda host, port: (host, port) == ("allowed.example", 80))
|
||||
|
||||
proxy = await asyncio.start_server(network_proxy.handle_proxy, "127.0.0.1", 0)
|
||||
proxy_port = proxy.sockets[0].getsockname()[1]
|
||||
try:
|
||||
reader, writer = await asyncio.open_connection("127.0.0.1", proxy_port)
|
||||
writer.write(b"POST http://allowed.example/first HTTP/1.1\r\nHost: allowed.example\r\nContent-Length: 4\r\nConnection: keep-alive\r\n\r\ndataGET http://denied.example/second HTTP/1.1\r\nHost: denied.example\r\n\r\n")
|
||||
await writer.drain()
|
||||
response = await asyncio.wait_for(reader.read(), timeout=2)
|
||||
writer.close()
|
||||
await writer.wait_closed()
|
||||
|
||||
upstream_request = await asyncio.wait_for(received, timeout=2)
|
||||
assert b"POST /first HTTP/1.1" in upstream_request
|
||||
assert b"Connection: close" in upstream_request
|
||||
assert b"Connection: keep-alive" not in upstream_request
|
||||
assert upstream_request.endswith(b"data")
|
||||
assert b"denied.example" not in upstream_request
|
||||
assert b"200 OK" in response
|
||||
finally:
|
||||
proxy.close()
|
||||
upstream.close()
|
||||
await proxy.wait_closed()
|
||||
await upstream.wait_closed()
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_sandbox_api_relay_requires_per_sandbox_token(monkeypatch) -> None:
|
||||
received: asyncio.Future[bytes] = asyncio.get_running_loop().create_future()
|
||||
|
||||
async def upstream_handler(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
|
||||
header = await reader.readuntil(b"\r\n\r\n")
|
||||
received.set_result(header)
|
||||
writer.write(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok")
|
||||
await writer.drain()
|
||||
writer.close()
|
||||
await writer.wait_closed()
|
||||
|
||||
upstream = await asyncio.start_server(upstream_handler, "127.0.0.1", 0)
|
||||
upstream_port = upstream.sockets[0].getsockname()[1]
|
||||
monkeypatch.setenv(network_proxy.RELAY_TOKEN_ENV, "test-relay-token")
|
||||
monkeypatch.setenv("DEERFLOW_SANDBOX_TARGET", f"127.0.0.1:{upstream_port}")
|
||||
relay = await asyncio.start_server(network_proxy.handle_relay, "127.0.0.1", 0)
|
||||
relay_port = relay.sockets[0].getsockname()[1]
|
||||
try:
|
||||
denied_reader, denied_writer = await asyncio.open_connection("127.0.0.1", relay_port)
|
||||
denied_writer.write(b"GET /v1/sandbox HTTP/1.1\r\nHost: sandbox\r\n\r\n")
|
||||
await denied_writer.drain()
|
||||
denied_response = await asyncio.wait_for(denied_reader.read(), timeout=2)
|
||||
denied_writer.close()
|
||||
await denied_writer.wait_closed()
|
||||
|
||||
assert b"403 Forbidden" in denied_response
|
||||
assert not received.done()
|
||||
|
||||
reader, writer = await asyncio.open_connection("127.0.0.1", relay_port)
|
||||
writer.write(b"GET /v1/sandbox HTTP/1.1\r\nHost: sandbox\r\n" + f"{network_proxy.RELAY_AUTH_HEADER}: test-relay-token\r\n\r\n".encode())
|
||||
await writer.drain()
|
||||
response = await asyncio.wait_for(reader.read(), timeout=2)
|
||||
writer.close()
|
||||
await writer.wait_closed()
|
||||
|
||||
assert b"200 OK" in response
|
||||
assert network_proxy.RELAY_AUTH_HEADER.encode() in await asyncio.wait_for(received, timeout=2)
|
||||
finally:
|
||||
relay.close()
|
||||
upstream.close()
|
||||
await relay.wait_closed()
|
||||
await upstream.wait_closed()
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_sandbox_api_relay_rejects_non_ascii_token(monkeypatch) -> None:
|
||||
monkeypatch.setenv(network_proxy.RELAY_TOKEN_ENV, "test-relay-token")
|
||||
relay = await asyncio.start_server(network_proxy.handle_relay, "127.0.0.1", 0)
|
||||
relay_port = relay.sockets[0].getsockname()[1]
|
||||
try:
|
||||
reader, writer = await asyncio.open_connection("127.0.0.1", relay_port)
|
||||
writer.write(b"GET /v1/sandbox HTTP/1.1\r\nHost: sandbox\r\n" + network_proxy.RELAY_AUTH_HEADER.encode() + b": \xff\r\n\r\n")
|
||||
await writer.drain()
|
||||
response = await asyncio.wait_for(reader.read(), timeout=2)
|
||||
writer.close()
|
||||
await writer.wait_closed()
|
||||
|
||||
assert b"403 Forbidden" in response
|
||||
finally:
|
||||
relay.close()
|
||||
await relay.wait_closed()
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
@pytest.mark.parametrize(
|
||||
"malformed_header",
|
||||
[
|
||||
b"Host : denied.example",
|
||||
b"Transfer-Encoding : chunked",
|
||||
b"Bad(Header): value",
|
||||
],
|
||||
)
|
||||
async def test_http_proxy_rejects_ambiguous_field_names_before_policy_check(monkeypatch, malformed_header: bytes) -> None:
|
||||
policy_checks: list[tuple[str, int]] = []
|
||||
|
||||
def fake_policy_allows(host: str, port: int) -> bool:
|
||||
policy_checks.append((host, port))
|
||||
return False
|
||||
|
||||
monkeypatch.setattr(network_proxy, "policy_allows", fake_policy_allows)
|
||||
|
||||
proxy = await asyncio.start_server(network_proxy.handle_proxy, "127.0.0.1", 0)
|
||||
proxy_port = proxy.sockets[0].getsockname()[1]
|
||||
try:
|
||||
reader, writer = await asyncio.open_connection("127.0.0.1", proxy_port)
|
||||
writer.write(b"GET http://allowed.example/ HTTP/1.1\r\nHost: allowed.example\r\n" + malformed_header + b"\r\n\r\n")
|
||||
await writer.drain()
|
||||
response = await asyncio.wait_for(reader.read(), timeout=2)
|
||||
writer.close()
|
||||
await writer.wait_closed()
|
||||
|
||||
assert b"400 Bad Request" in response
|
||||
assert policy_checks == []
|
||||
finally:
|
||||
proxy.close()
|
||||
await proxy.wait_closed()
|
||||
@ -75,6 +75,9 @@ def test_wait_for_sandbox_ready_bypasses_environment_proxy_for_docker_host(monke
|
||||
class FakeSession:
|
||||
trust_env = True
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.headers: dict[str, str] = {}
|
||||
|
||||
def __enter__(self):
|
||||
sessions.append(self)
|
||||
return self
|
||||
@ -89,9 +92,18 @@ def test_wait_for_sandbox_ready_bypasses_environment_proxy_for_docker_host(monke
|
||||
|
||||
monkeypatch.setattr(readiness.requests, "Session", FakeSession)
|
||||
|
||||
assert readiness.wait_for_sandbox_ready("http://host.docker.internal:8080", timeout=1) is True
|
||||
headers = {"X-DeerFlow-Relay-Token": "secret-token"}
|
||||
assert (
|
||||
readiness.wait_for_sandbox_ready(
|
||||
"http://host.docker.internal:8080",
|
||||
timeout=1,
|
||||
headers=headers,
|
||||
)
|
||||
is True
|
||||
)
|
||||
assert len(sessions) == 1
|
||||
assert sessions[0].trust_env is False
|
||||
assert sessions[0].headers == headers
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
@ -99,8 +111,10 @@ async def test_wait_for_sandbox_ready_async_uses_nonblocking_polling(monkeypatch
|
||||
calls: list[str] = []
|
||||
sleeps: list[float] = []
|
||||
clients: list[_FakeAsyncClient] = []
|
||||
client_headers: list[dict[str, str]] = []
|
||||
|
||||
def fake_client(*, timeout: float, trust_env: bool):
|
||||
def fake_client(*, timeout: float, trust_env: bool, headers: dict[str, str]):
|
||||
client_headers.append(headers)
|
||||
client = _FakeAsyncClient(
|
||||
responses=[SimpleNamespace(status_code=503), SimpleNamespace(status_code=200)],
|
||||
calls=calls,
|
||||
@ -118,11 +132,21 @@ async def test_wait_for_sandbox_ready_async_uses_nonblocking_polling(monkeypatch
|
||||
monkeypatch.setattr(readiness.requests, "get", lambda *args, **kwargs: (_ for _ in ()).throw(AssertionError("requests.get should not be used")))
|
||||
monkeypatch.setattr(readiness.time, "sleep", lambda *_args, **_kwargs: (_ for _ in ()).throw(AssertionError("time.sleep should not be used")))
|
||||
|
||||
assert await readiness.wait_for_sandbox_ready_async("http://sandbox", timeout=5, poll_interval=0.05) is True
|
||||
headers = {"X-DeerFlow-Relay-Token": "secret-token"}
|
||||
assert (
|
||||
await readiness.wait_for_sandbox_ready_async(
|
||||
"http://sandbox",
|
||||
timeout=5,
|
||||
poll_interval=0.05,
|
||||
headers=headers,
|
||||
)
|
||||
is True
|
||||
)
|
||||
|
||||
assert calls == ["http://sandbox/v1/sandbox", "http://sandbox/v1/sandbox"]
|
||||
assert sleeps == [0.05]
|
||||
assert clients[0].trust_env is False
|
||||
assert client_headers == [headers]
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
|
||||
@ -1306,7 +1306,8 @@ class TestChatModelStartHumanMessage:
|
||||
assert not any(e["event_type"] == "llm.human.input" for e in events)
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_hidden_human_input_response_is_captured(self, journal_setup):
|
||||
@pytest.mark.parametrize("source", ["ask_clarification", "sandbox_network"])
|
||||
async def test_hidden_human_input_response_is_captured(self, journal_setup, source):
|
||||
"""Hidden HumanInputCard replies are user-authored and must survive compaction."""
|
||||
from langchain_core.messages import HumanMessage
|
||||
|
||||
@ -1315,7 +1316,7 @@ class TestChatModelStartHumanMessage:
|
||||
content='For your clarification "Which environment?", my answer is: staging',
|
||||
additional_kwargs={
|
||||
"hide_from_ui": True,
|
||||
"human_input_response": self._human_input_response(),
|
||||
"human_input_response": self._human_input_response(source=source),
|
||||
},
|
||||
)
|
||||
j.on_chat_model_start({}, [[hidden_response]], run_id=uuid4(), tags=["lead_agent"])
|
||||
|
||||
@ -6,7 +6,7 @@ from typing import get_type_hints
|
||||
import pytest
|
||||
from langchain.agents.middleware import AgentMiddleware
|
||||
from langchain.tools import ToolRuntime
|
||||
from langchain_core.messages import ToolMessage
|
||||
from langchain_core.messages import HumanMessage, ToolMessage
|
||||
from langgraph.graph import END
|
||||
from langgraph.prebuilt.tool_node import ToolCallRequest
|
||||
from langgraph.runtime import Runtime
|
||||
@ -61,6 +61,36 @@ class _AgentSkillSyncProvider(_SyncProvider):
|
||||
self.skill_syncs.append((sandbox_id, thread_id, user_id, projection))
|
||||
|
||||
|
||||
class _NetworkPolicyProvider(_SyncProvider):
|
||||
def __init__(self) -> None:
|
||||
super().__init__()
|
||||
self.events: list[dict[str, object]] = []
|
||||
self.decisions: list[tuple[str, str, str]] = []
|
||||
self.consume_calls: list[str] = []
|
||||
self.deny_pending_calls: list[str] = []
|
||||
|
||||
def sandbox_network_mode(self) -> str:
|
||||
return "allowlist"
|
||||
|
||||
def consume_network_policy_events(self, sandbox_id: str) -> list[dict[str, object]]:
|
||||
self.consume_calls.append(sandbox_id)
|
||||
events, self.events = self.events, []
|
||||
return events
|
||||
|
||||
def deny_pending_network_policy_events(self, sandbox_id: str) -> bool:
|
||||
self.deny_pending_calls.append(sandbox_id)
|
||||
for event in self.events:
|
||||
request_id = event.get("request_id")
|
||||
if isinstance(request_id, str):
|
||||
self.decisions.append((sandbox_id, request_id, "deny"))
|
||||
self.events = []
|
||||
return True
|
||||
|
||||
def decide_network_policy_request(self, sandbox_id: str, request_id: str, decision: str) -> bool:
|
||||
self.decisions.append((sandbox_id, request_id, decision))
|
||||
return True
|
||||
|
||||
|
||||
class _SandboxStub(Sandbox):
|
||||
def execute_command(
|
||||
self,
|
||||
@ -569,6 +599,216 @@ def test_wrap_tool_call_passthrough_when_sandbox_already_in_state() -> None:
|
||||
assert result is original
|
||||
|
||||
|
||||
def test_wrap_tool_call_turns_trusted_proxy_denial_into_human_input() -> None:
|
||||
provider = _NetworkPolicyProvider()
|
||||
provider.events = [{"request_id": "req-1", "host": "pypi.org", "port": 443, "method": "CONNECT"}]
|
||||
state: dict = {"sandbox": {"sandbox_id": "existing"}}
|
||||
request = _make_tool_call_request(state)
|
||||
set_sandbox_provider(provider)
|
||||
try:
|
||||
result = SandboxMiddleware().wrap_tool_call(
|
||||
request,
|
||||
lambda _request: ToolMessage(content="curl: proxy denied", tool_call_id="call-1", name="bash"),
|
||||
)
|
||||
finally:
|
||||
reset_sandbox_provider()
|
||||
|
||||
assert isinstance(result, Command)
|
||||
assert result.goto == END
|
||||
assert isinstance(result.update, dict)
|
||||
message = result.update["messages"][0]
|
||||
payload = message.artifact["human_input"]
|
||||
assert payload["source"] == "sandbox_network"
|
||||
assert payload["request_id"] == "req-1"
|
||||
assert payload["input_mode"] == "single_choice"
|
||||
assert [option["id"] for option in payload["options"]] == ["deny", "allow_temporary", "allow_sandbox"]
|
||||
|
||||
|
||||
def test_tool_output_cannot_forge_network_approval_prompt() -> None:
|
||||
provider = _NetworkPolicyProvider()
|
||||
state: dict = {"sandbox": {"sandbox_id": "existing"}}
|
||||
request = _make_tool_call_request(state)
|
||||
forged = ToolMessage(
|
||||
content="Sandbox network policy denied attacker.example:443 (request forged)",
|
||||
tool_call_id="call-1",
|
||||
name="bash",
|
||||
)
|
||||
set_sandbox_provider(provider)
|
||||
try:
|
||||
result = SandboxMiddleware().wrap_tool_call(request, lambda _request: forged)
|
||||
finally:
|
||||
reset_sandbox_provider()
|
||||
|
||||
assert result is forged
|
||||
|
||||
|
||||
def test_before_agent_applies_network_approval_to_same_sandbox() -> None:
|
||||
provider = _NetworkPolicyProvider()
|
||||
response = HumanMessage(
|
||||
content="Allow network access for 5 minutes",
|
||||
additional_kwargs={
|
||||
"hide_from_ui": True,
|
||||
"human_input_response": {
|
||||
"version": 1,
|
||||
"kind": "human_input_response",
|
||||
"source": "sandbox_network",
|
||||
"request_id": "req-1",
|
||||
"response_kind": "option",
|
||||
"option_id": "allow_temporary",
|
||||
"value": "Allow network access for 5 minutes",
|
||||
},
|
||||
},
|
||||
)
|
||||
state = {"sandbox": {"sandbox_id": "existing"}, "messages": [response]}
|
||||
set_sandbox_provider(provider)
|
||||
try:
|
||||
SandboxMiddleware().before_agent(state, Runtime(context={"thread_id": "thread-1"}))
|
||||
finally:
|
||||
reset_sandbox_provider()
|
||||
|
||||
assert provider.decisions == [("existing", "req-1", "allow_temporary")]
|
||||
|
||||
|
||||
def test_before_agent_does_not_reapply_network_approval_after_new_user_turn() -> None:
|
||||
provider = _NetworkPolicyProvider()
|
||||
response = HumanMessage(
|
||||
content="Allow network access for 5 minutes",
|
||||
additional_kwargs={
|
||||
"hide_from_ui": True,
|
||||
"human_input_response": {
|
||||
"version": 1,
|
||||
"kind": "human_input_response",
|
||||
"source": "sandbox_network",
|
||||
"request_id": "req-1",
|
||||
"response_kind": "option",
|
||||
"option_id": "allow_temporary",
|
||||
"value": "Allow network access for 5 minutes",
|
||||
},
|
||||
},
|
||||
)
|
||||
state = {
|
||||
"sandbox": {"sandbox_id": "existing"},
|
||||
"messages": [response, HumanMessage(content="Now summarize the result")],
|
||||
}
|
||||
set_sandbox_provider(provider)
|
||||
try:
|
||||
SandboxMiddleware().before_agent(state, Runtime(context={"thread_id": "thread-1"}))
|
||||
finally:
|
||||
reset_sandbox_provider()
|
||||
|
||||
assert provider.decisions == []
|
||||
|
||||
|
||||
@pytest.mark.parametrize("context_key", ["disable_clarification", "non_interactive"])
|
||||
def test_sync_noninteractive_network_denial_is_recorded_without_prompt(context_key: str) -> None:
|
||||
provider = _NetworkPolicyProvider()
|
||||
provider.events = [{"request_id": "req-1", "host": "example.com", "port": 443, "method": "CONNECT"}]
|
||||
state: dict = {"sandbox": {"sandbox_id": "existing"}}
|
||||
request = _make_tool_call_request(state)
|
||||
request.runtime.context[context_key] = True
|
||||
original = ToolMessage(content="proxy denied", tool_call_id="call-1", name="bash")
|
||||
set_sandbox_provider(provider)
|
||||
try:
|
||||
result = SandboxMiddleware().wrap_tool_call(request, lambda _request: original)
|
||||
finally:
|
||||
reset_sandbox_provider()
|
||||
|
||||
assert result is original
|
||||
assert provider.decisions == [("existing", "req-1", "deny")]
|
||||
assert provider.deny_pending_calls == ["existing"]
|
||||
assert provider.consume_calls == []
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
@pytest.mark.parametrize("context_key", ["disable_clarification", "non_interactive"])
|
||||
async def test_async_noninteractive_network_denial_is_recorded_without_prompt(context_key: str) -> None:
|
||||
provider = _NetworkPolicyProvider()
|
||||
provider.events = [{"request_id": "req-1", "host": "example.com", "port": 443, "method": "CONNECT"}]
|
||||
state: dict = {"sandbox": {"sandbox_id": "existing"}}
|
||||
request = _make_tool_call_request(state)
|
||||
request.runtime.context[context_key] = True
|
||||
original = ToolMessage(content="proxy denied", tool_call_id="call-1", name="bash")
|
||||
|
||||
async def handler(_request: ToolCallRequest) -> ToolMessage:
|
||||
return original
|
||||
|
||||
set_sandbox_provider(provider)
|
||||
try:
|
||||
result = await SandboxMiddleware().awrap_tool_call(request, handler)
|
||||
finally:
|
||||
reset_sandbox_provider()
|
||||
|
||||
assert result is original
|
||||
assert provider.decisions == [("existing", "req-1", "deny")]
|
||||
assert provider.deny_pending_calls == ["existing"]
|
||||
assert provider.consume_calls == []
|
||||
|
||||
|
||||
def test_subagent_network_denial_fails_closed_without_prompt() -> None:
|
||||
provider = _NetworkPolicyProvider()
|
||||
provider.events = [{"request_id": "req-1", "host": "example.com", "port": 443, "method": "CONNECT"}]
|
||||
state: dict = {"sandbox": {"sandbox_id": "existing"}}
|
||||
request = _make_tool_call_request(state)
|
||||
request.runtime.context["is_subagent"] = True
|
||||
original = ToolMessage(content="proxy denied", tool_call_id="call-1", name="bash")
|
||||
set_sandbox_provider(provider)
|
||||
try:
|
||||
result = SandboxMiddleware().wrap_tool_call(request, lambda _request: original)
|
||||
finally:
|
||||
reset_sandbox_provider()
|
||||
|
||||
assert result is original
|
||||
assert provider.events == []
|
||||
assert provider.decisions == [("existing", "req-1", "deny")]
|
||||
assert provider.deny_pending_calls == ["existing"]
|
||||
assert provider.consume_calls == []
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_async_subagent_network_denial_fails_closed_without_prompt() -> None:
|
||||
provider = _NetworkPolicyProvider()
|
||||
provider.events = [{"request_id": "req-1", "host": "example.com", "port": 443, "method": "CONNECT"}]
|
||||
state: dict = {"sandbox": {"sandbox_id": "existing"}}
|
||||
request = _make_tool_call_request(state)
|
||||
request.runtime.context["is_subagent"] = True
|
||||
original = ToolMessage(content="proxy denied", tool_call_id="call-1", name="bash")
|
||||
|
||||
async def handler(_request: ToolCallRequest) -> ToolMessage:
|
||||
return original
|
||||
|
||||
set_sandbox_provider(provider)
|
||||
try:
|
||||
result = await SandboxMiddleware().awrap_tool_call(request, handler)
|
||||
finally:
|
||||
reset_sandbox_provider()
|
||||
|
||||
assert result is original
|
||||
assert provider.events == []
|
||||
assert provider.decisions == [("existing", "req-1", "deny")]
|
||||
assert provider.deny_pending_calls == ["existing"]
|
||||
assert provider.consume_calls == []
|
||||
|
||||
|
||||
def test_noninteractive_network_denial_atomically_drains_more_than_sixteen_hosts() -> None:
|
||||
provider = _NetworkPolicyProvider()
|
||||
provider.events = [{"request_id": f"req-{index}", "host": f"host-{index}.example", "port": 443, "method": "CONNECT"} for index in range(17)]
|
||||
state: dict = {"sandbox": {"sandbox_id": "existing"}}
|
||||
request = _make_tool_call_request(state)
|
||||
request.runtime.context["non_interactive"] = True
|
||||
original = ToolMessage(content="proxy denied", tool_call_id="call-1", name="bash")
|
||||
set_sandbox_provider(provider)
|
||||
try:
|
||||
result = SandboxMiddleware().wrap_tool_call(request, lambda _request: original)
|
||||
finally:
|
||||
reset_sandbox_provider()
|
||||
|
||||
assert result is original
|
||||
assert provider.events == []
|
||||
assert len(provider.decisions) == 17
|
||||
assert provider.deny_pending_calls == ["existing"]
|
||||
assert provider.consume_calls == []
|
||||
|
||||
|
||||
def test_wrap_tool_call_passthrough_when_handler_did_not_initialize_sandbox() -> None:
|
||||
middleware = SandboxMiddleware()
|
||||
state: dict = {}
|
||||
|
||||
38
backend/tests/test_sandbox_network_config.py
Normal file
38
backend/tests/test_sandbox_network_config.py
Normal file
@ -0,0 +1,38 @@
|
||||
import pytest
|
||||
from pydantic import ValidationError
|
||||
|
||||
from deerflow.config.sandbox_config import SandboxConfig, SandboxNetworkConfig
|
||||
|
||||
|
||||
def test_sandbox_network_defaults_to_open() -> None:
|
||||
config = SandboxConfig(use="test")
|
||||
|
||||
assert config.network.mode == "open"
|
||||
assert config.network.allow_domains == []
|
||||
assert config.network.approval == "prompt"
|
||||
|
||||
|
||||
def test_sandbox_network_normalizes_domains() -> None:
|
||||
config = SandboxNetworkConfig(allow_domains=["PyPI.org.", "*.PythonHosted.org", "pypi.org"])
|
||||
|
||||
assert config.allow_domains == ["pypi.org", "*.pythonhosted.org"]
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"domain",
|
||||
[
|
||||
"*",
|
||||
"https://pypi.org",
|
||||
"pypi.org:443",
|
||||
"foo.*.example.com",
|
||||
"../example.com",
|
||||
"127.0.0.1",
|
||||
"localhost",
|
||||
"foo..example.com",
|
||||
"-foo.example.com",
|
||||
"foo_.example.com",
|
||||
],
|
||||
)
|
||||
def test_sandbox_network_rejects_unsafe_domain_rules(domain: str) -> None:
|
||||
with pytest.raises(ValidationError, match="invalid sandbox network allowlist domain"):
|
||||
SandboxNetworkConfig(allow_domains=[domain])
|
||||
31
backend/tests/test_sandbox_network_proxy_workflow.py
Normal file
31
backend/tests/test_sandbox_network_proxy_workflow.py
Normal file
@ -0,0 +1,31 @@
|
||||
from pathlib import Path
|
||||
|
||||
import yaml
|
||||
|
||||
WORKFLOW = Path(__file__).resolve().parents[2] / ".github" / "workflows" / "sandbox-network-proxy-image.yaml"
|
||||
|
||||
|
||||
def test_pull_request_proxy_build_has_read_only_permissions():
|
||||
workflow = yaml.safe_load(WORKFLOW.read_text(encoding="utf-8"))
|
||||
validate = workflow["jobs"]["validate"]
|
||||
build = next(step for step in validate["steps"] if step["name"] == "Build image")
|
||||
|
||||
assert validate["if"] == "github.event_name == 'pull_request'"
|
||||
assert validate["permissions"] == {"contents": "read"}
|
||||
assert build["with"]["push"] is False
|
||||
|
||||
|
||||
def test_proxy_publish_credentials_are_gated_to_upstream_main_pushes():
|
||||
workflow = yaml.safe_load(WORKFLOW.read_text(encoding="utf-8"))
|
||||
publish = workflow["jobs"]["publish"]
|
||||
build = next(step for step in publish["steps"] if step["name"] == "Build and publish image")
|
||||
|
||||
assert publish["if"] == "github.event_name == 'push' && github.ref == 'refs/heads/main' && github.repository == 'bytedance/deer-flow'"
|
||||
assert publish["permissions"] == {
|
||||
"contents": "read",
|
||||
"packages": "write",
|
||||
"attestations": "write",
|
||||
"id-token": "write",
|
||||
}
|
||||
assert build["with"]["push"] is True
|
||||
assert "workflow_dispatch" not in WORKFLOW.read_text(encoding="utf-8")
|
||||
@ -14,11 +14,14 @@ import signal
|
||||
import threading
|
||||
import time
|
||||
from datetime import UTC, datetime
|
||||
from unittest.mock import MagicMock, patch
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from deerflow.community.aio_sandbox.aio_sandbox_provider import SandboxBeingDestroyedError
|
||||
from deerflow.community.aio_sandbox.aio_sandbox_provider import (
|
||||
SandboxBeingDestroyedError,
|
||||
SandboxPolicyReplacementDeferredError,
|
||||
)
|
||||
from deerflow.community.aio_sandbox.ownership import compute_lease_ttl
|
||||
from deerflow.community.aio_sandbox.sandbox_info import SandboxInfo
|
||||
|
||||
@ -175,6 +178,7 @@ def test_list_running_includes_containers_without_port(monkeypatch):
|
||||
assert len(infos) == 1
|
||||
assert infos[0].sandbox_id == "abc12345"
|
||||
assert infos[0].sandbox_url == ""
|
||||
assert infos[0].requires_replacement is True
|
||||
|
||||
|
||||
def test_list_running_handles_docker_failure(monkeypatch):
|
||||
@ -594,6 +598,146 @@ def test_reconcile_skips_container_owned_by_peer():
|
||||
assert shared.owner("shared01") == "worker-a"
|
||||
|
||||
|
||||
def test_reconcile_does_not_replace_mismatched_policy_while_peer_owns_container():
|
||||
"""A rolling upgrade must not delete an older Gateway's live sandbox."""
|
||||
shared = _make_shared_ownership_store()
|
||||
worker_a = _make_provider_for_reconciliation(worker_id="worker-a", store=shared)
|
||||
worker_b = _make_provider_for_reconciliation(worker_id="worker-b", store=shared)
|
||||
info = SandboxInfo(
|
||||
sandbox_id="rolling01",
|
||||
sandbox_url="http://localhost:8080",
|
||||
container_name="deer-flow-sandbox-rolling01",
|
||||
created_at=time.time() - 50,
|
||||
requires_replacement=True,
|
||||
)
|
||||
worker_a._publish_ownership(info.sandbox_id)
|
||||
worker_b._backend.list_running.return_value = [info]
|
||||
|
||||
worker_b._reconcile_orphans()
|
||||
|
||||
worker_b._backend.destroy.assert_not_called()
|
||||
assert info.sandbox_id not in worker_b._warm_pool
|
||||
assert shared.owner(info.sandbox_id) == "worker-a"
|
||||
|
||||
|
||||
def test_reconcile_replaces_mismatched_policy_after_orphan_grace_and_destroy_claim():
|
||||
"""A stale policy is replaced only after grace plus an exclusive teardown lease."""
|
||||
aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider")
|
||||
shared = _make_shared_ownership_store()
|
||||
worker = _make_provider_for_reconciliation(worker_id="worker-b", store=shared)
|
||||
info = SandboxInfo(
|
||||
sandbox_id="rolling02",
|
||||
sandbox_url="http://localhost:8080",
|
||||
container_name="deer-flow-sandbox-rolling02",
|
||||
created_at=time.time() - 50,
|
||||
requires_replacement=True,
|
||||
)
|
||||
worker._backend.list_running.return_value = [info]
|
||||
claims: list[bool] = []
|
||||
real_claim = worker._claim_ownership
|
||||
|
||||
def recording_claim(sandbox_id: str, *, for_destroy: bool = False) -> bool:
|
||||
claims.append(for_destroy)
|
||||
return real_claim(sandbox_id, for_destroy=for_destroy)
|
||||
|
||||
worker._claim_ownership = recording_claim
|
||||
now = time.time()
|
||||
|
||||
with patch.object(aio_mod.time, "time", return_value=now):
|
||||
worker._reconcile_orphans()
|
||||
|
||||
worker._backend.destroy.assert_not_called()
|
||||
assert claims == []
|
||||
|
||||
with patch.object(
|
||||
aio_mod.time,
|
||||
"time",
|
||||
return_value=now + compute_lease_ttl(worker._ownership_config) + 1,
|
||||
):
|
||||
worker._reconcile_orphans()
|
||||
|
||||
worker._backend.destroy.assert_called_once_with(info)
|
||||
assert claims == [True]
|
||||
assert info.sandbox_id not in worker._warm_pool
|
||||
|
||||
|
||||
def test_reconcile_does_not_replace_mismatched_policy_during_local_teardown():
|
||||
"""The provider's local reservation is the same-process half of the fence."""
|
||||
worker = _make_provider_for_reconciliation()
|
||||
info = SandboxInfo(
|
||||
sandbox_id="rolling03",
|
||||
sandbox_url="http://localhost:8080",
|
||||
container_name="deer-flow-sandbox-rolling03",
|
||||
requires_replacement=True,
|
||||
)
|
||||
worker._backend.list_running.return_value = [info]
|
||||
worker._local_teardown.add(info.sandbox_id)
|
||||
worker._claim_ownership = MagicMock(return_value=True)
|
||||
|
||||
worker._reconcile_orphans()
|
||||
|
||||
worker._claim_ownership.assert_not_called()
|
||||
worker._backend.destroy.assert_not_called()
|
||||
assert info.sandbox_id not in worker._warm_pool
|
||||
|
||||
|
||||
def test_discover_or_create_defers_mismatched_policy_owned_by_live_peer(tmp_path):
|
||||
"""The request path uses the same fence instead of deleting on discovery."""
|
||||
aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider")
|
||||
shared = _make_shared_ownership_store()
|
||||
worker_a = _make_provider_for_reconciliation(worker_id="worker-a", store=shared)
|
||||
worker_b = _make_provider_for_reconciliation(worker_id="worker-b", store=shared)
|
||||
info = SandboxInfo(
|
||||
sandbox_id="rolling04",
|
||||
sandbox_url="",
|
||||
container_name="deer-flow-sandbox-rolling04",
|
||||
requires_replacement=True,
|
||||
)
|
||||
worker_a._publish_ownership(info.sandbox_id)
|
||||
worker_b._backend.discover.return_value = info
|
||||
worker_b._create_sandbox = MagicMock(return_value=info.sandbox_id)
|
||||
worker_b._recheck_cached_sandbox = MagicMock(return_value=None)
|
||||
|
||||
paths = MagicMock()
|
||||
paths.thread_dir.return_value = tmp_path
|
||||
with patch.object(aio_mod, "get_paths", return_value=paths):
|
||||
with pytest.raises(SandboxPolicyReplacementDeferredError):
|
||||
worker_b._discover_or_create_with_lock("thread-rolling", info.sandbox_id, user_id="user-rolling")
|
||||
|
||||
worker_b._backend.destroy.assert_not_called()
|
||||
worker_b._create_sandbox.assert_not_called()
|
||||
assert shared.owner(info.sandbox_id) == "worker-a"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_async_discover_or_create_defers_mismatched_policy_owned_by_live_peer(tmp_path):
|
||||
"""The async request path must preserve the same replacement fence."""
|
||||
aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider")
|
||||
shared = _make_shared_ownership_store()
|
||||
worker_a = _make_provider_for_reconciliation(worker_id="worker-a", store=shared)
|
||||
worker_b = _make_provider_for_reconciliation(worker_id="worker-b", store=shared)
|
||||
info = SandboxInfo(
|
||||
sandbox_id="rolling05",
|
||||
sandbox_url="",
|
||||
container_name="deer-flow-sandbox-rolling05",
|
||||
requires_replacement=True,
|
||||
)
|
||||
worker_a._publish_ownership(info.sandbox_id)
|
||||
worker_b._backend.discover.return_value = info
|
||||
worker_b._create_sandbox_async = AsyncMock(return_value=info.sandbox_id)
|
||||
worker_b._recheck_cached_sandbox = MagicMock(return_value=None)
|
||||
|
||||
paths = MagicMock()
|
||||
paths.thread_dir.return_value = tmp_path
|
||||
with patch.object(aio_mod, "get_paths", return_value=paths):
|
||||
with pytest.raises(SandboxPolicyReplacementDeferredError):
|
||||
await worker_b._discover_or_create_with_lock_async("thread-rolling", info.sandbox_id, user_id="user-rolling")
|
||||
|
||||
worker_b._backend.destroy.assert_not_called()
|
||||
worker_b._create_sandbox_async.assert_not_awaited()
|
||||
assert shared.owner(info.sandbox_id) == "worker-a"
|
||||
|
||||
|
||||
def test_idle_reap_does_not_destroy_peer_owned_warm_entry():
|
||||
"""#4206: idle reaper must not stop a container another instance owns."""
|
||||
shared = _make_shared_ownership_store()
|
||||
@ -715,6 +859,7 @@ def test_acquire_fails_closed_when_ownership_cannot_be_published():
|
||||
worker = _make_provider_for_reconciliation(worker_id="worker-a")
|
||||
worker._ownership = MagicMock()
|
||||
worker._ownership.take.side_effect = OwnershipBackendError("store down")
|
||||
worker._ownership.claim.return_value = True
|
||||
|
||||
info = SandboxInfo(
|
||||
sandbox_id="new001",
|
||||
@ -726,8 +871,10 @@ def test_acquire_fails_closed_when_ownership_cannot_be_published():
|
||||
with pytest.raises(OwnershipBackendError):
|
||||
worker._register_created_sandbox("t1", "new001", info, user_id="u1")
|
||||
|
||||
# The just-created container must not be leaked as an unowned orphan.
|
||||
# Registration is fail-closed, and cleanup proceeds only after a separate
|
||||
# teardown claim succeeds.
|
||||
worker._backend.destroy.assert_called_once_with(info)
|
||||
worker._ownership.claim.assert_called_once_with("new001", for_destroy=True)
|
||||
assert "new001" not in worker._sandboxes
|
||||
|
||||
|
||||
@ -1119,12 +1266,13 @@ def test_lost_lease_drops_sandbox_without_destroying_container():
|
||||
|
||||
|
||||
def test_ownership_rollback_on_create_closes_the_client_it_drops():
|
||||
"""The rollback destroys the container; its host-side client must not leak (#2872)."""
|
||||
"""A fenced rollback destroys the container; its host-side client must not leak (#2872)."""
|
||||
from deerflow.community.aio_sandbox.ownership import OwnershipBackendError
|
||||
|
||||
worker = _make_provider_for_reconciliation(worker_id="worker-a")
|
||||
worker._ownership = MagicMock()
|
||||
worker._ownership.take.side_effect = OwnershipBackendError("store down")
|
||||
worker._ownership.claim.return_value = True
|
||||
info = SandboxInfo(
|
||||
sandbox_id="new002",
|
||||
sandbox_url="http://localhost:8080",
|
||||
@ -1145,6 +1293,39 @@ def test_ownership_rollback_on_create_closes_the_client_it_drops():
|
||||
worker._register_created_sandbox("t1", "new002", info, user_id="u1")
|
||||
|
||||
worker._backend.destroy.assert_called_once_with(info)
|
||||
worker._ownership.claim.assert_called_once_with("new002", for_destroy=True)
|
||||
assert created and created[0].close.call_count == 1
|
||||
|
||||
|
||||
def test_ownership_rollback_does_not_destroy_when_teardown_claim_is_unavailable():
|
||||
"""Unknown ownership is not permission to remove a possibly peer-created container."""
|
||||
from deerflow.community.aio_sandbox.ownership import OwnershipBackendError
|
||||
|
||||
worker = _make_provider_for_reconciliation(worker_id="worker-a")
|
||||
worker._ownership = MagicMock()
|
||||
worker._ownership.take.side_effect = OwnershipBackendError("store down")
|
||||
worker._ownership.claim.side_effect = OwnershipBackendError("store still down")
|
||||
info = SandboxInfo(
|
||||
sandbox_id="new003",
|
||||
sandbox_url="http://localhost:8080",
|
||||
container_name="deer-flow-sandbox-new003",
|
||||
created_at=time.time(),
|
||||
)
|
||||
|
||||
aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider")
|
||||
created: list[MagicMock] = []
|
||||
|
||||
def fake_aio_sandbox(**kwargs):
|
||||
sandbox = MagicMock()
|
||||
created.append(sandbox)
|
||||
return sandbox
|
||||
|
||||
with patch.object(aio_mod, "AioSandbox", side_effect=fake_aio_sandbox):
|
||||
with pytest.raises(OwnershipBackendError, match="store down"):
|
||||
worker._register_created_sandbox("t1", "new003", info, user_id="u1")
|
||||
|
||||
worker._backend.destroy.assert_not_called()
|
||||
worker._ownership.claim.assert_called_once_with("new003", for_destroy=True)
|
||||
assert created and created[0].close.call_count == 1
|
||||
|
||||
|
||||
@ -1761,14 +1942,15 @@ def test_reclaim_drops_a_container_a_peer_is_destroying():
|
||||
worker_a._backend.destroy.assert_not_called()
|
||||
|
||||
|
||||
def test_created_sandbox_is_rolled_back_when_a_peer_is_destroying_its_id():
|
||||
"""Rollback must cover a teardown marker, not just a store outage.
|
||||
def test_created_sandbox_is_not_rolled_back_when_a_peer_is_destroying_its_id():
|
||||
"""A failed registration must not bypass a peer's teardown marker.
|
||||
|
||||
`test_ownership_rollback_on_create_closes_the_client_it_drops` drives this
|
||||
path with `OwnershipBackendError` only. The comment says the teardown case is
|
||||
reachable too — a peer that died mid-stop leaves a `del:` marker until its
|
||||
TTL lapses — and without rollback the container we just started is leaked as
|
||||
an adoptable orphan.
|
||||
path with `OwnershipBackendError` only. The teardown case is reachable too:
|
||||
a peer that died mid-stop leaves a `del:` marker until its TTL lapses. A
|
||||
direct backend rollback here would ignore that ownership verdict and can
|
||||
remove resources the peer still owns; leave the untracked container for
|
||||
ownership-fenced reconciliation after the marker expires instead.
|
||||
"""
|
||||
shared = _make_shared_ownership_store()
|
||||
worker_a = _make_provider_for_reconciliation(worker_id="worker-a", store=shared)
|
||||
@ -1785,8 +1967,9 @@ def test_created_sandbox_is_rolled_back_when_a_peer_is_destroying_its_id():
|
||||
with pytest.raises(SandboxBeingDestroyedError):
|
||||
worker_a._register_created_sandbox("t1", "fresh01", info, user_id="u1")
|
||||
|
||||
worker_a._backend.destroy.assert_called_once_with(info)
|
||||
assert "fresh01" not in worker_a._sandboxes, "a container we could not own was handed out anyway"
|
||||
worker_a._backend.destroy.assert_not_called()
|
||||
assert "fresh01" not in worker_a._sandboxes
|
||||
assert shared.owner("fresh01") == "worker-b"
|
||||
|
||||
|
||||
def test_shutdown_does_not_stop_a_peers_warm_container():
|
||||
|
||||
@ -20,7 +20,7 @@
|
||||
# ============================================================================
|
||||
# Bump this number when the config schema changes.
|
||||
# Run `make config-upgrade` to merge new fields into your local config.yaml.
|
||||
config_version: 39
|
||||
config_version: 40
|
||||
|
||||
# ============================================================================
|
||||
# Logging
|
||||
@ -1338,7 +1338,8 @@ sandbox:
|
||||
|
||||
# Option 2: Container-based AIO Sandbox
|
||||
# Executes commands in isolated containers (Docker or Apple Container)
|
||||
# On macOS: Automatically prefers Apple Container if available, falls back to Docker
|
||||
# On macOS: Prefers Apple Container when available, but keeps Docker while this
|
||||
# sandbox prefix still has managed Docker containers to reconcile after a mode change
|
||||
# On other platforms: Uses Docker
|
||||
# Uncomment to use:
|
||||
# sandbox:
|
||||
@ -1365,6 +1366,23 @@ sandbox:
|
||||
# # Optional: Prefix for container names (default: deer-flow-sandbox)
|
||||
# # container_prefix: deer-flow-sandbox
|
||||
#
|
||||
# # Optional: outbound network policy for locally managed Docker sandboxes.
|
||||
# # open keeps the current behavior. isolated denies all outbound traffic.
|
||||
# # allowlist permits the listed HTTP/HTTPS domains; denied public domains can
|
||||
# # optionally be approved from the Human Input card. Restricted modes require
|
||||
# # Docker Engine 28+ and are not supported by Apple Container or provisioner mode.
|
||||
# # Private, loopback, link-local, multicast, and cloud metadata addresses are always denied.
|
||||
# network:
|
||||
# mode: allowlist # open | isolated | allowlist (default: open)
|
||||
# allow_domains:
|
||||
# - pypi.org
|
||||
# - files.pythonhosted.org
|
||||
# - registry.npmjs.org
|
||||
# - github.com
|
||||
# approval: prompt # deny | prompt
|
||||
# temporary_grant_ttl: 300 # 30-3600 seconds
|
||||
# proxy_image: ghcr.io/bytedance/deer-flow-sandbox-network-proxy:latest
|
||||
#
|
||||
# # Optional: Override whether the sandbox already sees the gateway's
|
||||
# # thread workspace/uploads/outputs through shared mounts.
|
||||
# # Omit this field to auto-detect from the backend (local containers: true;
|
||||
|
||||
@ -131,7 +131,7 @@ they resolve from the `secrets` map):
|
||||
|
||||
```yaml
|
||||
config: |
|
||||
config_version: 39
|
||||
config_version: 40
|
||||
models:
|
||||
- name: gpt-4
|
||||
use: langchain_openai:ChatOpenAI
|
||||
|
||||
@ -249,7 +249,7 @@ ingress:
|
||||
# -- DeerFlow config.yaml content. Secrets MUST stay as $VAR references — never
|
||||
# inline literal secret values here. The default enables provisioner sandbox.
|
||||
config: |
|
||||
config_version: 39
|
||||
config_version: 40
|
||||
log_level: info
|
||||
|
||||
models: []
|
||||
|
||||
11
docker/sandbox-network-proxy/Dockerfile
Normal file
11
docker/sandbox-network-proxy/Dockerfile
Normal file
@ -0,0 +1,11 @@
|
||||
ARG BASE_IMAGE=python:3.12-alpine
|
||||
FROM ${BASE_IMAGE}
|
||||
|
||||
# BASE_IMAGE is overridable for offline smoke tests. Production uses the small
|
||||
# official Python Alpine image; a cached Alpine-derived base can install Python.
|
||||
RUN if ! command -v python >/dev/null 2>&1; then apk add --no-cache python3 && ln -sf /usr/bin/python3 /usr/bin/python; fi
|
||||
|
||||
COPY backend/packages/harness/deerflow/community/aio_sandbox/network_proxy.py /opt/deerflow-network-proxy.py
|
||||
|
||||
USER 65532:65532
|
||||
CMD ["python", "/opt/deerflow-network-proxy.py", "serve"]
|
||||
@ -45,6 +45,37 @@ test("extractHumanInputRequest reads a valid tool artifact payload", () => {
|
||||
expect(extractHumanInputRequest(message)).toEqual(requestPayload);
|
||||
});
|
||||
|
||||
test("extractHumanInputRequest accepts sandbox network single-choice approvals", () => {
|
||||
const networkRequest = {
|
||||
...requestPayload,
|
||||
source: "sandbox_network",
|
||||
request_id: "network-request-1",
|
||||
title: "Sandbox network access",
|
||||
input_mode: "single_choice",
|
||||
options: [
|
||||
{ id: "deny", label: "Deny", value: "Deny network access" },
|
||||
{
|
||||
id: "allow_temporary",
|
||||
label: "Allow for 5 minutes",
|
||||
value: "Allow network access for 5 minutes",
|
||||
},
|
||||
{
|
||||
id: "allow_sandbox",
|
||||
label: "Allow for this sandbox",
|
||||
value: "Allow network access for this sandbox",
|
||||
},
|
||||
],
|
||||
};
|
||||
const message = {
|
||||
type: "tool",
|
||||
name: "bash",
|
||||
content: "network policy denied pypi.org:443",
|
||||
artifact: { human_input: networkRequest },
|
||||
} as unknown as Message;
|
||||
|
||||
expect(extractHumanInputRequest(message)).toEqual(networkRequest);
|
||||
});
|
||||
|
||||
test("extractHumanInputRequest rejects malformed artifacts", () => {
|
||||
const message = {
|
||||
type: "tool",
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user