diff --git a/doc/source/cluster/kubernetes/examples/ray-sandboxing.md b/doc/source/cluster/kubernetes/examples/ray-sandboxing.md index ab66427736a3..3f9d9f250957 100644 --- a/doc/source/cluster/kubernetes/examples/ray-sandboxing.md +++ b/doc/source/cluster/kubernetes/examples/ray-sandboxing.md @@ -146,6 +146,8 @@ res = ray.get(sb.exec.remote("python3 -c 'import urllib.request; urllib.request. print(res.exit_code) # Non-zero exit code ``` +With `network="public"`, each sandbox gets internet egress from its own private network namespace, bridged by [pasta](https://passt.top): sandboxes can bind the same port concurrently without conflicting, and can't reach each other, the Pod, or internal cluster services. This mode needs the `pasta` binary on the worker's `$PATH` and `/dev/net/tun` available in the Ray container (present on standard GKE `containerd` node pools). + --- ## (Optional) Step 5: Build a custom Ray image with pre-installed `runsc` @@ -161,11 +163,14 @@ FROM rayproject/ray:2.58.0-py312 USER root -# Install wget, download gVisor runsc, and install into system PATH +# Install wget, download gVisor runsc and pasta (for network="public"), +# and install both into the system PATH. passt.top static builds are +# x86_64-only; on arm64 nodes install the distro's passt package instead. RUN apt-get update && apt-get install -y --no-install-recommends wget && \ ARCH=$(uname -m) && \ wget "https://storage.googleapis.com/gvisor/releases/release/latest/${ARCH}/runsc" -O /usr/local/bin/runsc && \ - chmod a+rx /usr/local/bin/runsc && \ + wget "https://passt.top/builds/latest/x86_64/pasta" -O /usr/local/bin/pasta && \ + chmod a+rx /usr/local/bin/runsc /usr/local/bin/pasta && \ rm -rf /var/lib/apt/lists/* USER ray diff --git a/doc/source/ray-core/sandboxes.md b/doc/source/ray-core/sandboxes.md index 6dcbc5929e7c..11a8ecf09ba2 100644 --- a/doc/source/ray-core/sandboxes.md +++ b/doc/source/ray-core/sandboxes.md @@ -37,8 +37,9 @@ Ray Sandboxes need the following on every Ray node that runs a sandbox: * **Linux**: x86_64 or arm64. * **gVisor (`runsc`)**: Install the `runsc` binary on worker nodes and make it reachable from the system `$PATH`. * **Ray**: version 2.58.0 or later, which includes the `ray.experimental.sandbox` package. +* **pasta (`network="public"` only)**: The [passt](https://passt.top) package's `pasta` binary on the `$PATH`, plus `/dev/net/tun` in the worker's environment. pasta bridges each sandbox's private network namespace to the node. -To install `runsc` on a Linux worker node, see the [gVisor installation guide](https://gvisor.dev/docs/user_guide/install/). +To install `runsc` on a Linux worker node, see the [gVisor installation guide](https://gvisor.dev/docs/user_guide/install/). `pasta` ships as the `passt` package on Debian 12+/Ubuntu 23.04+ and Fedora, or as a [static build](https://passt.top/builds/latest/) (x86_64 only; on arm64 use the distro package or build from source — passt has no build dependencies). ## Usage patterns and examples @@ -268,7 +269,7 @@ Sandboxes support four network modes. The default is `none`, which follows the s | Mode | Network access | `/etc/resolv.conf` | Security property | | --- | --- | --- | --- | | `none` *(default)* | None | untouched | No egress. | -| `public` | Host egress | Generated from `dns` (default `8.8.8.8`, `1.1.1.1`), mounted read-only | Egress works, but the sandbox inherits nothing from the host's resolver configuration. No internal search domains, resolver addresses, or `ndots` options leak in, and the sandbox config stays portable across clusters. | +| `public` | Internet egress from a network namespace private to the sandbox, bridged by [pasta](https://passt.top) | Generated from `dns` (default `8.8.8.8`, `1.1.1.1`), mounted read-only | Ports and loopback are per-sandbox: a bind on `0.0.0.0` can't collide with, be reached by, or reach other sandboxes or node-local services, and there's no inbound path from the node or cluster. The sandbox also inherits nothing from the host's resolver configuration — no internal search domains, resolver addresses, or `ndots` options leak in, and the sandbox config stays portable across clusters. Requires `pasta` on the node; setting `RAY_SANDBOX_PUBLIC_HOST_NETNS=1` on workers reverts `public` to the worker's shared namespace. | | `host` | Full host network identity | Host's own file, mounted read-only (`dns=` overrides it) | Strictly more permissive than `public`. The sandbox can reach anything the node can reach, including internal networks and node-local services. Use `public` for untrusted code. | | `sandbox` | gVisor netstack | untouched | Requires `rootless=False`. runsc doesn't support the sandbox netstack in rootless mode. | @@ -347,6 +348,113 @@ Ray Sandboxes implement multi-layered defense-in-depth isolation: * **Network containment**: By default, `network="none"` disables all outbound network interfaces, which prevents untrusted code from making external API calls or scanning the internal cluster network. When internet access is needed, `network="public"` grants egress without handing over the host's resolver configuration or network identity; see [Networking and DNS](#networking-and-dns). * **Resource quotas**: cgroups enforce CPU quotas and memory limits, which prevents CPU starvation and out-of-memory (OOM) conditions from affecting other Ray actors. +## HTTP API service + +Ray Sandbox ships an experimental REST API service so you can manage sandboxes from outside the Ray cluster with nothing but an HTTP client and a bearer token. The service is a FastAPI app on Ray Serve (`ray.experimental.sandbox.http`). Each sandbox is held by a named, detached actor, so the service itself is stateless and its replicas can scale or restart without losing sandboxes. + +Image pulls and commands can far outlive an HTTP request and the load balancer in front of a deployed service, so creation and execution are asynchronous. `POST` returns immediately and clients poll, optionally long-polling with `wait_seconds` for up to 30 seconds per request. + +### Endpoints + +All endpoints sit under `/api/v1`. Except for `GET /health`, they require `Authorization: Bearer ` when a token is configured. + +| Method and path | Description | +| --- | --- | +| `GET /health` | Liveness probe; never requires auth. | +| `POST /sandboxes` | Create a sandbox. Returns `202` with `status: pending`. Poll until `running` or `error`. Send a `client_token` to make creation idempotent, so a retry returns `200` with the existing sandbox. | +| `GET /sandboxes?label=k=v` | List sandboxes, optionally filtered by labels. | +| `GET /sandboxes/{id}?wait_seconds=N` | Sandbox status; long-polls while it boots. | +| `DELETE /sandboxes/{id}` | Terminate the sandbox and its actor. Idempotent from any state. | +| `POST /sandboxes/{id}/execs` | Start a command. Returns `202` with an `exec_id`, or `409` while the sandbox isn't running. A string command runs under the sandbox's shell, `/bin/bash` by default and configurable per sandbox and per exec via `shell`. A list runs argv-style. | +| `GET /sandboxes/{id}/execs/{exec_id}?wait_seconds=N` | Exec status and result: `running`, `completed` with `exit_code`, `stdout`, and `stderr`, `timeout`, or `error`. Output is capped per stream by `max_output_bytes` with a loud truncation marker. | +| `PUT /sandboxes/{id}/files?path=/abs/path` | Write the raw request body to a file in the sandbox. Returns `413` above `max_file_bytes`. Pass `append=true` to extend the file, which lets clients chunk large uploads under proxy body-size limits. | +| `GET /sandboxes/{id}/files?path=/abs/path` | Read a file from the sandbox as `application/octet-stream`. | + +Errors use a JSON envelope of the form `{"error": {"code": "...", "message": "..."}}`. The codes are `401 unauthorized`, `404 sandbox_not_found`, `404 exec_not_found`, `404 file_not_found`, `409 conflict`, `409 unschedulable`, `400 invalid_request`, `413 payload_too_large`, and FastAPI's native `422` for schema violations. The full OpenAPI schema is served at `/openapi.json`. + +Server behavior worth knowing: + +* **TTL**: Every sandbox gets a TTL that reclaims both the sandbox and its hosting actor. Request it with `ttl_seconds`, capped and defaulted by the server's `max_ttl_seconds`. +* **Resources**: `resources` separates cluster reservations from in-sandbox cgroup caps. `cpu_request`, `memory_request_mb`, and `custom` Ray resources reserve cluster capacity, and custom resources such as `{"gvisor": 1}` pin sandboxes to runsc-equipped nodes. `cpu_limit` and `memory_limit_mb` become cgroup caps. Requests default to the limits. +* **Capabilities**: By default sandboxes get Docker's default Linux capability set so images behave the way they do under Docker. Ray's own default is far narrower and breaks `apt-get` and `tar`. The sets are written exactly, so `capabilities: []` runs the sandbox with no capabilities at all. +* **Network modes**: The modes are the Python API's, validated by Ray: `none` (the default), `public` for egress with generated DNS that `dns` overrides, `host`, and `sandbox`. See [Networking and DNS](#networking-and-dns). + +### Self-hosted quickstart + +On a Linux machine or cluster with `runsc` on `PATH`: + +```bash +pip install "ray[serve]" +export RAY_SANDBOX_API_TOKEN=dev-token # Optional. Unset disables app-level auth. +serve run ray.experimental.sandbox.http.app:build_app +``` + +```bash +curl -s -H "Authorization: Bearer dev-token" \ + -H "Content-Type: application/json" \ + -d '{"image": "busybox:latest", "readonly": false, "shell": "/bin/sh"}' \ + http://localhost:8000/api/v1/sandboxes +``` + +Builder arguments configure the server. See `ray.experimental.sandbox.http.schemas.SandboxAPISettings` for the full list. For example: + +```bash +serve run ray.experimental.sandbox.http.app:build_app max_ttl_seconds=86400 num_replicas=2 +``` + +### Deploying as an Anyscale service + +Build a cluster image whose worker nodes have `runsc`: + +```dockerfile +FROM anyscale/ray:2.58.0-py312 +RUN ARCH=$(uname -m | sed 's/arm64/aarch64/') && \ + curl -fsSL -o /usr/local/bin/runsc \ + "https://storage.googleapis.com/gvisor/releases/release/latest/${ARCH}/runsc" && \ + curl -fsSL -o /usr/local/bin/pasta \ + "https://passt.top/builds/latest/x86_64/pasta" && \ + chmod +x /usr/local/bin/runsc /usr/local/bin/pasta +``` + +The `pasta` binary is only needed for `network="public"`; passt.top publishes static builds for x86_64 only, so on arm64 nodes install the distro's `passt` package instead. + +Then deploy the builder as the service's application: + +```yaml +# service.yaml +name: ray-sandbox-api +image_uri: /ray-sandbox-api:latest +applications: + - name: sandbox-api + import_path: ray.experimental.sandbox.http.app:build_app + args: + max_ttl_seconds: 86400 +``` + +```bash +anyscale service deploy -f service.yaml +``` + +Anyscale services require their own bearer token at the platform edge, so leave `RAY_SANDBOX_API_TOKEN` unset and hand clients the service's base URL and token. Consumers such as the [Harbor](https://harborframework.com) `ray-sandbox` environment take exactly that pair as `RAY_SANDBOX_API_URL` and `RAY_SANDBOX_API_KEY`. + +### Local development loop on macOS + +`runsc` is Linux-only. Develop against the service in a privileged container: + +```bash +docker run --privileged -p 8000:8000 \ + -v ~/path/to/ray/python/ray/experimental/sandbox:/overlay:ro \ + rayproject/ray:nightly-py312 bash -lc ' + pip install "ray[serve]" && + SITE=$(python -c "import ray, os; print(os.path.dirname(ray.__file__))") && + cp -r /overlay/* "$SITE/experimental/sandbox/" && + ARCH=$(uname -m | sed "s/arm64/aarch64/") && + curl -fsSL -o /usr/local/bin/runsc "https://storage.googleapis.com/gvisor/releases/release/latest/${ARCH}/runsc" && + { curl -fsSL -o /usr/local/bin/pasta "https://passt.top/builds/latest/x86_64/pasta" || sudo apt-get install -y passt; } && + chmod +x /usr/local/bin/runsc /usr/local/bin/pasta 2>/dev/null; + RAY_SANDBOX_API_TOKEN=dev-token serve run --host 0.0.0.0 ray.experimental.sandbox.http.app:build_app' +``` + ## API reference For detailed signatures, parameters, and return types, see {ref}`ray-sandbox-ref`. @@ -356,6 +464,8 @@ For detailed signatures, parameters, and return types, see {ref}`ray-sandbox-ref * **`runsc` not found in `$PATH`**: Verify that gVisor's `runsc` binary is installed on all Ray worker nodes and sits in a directory on the system `$PATH`, such as `/usr/local/bin/runsc`. * **cgroup or permission errors**: In containerized environments such as Kubernetes without root permissions, keep the default `rootless=True`. Where cgroups are restricted, set `RAY_SANDBOX_IGNORE_CGROUPS=1`. * **Image pull failures**: Verify that the node can reach the container registry, such as Docker Hub or GHCR, or pre-populate the image cache directory at `/tmp/ray/sandbox/images`. When many nodes pull large images at once, Docker Hub's anonymous rate limits are a likely cause; see [Route Docker Hub pulls through a mirror](#route-docker-hub-pulls-through-a-mirror). +* **`pasta` not found for `network="public"`**: Install the passt package (or a [static build](https://passt.top/builds/latest/)) on worker nodes, enable the HTTP API's `auto_install_pasta` setting, or set `RAY_SANDBOX_PUBLIC_HOST_NETNS=1` on workers to run `public` sandboxes in the worker's shared network namespace (the pre-namespace behavior, without port isolation). +* **`public` sandboxes fail to start with a tap or namespace error**: pasta needs `/dev/net/tun` in the worker's environment and a seccomp policy that allows unprivileged user+network namespace creation (`unshare -Un true` must succeed as the Ray user). The pasta error appears in the sandbox's `runsc.stderr.log` and in the creation error message; fall back to the kill switch or `network="host"` until the node environment provides both. ## Next steps diff --git a/python/ray/experimental/sandbox/backend/base.py b/python/ray/experimental/sandbox/backend/base.py index 9eb1e52020fc..ff66fe363f3c 100644 --- a/python/ray/experimental/sandbox/backend/base.py +++ b/python/ray/experimental/sandbox/backend/base.py @@ -107,7 +107,11 @@ def exec_command( @abstractmethod def write_file( - self, sandbox_id: str, path: str, content: Union[str, bytes] + self, + sandbox_id: str, + path: str, + content: Union[str, bytes], + append: bool = False, ) -> None: """Write content to a file inside the sandbox. @@ -115,6 +119,7 @@ def write_file( sandbox_id: Unique string identifier of the sandbox. path: Target file path inside the sandbox environment. content: Text string or raw bytes to write. + append: Append to the file instead of truncating it. """ pass diff --git a/python/ray/experimental/sandbox/backend/gvisor.py b/python/ray/experimental/sandbox/backend/gvisor.py index 442808799c32..89e0541cb4ef 100644 --- a/python/ray/experimental/sandbox/backend/gvisor.py +++ b/python/ray/experimental/sandbox/backend/gvisor.py @@ -1,10 +1,13 @@ import json import logging import os +import shlex import shutil +import signal import subprocess import time import uuid +from pathlib import Path from typing import Callable, Dict, List, Optional, Union from ray.experimental.sandbox.backend.base import ( @@ -31,6 +34,48 @@ # Directory to store sandbox states, container images and overlay filesystem. _RAY_SANDBOX_DIR = "/tmp/ray/sandbox" +# network="public" gives each sandbox a private user+net namespace pair, +# bridged by pasta (passt) user-mode networking. Topology (the rootless +# Podman shape): a tiny holder process unshares the namespaces and sleeps; +# pasta, launched from the pod side so its uplink is the pod's real +# interface, attaches to the holder's namespaces and daemonizes; runsc runs +# inside via nsenter, mapped to root in the user namespace (so no +# --rootless — runsc would nest a second user namespace whose /proc +# magic-link derefs the gofer cannot perform). runsc still gets +# --network=host, but "host" is now private to this sandbox: a bind on +# 0.0.0.0 cannot collide with or be reached by the pod or other sandboxes, +# while egress flows out through pasta's tap. Mount and pid namespaces stay +# shared, so the bundle and the runsc control sockets under _RUNSC_ROOT +# keep working for pod-side state/exec/kill/delete. The namespaces are +# anonymous — held by the process group, freed with it. +# These flags are the isolation property; tests pin the exact list: +# --config-net copy the pod interface's addressing/routes onto the tap. +# -t/-u none never republish namespace binds on the pod (the "auto" +# default would recreate the cross-sandbox collision). +# -T/-U none no loopback splicing: pod-local services stay +# unreachable from the sandbox's 127.0.0.1. +# --no-map-gw don't remap gateway-addressed traffic to the pod +# loopback (closes the remaining sandbox->pod path). +# -4 IPv4 only, matching the generated resolv.conf. +_PASTA_FLAGS = [ + "--config-net", + "-t", + "none", + "-u", + "none", + "-T", + "none", + "-U", + "none", + "--no-map-gw", + "-4", +] + +# Kill switch: set to "1" on workers to run network="public" sandboxes in +# the worker's own network namespace (the pre-pasta behavior, where binds +# are shared across sandboxes) without a code deploy. +_PUBLIC_HOST_NETNS_ENV = "RAY_SANDBOX_PUBLIC_HOST_NETNS" + class GVisorSandboxBackend(BaseSandboxBackend): """gVisor sandbox backend running a single persistent container instance per sandbox locally via runsc.""" @@ -46,6 +91,19 @@ def create_sandbox(self, config: SandboxConfig) -> str: "gVisor executable 'runsc' not found in PATH. " "Please install gVisor (runsc) on the node." ) + use_pasta = self._uses_pasta_netns(config) + if use_pasta: + missing = [b for b in ("pasta", "nsenter") if not shutil.which(b)] + if missing: + raise SandboxCreationError( + "network='public' isolates each sandbox in its own " + "network namespace via pasta (passt), but " + f"{', '.join(repr(b) for b in missing)} was not found in " + "PATH. Install the passt package (and util-linux) on the " + "node image, enable the auto_install_pasta server " + f"setting, or set {_PUBLIC_HOST_NETNS_ENV}=1 on workers " + "to restore the previous shared-host-network behavior." + ) sandbox_uuid = uuid.uuid4().hex[:12] sandbox_id = f"ray-sandbox-{sandbox_uuid}" @@ -100,28 +158,25 @@ def create_sandbox(self, config: SandboxConfig) -> str: dns=config.dns, _oci_spec_transform_fn=config._oci_spec_transform_fn, ) - run_args = self._runsc_base_args(config) - if config.network: - # "public" = host egress + generated resolv.conf (handled in the - # OCI bundle); runsc itself just sees host networking. - runsc_network = "host" if config.network == "public" else config.network - run_args.extend(["--network", runsc_network]) overlay_dir = os.path.join(root_dir, "overlay") os.makedirs(overlay_dir, mode=0o777, exist_ok=True) - run_args.append(f"--overlay2=root:dir={overlay_dir}") - run_args.extend(["run", "--bundle", root_dir, sandbox_id]) + run_args = self._build_run_command(config, root_dir, overlay_dir, sandbox_id) stderr_log_path = os.path.join(root_dir, "runsc.stderr.log") stderr_file = open(stderr_log_path, "w+", encoding="utf-8") + # start_new_session puts the namespace holder, pasta, and runsc run + # in one process group so cleanup can kill the whole tree; they share + # the stderr log so startup failures (missing /dev/net/tun, no + # uplink) surface through the SandboxCreationError path below. proc = subprocess.Popen( run_args, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=stderr_file, + start_new_session=True, ) start_time = time.time() timeout = config.timeout_seconds - state_args = self._runsc_base_args(config) + ["state", sandbox_id] try: while True: @@ -132,6 +187,7 @@ def create_sandbox(self, config: SandboxConfig) -> str: f"gVisor container failed to start: {stderr_str}" ) + state_args = self._runsc_base_args(config) + ["state", sandbox_id] res = subprocess.run(state_args, capture_output=True, text=True) if res.returncode == 0: try: @@ -148,24 +204,23 @@ def create_sandbox(self, config: SandboxConfig) -> str: raise pass + # Check the deadline only after polling state, so a sandbox + # that reached 'running' just as the deadline passed is + # observed and kept rather than torn down unpolled. if time.time() - start_time > timeout: - proc.kill() - proc.communicate() raise SandboxTimeoutError( f"gVisor container '{sandbox_id}' failed to reach 'running' state within {timeout} seconds." ) time.sleep(0.1) except Exception: - if proc and proc.poll() is None: - proc.kill() - try: - proc.communicate(timeout=2) - except subprocess.TimeoutExpired: - pass + # Delete runsc's container state, then kill the whole group: + # under pasta, a bare proc.kill() would orphan the namespace + # holder and the pasta daemon. + self._delete_container_state(config, sandbox_id) + self._terminate_tree(proc) + self._kill_pasta(root_dir) stderr_file.close() - del_args = self._runsc_base_args(config) + ["delete", sandbox_id] - subprocess.run(del_args, capture_output=True) shutil.rmtree(root_dir, ignore_errors=True) raise @@ -174,6 +229,8 @@ def create_sandbox(self, config: SandboxConfig) -> str: "workdir": workdir_path, "cwd": container_cwd, "config": config, + # The process group leader whose tree holds the sandbox — and, + # for network="public", the namespace holder and pasta daemon. "proc": proc, "stderr_file": stderr_file, "status": SandboxStatus.RUNNING, @@ -196,12 +253,17 @@ def delete_sandbox(self, sandbox_id: str) -> None: except subprocess.TimeoutExpired: pass - if proc and proc.poll() is None: - proc.terminate() - try: - proc.communicate(timeout=2) - except subprocess.TimeoutExpired: - proc.kill() + self._delete_container_state(config, sandbox_id) + + # Always take the whole group: after `runsc run` exits, the + # namespace holder and pasta daemon (network="public") are + # still alive in it. + if proc: + self._terminate_tree(proc) + + # pasta daemonizes out of that group, so reap it by its + # recorded pid (no-op when no pasta ran). + self._kill_pasta(root_dir) if stderr_file: try: @@ -209,12 +271,47 @@ def delete_sandbox(self, sandbox_id: str) -> None: except Exception: pass - del_args = self._runsc_base_args(config) - del_args.extend(["delete", sandbox_id]) - subprocess.run(del_args, capture_output=True) - shutil.rmtree(root_dir, ignore_errors=True) + def _resolve_exec_user(self, user: str, image: str) -> str: + """Turn a user name or uid[:gid] into runsc exec's numeric form. + + runsc only accepts numeric ids; names are resolved against the + image's own /etc/passwd (and the login group via /etc/group). + + Args: + user: Numeric uid, "uid:gid", or a user name from the image. + image: The sandbox's image, locating the extracted rootfs. + + Returns: + A "uid" or "uid:gid" string runsc accepts. + + Raises: + SandboxExecError: When a named user is not in the image's + /etc/passwd. + """ + head, sep, specified_gid = user.partition(":") + if head.isdigit(): + return user + rootfs = os.path.join(self._image_manager.get_image_dir(image), "rootfs") + try: + passwd = Path(os.path.join(rootfs, "etc", "passwd")).read_text( + encoding="utf-8", errors="replace" + ) + except OSError: + passwd = "" + for line in passwd.splitlines(): + parts = line.split(":") + # Match on the name alone; when the caller wrote "name:gid", + # honor the explicit gid instead of the passwd login group. + if len(parts) >= 4 and parts[0] == head: + gid = specified_gid if sep and specified_gid else parts[3] + return f"{parts[2]}:{gid}" + raise SandboxExecError( + f"user {user!r} not found in the image's /etc/passwd; " + "pass a numeric uid or uid:gid instead" + ) + def exec_command( self, sandbox_id: str, @@ -223,6 +320,7 @@ def exec_command( cwd: Optional[str] = None, env: Optional[Dict[str, str]] = None, shell: Optional[str] = None, + user: Optional[str] = None, ) -> ExecResult: """Execute a process inside the running gVisor sandbox instance via runsc exec.""" meta = self._get_metadata_or_raise(sandbox_id) @@ -237,6 +335,8 @@ def exec_command( # Production execution against running container via `runsc exec` runsc_args = self._runsc_base_args(config) runsc_args.extend(["exec", "-cwd", exec_cwd]) + if user is not None: + runsc_args.extend(["-user", self._resolve_exec_user(user, config.image)]) if env: for k, v in env.items(): runsc_args.extend(["-env", f"{k}={v}"]) @@ -278,9 +378,13 @@ def exec_command( raise SandboxExecError(f"gVisor exec failed: {err}") from err def write_file( - self, sandbox_id: str, path: str, content: Union[str, bytes] + self, + sandbox_id: str, + path: str, + content: Union[str, bytes], + append: bool = False, ) -> None: - """Write content to a file inside the local gVisor sandbox directory.""" + """Write (or append) content to a file inside the sandbox.""" meta = self._get_metadata_or_raise(sandbox_id) config: SandboxConfig = meta["config"] @@ -294,7 +398,9 @@ def write_file( sandbox_id, "/bin/sh", "-c", - 'mkdir -p "$(dirname "$1")" && cat > "$1"', + 'mkdir -p "$(dirname "$1")" && cat >> "$1"' + if append + else 'mkdir -p "$(dirname "$1")" && cat > "$1"', "--", path, ] @@ -357,6 +463,111 @@ def _runsc_base_args(self, config: SandboxConfig) -> List[str]: args.extend(["--root", _RUNSC_ROOT]) return args + def _uses_pasta_netns(self, config: SandboxConfig) -> bool: + """Whether this sandbox gets a private network namespace via pasta.""" + return ( + config.network == "public" and os.environ.get(_PUBLIC_HOST_NETNS_ENV) != "1" + ) + + def _delete_container_state(self, config: SandboxConfig, sandbox_id: str) -> None: + """Best-effort ``runsc delete -force`` for teardown paths. + + Bounded by a timeout: a wedged gVisor (the usual reason a create + timed out) must not block the process-group kill and pasta reap + that follow, which is what actually frees the sandbox. + """ + del_args = self._runsc_base_args(config) + ["delete", "-force", sandbox_id] + try: + subprocess.run(del_args, capture_output=True, timeout=10) + except subprocess.TimeoutExpired: + pass + + def _build_run_command( + self, config: SandboxConfig, root_dir: str, overlay_dir: str, sandbox_id: str + ) -> List[str]: + """Build the full `runsc run` argv, pasta-wrapped for network="public". + + Pure argv construction (no filesystem side effects) so tests can + assert the exact command without runsc or pasta installed. + """ + args = self._runsc_base_args(config) + use_pasta = self._uses_pasta_netns(config) + if use_pasta: + # runsc runs as mapped root inside the holder's user namespace; + # --rootless would nest a second user namespace whose + # /proc//root magic links the gofer cannot dereference. + args = [a for a in args if a != "--rootless"] + if config.network: + # "public" = host egress + generated resolv.conf (handled in the + # OCI bundle); runsc itself just sees host networking — of the + # per-sandbox namespace when wrapped, of the worker otherwise. + runsc_network = "host" if config.network == "public" else config.network + args.extend(["--network", runsc_network]) + args.append(f"--overlay2=root:dir={overlay_dir}") + args.extend(["run", "--bundle", root_dir, sandbox_id]) + if use_pasta: + pidfile = shlex.quote(os.path.join(root_dir, "netns.pid")) + pasta_pidfile = shlex.quote(os.path.join(root_dir, "pasta.pid")) + runsc = " ".join(shlex.quote(a) for a in args) + pasta = " ".join(["pasta", *_PASTA_FLAGS]) + script = ( + # The holder pins the namespaces for the sandbox's lifetime; + # --kill-child ties it to this script's process group. + "unshare --user --map-root-user --net --fork --kill-child " + f"bash -c 'echo $$ > {pidfile}; exec sleep infinity' & " + "HOLDER=$!; " + # Stop waiting the moment the holder dies (permission/seccomp + # failure) instead of spinning the full 10s, then require a + # non-empty NSPID so a lost holder can't yield /proc//ns/net. + f"for i in $(seq 1 100); do [ -s {pidfile} ] && break; " + "kill -0 $HOLDER 2>/dev/null || break; sleep 0.1; done; " + f"NSPID=$(cat {pidfile} 2>/dev/null); " + '[ -n "$NSPID" ] || { echo "netns holder failed to start" >&2; exit 1; }; ' + # pasta runs from the pod side (its uplink is the pod's real + # interface), attaches to the holder's namespaces, and + # daemonizes; --pid records the daemon so teardown can reap + # it (it escapes _terminate_tree's process group via setsid). + f"{pasta} --netns /proc/$NSPID/ns/net --userns /proc/$NSPID/ns/user " + f"--pid {pasta_pidfile} && " + f"exec nsenter --preserve-credentials -U -n -t $NSPID -- {runsc}" + ) + return ["bash", "-c", script] + return args + + def _kill_pasta(self, root_dir: str) -> None: + """SIGKILL the pasta daemon recorded for this sandbox, if any. + + pasta daemonizes with ``setsid()``, so it leaves the process group + that :meth:`_terminate_tree` reaps and would otherwise leak on passt + builds that don't self-exit when the namespaces empty. Its pid is in + ``{root_dir}/pasta.pid`` (written via ``pasta --pid`` in + :meth:`_build_run_command`); a missing file means no pasta ran. + """ + try: + pid = int(Path(os.path.join(root_dir, "pasta.pid")).read_text().strip()) + except (OSError, ValueError): + return + try: + os.kill(pid, signal.SIGKILL) + except (ProcessLookupError, PermissionError): + pass + + def _terminate_tree(self, proc: subprocess.Popen) -> None: + """SIGKILL the sandbox process group and reap the Popen. + + The run Popen is started with ``start_new_session=True``, so its pid + is the group id for pasta, runsc run, and the sandbox process. + """ + try: + os.killpg(proc.pid, signal.SIGKILL) + except (ProcessLookupError, PermissionError): + if proc.poll() is None: + proc.kill() + try: + proc.communicate(timeout=2) + except (subprocess.TimeoutExpired, ValueError): + pass + def _resolve_path(self, root_dir: str, relative_or_abs_path: str) -> str: clean_path = relative_or_abs_path.lstrip("/") return os.path.join(root_dir, clean_path) diff --git a/python/ray/experimental/sandbox/config.py b/python/ray/experimental/sandbox/config.py index 42611300af0f..d6b45645a932 100644 --- a/python/ray/experimental/sandbox/config.py +++ b/python/ray/experimental/sandbox/config.py @@ -2,8 +2,11 @@ from dataclasses import dataclass, field from typing import Callable, Dict, List, Optional, Union -# Sandbox network modes. All but "public" map directly to runsc --network; -# "public" is host egress plus a generated, host-independent resolv.conf. +# Sandbox network modes. "none", "host", and "sandbox" map directly to runsc +# --network; "public" runs runsc with host networking inside a per-sandbox +# network namespace bridged by pasta (internet egress only; ports and +# loopback are private to the sandbox) plus a generated, host-independent +# resolv.conf. VALID_NETWORK_MODES = ("none", "public", "host", "sandbox") # Default resolvers for network="public" (Google and Cloudflare public DNS). @@ -94,10 +97,17 @@ class SandboxConfig: timeout_seconds: Timeout in seconds for sandbox creation. rootless: If True, run gVisor in rootless mode (default: True). network: Network mode (default: "none" — no network access). - "public" (recommended for internet access) gives host egress with - a generated /etc/resolv.conf from ``dns``, inheriting nothing - from the host resolver. "host" gives full host network identity, - including the host's resolv.conf and internal networks. + "public" (recommended for internet access) gives internet egress + from a network namespace private to the sandbox, bridged by + pasta: ports and loopback are per-sandbox, nothing in the + sandbox is reachable from the host or from other sandboxes, and + /etc/resolv.conf is generated from ``dns``, inheriting nothing + from the host resolver. Requires the ``pasta`` binary on the + node; setting ``RAY_SANDBOX_PUBLIC_HOST_NETNS=1`` on workers + reverts "public" to the worker's shared namespace. "host" gives + full host network identity — the host's resolv.conf, internal + networks, and a port space shared with the worker and every + other host-mode sandbox. "sandbox" uses gVisor's netstack and requires ``rootless=False``. dns: Nameserver IPs for a generated /etc/resolv.conf, mounted read-only (like ``docker --dns``); useful when public DNS is diff --git a/python/ray/experimental/sandbox/http/BUILD.bazel b/python/ray/experimental/sandbox/http/BUILD.bazel new file mode 100644 index 000000000000..467628825c15 --- /dev/null +++ b/python/ray/experimental/sandbox/http/BUILD.bazel @@ -0,0 +1,44 @@ +load("@rules_python//python:defs.bzl", "py_library") +load("//bazel:python.bzl", "py_test_module_list") + +py_library( + name = "sandbox_http_lib", + srcs = glob( + ["**/*.py"], + exclude = ["tests/**/*.py"], + ), + visibility = [ + "//python/ray/experimental/sandbox/http:__pkg__", + "//python/ray/experimental/sandbox/http:__subpackages__", + ], + deps = [ + "//python/ray/experimental/sandbox:sandbox_lib", + ], +) + +py_library( + name = "conftest", + srcs = ["tests/conftest.py"], + deps = [ + "//python/ray/tests:conftest", + ], +) + +py_test_module_list( + size = "medium", + files = glob( + ["tests/**/*.py"], + exclude = [ + "tests/conftest.py", + "tests/__init__.py", + ], + ), + tags = [ + "sandbox_tests", + "team:core", + ], + deps = [ + ":conftest", + ":sandbox_http_lib", + ], +) diff --git a/python/ray/experimental/sandbox/http/__init__.py b/python/ray/experimental/sandbox/http/__init__.py new file mode 100644 index 000000000000..87ee7fba319d --- /dev/null +++ b/python/ray/experimental/sandbox/http/__init__.py @@ -0,0 +1,35 @@ +"""HTTP API service for Ray Sandbox. + +This subpackage exposes ``ray.experimental.sandbox`` over a versioned REST +API (``/api/v1``) served by Ray Serve, so sandboxes can be managed from +outside the Ray cluster with nothing but an HTTP client and a bearer token. + +Importing this package requires the Serve extra +(``pip install "ray[serve]"``); the base ``ray.experimental.sandbox`` +package deliberately never imports it. +""" + +from ray.experimental.sandbox.http.app import build_app, create_app +from ray.experimental.sandbox.http.host import SandboxHost +from ray.experimental.sandbox.http.schemas import ( + DOCKER_DEFAULT_CAPABILITIES, + CreateSandboxRequest, + ExecInfo, + ResourceSpec, + SandboxAPISettings, + SandboxInfo, + StartExecRequest, +) + +__all__ = [ + "build_app", + "create_app", + "SandboxHost", + "DOCKER_DEFAULT_CAPABILITIES", + "CreateSandboxRequest", + "ExecInfo", + "ResourceSpec", + "SandboxAPISettings", + "SandboxInfo", + "StartExecRequest", +] diff --git a/python/ray/experimental/sandbox/http/app.py b/python/ray/experimental/sandbox/http/app.py new file mode 100644 index 000000000000..aeafac7eba3c --- /dev/null +++ b/python/ray/experimental/sandbox/http/app.py @@ -0,0 +1,604 @@ +"""FastAPI application and Ray Serve builder for the Ray Sandbox HTTP API. + +Deploy on a Ray cluster whose worker nodes have gVisor's ``runsc`` on PATH: + + serve run ray.experimental.sandbox.http.app:build_app + +or as an Anyscale service (see ``doc/source/ray-core/sandboxes.md``). Bearer +auth is enforced when the environment variable named by +``SandboxAPISettings.token_env_var`` (default ``RAY_SANDBOX_API_TOKEN``) is +set; an Anyscale service can leave it unset because the platform edge already +requires the service's own bearer token. + +The service holds no state: every sandbox lives in a named detached +``SandboxHost`` actor and every replica resolves them by name, so replicas +can scale or restart freely. +""" + +import asyncio +import hashlib +import hmac +import logging +import os +import uuid +from datetime import datetime, timedelta, timezone +from typing import Any, Awaitable, Dict, List, Optional + +from fastapi import APIRouter, Depends, FastAPI, Query, Request, Response +from fastapi.responses import JSONResponse + +from ray.experimental.sandbox.http.host import SandboxHost +from ray.experimental.sandbox.http.schemas import ( + CreateSandboxRequest, + ExecInfo, + ExecStarted, + SandboxAPISettings, + SandboxInfo, + SandboxList, + StartExecRequest, +) +from ray.util.annotations import PublicAPI + +logger = logging.getLogger(__name__) + +SANDBOX_ID_PREFIX = "sb-" + +_MAX_WAIT_SECONDS = 30.0 + +_WAIT_QUERY = Query( + default=0.0, + ge=0.0, + le=_MAX_WAIT_SECONDS, + description="Long-poll for up to this many seconds for a state change.", +) + + +class _ApiError(Exception): + """Maps to the JSON error envelope; raised by handlers, caught app-wide.""" + + def __init__(self, status_code: int, code: str, message: str) -> None: + super().__init__(message) + self.status_code = status_code + self.code = code + self.message = message + + +def _sandbox_not_found(sandbox_id: str) -> _ApiError: + return _ApiError(404, "sandbox_not_found", f"no sandbox with id {sandbox_id!r}") + + +class _SchedulingTimeout(_ApiError): + """The sandbox's hosting actor has not been scheduled within the grace.""" + + def __init__(self, sandbox_id: str) -> None: + super().__init__( + 409, + "conflict", + f"sandbox {sandbox_id} is not ready yet (its actor is still " + "being scheduled); retry shortly", + ) + + +def _is_unschedulable(exc: BaseException) -> bool: + """True when Ray reports the actor can never fit the cluster's resources. + + Matched by class name for the same reason as ``_is_actor_gone``. + """ + names = {type(exc).__name__, *(base.__name__ for base in type(exc).__mro__)} + return any("ActorUnschedulableError" in name for name in names) + + +def _is_actor_gone(exc: BaseException) -> bool: + """True when a remote call failed because the actor no longer exists. + + Matched by class name because Ray re-raises remote failures as + dynamically-built subclasses, and so this module stays importable (and + unit-testable) without a live Ray context. + """ + names = {type(exc).__name__, *(base.__name__ for base in type(exc).__mro__)} + return any( + "RayActorError" in name + or "ActorDiedError" in name + or "ActorUnavailableError" in name + for name in names + ) + + +async def _actor_call(sandbox_id: str, awaitable: Awaitable[Any]) -> Any: + """Await a SandboxHost call, mapping a dead actor to 404.""" + try: + return await awaitable + except Exception as exc: + if _is_actor_gone(exc): + raise _sandbox_not_found(sandbox_id) from exc + if _is_unschedulable(exc): + # The requested cpu/memory shape cannot fit any node the cluster + # can offer: a permanent condition, not a scheduling wait. + raise _ApiError( + 409, + "unschedulable", + f"sandbox {sandbox_id} cannot be scheduled: {str(exc)[:300]}", + ) from exc + raise + + +def _sandbox_name_for_token(client_token: str) -> str: + digest = hashlib.sha256(client_token.encode("utf-8")).hexdigest() + return f"{SANDBOX_ID_PREFIX}{digest[:12]}" + + +def _parse_label_filters(labels: List[str]) -> Dict[str, str]: + filters: Dict[str, str] = {} + for item in labels: + key, sep, value = item.partition("=") + if not sep or not key: + raise _ApiError( + 400, + "invalid_request", + f"label filter {item!r} must have the form key=value", + ) + filters[key] = value + return filters + + +@PublicAPI(stability="alpha") +class RayActorHandleResolver: + """Creates and resolves the named detached SandboxHost actors. + + The default (and only production) resolver; tests inject a fake with the + same four methods so the HTTP layer runs without a Ray cluster. + """ + + def __init__(self, settings: SandboxAPISettings) -> None: + self._settings = settings + + def create( + self, + name: str, + actor_options: Dict[str, Any], + ctor_kwargs: Dict[str, Any], + ) -> Any: + import ray + + return ( + ray.remote(SandboxHost) + .options( + name=name, + namespace=self._settings.namespace, + lifetime="detached", + # Atomic get-or-create: two replicas racing on the same + # client_token converge on one actor. boot() is idempotent. + get_if_exists=True, + **actor_options, + ) + .remote(**ctor_kwargs) + ) + + def get(self, name: str) -> Optional[Any]: + import ray + + try: + return ray.get_actor(name, namespace=self._settings.namespace) + except ValueError: + return None + + def list_names(self) -> List[str]: + from ray.util import list_named_actors + + names: List[str] = [] + for entry in list_named_actors(all_namespaces=True): + if entry.get("namespace") == self._settings.namespace and entry.get( + "name", "" + ).startswith(SANDBOX_ID_PREFIX): + names.append(entry["name"]) + return names + + def kill(self, handle: Any) -> None: + import ray + + try: + ray.kill(handle) + except Exception as exc: + logger.debug("Failed to kill sandbox actor: %s", exc) + + +@PublicAPI(stability="alpha") +def create_app( + settings: Optional[SandboxAPISettings] = None, + *, + handle_resolver: Optional[Any] = None, +) -> FastAPI: + """Build the FastAPI app. + + Args: + settings: Server settings; defaults are production-safe. + handle_resolver: Test seam; anything with the + ``RayActorHandleResolver`` method surface. Defaults to the real + Ray-backed resolver. + + Returns: + The configured FastAPI application. + """ + settings = settings or SandboxAPISettings() + resolver = handle_resolver or RayActorHandleResolver(settings) + token = os.environ.get(settings.token_env_var) or None + + async def _bounded_call( + sandbox_id: str, awaitable: Awaitable[Any], extra_wait: float = 0.0 + ) -> Any: + """Await a SandboxHost call, but never block behind actor scheduling. + + A named detached actor exists the moment it is created, but on a + cluster that is scaling up it may not be *scheduled* for a while — + and a call to an unscheduled actor blocks indefinitely. Cap the wait + at the request's own long-poll budget plus a grace period. + """ + try: + return await asyncio.wait_for( + _actor_call(sandbox_id, awaitable), + timeout=extra_wait + settings.scheduling_grace_seconds, + ) + except asyncio.TimeoutError: + raise _SchedulingTimeout(sandbox_id) + + async def _describe_or_pending( + sandbox_id: str, handle: Any, wait_seconds: float = 0.0 + ) -> Dict[str, Any]: + try: + return await _bounded_call( + sandbox_id, + handle.describe.remote(wait_seconds=wait_seconds), + extra_wait=wait_seconds, + ) + except _SchedulingTimeout: + # The actor owns the spec, so report a shape-complete pending + # info with the fields only it knows left empty. + return { + "sandbox_id": sandbox_id, + "status": "pending", + "image": "", + "created_at": datetime.now(timezone.utc).isoformat(), + "ttl_seconds": None, + "expires_at": None, + "network": "none", + "labels": {}, + "error": None, + } + + async def require_bearer_token(request: Request) -> None: + if token is None: + return + provided = request.headers.get("authorization", "") + if not hmac.compare_digest(provided, f"Bearer {token}"): + raise _ApiError(401, "unauthorized", "invalid or missing bearer token") + + public = APIRouter(prefix="/api/v1") + v1 = APIRouter(prefix="/api/v1", dependencies=[Depends(require_bearer_token)]) + + @public.get("/health") + async def health() -> Dict[str, str]: + return {"status": "ok"} + + # ------------------------------------------------------------------ + # Sandboxes + # ------------------------------------------------------------------ + + @v1.post("/sandboxes", response_model=SandboxInfo, status_code=202) + async def create_sandbox(request: CreateSandboxRequest) -> JSONResponse: + if ( + request.ttl_seconds is not None + and request.ttl_seconds > settings.max_ttl_seconds + ): + raise _ApiError( + 400, + "invalid_request", + f"ttl_seconds may not exceed {settings.max_ttl_seconds}", + ) + # No TTL still gets the server-wide bound so abandoned sandboxes are + # always reclaimed eventually. + effective_ttl = ( + request.ttl_seconds + if request.ttl_seconds is not None + else settings.max_ttl_seconds + ) + + if request.client_token is not None: + sandbox_id = _sandbox_name_for_token(request.client_token) + existing = resolver.get(sandbox_id) + if existing is not None: + try: + info = await _describe_or_pending(sandbox_id, existing) + return JSONResponse(status_code=200, content=info) + except _ApiError as exc: + if exc.code != "sandbox_not_found": + raise + # The previous actor died (node loss, OOM). Clear it and + # fall through to create a fresh sandbox under the same + # name, keeping the token idempotent across failures. + resolver.kill(existing) + else: + sandbox_id = f"{SANDBOX_ID_PREFIX}{uuid.uuid4().hex[:12]}" + + resources = request.resources + cpu_request = resources.cpu_request if resources else None + cpu_limit = resources.cpu_limit if resources else None + memory_request_mb = resources.memory_request_mb if resources else None + memory_limit_mb = resources.memory_limit_mb if resources else None + # A capped sandbox should also be scheduled onto capacity that can + # honor the cap, so requests default to the limits. + if cpu_request is None: + cpu_request = cpu_limit + if memory_request_mb is None: + memory_request_mb = memory_limit_mb + + actor_options: Dict[str, Any] = { + "num_cpus": ( + cpu_request + if cpu_request is not None + else settings.default_actor_num_cpus + ), + } + if memory_request_mb is not None: + actor_options["memory"] = memory_request_mb * 1024 * 1024 + if resources is not None and resources.custom: + actor_options["resources"] = dict(resources.custom) + + capabilities = ( + request.capabilities + if request.capabilities is not None + else list(settings.default_capabilities) + ) + spec = { + "image": request.image, + "env": request.env, + "workdir": request.workdir, + "ttl_seconds": effective_ttl, + "network": request.network, + "dns": request.dns, + "shell": request.shell, + "rootless": request.rootless, + "readonly": request.readonly, + "capabilities": capabilities, + "cpu_limit": cpu_limit, + "memory_limit_mb": memory_limit_mb, + "image_pull_timeout_seconds": request.image_pull_timeout_seconds, + "start_timeout_seconds": request.start_timeout_seconds, + "labels": request.labels, + } + host_settings = { + "max_output_bytes": settings.max_output_bytes, + "max_exec_history": settings.max_exec_history, + "auto_install_runsc": settings.auto_install_runsc, + "auto_install_pasta": settings.auto_install_pasta, + } + handle = resolver.create( + sandbox_id, + actor_options, + { + "sandbox_id": sandbox_id, + "spec": spec, + "settings": host_settings, + }, + ) + # Fire-and-forget: boot progress and failures are reported through + # describe(), never through this call's result. The response is + # synthesized from the request rather than fetched from the actor so + # that creation never blocks behind actor scheduling — on a saturated + # cluster the new actor may legitimately be queued for a while. + handle.boot.remote() + created_at = datetime.now(timezone.utc) + info = { + "sandbox_id": sandbox_id, + "status": "pending", + "image": request.image, + "created_at": created_at.isoformat(), + "ttl_seconds": effective_ttl, + "expires_at": (created_at + timedelta(seconds=effective_ttl)).isoformat(), + "network": request.network, + "labels": request.labels, + "error": None, + } + return JSONResponse(status_code=202, content=info) + + @v1.get("/sandboxes", response_model=SandboxList) + async def list_sandboxes( + label: List[str] = Query(default=[]), + ) -> Dict[str, Any]: + filters = _parse_label_filters(label) + named = [(name, resolver.get(name)) for name in resolver.list_names()] + # Concurrently: sequential describes would make the listing scale + # linearly with the number of sandboxes. + results = await asyncio.gather( + *( + _describe_or_pending(name, handle) + for name, handle in named + if handle is not None + ), + return_exceptions=True, + ) + sandboxes: List[Dict[str, Any]] = [] + for info in results: + if isinstance(info, _ApiError) and info.code == "sandbox_not_found": + continue + if isinstance(info, BaseException): + raise info + if all(info.get("labels", {}).get(k) == v for k, v in filters.items()): + sandboxes.append(info) + return {"sandboxes": sandboxes} + + @v1.get("/sandboxes/{sandbox_id}", response_model=SandboxInfo) + async def get_sandbox( + sandbox_id: str, wait_seconds: float = _WAIT_QUERY + ) -> Dict[str, Any]: + handle = resolver.get(sandbox_id) + if handle is None: + raise _sandbox_not_found(sandbox_id) + return await _describe_or_pending(sandbox_id, handle, wait_seconds) + + @v1.delete("/sandboxes/{sandbox_id}") + async def delete_sandbox(sandbox_id: str) -> Dict[str, str]: + handle = resolver.get(sandbox_id) + if handle is not None: + try: + await handle.terminate.remote() + except Exception as exc: + if not _is_actor_gone(exc): + raise + resolver.kill(handle) + # Idempotent: deleting an unknown or already-gone sandbox succeeds. + return {"sandbox_id": sandbox_id, "status": "terminated"} + + # ------------------------------------------------------------------ + # Execs + # ------------------------------------------------------------------ + + @v1.post( + "/sandboxes/{sandbox_id}/execs", + response_model=ExecStarted, + status_code=202, + ) + async def start_exec(sandbox_id: str, request: StartExecRequest) -> Dict[str, Any]: + if ( + request.timeout_seconds is not None + and request.timeout_seconds > settings.max_exec_timeout_seconds + ): + raise _ApiError( + 400, + "invalid_request", + f"timeout_seconds may not exceed {settings.max_exec_timeout_seconds}", + ) + handle = resolver.get(sandbox_id) + if handle is None: + raise _sandbox_not_found(sandbox_id) + result = await _bounded_call( + sandbox_id, + handle.start_exec.remote( + command=request.command, + cwd=request.cwd, + env=request.env, + timeout_seconds=request.timeout_seconds, + shell=request.shell, + user=request.user, + ), + ) + if result.get("error_code") == "conflict": + raise _ApiError(409, "conflict", result["message"]) + return result + + @v1.get("/sandboxes/{sandbox_id}/execs/{exec_id}", response_model=ExecInfo) + async def get_exec( + sandbox_id: str, exec_id: str, wait_seconds: float = _WAIT_QUERY + ) -> Dict[str, Any]: + handle = resolver.get(sandbox_id) + if handle is None: + raise _sandbox_not_found(sandbox_id) + result = await _bounded_call( + sandbox_id, + handle.get_exec.remote(exec_id, wait_seconds=wait_seconds), + extra_wait=wait_seconds, + ) + if result.get("error_code") == "exec_not_found": + raise _ApiError(404, "exec_not_found", result["message"]) + return result + + # ------------------------------------------------------------------ + # Files + # ------------------------------------------------------------------ + + def _validate_file_path(path: str) -> None: + if not path.startswith("/"): + raise _ApiError(400, "invalid_request", "file path must be absolute") + + @v1.put("/sandboxes/{sandbox_id}/files", status_code=204) + async def put_file( + sandbox_id: str, + request: Request, + path: str = Query(min_length=1), + append: bool = Query( + default=False, + description=( + "Append to the file instead of truncating it. Lets clients " + "chunk large uploads under proxy body-size limits." + ), + ), + ) -> Response: + _validate_file_path(path) + body = await request.body() + if len(body) > settings.max_file_bytes: + raise _ApiError( + 413, + "payload_too_large", + f"file body may not exceed {settings.max_file_bytes} bytes", + ) + handle = resolver.get(sandbox_id) + if handle is None: + raise _sandbox_not_found(sandbox_id) + result = await _bounded_call( + sandbox_id, handle.write_file.remote(path, body, append=append) + ) + if result.get("error_code") == "conflict": + raise _ApiError(409, "conflict", result["message"]) + return Response(status_code=204) + + @v1.get("/sandboxes/{sandbox_id}/files") + async def get_file(sandbox_id: str, path: str = Query(min_length=1)) -> Response: + _validate_file_path(path) + handle = resolver.get(sandbox_id) + if handle is None: + raise _sandbox_not_found(sandbox_id) + result = await _bounded_call(sandbox_id, handle.read_file.remote(path)) + if result.get("error_code") == "conflict": + raise _ApiError(409, "conflict", result["message"]) + if result.get("error_code") == "file_not_found": + raise _ApiError(404, "file_not_found", result["message"]) + return Response( + content=result["content"], media_type="application/octet-stream" + ) + + app = FastAPI(title="Ray Sandbox API", version="1.0.0") + app.include_router(public) + app.include_router(v1) + + @app.exception_handler(_ApiError) + async def api_error_handler(request: Request, exc: _ApiError) -> JSONResponse: + return JSONResponse( + status_code=exc.status_code, + content={"error": {"code": exc.code, "message": exc.message}}, + ) + + @app.exception_handler(Exception) + async def internal_error_handler(request: Request, exc: Exception) -> JSONResponse: + logger.exception("Unhandled error serving %s", request.url.path) + return JSONResponse( + status_code=500, + content={"error": {"code": "internal", "message": "internal server error"}}, + ) + + return app + + +@PublicAPI(stability="alpha") +def build_app(args: Optional[Dict[str, Any]] = None) -> Any: + """Ray Serve application builder. + + Usable directly with ``serve run`` (builder args become + :class:`SandboxAPISettings` fields):: + + serve run ray.experimental.sandbox.http.app:build_app + + or from an Anyscale service config via ``import_path``. + """ + from ray import serve + + settings = SandboxAPISettings(**(args or {})) + fastapi_app = create_app(settings) + + # The API is long-poll based (requests deliberately hold a slot for up + # to ~30s), so the replica must admit far more concurrent requests than + # Serve's default allows. + @serve.deployment(name="RaySandboxAPI", max_ongoing_requests=1000) + @serve.ingress(fastapi_app) + class SandboxAPIIngress: + pass + + return SandboxAPIIngress.options(num_replicas=settings.num_replicas).bind() diff --git a/python/ray/experimental/sandbox/http/host.py b/python/ray/experimental/sandbox/http/host.py new file mode 100644 index 000000000000..7c686e0a399e --- /dev/null +++ b/python/ray/experimental/sandbox/http/host.py @@ -0,0 +1,592 @@ +"""Per-sandbox actor backing the Ray Sandbox HTTP API. + +Each API sandbox is one ``SandboxHost``, created by the HTTP layer as a +*named, detached* Ray actor (name = sandbox id). The detached actors are the +API's registry: any Serve replica resolves a sandbox with ``ray.get_actor`` +and the service keeps no state of its own, so replicas can restart or scale +without losing sandboxes. + +``SandboxHost`` composes :class:`~ray.experimental.sandbox.runtime.SandboxRuntime` +rather than the ``ray.experimental.sandbox.Sandbox`` actor because the API +needs behavior the upstream actor does not provide: + +* **Boot in the background** — the upstream actor pulls the image and boots + the container inside ``__init__``, so creation errors only surface on the + first method call. Here ``__init__`` is trivial and ``boot()`` runs as a + background task, making progress (``pending -> pulling -> starting -> + running``) and failures pollable over HTTP. +* **Exec as jobs** — commands can outrun any HTTP request (and the load + balancers in front of an Anyscale service), so ``start_exec`` returns an id + immediately and results are polled. +* **A TTL that reclaims everything** — the upstream TTL timer deletes the + sandbox but leaks the hosting actor and its resource reservation; this one + deletes the sandbox and then kills its own actor. + +Capability grants and host-network behavior (netns, resolv.conf) are plain +``SandboxConfig`` fields handled by the core runtime; nothing is patched here. + +Cross-actor control flow uses plain dicts (``{"error_code": ...}``) instead +of exceptions: Ray re-raises remote exceptions as dynamically-built +``RayTaskError`` subclasses, which makes matching them in the HTTP layer +fragile. Unexpected exceptions still propagate and map to HTTP 500. +""" + +import asyncio +import logging +import os +import platform +import shutil +import subprocess +import sys +import urllib.request +import uuid +from collections import OrderedDict +from datetime import datetime, timedelta, timezone +from typing import Any, Callable, Dict, List, Optional, Tuple, Union + +from ray.experimental.sandbox.exceptions import SandboxError, SandboxTimeoutError +from ray.experimental.sandbox.http.schemas import DOCKER_DEFAULT_CAPABILITIES +from ray.util.annotations import DeveloperAPI + +logger = logging.getLogger(__name__) + +_TERMINAL_EXEC_STATUSES = ("completed", "timeout", "error") +_BOOTING_STATUSES = ("pending", "pulling", "starting") + + +def _truncate_output(text: str, max_bytes: int) -> Tuple[str, bool]: + """Cap *text* at *max_bytes* of UTF-8, with a loud trailing marker.""" + data = text.encode("utf-8", errors="replace") + if len(data) <= max_bytes: + return text, False + clipped = data[:max_bytes].decode("utf-8", errors="replace") + return ( + clipped + f"\n[truncated by ray-sandbox: output exceeded {max_bytes} bytes]", + True, + ) + + +def _ensure_runsc_installed() -> None: + """Download runsc from the official gVisor release bucket onto this node. + + Opt-in via the ``auto_install_runsc`` server setting, for running the + service on stock node images. The download lands in a temp directory + prepended to this process's PATH, which the runsc subprocesses inherit. + """ + if shutil.which("runsc"): + return + # One shared, cached location per node: repeated boots reuse it instead + # of leaking a ~40MB download per sandbox, and concurrent downloaders + # converge through the atomic rename. + shared_bin = "/tmp/ray-sandbox-runsc" + runsc_path = os.path.join(shared_bin, "runsc") + if not os.path.exists(runsc_path): + os.makedirs(shared_bin, mode=0o755, exist_ok=True) + arch = ( + "aarch64" + if platform.machine().lower() in ("aarch64", "arm64") + else "x86_64" + ) + url = ( + "https://storage.googleapis.com/gvisor/releases/release/latest/" + f"{arch}/runsc" + ) + logger.info("runsc not found on this node; downloading from %s", url) + tmp_path = f"{runsc_path}.tmp.{os.getpid()}" + urllib.request.urlretrieve(url, tmp_path) + os.chmod(tmp_path, 0o755) + os.replace(tmp_path, runsc_path) + if shared_bin not in os.environ.get("PATH", "").split(os.pathsep): + os.environ["PATH"] = f"{shared_bin}{os.pathsep}{os.environ.get('PATH', '')}" + + +def _ensure_pasta_installed() -> None: + """Download a static pasta (passt) build onto this node. + + Opt-in via the ``auto_install_pasta`` server setting. pasta provides the + per-sandbox network namespaces behind ``network="public"``. passt.top + publishes static builds for x86_64 only, and only an unversioned + "latest" (same trust posture as the runsc "latest" URL above); prefer + baking the distro's passt package into the node image for production. + ``RAY_SANDBOX_PASTA_URL`` overrides the download URL, e.g. for a + mirrored, pinned, or non-x86_64 build. + """ + if shutil.which("pasta"): + return + shared_bin = "/tmp/ray-sandbox-pasta" + pasta_path = os.path.join(shared_bin, "pasta") + if not os.path.exists(pasta_path): + url = os.environ.get("RAY_SANDBOX_PASTA_URL") + if not url: + if platform.machine().lower() not in ("x86_64", "amd64"): + raise RuntimeError( + "auto_install_pasta: no static pasta build is published " + f"for {platform.machine()}; install the distro's passt " + "package on the node image or set RAY_SANDBOX_PASTA_URL." + ) + url = "https://passt.top/builds/latest/x86_64/pasta" + os.makedirs(shared_bin, mode=0o755, exist_ok=True) + logger.info("pasta not found on this node; downloading from %s", url) + tmp_path = f"{pasta_path}.tmp.{os.getpid()}" + urllib.request.urlretrieve(url, tmp_path) + os.chmod(tmp_path, 0o755) + os.replace(tmp_path, pasta_path) + if shared_bin not in os.environ.get("PATH", "").split(os.pathsep): + os.environ["PATH"] = f"{shared_bin}{os.pathsep}{os.environ.get('PATH', '')}" + + +def _ensure_tun_device() -> None: + """Best-effort creation of /dev/net/tun for pasta. + + Kubernetes gives each container a fresh minimal /dev without the tun + device, and pasta needs it to build the per-sandbox tap. Creating the + node takes CAP_MKNOD (via passwordless sudo, standard on Anyscale node + images); *opening* it is allowed by the default device cgroup, so no + securityContext change is needed. No-op when the node already exists; + on failure pasta reports the missing device itself. + """ + if sys.platform != "linux" or os.path.exists("/dev/net/tun"): + return + for cmd in ( + ["sudo", "-n", "mkdir", "-p", "/dev/net"], + ["sudo", "-n", "mknod", "/dev/net/tun", "c", "10", "200"], + ["sudo", "-n", "chmod", "0666", "/dev/net/tun"], + ): + res = subprocess.run(cmd, capture_output=True) + if res.returncode != 0: + logger.warning( + "Could not create /dev/net/tun (%s: %s); network='public' " + "sandboxes need it for pasta.", + " ".join(cmd), + res.stderr.decode(errors="replace").strip(), + ) + return + + +class _ExecJob: + """One submitted command and its (eventual) result.""" + + def __init__(self, exec_id: str) -> None: + self.exec_id = exec_id + self.status = "running" + self.exit_code: Optional[int] = None + self.stdout: Optional[str] = None + self.stderr: Optional[str] = None + self.stdout_truncated = False + self.stderr_truncated = False + self.duration_seconds: Optional[float] = None + self.error: Optional[str] = None + self.done = asyncio.Event() + + def to_dict(self) -> Dict[str, Any]: + return { + "exec_id": self.exec_id, + "status": self.status, + "exit_code": self.exit_code, + "stdout": self.stdout, + "stderr": self.stderr, + "stdout_truncated": self.stdout_truncated, + "stderr_truncated": self.stderr_truncated, + "duration_seconds": self.duration_seconds, + "error": self.error, + } + + +@DeveloperAPI +class SandboxHost: + """Hosts one gVisor sandbox for the HTTP API. + + Instantiated by the HTTP layer via ``ray.remote(SandboxHost)`` as a named + detached async actor; unit tests instantiate it directly with a fake + runtime factory, so nothing here may assume a Ray context except the + self-destruct path (which degrades to a no-op outside an actor). + + Args: + sandbox_id: The API-level sandbox id; also the detached actor's name. + spec: Sandbox creation spec (validated request data, plain dict). + settings: Server limits (``max_output_bytes``, ``max_exec_history``). + runtime_factory: Test seam; defaults to ``SandboxRuntime``. + """ + + def __init__( + self, + sandbox_id: str, + spec: Dict[str, Any], + settings: Dict[str, Any], + runtime_factory: Optional[Callable[[], Any]] = None, + ) -> None: + self._sandbox_id = sandbox_id + self._spec = spec + self._max_output_bytes = int(settings.get("max_output_bytes", 10 * 1024**2)) + self._max_exec_history = int(settings.get("max_exec_history", 256)) + self._auto_install_runsc = bool(settings.get("auto_install_runsc", False)) + self._auto_install_pasta = bool(settings.get("auto_install_pasta", False)) + self._runtime_factory = runtime_factory + self._runtime: Optional[Any] = None + self._instance_id: Optional[str] = None + self._status = "pending" + self._error: Optional[str] = None + self._created_at = datetime.now(timezone.utc) + self._status_changed = asyncio.Event() + self._execs: "OrderedDict[str, _ExecJob]" = OrderedDict() + self._exec_tasks: Dict[str, asyncio.Task] = {} + self._ttl_task: Optional[asyncio.Task] = None + self._boot_started = False + + # ------------------------------------------------------------------ + # Lifecycle + # ------------------------------------------------------------------ + + async def boot(self) -> None: + """Pull the image and start the sandbox, recording progress. + + Fired by the HTTP layer right after actor creation and never awaited + for its result; every outcome (including failure) lands in the status + this actor reports. Idempotent so a lost-then-retried create (via + ``get_if_exists``) cannot boot twice. + """ + if self._boot_started: + return + self._boot_started = True + + ttl_seconds = self._spec.get("ttl_seconds") + if ttl_seconds is not None: + # Started before the boot work so even a sandbox stuck in a + # failing boot is eventually reclaimed. + self._ttl_task = asyncio.create_task(self._ttl_watchdog(ttl_seconds)) + + try: + if self._auto_install_runsc: + await asyncio.to_thread(_ensure_runsc_installed) + if self._auto_install_pasta: + await asyncio.to_thread(_ensure_pasta_installed) + if self._spec.get("network") == "public": + await asyncio.to_thread(_ensure_tun_device) + factory = self._runtime_factory + if factory is None: + from ray.experimental.sandbox.runtime import SandboxRuntime + + factory = SandboxRuntime + self._runtime = factory() + + self._set_status("pulling") + await asyncio.to_thread( + self._runtime.pull_image, + self._spec["image"], + timeout_seconds=float( + self._spec.get("image_pull_timeout_seconds", 600.0) + ), + ) + + self._set_status("starting") + capabilities = self._spec.get("capabilities") + if capabilities is None: + capabilities = list(DOCKER_DEFAULT_CAPABILITIES) + + memory_limit_mb = self._spec.get("memory_limit_mb") + cpu_limit = self._spec.get("cpu_limit") + create_kwargs: Dict[str, Any] = {} + if self._spec.get("shell") is not None: + # Omitted rather than passed as None: SandboxConfig.shell is + # a plain str with a /bin/bash default. + create_kwargs["shell"] = self._spec["shell"] + # Requests and limits are deliberately decoupled: cpu_request / + # memory_request_mb size the hosting actor (cluster scheduling) + # and only cpu_limit / memory_limit_mb become cgroup caps — + # unlike the upstream Sandbox actor, which infers a cpu quota + # from its assigned resources. + self._instance_id = await asyncio.to_thread( + self._runtime.create, + self._spec["image"], + # 0 means "no cgroup limit" to Ray Sandbox. Note that it still + # derives a cpu quota from the hosting actor's assigned CPUs + # when no explicit limit is set. + cpu=float(cpu_limit) if cpu_limit is not None else 0.0, + memory=f"{memory_limit_mb}Mi" if memory_limit_mb is not None else 0, + env=dict(self._spec.get("env") or {}), + workdir=self._spec.get("workdir"), + # Writability is the runtime's explicit contract: a scratch + # dir exists only for an explicitly passed workdir on a + # readonly rootfs; readonly=False sandboxes are fully + # writable with image WORKDIR content visible. + # The API owns the TTL (see _ttl_watchdog); the upstream + # runtime stores but never enforces this, and the upstream + # actor's timer would reclaim only the sandbox, not the actor. + ttl_seconds=None, + timeout_seconds=float(self._spec.get("start_timeout_seconds", 60.0)), + rootless=bool(self._spec.get("rootless", True)), + network=self._spec.get("network", "none"), + dns=self._spec.get("dns"), + capabilities=capabilities, + readonly=bool(self._spec.get("readonly", True)), + **create_kwargs, + ) + + self._set_status("running") + logger.info( + "Sandbox %s running (instance %s, image %s)", + self._sandbox_id, + self._instance_id, + self._spec["image"], + ) + except Exception as exc: + # The actor stays alive holding the error so clients can read it; + # DELETE or the TTL reclaims it. + logger.warning("Sandbox %s failed to boot: %s", self._sandbox_id, exc) + self._error = str(exc) + self._set_status("error") + await self._delete_sandbox_instance() + + async def _ttl_watchdog(self, ttl_seconds: float) -> None: + await asyncio.sleep(ttl_seconds) + logger.info( + "Sandbox %s reached its TTL (%ss); terminating", + self._sandbox_id, + ttl_seconds, + ) + await self._shutdown() + self._self_destruct() + + def _self_destruct(self) -> None: + """Kill this actor so the TTL reclaims its name and reservation. + + ``ray.kill`` on the self-handle (rather than ``exit_actor``) because + the watchdog runs as a self-spawned asyncio task, outside any Ray + method invocation, where ``exit_actor``'s control-flow exception has + nothing to catch it. No-op outside an actor (unit tests). + """ + try: + import ray + + handle = ray.get_runtime_context().current_actor + ray.kill(handle) + except Exception: + logger.debug( + "Sandbox %s host is not a Ray actor; skipping self-destruct", + self._sandbox_id, + ) + + async def _shutdown(self) -> None: + """Cancel work and delete the sandbox instance. Idempotent.""" + if self._ttl_task is not None and self._ttl_task is not asyncio.current_task(): + self._ttl_task.cancel() + self._ttl_task = None + for task in self._exec_tasks.values(): + task.cancel() + self._exec_tasks.clear() + for job in self._execs.values(): + if job.status == "running": + job.status = "error" + job.error = "sandbox terminated while the command was running" + job.done.set() + await self._delete_sandbox_instance() + self._set_status("terminated") + + async def _delete_sandbox_instance(self) -> None: + if self._runtime is None or self._instance_id is None: + return + instance_id, self._instance_id = self._instance_id, None + try: + await asyncio.to_thread(self._runtime.delete, instance_id) + except Exception as exc: + logger.warning("Failed to delete sandbox instance %s: %s", instance_id, exc) + + async def terminate(self) -> Dict[str, Any]: + """Delete the sandbox and mark this host terminated. + + The HTTP layer kills the actor afterwards; splitting the two keeps + this method's reply deliverable. + """ + await self._shutdown() + return {"ok": True} + + # ------------------------------------------------------------------ + # Introspection + # ------------------------------------------------------------------ + + def _set_status(self, status: str) -> None: + self._status = status + # Replace-then-set so current waiters wake once and later waiters + # block on a fresh event. + event, self._status_changed = self._status_changed, asyncio.Event() + event.set() + + def _info(self) -> Dict[str, Any]: + ttl_seconds = self._spec.get("ttl_seconds") + expires_at = ( + self._created_at + timedelta(seconds=ttl_seconds) + if ttl_seconds is not None + else None + ) + return { + "sandbox_id": self._sandbox_id, + "status": self._status, + "image": self._spec["image"], + "created_at": self._created_at.isoformat(), + "ttl_seconds": ttl_seconds, + "expires_at": expires_at.isoformat() if expires_at else None, + "network": self._spec.get("network", "none"), + "labels": dict(self._spec.get("labels") or {}), + "error": self._error, + } + + async def describe(self, wait_seconds: float = 0.0) -> Dict[str, Any]: + """Report sandbox state, optionally long-polling while it boots.""" + if wait_seconds > 0 and self._status in _BOOTING_STATUSES: + event = self._status_changed + try: + await asyncio.wait_for(event.wait(), timeout=wait_seconds) + except asyncio.TimeoutError: + pass + return self._info() + + # ------------------------------------------------------------------ + # Exec jobs + # ------------------------------------------------------------------ + + async def start_exec( + self, + command: Union[str, List[str]], + cwd: Optional[str] = None, + env: Optional[Dict[str, str]] = None, + timeout_seconds: Optional[float] = None, + shell: Optional[str] = None, + user: Optional[str] = None, + ) -> Dict[str, Any]: + if self._status != "running": + return { + "error_code": "conflict", + "message": ( + f"sandbox {self._sandbox_id} is {self._status}, not running" + ), + } + exec_id = f"ex-{uuid.uuid4().hex[:12]}" + job = _ExecJob(exec_id) + self._execs[exec_id] = job + self._prune_exec_history() + task = asyncio.create_task( + self._run_exec(job, command, cwd, env, timeout_seconds, shell, user) + ) + self._exec_tasks[exec_id] = task + task.add_done_callback(lambda _: self._exec_tasks.pop(exec_id, None)) + return {"exec_id": exec_id, "status": job.status} + + def _prune_exec_history(self) -> None: + # Evict oldest *finished* jobs beyond the cap; running jobs must stay + # addressable, so with a pathological number in flight the dict may + # exceed the cap rather than lose one. + finished = [ + exec_id + for exec_id, job in self._execs.items() + if job.status in _TERMINAL_EXEC_STATUSES + ] + excess = len(self._execs) - self._max_exec_history + for exec_id in finished[: max(0, excess)]: + del self._execs[exec_id] + + async def _run_exec( + self, + job: _ExecJob, + command: Union[str, List[str]], + cwd: Optional[str], + env: Optional[Dict[str, str]], + timeout_seconds: Optional[float], + shell: Optional[str], + user: Optional[str], + ) -> None: + try: + result = await self._runtime.exec_async( + self._instance_id, + command, + timeout=timeout_seconds, + cwd=cwd, + env=env or None, + shell=shell, + user=user, + ) + except asyncio.CancelledError: + # _shutdown already marked the job; just stop. + raise + except SandboxTimeoutError: + job.status = "timeout" + job.error = f"command timed out after {timeout_seconds} seconds" + except SandboxError as exc: + job.status = "error" + job.error = str(exc) + except Exception as exc: + logger.warning( + "Exec %s in sandbox %s failed unexpectedly: %s", + job.exec_id, + self._sandbox_id, + exc, + ) + job.status = "error" + job.error = str(exc) + else: + job.status = "completed" + job.exit_code = result.exit_code + job.stdout, job.stdout_truncated = _truncate_output( + result.stdout, self._max_output_bytes + ) + job.stderr, job.stderr_truncated = _truncate_output( + result.stderr, self._max_output_bytes + ) + job.duration_seconds = result.duration_seconds + finally: + if not job.done.is_set(): + job.done.set() + + async def get_exec(self, exec_id: str, wait_seconds: float = 0.0) -> Dict[str, Any]: + job = self._execs.get(exec_id) + if job is None: + return { + "error_code": "exec_not_found", + "message": f"unknown exec id {exec_id!r}", + } + if wait_seconds > 0 and job.status == "running": + try: + await asyncio.wait_for(job.done.wait(), timeout=wait_seconds) + except asyncio.TimeoutError: + pass + return job.to_dict() + + # ------------------------------------------------------------------ + # Files + # ------------------------------------------------------------------ + + async def write_file( + self, path: str, content: bytes, append: bool = False + ) -> Dict[str, Any]: + if self._status != "running": + return { + "error_code": "conflict", + "message": ( + f"sandbox {self._sandbox_id} is {self._status}, not running" + ), + } + await asyncio.to_thread( + self._runtime.write_file, + self._instance_id, + path, + content, + append, + ) + return {"ok": True} + + async def read_file(self, path: str) -> Dict[str, Any]: + if self._status != "running": + return { + "error_code": "conflict", + "message": ( + f"sandbox {self._sandbox_id} is {self._status}, not running" + ), + } + try: + content = await asyncio.to_thread( + self._runtime.read_file, self._instance_id, path + ) + except SandboxError as exc: + # Upstream read_file shells out to `cat`; a missing file is the + # overwhelmingly common failure, so report it as such. + return {"error_code": "file_not_found", "message": str(exc)} + return {"ok": True, "content": content} diff --git a/python/ray/experimental/sandbox/http/schemas.py b/python/ray/experimental/sandbox/http/schemas.py new file mode 100644 index 000000000000..564dbfef1381 --- /dev/null +++ b/python/ray/experimental/sandbox/http/schemas.py @@ -0,0 +1,320 @@ +"""Pydantic request/response models for the Ray Sandbox HTTP API (v1). + +These models are the API contract: the OpenAPI schema FastAPI generates from +them is what clients (e.g. the Harbor ``ray-sandbox`` environment) are written +against. Changing a field here is a contract change; update the snapshot test +in ``tests/test_http_schemas.py`` deliberately when you do. +""" + +from datetime import datetime +from typing import Dict, List, Literal, Optional, Union + +from pydantic import BaseModel, Field, field_validator + +# Re-exported so the wire contract documents its default in one place; the +# canonical definition (and rationale) lives in the core sandbox config. +# The API defaults to Docker's set so images behave the way they do under +# Docker; the sets are written exactly, so ``capabilities: []`` runs the +# sandbox with no capabilities at all. +from ray.experimental.sandbox.config import ( # noqa: E402 + DOCKER_DEFAULT_CAPABILITIES, + VALID_NETWORK_MODES, +) +from ray.util.annotations import PublicAPI + +SandboxStatusName = Literal[ + "pending", "pulling", "starting", "running", "error", "terminated" +] +ExecStatusName = Literal["running", "completed", "timeout", "error"] + +_MAX_LABELS = 16 +_MAX_LABEL_LENGTH = 256 + + +@PublicAPI(stability="alpha") +class ResourceSpec(BaseModel): + """Resource requests (cluster reservation) and limits (in-sandbox cgroup). + + Requests reserve capacity on the Ray cluster for the actor hosting the + sandbox; limits cap the sandbox itself via its cgroup. When only a limit + is given the request defaults to it, so a capped sandbox is also scheduled + onto capacity that can honor the cap. + + Caveat for CPU: Ray Sandbox derives a cgroup cpu quota from the hosting + actor's assigned CPUs whenever no explicit cpu limit is set, so a + request-only CPU spec also ends up capped at the requested value. Memory + has no such coupling. + """ + + cpu_request: Optional[float] = Field(default=None, gt=0) + cpu_limit: Optional[float] = Field(default=None, gt=0) + memory_request_mb: Optional[int] = Field(default=None, gt=0) + memory_limit_mb: Optional[int] = Field(default=None, gt=0) + custom: Optional[Dict[str, float]] = Field( + default=None, + description=( + "Custom Ray resources required to schedule the sandbox's hosting " + 'actor, e.g. {"gvisor": 1} to pin sandboxes to runsc-equipped nodes.' + ), + ) + + +@PublicAPI(stability="alpha") +class CreateSandboxRequest(BaseModel): + """Request body for ``POST /api/v1/sandboxes``.""" + + image: str = Field( + min_length=1, + description=( + "Container image reference (e.g. 'python:3.12-slim'). Pulled " + "anonymously from the registry; the image must be public." + ), + ) + env: Dict[str, str] = Field(default_factory=dict) + workdir: Optional[str] = Field( + default=None, + description=( + "Sandbox-level working directory; also the default cwd for execs. " + "None uses the image's own WORKDIR, like Docker." + ), + ) + ttl_seconds: Optional[int] = Field( + default=3600, + ge=1, + description=( + "Auto-cleanup TTL. The sandbox and its hosting actor are " + "terminated this many seconds after creation. null disables the " + "TTL (subject to the server's max_ttl_seconds cap)." + ), + ) + network: str = Field( + default="none", + description=( + "Sandbox network mode; one of " + f"{', '.join(VALID_NETWORK_MODES)}. 'public' gives internet " + "egress from a network namespace private to the sandbox " + "(per-sandbox ports and loopback)." + ), + ) + dns: Optional[List[str]] = Field( + default=None, + description=( + "Nameserver IPs for a generated /etc/resolv.conf (mirroring " + "docker --dns). Defaults to public resolvers for network=" + "'public'; overrides the host file for network='host'." + ), + ) + shell: Optional[str] = Field( + default=None, + description=( + "Shell that runs string exec commands (e.g. '/bin/sh' for " + "images without bash). null uses the sandbox default, /bin/bash." + ), + ) + rootless: bool = True + readonly: bool = True + resources: Optional[ResourceSpec] = None + labels: Dict[str, str] = Field(default_factory=dict) + capabilities: Optional[List[str]] = Field( + default=None, + description=( + "Linux capabilities granted to the sandbox (the sets are " + "written exactly). null applies the server default (Docker's " + "14-capability set); [] runs with no capabilities." + ), + ) + image_pull_timeout_seconds: float = Field(default=600.0, gt=0) + start_timeout_seconds: float = Field(default=60.0, gt=0) + client_token: Optional[str] = Field( + default=None, + min_length=1, + max_length=128, + description=( + "Idempotency token. Creates with the same token resolve to the " + "same sandbox, so a client that lost a create response can retry " + "without leaking a duplicate sandbox." + ), + ) + + @field_validator("network") + @classmethod + def _validate_network(cls, network: str) -> str: + # Sourced from the core sandbox config so a new Ray mode is accepted + # here without an API change. + if network not in VALID_NETWORK_MODES: + raise ValueError(f"network must be one of {VALID_NETWORK_MODES}") + return network + + @field_validator("labels") + @classmethod + def _validate_labels(cls, labels: Dict[str, str]) -> Dict[str, str]: + if len(labels) > _MAX_LABELS: + raise ValueError(f"at most {_MAX_LABELS} labels are allowed") + for key, value in labels.items(): + if not key or len(key) > _MAX_LABEL_LENGTH: + raise ValueError(f"label keys must be 1-{_MAX_LABEL_LENGTH} characters") + if len(value) > _MAX_LABEL_LENGTH: + raise ValueError( + f"label values must be at most {_MAX_LABEL_LENGTH} characters" + ) + return labels + + +@PublicAPI(stability="alpha") +class SandboxInfo(BaseModel): + """Sandbox state as reported by ``GET /api/v1/sandboxes/{id}``.""" + + sandbox_id: str + status: SandboxStatusName + image: str + created_at: datetime + ttl_seconds: Optional[int] = None + expires_at: Optional[datetime] = None + network: str + labels: Dict[str, str] = Field(default_factory=dict) + error: Optional[str] = Field( + default=None, description="Failure detail when status is 'error'." + ) + + +@PublicAPI(stability="alpha") +class SandboxList(BaseModel): + """Response body for ``GET /api/v1/sandboxes``.""" + + sandboxes: List[SandboxInfo] + + +@PublicAPI(stability="alpha") +class StartExecRequest(BaseModel): + """Request body for ``POST /api/v1/sandboxes/{id}/execs``. + + A string command runs under the sandbox's shell (``/bin/bash`` unless + the sandbox or this request configures another); a list is executed + argv-style without a shell. + """ + + command: Union[str, List[str]] + cwd: Optional[str] = None + env: Dict[str, str] = Field(default_factory=dict) + timeout_seconds: Optional[float] = Field(default=None, gt=0) + shell: Optional[str] = Field( + default=None, + description="Override the sandbox's shell for this command only.", + ) + user: Optional[str] = Field( + default=None, + description=( + "Run as this user: a numeric uid, 'uid:gid', or a name from " + "the image's /etc/passwd. Default: the image's user." + ), + ) + + @field_validator("command") + @classmethod + def _validate_command(cls, command: Union[str, List[str]]) -> Union[str, List[str]]: + if isinstance(command, str): + if not command.strip(): + raise ValueError("command must be non-empty") + elif not command or not all(isinstance(part, str) for part in command): + raise ValueError("argv command must be a non-empty list of strings") + return command + + +@PublicAPI(stability="alpha") +class ExecStarted(BaseModel): + """Response body for ``POST /api/v1/sandboxes/{id}/execs``.""" + + exec_id: str + status: ExecStatusName + + +@PublicAPI(stability="alpha") +class ExecInfo(BaseModel): + """Exec state as reported by ``GET /api/v1/sandboxes/{id}/execs/{exec_id}``.""" + + exec_id: str + status: ExecStatusName + exit_code: Optional[int] = None + stdout: Optional[str] = None + stderr: Optional[str] = None + stdout_truncated: bool = False + stderr_truncated: bool = False + duration_seconds: Optional[float] = None + error: Optional[str] = Field( + default=None, description="Failure detail when status is 'timeout' or 'error'." + ) + + +@PublicAPI(stability="alpha") +class SandboxAPISettings(BaseModel): + """Server-side settings, passed as Serve application builder args.""" + + namespace: str = Field( + default="ray_sandbox_api", + description="Ray namespace holding the detached per-sandbox actors.", + ) + token_env_var: str = Field( + default="RAY_SANDBOX_API_TOKEN", + description=( + "Environment variable read at app construction for the bearer " + "token. Unset/empty disables the app-level check (an Anyscale " + "service already enforces its own bearer token at the platform " + "edge)." + ), + ) + num_replicas: int = Field(default=1, ge=1) + max_output_bytes: int = Field( + default=10 * 1024 * 1024, + gt=0, + description="Per-stream cap on retained exec stdout/stderr.", + ) + max_file_bytes: int = Field( + default=256 * 1024 * 1024, + gt=0, + description="Cap on file upload body size (HTTP 413 above it).", + ) + max_ttl_seconds: int = Field(default=7 * 24 * 3600, gt=0) + max_exec_timeout_seconds: float = Field(default=6 * 3600.0, gt=0) + max_exec_history: int = Field( + default=256, + gt=0, + description="Completed exec records retained per sandbox.", + ) + scheduling_grace_seconds: float = Field( + default=15.0, + gt=0, + description=( + "How long a request may wait on a sandbox whose hosting actor " + "has not been scheduled yet (e.g. the cluster is scaling up) " + "before reporting the sandbox as pending instead of blocking." + ), + ) + default_actor_num_cpus: float = Field( + default=1.0, + ge=0, + description="Actor CPU reservation when the request has no cpu_request.", + ) + default_capabilities: List[str] = Field( + default_factory=lambda: list(DOCKER_DEFAULT_CAPABILITIES) + ) + auto_install_runsc: bool = Field( + default=False, + description=( + "Download gVisor's runsc from the official release bucket onto " + "any node that lacks it, at first sandbox boot. Lets the service " + "run on stock images (e.g. an unmodified Anyscale cluster image) " + "at the cost of a one-time ~40MB download per node. Prefer " + "baking runsc into the node image for production." + ), + ) + auto_install_pasta: bool = Field( + default=False, + description=( + "Download a static pasta (passt) build from passt.top onto any " + "node that lacks it, at first sandbox boot. pasta provides the " + "per-sandbox network namespaces of network='public'. Static " + "builds exist for x86_64 only (RAY_SANDBOX_PASTA_URL overrides " + "the source); prefer baking the distro's passt package into the " + "node image for production." + ), + ) diff --git a/python/ray/experimental/sandbox/http/tests/__init__.py b/python/ray/experimental/sandbox/http/tests/__init__.py new file mode 100644 index 000000000000..e69de29bb2d1 diff --git a/python/ray/experimental/sandbox/http/tests/conftest.py b/python/ray/experimental/sandbox/http/tests/conftest.py new file mode 100644 index 000000000000..dbe72807693f --- /dev/null +++ b/python/ray/experimental/sandbox/http/tests/conftest.py @@ -0,0 +1,267 @@ +"""Fixtures for the Ray Sandbox HTTP API tests. + +Unlike ``sandbox/tests/conftest.py`` there is no TEST_SANDBOX gate here: the +unit tests fake the sandbox runtime and the actor layer, so they run on any +platform with no cluster and no runsc. Only ``test_http_integration.py`` +needs the real thing and gates itself. +""" + +import asyncio +import os +import platform +import shutil +import tempfile +import threading +import time +import urllib.request +from typing import Any, Dict, List, Optional, Union + +import pytest + +from ray.experimental.sandbox.exceptions import ( + SandboxExecError, + SandboxTimeoutError, +) +from ray.experimental.sandbox.http.host import SandboxHost + + +def _sandbox_test_enabled() -> bool: + try: + from ray._private.test_utils import sandbox_test_enabled + except ImportError: + return os.environ.get("TEST_SANDBOX") == "1" + return sandbox_test_enabled() + + +class FakeExecResult: + def __init__( + self, + exit_code: int = 0, + stdout: str = "", + stderr: str = "", + duration_seconds: float = 0.01, + ) -> None: + self.exit_code = exit_code + self.stdout = stdout + self.stderr = stderr + self.duration_seconds = duration_seconds + + +class FakeSandboxRuntime: + """Scriptable stand-in for ``ray.experimental.sandbox.SandboxRuntime``.""" + + def __init__(self) -> None: + self.instance_id = "ray-sandbox-fake0001" + self.pull_calls: List[Dict[str, Any]] = [] + self.create_calls: List[Dict[str, Any]] = [] + self.exec_calls: List[Dict[str, Any]] = [] + self.written_files: Dict[str, bytes] = {} + self.readable_files: Dict[str, bytes] = {} + self.deleted: List[str] = [] + self.pull_error: Optional[Exception] = None + self.create_error: Optional[Exception] = None + self.exec_error: Optional[Exception] = None + self.exec_results: List[FakeExecResult] = [] + # When set, pull_image blocks until the event fires, holding the + # sandbox in the "pulling" state for conflict tests. + self.pull_gate: Optional[threading.Event] = None + + def pull_image(self, image: str, timeout_seconds: float = 120.0) -> str: + self.pull_calls.append({"image": image, "timeout_seconds": timeout_seconds}) + if self.pull_gate is not None: + if not self.pull_gate.wait(timeout=30): + raise RuntimeError("pull gate never released") + if self.pull_error is not None: + raise self.pull_error + return f"/tmp/fake-images/{image}" + + def create(self, image: str, **kwargs: Any) -> str: + self.create_calls.append({"image": image, **kwargs}) + if self.create_error is not None: + raise self.create_error + return self.instance_id + + def exec( + self, + instance_id: str, + command: Union[str, List[str]], + timeout: Optional[float] = None, + cwd: Optional[str] = None, + env: Optional[Dict[str, str]] = None, + shell: Optional[str] = None, + user: Optional[str] = None, + ) -> FakeExecResult: + self.exec_calls.append( + { + "instance_id": instance_id, + "command": command, + "timeout": timeout, + "cwd": cwd, + "env": env, + "shell": shell, + "user": user, + } + ) + if self.exec_error is not None: + raise self.exec_error + if self.exec_results: + return self.exec_results.pop(0) + return FakeExecResult() + + async def exec_async( + self, + instance_id: str, + command: Union[str, List[str]], + timeout: Optional[float] = None, + cwd: Optional[str] = None, + env: Optional[Dict[str, str]] = None, + shell: Optional[str] = None, + user: Optional[str] = None, + ) -> FakeExecResult: + return await asyncio.to_thread( + self.exec, + instance_id, + command, + timeout=timeout, + cwd=cwd, + env=env, + shell=shell, + user=user, + ) + + def write_file( + self, + instance_id: str, + path: str, + content: Union[str, bytes], + append: bool = False, + ) -> None: + if isinstance(content, str): + content = content.encode("utf-8") + if append and path in self.written_files: + self.written_files[path] += content + else: + self.written_files[path] = content + + def read_file(self, instance_id: str, path: str) -> bytes: + if path not in self.readable_files: + raise SandboxExecError(f"cat: {path}: No such file or directory") + return self.readable_files[path] + + def delete(self, instance_id: str) -> None: + self.deleted.append(instance_id) + + +class _FakeRemoteMethod: + """Mimics ``handle.method.remote(...)``: schedules eagerly, returns an awaitable.""" + + def __init__(self, fn: Any) -> None: + self._fn = fn + + def remote(self, *args: Any, **kwargs: Any) -> "asyncio.Task": + return asyncio.get_running_loop().create_task(self._fn(*args, **kwargs)) + + +class FakeHandle: + def __init__(self, host: SandboxHost) -> None: + self.host = host + + def __getattr__(self, name: str) -> _FakeRemoteMethod: + return _FakeRemoteMethod(getattr(self.host, name)) + + +class FakeResolver: + """In-process stand-in for ``RayActorHandleResolver``.""" + + def __init__(self) -> None: + self.handles: Dict[str, FakeHandle] = {} + self.runtimes: List[FakeSandboxRuntime] = [] + self.create_options: List[Dict[str, Any]] = [] + self.killed: List[str] = [] + # Applied to the next runtime a created host builds. + self.next_runtime: Optional[FakeSandboxRuntime] = None + + def _runtime_factory(self) -> FakeSandboxRuntime: + runtime = self.next_runtime or FakeSandboxRuntime() + self.next_runtime = None + self.runtimes.append(runtime) + return runtime + + def create( + self, + name: str, + actor_options: Dict[str, Any], + ctor_kwargs: Dict[str, Any], + ) -> FakeHandle: + self.create_options.append({"name": name, **actor_options}) + # get_if_exists semantics: racing creates converge on one host. + if name in self.handles: + return self.handles[name] + host = SandboxHost(runtime_factory=self._runtime_factory, **ctor_kwargs) + handle = FakeHandle(host) + self.handles[name] = handle + return handle + + def get(self, name: str) -> Optional[FakeHandle]: + return self.handles.get(name) + + def list_names(self) -> List[str]: + return list(self.handles) + + def kill(self, handle: FakeHandle) -> None: + for name, existing in list(self.handles.items()): + if existing is handle: + del self.handles[name] + self.killed.append(name) + + +@pytest.fixture +def fake_resolver() -> FakeResolver: + return FakeResolver() + + +@pytest.fixture(scope="session", autouse=True) +def ensure_runsc(): + """Provision runsc for the integration test, mirroring sandbox/tests.""" + if not _sandbox_test_enabled(): + return + + os.environ["RAY_SANDBOX_IGNORE_CGROUPS"] = "1" + + if not shutil.which("runsc"): + temp_bin = tempfile.mkdtemp() + os.chmod(temp_bin, 0o755) + runsc_path = os.path.join(temp_bin, "runsc") + arch = ( + "aarch64" + if platform.machine().lower() in ("aarch64", "arm64") + else "x86_64" + ) + url = f"https://storage.googleapis.com/gvisor/releases/release/latest/{arch}/runsc" + try: + urllib.request.urlretrieve(url, runsc_path) + os.chmod(runsc_path, 0o755) + os.environ["PATH"] = f"{temp_bin}:{os.environ.get('PATH', '')}" + except Exception as e: + pytest.skip(f"Failed to install runsc for sandbox tests: {e}") + + +def wait_until(predicate, timeout: float = 10.0, interval: float = 0.02) -> None: + """Poll *predicate* until true or fail the test.""" + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if predicate(): + return + time.sleep(interval) + raise AssertionError("condition not met within timeout") + + +__all__ = [ + "FakeExecResult", + "FakeSandboxRuntime", + "FakeHandle", + "FakeResolver", + "SandboxExecError", + "SandboxTimeoutError", + "wait_until", +] diff --git a/python/ray/experimental/sandbox/http/tests/test_http_app.py b/python/ray/experimental/sandbox/http/tests/test_http_app.py new file mode 100644 index 000000000000..20360153577d --- /dev/null +++ b/python/ray/experimental/sandbox/http/tests/test_http_app.py @@ -0,0 +1,620 @@ +"""HTTP-layer tests: FastAPI app + fake resolver + real SandboxHost logic. + +No Ray cluster and no runsc: the resolver seam replaces actor creation with +in-process ``SandboxHost`` instances driven by ``FakeSandboxRuntime``. The +``TestClient`` context manager keeps one event loop alive across requests so +background boot tasks progress between calls, exactly like a live server. +""" + +import sys + +import pytest +from fastapi.testclient import TestClient + +from ray.experimental.sandbox.http.app import create_app +from ray.experimental.sandbox.http.schemas import SandboxAPISettings +from ray.experimental.sandbox.http.tests.conftest import ( + FakeExecResult, + FakeResolver, + FakeSandboxRuntime, +) + +BASE = "/api/v1" + + +def _client(resolver: FakeResolver, settings: SandboxAPISettings = None) -> TestClient: + app = create_app(settings or SandboxAPISettings(), handle_resolver=resolver) + return TestClient(app) + + +def _create_sandbox(client: TestClient, **overrides) -> dict: + body = {"image": "python:3.12-slim", "readonly": False, **overrides} + response = client.post(f"{BASE}/sandboxes", json=body) + assert response.status_code == 202, response.text + return response.json() + + +def _wait_running(client: TestClient, sandbox_id: str) -> dict: + for _ in range(100): + response = client.get( + f"{BASE}/sandboxes/{sandbox_id}", params={"wait_seconds": 1} + ) + assert response.status_code == 200, response.text + info = response.json() + if info["status"] not in ("pending", "pulling", "starting"): + return info + raise AssertionError("sandbox never left its boot states") + + +# ---------------------------------------------------------------------- +# Auth +# ---------------------------------------------------------------------- + + +def test_bearer_auth_enforced_when_token_configured( + fake_resolver: FakeResolver, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setenv("RAY_SANDBOX_API_TOKEN", "sekret") + with _client(fake_resolver) as client: + # Health stays public. + assert client.get(f"{BASE}/health").status_code == 200 + + response = client.get(f"{BASE}/sandboxes") + assert response.status_code == 401 + assert response.json()["error"]["code"] == "unauthorized" + + response = client.get( + f"{BASE}/sandboxes", headers={"Authorization": "Bearer wrong"} + ) + assert response.status_code == 401 + + response = client.get( + f"{BASE}/sandboxes", headers={"Authorization": "Bearer sekret"} + ) + assert response.status_code == 200 + + +def test_auth_disabled_without_token( + fake_resolver: FakeResolver, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.delenv("RAY_SANDBOX_API_TOKEN", raising=False) + with _client(fake_resolver) as client: + assert client.get(f"{BASE}/sandboxes").status_code == 200 + + +# ---------------------------------------------------------------------- +# Sandbox lifecycle +# ---------------------------------------------------------------------- + + +def test_create_then_poll_until_running(fake_resolver: FakeResolver) -> None: + with _client(fake_resolver) as client: + info = _create_sandbox(client, labels={"team": "eval"}) + # The boot task starts immediately, so the 202 body reports whatever + # early state it reached — any pre-terminal status is legitimate. + assert info["status"] in ("pending", "pulling", "starting", "running") + assert info["sandbox_id"].startswith("sb-") + assert info["image"] == "python:3.12-slim" + assert info["labels"] == {"team": "eval"} + + running = _wait_running(client, info["sandbox_id"]) + assert running["status"] == "running" + assert running["error"] is None + + (runtime,) = fake_resolver.runtimes + assert runtime.create_calls[0]["readonly"] is False + + +def test_create_maps_resources_to_actor_options_and_limits( + fake_resolver: FakeResolver, +) -> None: + with _client(fake_resolver) as client: + info = _create_sandbox( + client, + resources={ + "cpu_request": 0.5, + "cpu_limit": 2.0, + "memory_request_mb": 256, + "memory_limit_mb": 2048, + "custom": {"gvisor": 1.0}, + }, + ) + _wait_running(client, info["sandbox_id"]) + + (options,) = fake_resolver.create_options + assert options["num_cpus"] == 0.5 + assert options["memory"] == 256 * 1024 * 1024 + assert options["resources"] == {"gvisor": 1.0} + + (runtime,) = fake_resolver.runtimes + create_call = runtime.create_calls[0] + assert create_call["cpu"] == 2.0 + assert create_call["memory"] == "2048Mi" + + +def test_create_defaults_request_to_limit(fake_resolver: FakeResolver) -> None: + with _client(fake_resolver) as client: + _create_sandbox(client, resources={"cpu_limit": 4.0, "memory_limit_mb": 512}) + + (options,) = fake_resolver.create_options + assert options["num_cpus"] == 4.0 + assert options["memory"] == 512 * 1024 * 1024 + + +def test_create_default_actor_cpus_without_resources( + fake_resolver: FakeResolver, +) -> None: + with _client(fake_resolver) as client: + _create_sandbox(client) + + (options,) = fake_resolver.create_options + assert options["num_cpus"] == 1.0 + assert "memory" not in options + + +def test_create_ttl_over_cap_rejected(fake_resolver: FakeResolver) -> None: + settings = SandboxAPISettings(max_ttl_seconds=100) + with _client(fake_resolver, settings) as client: + response = client.post( + f"{BASE}/sandboxes", + json={"image": "x", "ttl_seconds": 101}, + ) + assert response.status_code == 400 + assert response.json()["error"]["code"] == "invalid_request" + + +def test_create_null_ttl_clamped_to_server_cap( + fake_resolver: FakeResolver, +) -> None: + settings = SandboxAPISettings(max_ttl_seconds=1234) + with _client(fake_resolver, settings) as client: + info = _create_sandbox(client, ttl_seconds=None) + assert info["ttl_seconds"] == 1234 + + +def test_create_missing_image_is_422(fake_resolver: FakeResolver) -> None: + with _client(fake_resolver) as client: + assert client.post(f"{BASE}/sandboxes", json={}).status_code == 422 + + +def test_client_token_makes_create_idempotent( + fake_resolver: FakeResolver, +) -> None: + with _client(fake_resolver) as client: + first = client.post( + f"{BASE}/sandboxes", + json={"image": "x", "client_token": "trial-42"}, + ) + assert first.status_code == 202 + second = client.post( + f"{BASE}/sandboxes", + json={"image": "x", "client_token": "trial-42"}, + ) + assert second.status_code == 200 + assert second.json()["sandbox_id"] == first.json()["sandbox_id"] + assert len(fake_resolver.runtimes) == 1 + + +def test_get_unknown_sandbox_404(fake_resolver: FakeResolver) -> None: + with _client(fake_resolver) as client: + response = client.get(f"{BASE}/sandboxes/sb-missing") + assert response.status_code == 404 + assert response.json()["error"]["code"] == "sandbox_not_found" + + +def test_delete_is_idempotent_and_kills_actor( + fake_resolver: FakeResolver, +) -> None: + with _client(fake_resolver) as client: + info = _create_sandbox(client) + sandbox_id = info["sandbox_id"] + _wait_running(client, sandbox_id) + + response = client.delete(f"{BASE}/sandboxes/{sandbox_id}") + assert response.status_code == 200 + assert response.json() == { + "sandbox_id": sandbox_id, + "status": "terminated", + } + assert fake_resolver.killed == [sandbox_id] + (runtime,) = fake_resolver.runtimes + assert runtime.deleted == [runtime.instance_id] + + # Gone now, and deleting again still succeeds. + assert client.get(f"{BASE}/sandboxes/{sandbox_id}").status_code == 404 + assert client.delete(f"{BASE}/sandboxes/{sandbox_id}").status_code == 200 + + +def test_boot_failure_surfaces_as_error_status( + fake_resolver: FakeResolver, +) -> None: + runtime = FakeSandboxRuntime() + runtime.pull_error = RuntimeError("no such image") + fake_resolver.next_runtime = runtime + with _client(fake_resolver) as client: + info = _create_sandbox(client) + final = _wait_running(client, info["sandbox_id"]) + assert final["status"] == "error" + assert "no such image" in final["error"] + + +def test_list_sandboxes_with_label_filter(fake_resolver: FakeResolver) -> None: + with _client(fake_resolver) as client: + a = _create_sandbox(client, labels={"job": "j1", "role": "env"}) + b = _create_sandbox(client, labels={"job": "j2"}) + + response = client.get(f"{BASE}/sandboxes") + assert response.status_code == 200 + assert {s["sandbox_id"] for s in response.json()["sandboxes"]} == { + a["sandbox_id"], + b["sandbox_id"], + } + + response = client.get(f"{BASE}/sandboxes", params=[("label", "job=j1")]) + assert [s["sandbox_id"] for s in response.json()["sandboxes"]] == [ + a["sandbox_id"] + ] + + response = client.get(f"{BASE}/sandboxes", params=[("label", "nonsense")]) + assert response.status_code == 400 + + +# ---------------------------------------------------------------------- +# Execs +# ---------------------------------------------------------------------- + + +def test_exec_submit_and_poll(fake_resolver: FakeResolver) -> None: + runtime = FakeSandboxRuntime() + runtime.exec_results = [FakeExecResult(exit_code=3, stdout="done")] + fake_resolver.next_runtime = runtime + with _client(fake_resolver) as client: + info = _create_sandbox(client) + sandbox_id = info["sandbox_id"] + _wait_running(client, sandbox_id) + + response = client.post( + f"{BASE}/sandboxes/{sandbox_id}/execs", + json={ + "command": ["bash", "-c", "exit 3"], + "cwd": "/app", + "env": {"A": "1"}, + "timeout_seconds": 5, + }, + ) + assert response.status_code == 202, response.text + exec_id = response.json()["exec_id"] + assert exec_id.startswith("ex-") + + response = client.get( + f"{BASE}/sandboxes/{sandbox_id}/execs/{exec_id}", + params={"wait_seconds": 10}, + ) + assert response.status_code == 200 + result = response.json() + assert result["status"] == "completed" + assert result["exit_code"] == 3 + assert result["stdout"] == "done" + + exec_call = runtime.exec_calls[-1] + assert exec_call["command"] == ["bash", "-c", "exit 3"] + assert exec_call["cwd"] == "/app" + assert exec_call["env"] == {"A": "1"} + assert exec_call["timeout"] == 5 + + +def test_exec_on_unknown_sandbox_404(fake_resolver: FakeResolver) -> None: + with _client(fake_resolver) as client: + response = client.post( + f"{BASE}/sandboxes/sb-missing/execs", json={"command": "true"} + ) + assert response.status_code == 404 + + +def test_unknown_exec_id_404(fake_resolver: FakeResolver) -> None: + with _client(fake_resolver) as client: + info = _create_sandbox(client) + sandbox_id = info["sandbox_id"] + _wait_running(client, sandbox_id) + response = client.get(f"{BASE}/sandboxes/{sandbox_id}/execs/ex-nope") + assert response.status_code == 404 + assert response.json()["error"]["code"] == "exec_not_found" + + +def test_exec_while_booting_is_409(fake_resolver: FakeResolver) -> None: + import threading + + runtime = FakeSandboxRuntime() + runtime.pull_gate = threading.Event() + fake_resolver.next_runtime = runtime + with _client(fake_resolver) as client: + info = _create_sandbox(client) + sandbox_id = info["sandbox_id"] + + response = client.post( + f"{BASE}/sandboxes/{sandbox_id}/execs", json={"command": "true"} + ) + assert response.status_code == 409 + assert response.json()["error"]["code"] == "conflict" + + runtime.pull_gate.set() + _wait_running(client, sandbox_id) + response = client.post( + f"{BASE}/sandboxes/{sandbox_id}/execs", json={"command": "true"} + ) + assert response.status_code == 202 + + +def test_exec_timeout_over_cap_rejected(fake_resolver: FakeResolver) -> None: + settings = SandboxAPISettings(max_exec_timeout_seconds=10.0) + with _client(fake_resolver, settings) as client: + info = _create_sandbox(client) + sandbox_id = info["sandbox_id"] + _wait_running(client, sandbox_id) + response = client.post( + f"{BASE}/sandboxes/{sandbox_id}/execs", + json={"command": "true", "timeout_seconds": 11}, + ) + assert response.status_code == 400 + + +def test_empty_command_is_422(fake_resolver: FakeResolver) -> None: + with _client(fake_resolver) as client: + info = _create_sandbox(client) + sandbox_id = info["sandbox_id"] + _wait_running(client, sandbox_id) + response = client.post( + f"{BASE}/sandboxes/{sandbox_id}/execs", json={"command": " "} + ) + assert response.status_code == 422 + + +# ---------------------------------------------------------------------- +# Files +# ---------------------------------------------------------------------- + + +def test_file_put_and_get_roundtrip(fake_resolver: FakeResolver) -> None: + runtime = FakeSandboxRuntime() + runtime.readable_files["/data/report.bin"] = b"\x00\x01binary" + fake_resolver.next_runtime = runtime + with _client(fake_resolver) as client: + info = _create_sandbox(client) + sandbox_id = info["sandbox_id"] + _wait_running(client, sandbox_id) + + response = client.put( + f"{BASE}/sandboxes/{sandbox_id}/files", + params={"path": "/data/in.bin"}, + content=b"payload-bytes", + ) + assert response.status_code == 204 + assert runtime.written_files["/data/in.bin"] == b"payload-bytes" + + response = client.get( + f"{BASE}/sandboxes/{sandbox_id}/files", + params={"path": "/data/report.bin"}, + ) + assert response.status_code == 200 + assert response.content == b"\x00\x01binary" + assert response.headers["content-type"].startswith("application/octet-stream") + + +def test_file_get_missing_404(fake_resolver: FakeResolver) -> None: + with _client(fake_resolver) as client: + info = _create_sandbox(client) + sandbox_id = info["sandbox_id"] + _wait_running(client, sandbox_id) + response = client.get( + f"{BASE}/sandboxes/{sandbox_id}/files", params={"path": "/nope"} + ) + assert response.status_code == 404 + assert response.json()["error"]["code"] == "file_not_found" + + +def test_file_put_too_large_413(fake_resolver: FakeResolver) -> None: + settings = SandboxAPISettings(max_file_bytes=8) + with _client(fake_resolver, settings) as client: + info = _create_sandbox(client) + sandbox_id = info["sandbox_id"] + _wait_running(client, sandbox_id) + response = client.put( + f"{BASE}/sandboxes/{sandbox_id}/files", + params={"path": "/big"}, + content=b"123456789", + ) + assert response.status_code == 413 + assert response.json()["error"]["code"] == "payload_too_large" + + +def test_file_relative_path_400(fake_resolver: FakeResolver) -> None: + with _client(fake_resolver) as client: + info = _create_sandbox(client) + sandbox_id = info["sandbox_id"] + response = client.put( + f"{BASE}/sandboxes/{sandbox_id}/files", + params={"path": "relative/path"}, + content=b"x", + ) + assert response.status_code == 400 + assert response.json()["error"]["code"] == "invalid_request" + + +def test_file_ops_while_booting_409(fake_resolver: FakeResolver) -> None: + import threading + + runtime = FakeSandboxRuntime() + runtime.pull_gate = threading.Event() + fake_resolver.next_runtime = runtime + try: + with _client(fake_resolver) as client: + info = _create_sandbox(client) + sandbox_id = info["sandbox_id"] + response = client.put( + f"{BASE}/sandboxes/{sandbox_id}/files", + params={"path": "/x"}, + content=b"x", + ) + assert response.status_code == 409 + finally: + runtime.pull_gate.set() + + +class _UnschedulableHandle: + """Mimics a handle whose actor Ray reports as permanently unschedulable.""" + + def __getattr__(self, name: str): + class _Method: + def remote(self, *args, **kwargs): + import asyncio + + async def _raise(): + exc_type = type("ActorUnschedulableError", (Exception,), {}) + raise exc_type("resource shapes cannot fit the cluster") + + return asyncio.get_running_loop().create_task(_raise()) + + return _Method() + + +def test_unschedulable_actor_maps_to_409(fake_resolver: FakeResolver) -> None: + client = _client(fake_resolver, _fast_settings()) + fake_resolver.handles["sb-unsched0001"] = _UnschedulableHandle() + + response = client.get(f"{BASE}/sandboxes/sb-unsched0001") + + assert response.status_code == 409, response.text + assert response.json()["error"]["code"] == "unschedulable" + + +class _StalledHandle: + """Mimics a handle to a created-but-unscheduled detached actor: every + remote call returns an awaitable that never resolves.""" + + def __getattr__(self, name: str): + class _Method: + def remote(self, *args, **kwargs): + import asyncio + + return asyncio.get_running_loop().create_future() + + return _Method() + + +def _fast_settings() -> SandboxAPISettings: + return SandboxAPISettings(scheduling_grace_seconds=0.2) + + +def test_get_sandbox_reports_pending_while_actor_is_scheduling( + fake_resolver: FakeResolver, +) -> None: + client = _client(fake_resolver, _fast_settings()) + fake_resolver.handles["sb-stalled0001"] = _StalledHandle() + + response = client.get(f"{BASE}/sandboxes/sb-stalled0001") + + assert response.status_code == 200, response.text + info = response.json() + assert info["status"] == "pending" + assert info["sandbox_id"] == "sb-stalled0001" + + +def test_exec_on_scheduling_actor_is_409(fake_resolver: FakeResolver) -> None: + client = _client(fake_resolver, _fast_settings()) + fake_resolver.handles["sb-stalled0001"] = _StalledHandle() + + response = client.post( + f"{BASE}/sandboxes/sb-stalled0001/execs", json={"command": "echo hi"} + ) + + assert response.status_code == 409, response.text + assert "scheduled" in response.json()["error"]["message"] + + +def test_list_includes_scheduling_sandboxes_as_pending( + fake_resolver: FakeResolver, +) -> None: + client = _client(fake_resolver, _fast_settings()) + _create_sandbox(client) + fake_resolver.handles["sb-stalled0001"] = _StalledHandle() + + response = client.get(f"{BASE}/sandboxes") + + assert response.status_code == 200, response.text + statuses = {s["sandbox_id"]: s["status"] for s in response.json()["sandboxes"]} + assert statuses["sb-stalled0001"] == "pending" + + +def test_shell_passthrough_to_runtime(fake_resolver: FakeResolver) -> None: + """The create-level and per-exec shell fields reach the sandbox runtime.""" + with _client(fake_resolver) as client: + info = _create_sandbox(client, shell="/bin/sh") + _wait_running(client, info["sandbox_id"]) + + response = client.post( + f"{BASE}/sandboxes/{info['sandbox_id']}/execs", + json={"command": "echo hi", "shell": "/bin/dash"}, + ) + assert response.status_code == 202, response.text + exec_id = response.json()["exec_id"] + result = client.get( + f"{BASE}/sandboxes/{info['sandbox_id']}/execs/{exec_id}", + params={"wait_seconds": 5}, + ).json() + assert result["status"] == "completed" + + runtime = fake_resolver.runtimes[0] + assert runtime.create_calls[0]["shell"] == "/bin/sh" + assert runtime.exec_calls[0]["shell"] == "/bin/dash" + + +def test_exec_user_passthrough(fake_resolver: FakeResolver) -> None: + with _client(fake_resolver) as client: + info = _create_sandbox(client) + _wait_running(client, info["sandbox_id"]) + + response = client.post( + f"{BASE}/sandboxes/{info['sandbox_id']}/execs", + json={"command": "id", "user": "1000:1000"}, + ) + assert response.status_code == 202, response.text + exec_id = response.json()["exec_id"] + client.get( + f"{BASE}/sandboxes/{info['sandbox_id']}/execs/{exec_id}", + params={"wait_seconds": 5}, + ) + assert fake_resolver.runtimes[0].exec_calls[0]["user"] == "1000:1000" + + +def test_invalid_network_mode_is_422(fake_resolver: FakeResolver) -> None: + client = _client(fake_resolver) + response = client.post( + f"{BASE}/sandboxes", json={"image": "python:3.12", "network": "bridge"} + ) + assert response.status_code == 422 + + +def test_chunked_file_upload_via_append(fake_resolver: FakeResolver) -> None: + """PUT ?append=true extends the file, so clients can chunk large uploads + under proxy body-size limits.""" + with _client(fake_resolver) as client: + info = _create_sandbox(client) + _wait_running(client, info["sandbox_id"]) + base = f"{BASE}/sandboxes/{info['sandbox_id']}/files" + + assert ( + client.put(base, params={"path": "/big"}, content=b"aaa").status_code == 204 + ) + assert ( + client.put( + base, params={"path": "/big", "append": "true"}, content=b"bbb" + ).status_code + == 204 + ) + + runtime = fake_resolver.runtimes[0] + assert runtime.written_files["/big"] == b"aaabbb" + + +if __name__ == "__main__": + sys.exit(pytest.main(["-v", __file__])) diff --git a/python/ray/experimental/sandbox/http/tests/test_http_integration.py b/python/ray/experimental/sandbox/http/tests/test_http_integration.py new file mode 100644 index 000000000000..bfe313f94ef8 --- /dev/null +++ b/python/ray/experimental/sandbox/http/tests/test_http_integration.py @@ -0,0 +1,172 @@ +"""End-to-end test of the HTTP API against a real Serve deployment. + +Requires TEST_SANDBOX=1, a local Ray, and runsc (the session fixture in +conftest.py downloads one on Linux). Runs in the Buildkite sandbox job, which +is privileged and sets TEST_SANDBOX=1. +""" + +import os +import shutil +import sys +import time + +import pytest + + +def _sandbox_test_enabled() -> bool: + try: + from ray._private.test_utils import sandbox_test_enabled + except ImportError: + return os.environ.get("TEST_SANDBOX") == "1" + return sandbox_test_enabled() + + +pytestmark = pytest.mark.skipif( + not _sandbox_test_enabled(), + reason="Sandbox tests are only run when TEST_SANDBOX=1", +) + +_TOKEN = "integration-test-token" +_IMAGE = "busybox:latest" + + +@pytest.fixture(scope="module") +def api_base_url(): + if not shutil.which("runsc"): + pytest.skip("runsc is not on PATH") + + # Ray Serve is stripped from the core sandbox CI image (the sandbox + # test job installs with --install-mask all-ray-libraries): the module + # stays importable but loses serve.run, so this end-to-end test only + # runs where Serve is fully installed, e.g. a dev container. + serve = pytest.importorskip("ray.serve") + if not hasattr(serve, "run"): + pytest.skip("ray.serve is present but not fully installed") + + os.environ["RAY_SANDBOX_API_TOKEN"] = _TOKEN + + import ray + from ray.experimental.sandbox.http.app import build_app + + ray.init() + serve.run(build_app({}), name="sandbox-api-integration") + try: + yield "http://127.0.0.1:8000/api/v1" + finally: + serve.shutdown() + ray.shutdown() + + +def _wait_for(fetch, accept, timeout: float = 300.0): + deadline = time.monotonic() + timeout + last = None + while time.monotonic() < deadline: + last = fetch() + if accept(last): + return last + time.sleep(1) + raise AssertionError(f"timed out waiting; last state: {last}") + + +def test_auth_is_enforced(api_base_url): + import httpx + + with httpx.Client(timeout=30) as anonymous: + assert anonymous.get(f"{api_base_url}/health").status_code == 200 + response = anonymous.get(f"{api_base_url}/sandboxes") + assert response.status_code == 401 + assert response.json()["error"]["code"] == "unauthorized" + + +def test_full_sandbox_lifecycle(api_base_url): + import httpx + + headers = {"Authorization": f"Bearer {_TOKEN}"} + with httpx.Client(headers=headers, timeout=60) as client: + response = client.post( + f"{api_base_url}/sandboxes", + json={ + "image": _IMAGE, + "readonly": False, + "network": "none", + "ttl_seconds": 600, + "labels": {"suite": "integration"}, + }, + ) + assert response.status_code == 202, response.text + sandbox_id = response.json()["sandbox_id"] + + info = _wait_for( + lambda: client.get( + f"{api_base_url}/sandboxes/{sandbox_id}", + params={"wait_seconds": 10}, + ).json(), + lambda i: i["status"] in ("running", "error"), + ) + assert info["status"] == "running", info + + # Exec: submit then poll. + response = client.post( + f"{api_base_url}/sandboxes/{sandbox_id}/execs", + json={ + "command": ["sh", "-c", "echo hello-from-sandbox && exit 4"], + "timeout_seconds": 60, + }, + ) + assert response.status_code == 202, response.text + exec_id = response.json()["exec_id"] + result = _wait_for( + lambda: client.get( + f"{api_base_url}/sandboxes/{sandbox_id}/execs/{exec_id}", + params={"wait_seconds": 10}, + ).json(), + lambda r: r["status"] != "running", + timeout=120, + ) + assert result["status"] == "completed", result + assert result["exit_code"] == 4 + assert "hello-from-sandbox" in result["stdout"] + + # Files: round-trip through the API and verify inside the sandbox. + payload = b"file-payload-123" + response = client.put( + f"{api_base_url}/sandboxes/{sandbox_id}/files", + params={"path": "/tmp/in.txt"}, + content=payload, + ) + assert response.status_code == 204, response.text + response = client.get( + f"{api_base_url}/sandboxes/{sandbox_id}/files", + params={"path": "/tmp/in.txt"}, + ) + assert response.status_code == 200 + assert response.content == payload + + response = client.get( + f"{api_base_url}/sandboxes/{sandbox_id}/files", + params={"path": "/tmp/does-not-exist"}, + ) + assert response.status_code == 404 + + # Listing sees it. + response = client.get( + f"{api_base_url}/sandboxes", params=[("label", "suite=integration")] + ) + assert sandbox_id in {s["sandbox_id"] for s in response.json()["sandboxes"]} + + # Delete is effective and idempotent. + assert ( + client.delete(f"{api_base_url}/sandboxes/{sandbox_id}").status_code == 200 + ) + _wait_for( + lambda: client.get(f"{api_base_url}/sandboxes/{sandbox_id}").status_code, + lambda code: code == 404, + timeout=60, + ) + assert ( + client.delete(f"{api_base_url}/sandboxes/{sandbox_id}").status_code == 200 + ) + + +if __name__ == "__main__": + sys.exit(pytest.main(["-v", "-s", __file__])) diff --git a/python/ray/experimental/sandbox/http/tests/test_http_schemas.py b/python/ray/experimental/sandbox/http/tests/test_http_schemas.py new file mode 100644 index 000000000000..6f87cc29e608 --- /dev/null +++ b/python/ray/experimental/sandbox/http/tests/test_http_schemas.py @@ -0,0 +1,147 @@ +"""Contract snapshot for the Ray Sandbox HTTP API. + +Pins the v1 surface (paths, methods, and the fields of the wire models) so a +change to it is a deliberate act — clients like Harbor's ``ray-sandbox`` +environment are written against exactly this shape. If this test fails, you +are changing the API contract: update it and the consumers together. +""" + +import sys + +import pytest + +from ray.experimental.sandbox.http.app import create_app +from ray.experimental.sandbox.http.schemas import SandboxAPISettings +from ray.experimental.sandbox.http.tests.conftest import FakeResolver + +EXPECTED_OPERATIONS = { + ("/api/v1/health", "get"), + ("/api/v1/sandboxes", "get"), + ("/api/v1/sandboxes", "post"), + ("/api/v1/sandboxes/{sandbox_id}", "get"), + ("/api/v1/sandboxes/{sandbox_id}", "delete"), + ("/api/v1/sandboxes/{sandbox_id}/execs", "post"), + ("/api/v1/sandboxes/{sandbox_id}/execs/{exec_id}", "get"), + ("/api/v1/sandboxes/{sandbox_id}/files", "put"), + ("/api/v1/sandboxes/{sandbox_id}/files", "get"), +} + +EXPECTED_MODEL_FIELDS = { + "CreateSandboxRequest": { + "image", + "env", + "workdir", + "ttl_seconds", + "network", + "rootless", + "readonly", + "resources", + "labels", + "capabilities", + "image_pull_timeout_seconds", + "start_timeout_seconds", + "client_token", + "dns", + "shell", + }, + "ResourceSpec": { + "cpu_request", + "cpu_limit", + "memory_request_mb", + "memory_limit_mb", + "custom", + }, + "SandboxInfo": { + "sandbox_id", + "status", + "image", + "created_at", + "ttl_seconds", + "expires_at", + "network", + "labels", + "error", + }, + "StartExecRequest": { + "command", + "cwd", + "env", + "timeout_seconds", + "shell", + "user", + }, + "ExecStarted": {"exec_id", "status"}, + "ExecInfo": { + "exec_id", + "status", + "exit_code", + "stdout", + "stderr", + "stdout_truncated", + "stderr_truncated", + "duration_seconds", + "error", + }, +} + +_HTTP_METHODS = {"get", "post", "put", "delete", "patch", "head", "options"} + + +def _openapi() -> dict: + app = create_app(SandboxAPISettings(), handle_resolver=FakeResolver()) + return app.openapi() + + +def test_v1_operations_are_exactly_the_contract() -> None: + schema = _openapi() + operations = { + (path, method) + for path, item in schema["paths"].items() + for method in item + if method in _HTTP_METHODS + } + assert operations == EXPECTED_OPERATIONS + + +def test_wire_models_carry_exactly_the_contract_fields() -> None: + schema = _openapi() + components = schema["components"]["schemas"] + for model, expected_fields in EXPECTED_MODEL_FIELDS.items(): + assert model in components, f"model {model} missing from the OpenAPI schema" + actual = set(components[model].get("properties", {})) + assert actual == expected_fields, f"{model} fields drifted" + + +def test_sandbox_status_enum_is_stable() -> None: + schema = _openapi() + components = schema["components"]["schemas"] + status = components["SandboxInfo"]["properties"]["status"] + # pydantic may inline or $ref the literal; normalize. + enum = status.get("enum") or components.get( + status.get("$ref", "").rsplit("/", 1)[-1], {} + ).get("enum") + assert set(enum) == { + "pending", + "pulling", + "starting", + "running", + "error", + "terminated", + } + + +def test_network_modes_track_the_core_config() -> None: + """Mode validation is delegated to the core config's list, so a new Ray + mode is accepted here without an API change.""" + from ray.experimental.sandbox.config import VALID_NETWORK_MODES + from ray.experimental.sandbox.http.schemas import CreateSandboxRequest + + for mode in VALID_NETWORK_MODES: + request = CreateSandboxRequest(image="python:3.12", network=mode) + assert request.network == mode + with pytest.raises(ValueError, match="network must be one of"): + CreateSandboxRequest(image="python:3.12", network="bridge") + + +if __name__ == "__main__": + sys.exit(pytest.main(["-v", __file__])) diff --git a/python/ray/experimental/sandbox/http/tests/test_sandbox_host.py b/python/ray/experimental/sandbox/http/tests/test_sandbox_host.py new file mode 100644 index 000000000000..5998b85782c0 --- /dev/null +++ b/python/ray/experimental/sandbox/http/tests/test_sandbox_host.py @@ -0,0 +1,462 @@ +"""Unit tests for SandboxHost against a fake runtime (no Ray, no runsc).""" + +import asyncio +import sys +import threading + +import pytest + +from ray.experimental.sandbox.exceptions import ( + SandboxCreationError, + SandboxExecError, + SandboxTimeoutError, +) +from ray.experimental.sandbox.http.host import SandboxHost +from ray.experimental.sandbox.http.schemas import DOCKER_DEFAULT_CAPABILITIES +from ray.experimental.sandbox.http.tests.conftest import ( + FakeExecResult, + FakeSandboxRuntime, +) + + +def _make_host( + runtime: FakeSandboxRuntime, + *, + spec_overrides=None, + settings_overrides=None, +) -> SandboxHost: + spec = { + "image": "python:3.12-slim", + "env": {"TASK_VAR": "1"}, + "workdir": "/", + "ttl_seconds": 3600, + "network": "none", + "rootless": True, + "readonly": False, + "capabilities": list(DOCKER_DEFAULT_CAPABILITIES), + "cpu_limit": 2.0, + "memory_limit_mb": 1024, + "image_pull_timeout_seconds": 300.0, + "start_timeout_seconds": 45.0, + "labels": {"harbor-session-id": "abc"}, + } + spec.update(spec_overrides or {}) + settings = {"max_output_bytes": 10 * 1024 * 1024, "max_exec_history": 256} + settings.update(settings_overrides or {}) + return SandboxHost( + sandbox_id="sb-test00000001", + spec=spec, + settings=settings, + runtime_factory=lambda: runtime, + ) + + +# ---------------------------------------------------------------------- +# Boot +# ---------------------------------------------------------------------- + + +def test_boot_reaches_running_with_expected_create_kwargs() -> None: + runtime = FakeSandboxRuntime() + host = _make_host(runtime) + + async def scenario() -> dict: + await host.boot() + return await host.describe() + + info = asyncio.run(scenario()) + assert info["status"] == "running" + assert info["error"] is None + assert info["image"] == "python:3.12-slim" + assert info["labels"] == {"harbor-session-id": "abc"} + assert info["ttl_seconds"] == 3600 + assert info["expires_at"] is not None + + assert runtime.pull_calls == [ + {"image": "python:3.12-slim", "timeout_seconds": 300.0} + ] + (create_call,) = runtime.create_calls + assert create_call["image"] == "python:3.12-slim" + assert create_call["cpu"] == 2.0 + assert create_call["memory"] == "1024Mi" + assert create_call["env"] == {"TASK_VAR": "1"} + assert create_call["workdir"] == "/" + # The API owns the TTL; the upstream runtime must not double-manage it. + assert create_call["ttl_seconds"] is None + # Writability follows the runtime's (readonly, workdir) contract; no + # extra knobs are forwarded. + assert "mount_workdir" not in create_call + assert create_call["timeout_seconds"] == 45.0 + assert create_call["rootless"] is True + assert create_call["network"] == "none" + assert create_call["readonly"] is False + assert create_call["capabilities"] == list(DOCKER_DEFAULT_CAPABILITIES) + assert "_oci_spec_transform_fn" not in create_call + + +def test_boot_failure_is_pollable_not_raised() -> None: + runtime = FakeSandboxRuntime() + runtime.pull_error = SandboxCreationError("registry says no") + host = _make_host(runtime) + + async def scenario() -> dict: + await host.boot() + return await host.describe() + + info = asyncio.run(scenario()) + assert info["status"] == "error" + assert "registry says no" in info["error"] + # Nothing was created, so nothing to delete. + assert runtime.deleted == [] + + +def test_boot_create_failure_cleans_up_nothing_and_reports() -> None: + runtime = FakeSandboxRuntime() + runtime.create_error = SandboxCreationError("runsc not found") + host = _make_host(runtime) + + async def scenario() -> dict: + await host.boot() + return await host.describe() + + info = asyncio.run(scenario()) + assert info["status"] == "error" + assert "runsc not found" in info["error"] + + +def test_boot_is_idempotent() -> None: + runtime = FakeSandboxRuntime() + host = _make_host(runtime) + + async def scenario() -> None: + await asyncio.gather(host.boot(), host.boot()) + await host.boot() + + asyncio.run(scenario()) + assert len(runtime.pull_calls) == 1 + assert len(runtime.create_calls) == 1 + + +def test_describe_long_poll_wakes_on_status_change() -> None: + runtime = FakeSandboxRuntime() + runtime.pull_gate = threading.Event() + host = _make_host(runtime) + + async def scenario() -> dict: + boot_task = asyncio.create_task(host.boot()) + # Wait until the pull is actually in flight. + while not runtime.pull_calls: + await asyncio.sleep(0.01) + assert (await host.describe())["status"] == "pulling" + + waiter = asyncio.create_task(host.describe(wait_seconds=10.0)) + await asyncio.sleep(0.05) + runtime.pull_gate.set() + info = await asyncio.wait_for(waiter, timeout=5.0) + await boot_task + return info + + info = asyncio.run(scenario()) + # The long-poll woke on the pulling->starting transition (or later). + assert info["status"] in ("starting", "running") + + +# ---------------------------------------------------------------------- +# Capabilities and network passthrough (behavior itself is covered by the +# core sandbox tests; the host only forwards SandboxConfig fields) +# ---------------------------------------------------------------------- + + +def test_network_mode_passed_through_natively() -> None: + runtime = FakeSandboxRuntime() + host = _make_host(runtime, spec_overrides={"network": "sandbox"}) + asyncio.run(host.boot()) + + (create_call,) = runtime.create_calls + assert create_call["network"] == "sandbox" + + +def test_explicit_capabilities_passed_through() -> None: + runtime = FakeSandboxRuntime() + host = _make_host(runtime, spec_overrides={"capabilities": ["CAP_CHOWN"]}) + asyncio.run(host.boot()) + + (create_call,) = runtime.create_calls + assert create_call["capabilities"] == ["CAP_CHOWN"] + + +def test_missing_capabilities_default_to_docker_set() -> None: + runtime = FakeSandboxRuntime() + host = _make_host(runtime, spec_overrides={"capabilities": None}) + asyncio.run(host.boot()) + + (create_call,) = runtime.create_calls + assert create_call["capabilities"] == list(DOCKER_DEFAULT_CAPABILITIES) + + +# ---------------------------------------------------------------------- +# Exec jobs +# ---------------------------------------------------------------------- + + +def test_exec_job_completes_with_passthrough_args() -> None: + runtime = FakeSandboxRuntime() + runtime.exec_results = [ + FakeExecResult(exit_code=7, stdout="out", stderr="err", duration_seconds=1.5) + ] + host = _make_host(runtime) + + async def scenario() -> dict: + await host.boot() + started = await host.start_exec( + ["bash", "-c", "exit 7"], + cwd="/app", + env={"K": "V"}, + timeout_seconds=30.0, + ) + assert started["status"] == "running" + return await host.get_exec(started["exec_id"], wait_seconds=10.0) + + info = asyncio.run(scenario()) + assert info["status"] == "completed" + assert info["exit_code"] == 7 + assert info["stdout"] == "out" + assert info["stderr"] == "err" + assert info["duration_seconds"] == 1.5 + assert info["stdout_truncated"] is False + + exec_call = runtime.exec_calls[-1] + assert exec_call["command"] == ["bash", "-c", "exit 7"] + assert exec_call["cwd"] == "/app" + assert exec_call["env"] == {"K": "V"} + assert exec_call["timeout"] == 30.0 + + +def test_exec_timeout_maps_to_timeout_status() -> None: + runtime = FakeSandboxRuntime() + runtime.exec_error = SandboxTimeoutError("too slow") + host = _make_host(runtime) + + async def scenario() -> dict: + await host.boot() + started = await host.start_exec("sleep 100", timeout_seconds=1.0) + return await host.get_exec(started["exec_id"], wait_seconds=10.0) + + info = asyncio.run(scenario()) + assert info["status"] == "timeout" + assert "timed out after 1.0 seconds" in info["error"] + assert info["exit_code"] is None + + +@pytest.mark.parametrize( + "error", [SandboxExecError("backend broke"), ValueError("surprise")] +) +def test_exec_errors_map_to_error_status(error: Exception) -> None: + runtime = FakeSandboxRuntime() + runtime.exec_error = error + host = _make_host(runtime) + + async def scenario() -> dict: + await host.boot() + started = await host.start_exec("true") + return await host.get_exec(started["exec_id"], wait_seconds=10.0) + + info = asyncio.run(scenario()) + assert info["status"] == "error" + assert str(error) in info["error"] + + +def test_exec_output_truncated_with_marker() -> None: + runtime = FakeSandboxRuntime() + runtime.exec_results = [FakeExecResult(stdout="x" * 100, stderr="y")] + host = _make_host(runtime, settings_overrides={"max_output_bytes": 10}) + + async def scenario() -> dict: + await host.boot() + started = await host.start_exec("true") + return await host.get_exec(started["exec_id"], wait_seconds=10.0) + + info = asyncio.run(scenario()) + assert info["stdout_truncated"] is True + assert info["stdout"].startswith("x" * 10) + assert "[truncated by ray-sandbox" in info["stdout"] + assert info["stderr_truncated"] is False + assert info["stderr"] == "y" + + +def test_exec_rejected_while_not_running() -> None: + runtime = FakeSandboxRuntime() + host = _make_host(runtime) + + async def scenario() -> dict: + return await host.start_exec("true") + + result = asyncio.run(scenario()) + assert result["error_code"] == "conflict" + assert "pending" in result["message"] + + +def test_get_exec_unknown_id() -> None: + runtime = FakeSandboxRuntime() + host = _make_host(runtime) + + async def scenario() -> dict: + await host.boot() + return await host.get_exec("ex-doesnotexist") + + result = asyncio.run(scenario()) + assert result["error_code"] == "exec_not_found" + + +def test_exec_history_evicts_oldest_finished() -> None: + runtime = FakeSandboxRuntime() + host = _make_host(runtime, settings_overrides={"max_exec_history": 2}) + + async def scenario() -> tuple: + await host.boot() + ids = [] + for _ in range(3): + started = await host.start_exec("true") + await host.get_exec(started["exec_id"], wait_seconds=10.0) + ids.append(started["exec_id"]) + first = await host.get_exec(ids[0]) + last = await host.get_exec(ids[-1]) + return first, last + + first, last = asyncio.run(scenario()) + assert first["error_code"] == "exec_not_found" + assert last["status"] == "completed" + + +# ---------------------------------------------------------------------- +# Files +# ---------------------------------------------------------------------- + + +def test_file_roundtrip_and_not_found() -> None: + runtime = FakeSandboxRuntime() + runtime.readable_files["/data/in.txt"] = b"hello" + host = _make_host(runtime) + + async def scenario() -> tuple: + await host.boot() + wrote = await host.write_file("/data/out.txt", b"payload") + read = await host.read_file("/data/in.txt") + missing = await host.read_file("/data/nope.txt") + return wrote, read, missing + + wrote, read, missing = asyncio.run(scenario()) + assert wrote == {"ok": True} + assert runtime.written_files["/data/out.txt"] == b"payload" + assert read["ok"] is True + assert read["content"] == b"hello" + assert missing["error_code"] == "file_not_found" + + +def test_file_ops_conflict_before_running() -> None: + runtime = FakeSandboxRuntime() + host = _make_host(runtime) + + async def scenario() -> tuple: + return ( + await host.write_file("/x", b"1"), + await host.read_file("/x"), + ) + + wrote, read = asyncio.run(scenario()) + assert wrote["error_code"] == "conflict" + assert read["error_code"] == "conflict" + + +# ---------------------------------------------------------------------- +# TTL and termination +# ---------------------------------------------------------------------- + + +def test_ttl_deletes_sandbox_and_marks_terminated() -> None: + runtime = FakeSandboxRuntime() + host = _make_host(runtime, spec_overrides={"ttl_seconds": 0.05}) + + async def scenario() -> dict: + await host.boot() + await asyncio.sleep(0.3) + return await host.describe() + + info = asyncio.run(scenario()) + assert info["status"] == "terminated" + assert runtime.deleted == [runtime.instance_id] + + +def test_terminate_cancels_running_exec_and_deletes() -> None: + runtime = FakeSandboxRuntime() + exec_gate = threading.Event() + original_exec = runtime.exec + + def blocking_exec(*args, **kwargs): + exec_gate.wait(timeout=30) + return original_exec(*args, **kwargs) + + runtime.exec = blocking_exec + host = _make_host(runtime) + + async def scenario() -> tuple: + await host.boot() + started = await host.start_exec("sleep forever") + await asyncio.sleep(0.05) + try: + result = await host.terminate() + info = await host.describe() + exec_info = await host.get_exec(started["exec_id"]) + finally: + exec_gate.set() + return result, info, exec_info + + result, info, exec_info = asyncio.run(scenario()) + assert result == {"ok": True} + assert info["status"] == "terminated" + assert runtime.deleted == [runtime.instance_id] + assert exec_info["status"] == "error" + assert "terminated" in exec_info["error"] + + +def test_terminate_is_idempotent() -> None: + runtime = FakeSandboxRuntime() + host = _make_host(runtime) + + async def scenario() -> None: + await host.boot() + await host.terminate() + await host.terminate() + + asyncio.run(scenario()) + assert runtime.deleted == [runtime.instance_id] + + +def test_shell_passthrough() -> None: + """A sandbox-level shell reaches runtime.create only when set, and a + per-exec shell reaches exec_async.""" + runtime = FakeSandboxRuntime() + host = _make_host(runtime, spec_overrides={"shell": "/bin/sh"}) + + async def scenario() -> dict: + await host.boot() + started = await host.start_exec("echo hi", shell="/bin/dash") + return await host.get_exec(started["exec_id"], wait_seconds=5) + + result = asyncio.run(scenario()) + assert result["status"] == "completed" + assert runtime.create_calls[0]["shell"] == "/bin/sh" + assert runtime.exec_calls[0]["shell"] == "/bin/dash" + + +def test_no_shell_in_spec_keeps_the_runtime_default() -> None: + runtime = FakeSandboxRuntime() + host = _make_host(runtime) + + asyncio.run(host.boot()) + # Omitted, not None: SandboxConfig.shell keeps its /bin/bash default. + assert "shell" not in runtime.create_calls[0] + + +if __name__ == "__main__": + sys.exit(pytest.main(["-v", __file__])) diff --git a/python/ray/experimental/sandbox/runtime.py b/python/ray/experimental/sandbox/runtime.py index 2e24dca6ad00..da0f21b52f05 100644 --- a/python/ray/experimental/sandbox/runtime.py +++ b/python/ray/experimental/sandbox/runtime.py @@ -141,6 +141,7 @@ def exec( cwd: Optional[str] = None, env: Optional[Dict[str, str]] = None, shell: Optional[str] = None, + user: Optional[str] = None, ) -> ExecResult: """Execute a command inside the specified sandbox. @@ -152,6 +153,8 @@ def exec( env: Environment variables to set for the command. shell: Optional shell for string commands, overriding the sandbox's configured shell (default /bin/bash). + user: Optional user to run as: a numeric uid, "uid:gid", or a + name from the image's /etc/passwd (default: the image user). Returns: ExecResult containing exit code, stdout, and stderr. @@ -163,6 +166,7 @@ def exec( cwd=cwd, env=env, shell=shell, + user=user, ) async def exec_async( @@ -173,6 +177,7 @@ async def exec_async( cwd: Optional[str] = None, env: Optional[Dict[str, str]] = None, shell: Optional[str] = None, + user: Optional[str] = None, ) -> ExecResult: """Execute a command inside the specified sandbox asynchronously. @@ -184,6 +189,8 @@ async def exec_async( env: Environment variables to set for the command. shell: Optional shell for string commands, overriding the sandbox's configured shell (default /bin/bash). + user: Optional user to run as: a numeric uid, "uid:gid", or a + name from the image's /etc/passwd (default: the image user). Returns: ExecResult containing exit code, stdout, and stderr. @@ -196,6 +203,7 @@ async def exec_async( cwd=cwd, env=env, shell=shell, + user=user, ) def upload_file(self, instance_id: str, local_path: str, remote_path: str) -> None: @@ -228,7 +236,11 @@ def download_file( f.write(content) def write_file( - self, instance_id: str, path: str, content: Union[str, bytes] + self, + instance_id: str, + path: str, + content: Union[str, bytes], + append: bool = False, ) -> None: """Write string or binary content directly to a file inside the sandbox. @@ -236,8 +248,9 @@ def write_file( instance_id: Unique identifier of the sandbox instance. path: Destination file path inside the sandbox. content: String or binary content to write into the file. + append: Append to the file instead of truncating it. """ - self._backend.write_file(instance_id, path, content) + self._backend.write_file(instance_id, path, content, append=append) def read_file(self, instance_id: str, path: str) -> bytes: """Read binary content from a file inside the sandbox. diff --git a/python/ray/experimental/sandbox/tests/conftest.py b/python/ray/experimental/sandbox/tests/conftest.py index f52c7308b9c5..52b65d53d01b 100644 --- a/python/ray/experimental/sandbox/tests/conftest.py +++ b/python/ray/experimental/sandbox/tests/conftest.py @@ -8,6 +8,11 @@ from ray._private.test_utils import sandbox_test_enabled +_RUNSC_URL = ( + "https://storage.googleapis.com/gvisor/releases/release/latest/{arch}/runsc" +) +_PASTA_URL = "https://passt.top/builds/latest/x86_64/pasta" + def pytest_runtest_setup(item): if not sandbox_test_enabled(): @@ -15,26 +20,80 @@ def pytest_runtest_setup(item): os.environ["RAY_SANDBOX_IGNORE_CGROUPS"] = "1" +def _install_on_path(name: str, url: str) -> None: + """Fetch a static binary into a temp dir prepended to PATH, or skip.""" + if shutil.which(name): + return + bin_dir = tempfile.mkdtemp() + os.chmod(bin_dir, 0o755) + binary = os.path.join(bin_dir, name) + try: + urllib.request.urlretrieve(url, binary) + except Exception as e: + pytest.skip(f"Failed to install {name} for sandbox tests: {e}") + os.chmod(binary, 0o755) + os.environ["PATH"] = f"{bin_dir}:{os.environ.get('PATH', '')}" + + @pytest.fixture(scope="session", autouse=True) def ensure_runsc(): if not sandbox_test_enabled(): return - os.environ["RAY_SANDBOX_IGNORE_CGROUPS"] = "1" + arch = "aarch64" if platform.machine().lower() in ("aarch64", "arm64") else "x86_64" + _install_on_path("runsc", _RUNSC_URL.format(arch=arch)) + + +def _public_netns_supported() -> bool: + """Whether this host can actually bring a network="public" sandbox up. + + The path parks a sandbox's network+user namespaces in an + ``unshare --user --net`` holder and has pasta and a non-rootless runsc + re-enter them via ``nsenter -U -n``. Some sandboxed CI environments permit + a single unprivileged user namespace (enough for the rootless sandbox + tests) and even entering another process's, yet still forbid the *nested* + user namespace runsc opens when it drops the sandbox process to ``nobody`` + -- which only surfaces once the sandbox boots, as ``Started as root, will + change to nobody. Couldn't open user namespace ...: Permission denied``. A + namespace-entry probe passes there and the tests then fail, so instead + bring a throwaway busybox sandbox all the way up and tear it down: only the + real path exercises that nested open. runsc and pasta must already be on + PATH. + """ + from ray.experimental.sandbox.backend.gvisor import GVisorSandboxBackend + from ray.experimental.sandbox.config import GVisorSandboxConfig + from ray.experimental.sandbox.exceptions import SandboxError + + backend = GVisorSandboxBackend() + try: + sandbox_id = backend.create_sandbox( + GVisorSandboxConfig( + image="busybox:latest", shell="/bin/sh", network="public" + ) + ) + except SandboxError: + return False + backend.delete_sandbox(sandbox_id) + return True + + +@pytest.fixture(scope="session") +def ensure_pasta(ensure_runsc): + """pasta plus a host that can actually run the network="public" path. - if not shutil.which("runsc"): - temp_bin = tempfile.mkdtemp() - os.chmod(temp_bin, 0o755) - runsc_path = os.path.join(temp_bin, "runsc") - arch = ( - "aarch64" - if platform.machine().lower() in ("aarch64", "arm64") - else "x86_64" + Requested rather than autouse, so only the netns tests skip when no static + pasta build exists for the arch or the environment forbids the per-sandbox + user+network namespace the path relies on. pasta and runsc are installed + first because the support probe boots a real sandbox. + """ + if not shutil.which("pasta") and platform.machine().lower() not in ( + "x86_64", + "amd64", + ): + pytest.skip("no static pasta build for this arch; install passt") + _install_on_path("pasta", _PASTA_URL) + if not _public_netns_supported(): + pytest.skip( + 'network="public" needs a per-sandbox user+network namespace this ' + "environment forbids (nested user namespace denied at sandbox boot)" ) - url = f"https://storage.googleapis.com/gvisor/releases/release/latest/{arch}/runsc" - try: - urllib.request.urlretrieve(url, runsc_path) - os.chmod(runsc_path, 0o755) - os.environ["PATH"] = f"{temp_bin}:{os.environ.get('PATH', '')}" - except Exception as e: - pytest.skip(f"Failed to install runsc for sandbox tests: {e}") diff --git a/python/ray/experimental/sandbox/tests/test_gvisor_backend.py b/python/ray/experimental/sandbox/tests/test_gvisor_backend.py index 8684755f4e6a..4f6bbf835a5d 100644 --- a/python/ray/experimental/sandbox/tests/test_gvisor_backend.py +++ b/python/ray/experimental/sandbox/tests/test_gvisor_backend.py @@ -1,5 +1,8 @@ import os +import socket +import subprocess import sys +from pathlib import Path import pytest @@ -11,6 +14,7 @@ from ray.experimental.sandbox.config import GVisorSandboxConfig from ray.experimental.sandbox.exceptions import ( SandboxCreationError, + SandboxExecError, SandboxNotFoundError, ) from ray.experimental.sandbox.runtime import SandboxRuntime @@ -305,5 +309,238 @@ def test_image_workdir_sets_cwd_without_becoming_writable(): runtime.delete(instance_id) +def test_resolve_exec_user(tmp_path, monkeypatch): + """Numeric users pass through; names resolve via the image's passwd.""" + backend = GVisorSandboxBackend() + img_dir = tmp_path / "img" + (img_dir / "rootfs" / "etc").mkdir(parents=True) + (img_dir / "rootfs" / "etc" / "passwd").write_text( + "root:x:0:0:root:/root:/bin/bash\n" + "postfix:x:102:104::/var/spool/postfix:/usr/sbin/nologin\n" + ) + monkeypatch.setattr( + backend._image_manager, "get_image_dir", lambda image: str(img_dir) + ) + + assert backend._resolve_exec_user("1000", "img") == "1000" + assert backend._resolve_exec_user("1000:1000", "img") == "1000:1000" + assert backend._resolve_exec_user("postfix", "img") == "102:104" + # "name:gid" resolves the name and honors the explicit gid. + assert backend._resolve_exec_user("postfix:999", "img") == "102:999" + with pytest.raises(SandboxExecError): + backend._resolve_exec_user("nosuch", "img") + + +def test_kill_pasta_missing_pidfile_is_noop(tmp_path): + """No pasta.pid (the non-public path) means nothing to reap.""" + GVisorSandboxBackend()._kill_pasta(str(tmp_path)) + + +def test_kill_pasta_signals_recorded_pid(tmp_path): + """A recorded pasta pid is SIGKILLed on teardown.""" + proc = subprocess.Popen(["sleep", "30"]) + (tmp_path / "pasta.pid").write_text(f"{proc.pid}\n") + GVisorSandboxBackend()._kill_pasta(str(tmp_path)) + assert proc.wait(timeout=5) != 0 + + +_PUBLIC_HOST_NETNS_ENV = "RAY_SANDBOX_PUBLIC_HOST_NETNS" + + +@pytest.fixture +def no_host_netns(monkeypatch): + """Default posture: the shared-host-network kill switch is unset.""" + monkeypatch.delenv(_PUBLIC_HOST_NETNS_ENV, raising=False) + + +def _public_config() -> GVisorSandboxConfig: + return GVisorSandboxConfig( + image="busybox:latest", shell="/bin/sh", network="public" + ) + + +def _run_argv(network: str, rootless: bool = True, **backend_kwargs) -> list: + """`_build_run_command` over fixed paths — pure argv, no side effects.""" + backend = GVisorSandboxBackend(**backend_kwargs) + cfg = GVisorSandboxConfig( + image="busybox:latest", network=network, rootless=rootless + ) + return backend._build_run_command(cfg, "/tmp/rd", "/tmp/rd/overlay", "sb-1") + + +def _host_ip() -> str: + """The worker's primary IPv4, which pasta copies onto each sandbox's tap.""" + with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock: + sock.connect(("8.8.8.8", 80)) + return sock.getsockname()[0] + + +def _pasta_pids() -> set: + """PIDs of running pasta processes.""" + pids = set() + for entry in Path("/proc").iterdir(): + if not entry.name.isdigit(): + continue + try: + argv0 = (entry / "cmdline").read_bytes().split(b"\0", 1)[0] + except OSError: + continue + if os.path.basename(argv0) == b"pasta": + pids.add(entry.name) + return pids + + +def test_build_run_command_public_wraps_with_pasta(no_host_netns): + """The pasta flags are the isolation property and the chain's shape is the + topology, so pin both: holder namespaces first, pasta attached from the pod + side, runsc entered as mapped root (never --rootless).""" + cmd = _run_argv("public") + assert cmd[:2] == ["bash", "-c"] + script = cmd[2] + + assert script.startswith( + "unshare --user --map-root-user --net --fork --kill-child " + ) + assert script.endswith("run --bundle /tmp/rd sb-1") + for fragment in ( + "/tmp/rd/netns.pid", + "pasta --config-net -t none -u none -T none -U none --no-map-gw -4 " + "--netns /proc/$NSPID/ns/net --userns /proc/$NSPID/ns/user", + # pasta's daemon pid is recorded so teardown can reap it. + "--pid /tmp/rd/pasta.pid", + # The wait loop fast-fails if the holder dies and refuses an empty + # NSPID (which would resolve to /proc//ns/net). + "kill -0 $HOLDER", + '[ -n "$NSPID" ]', + "exec nsenter --preserve-credentials -U -n -t $NSPID -- runsc", + "--network host", + "--overlay2=root:dir=/tmp/rd/overlay", + ): + assert fragment in script, fragment + assert "--rootless" not in script + + +@pytest.mark.parametrize( + "network,rootless", [("none", True), ("host", True), ("sandbox", False)] +) +def test_build_run_command_other_modes_unwrapped(no_host_netns, network, rootless): + """Every mode but "public" keeps today's bare runsc invocation.""" + cmd = _run_argv(network, rootless=rootless) + assert cmd[0] == "runsc" + assert "pasta" not in cmd + assert cmd[cmd.index("--network") + 1] == network + assert cmd[-4:] == ["run", "--bundle", "/tmp/rd", "sb-1"] + + +def test_public_host_netns_kill_switch(monkeypatch): + """The env kill switch restores the pre-netns shared-host behavior.""" + monkeypatch.setenv(_PUBLIC_HOST_NETNS_ENV, "1") + cmd = _run_argv("public") + assert cmd[0] == "runsc" + assert "pasta" not in cmd + assert cmd[cmd.index("--network") + 1] == "host" + + +def test_create_sandbox_requires_pasta(no_host_netns, monkeypatch): + """A missing pasta fails fast — before the image pull — with remediation.""" + + class _NoPullImageManager: + def pull_image(self, *args, **kwargs): + raise AssertionError("image pull must not run when pasta is missing") + + monkeypatch.setattr( + "ray.experimental.sandbox.backend.gvisor.shutil.which", + lambda name: None if name == "pasta" else f"/usr/bin/{name}", + ) + backend = GVisorSandboxBackend(image_manager=_NoPullImageManager()) + with pytest.raises(SandboxCreationError) as err: + backend.create_sandbox(_public_config()) + assert all( + hint in str(err.value) + for hint in ("pasta", "auto_install_pasta", _PUBLIC_HOST_NETNS_ENV) + ) + + +def test_netns_concurrent_same_port_bind_and_isolation(ensure_pasta): + """Two "public" sandboxes both bind 0.0.0.0:2222 (the terminal-bench QEMU + hostfwd contract): each reaches its own listener, the bind never surfaces in + the worker's namespace, and neither sandbox can reach the other's.""" + backend = GVisorSandboxBackend() + sb1, sb2 = (backend.create_sandbox(_public_config()) for _ in range(2)) + tokens = {sb1: "SB1-TOKEN", sb2: "SB2-TOKEN"} + try: + for sb, token in tokens.items(): + # /tmp stays a writable tmpfs on the readonly rootfs. + backend.write_file(sb, "/tmp/www/token", token) + # busybox httpd daemonizes; sharing one netns, the second bind + # would fail with EADDRINUSE. + res = backend.exec_command(sb, "httpd -p 2222 -h /tmp/www", timeout=30) + assert res.exit_code == 0, res.stderr + + for sb, token in tokens.items(): + res = backend.exec_command( + sb, "wget -q -T 5 -O - http://127.0.0.1:2222/token", timeout=30 + ) + assert res.exit_code == 0, res.stderr + assert token in res.stdout + + # The worker's own namespace must see nothing on 2222. + host_ip = _host_ip() + for target in ("127.0.0.1", host_ip): + with pytest.raises(OSError): + socket.create_connection((target, 2222), timeout=3).close() + + # No address names one sandbox from another: pasta --config-net gives + # every sandbox the worker's own IP, so from sb2 that IP is sb2 itself. + res = backend.exec_command( + sb2, f"wget -q -T 3 -O - http://{host_ip}:2222/token", timeout=30 + ) + assert tokens[sb1] not in res.stdout + assert res.exit_code != 0 or tokens[sb2] in res.stdout + finally: + for sb in (sb1, sb2): + backend.delete_sandbox(sb) + + +def test_netns_egress_and_dns(ensure_pasta): + """Egress and the generated resolv.conf work from inside the netns.""" + backend = GVisorSandboxBackend() + sb = backend.create_sandbox(_public_config()) + try: + res = backend.exec_command( + sb, "wget -q -T 15 -O - http://example.com", timeout=60 + ) + assert res.exit_code == 0, res.stderr + assert "Example" in res.stdout + finally: + backend.delete_sandbox(sb) + + +def test_netns_teardown_reaps_pasta(ensure_pasta): + """delete_sandbox ends the pasta process tree and removes all state.""" + backend = GVisorSandboxBackend() + sb = backend.create_sandbox(_public_config()) + meta = backend._sandbox_metadata[sb] + assert meta["proc"].poll() is None + + backend.delete_sandbox(sb) + assert meta["proc"].poll() is not None + assert not os.path.exists(meta["root_dir"]) + assert sb not in backend._sandbox_metadata + + +def test_netns_create_failure_leaves_no_pasta(ensure_pasta): + """A failed create (bad image) leaves no pasta process behind.""" + before = _pasta_pids() + backend = GVisorSandboxBackend() + with pytest.raises(SandboxCreationError): + backend.create_sandbox( + GVisorSandboxConfig( + image="nonexistent_invalid_image_12345:latest", network="public" + ) + ) + assert _pasta_pids() == before + + if __name__ == "__main__": sys.exit(pytest.main(["-v", __file__]))