You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by ka...@apache.org on 2026/10/14 11:13:30 UTC
(airflow) branch main updated: Add `OpenShell` backend for sandbox tools (#73961)
This is an automated email from the ASF dual-hosted git repository.
kaxil pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 98391ff597b Add `OpenShell` backend for sandbox tools (#73961)
98391ff597b is described below
commit 98391ff597b37d27e8dca26a6e4965c7410b0eb1
Author: Yossi Eliaz <zo...@users.noreply.github.com>
AuthorDate: Wed Oct 14 14:13:20 2026 +0300
Add `OpenShell` backend for sandbox tools (#73961)
* Add OpenShell backend for sandbox tools
Add OpenShellSandboxBackend, a self-hosted backend that runs SandboxToolset
sandboxes through an NVIDIA OpenShell gateway on Docker, Podman or
Kubernetes. OpenShell confines each sandbox with Landlock and seccomp, gives
it no network interface of its own and mediates egress in a per-sandbox
supervisor, so the toolset's deny-all default and a hostname allowlist on
port 443 are enforced by the runtime rather than approximated.
The backend keeps the fail-closed contract itself instead of trusting the
gateway. It reads the effective policy back after create, and before and
after every command, and destroys the sandbox when the policy differs from
the request, comes from a gateway-wide override, or can be widened by
auto-approved proposals.
It also works around gaps measured in the 0.1.2 SDK and gateway: an exec
timeout that leaves the command running (a guest wrapper kills the
command's session), client-side buffering of all output (events are
streamed into per-stream caps), the 32 KiB argument limit (commands and
file content go over stdin), the 19-character name limit, proxy and CA
variables the supervisor silently rewrites (refused), and gateway restarts
(reported as an unknown outcome, never as a success).
Credentials are ambient, read from the OpenShell CLI's gateway
registration, so no connection type is added. The extra states protobuf and
grpcio floors that the SDK wheel under-declares, and the root pyproject
carries a temporary exclude-newer override for the two-day-old release.
* Drop the expired openshell exclude-newer pin
openshell 0.1.2 is now outside the rolling 4-day window, so the per-package
cutoff the pin comment scheduled for removal on 2026-10-03 is no longer needed.
* Fix OpenShell long-line output, retry drops and backend copying
A single output line longer than the cap came back as no output at all,
because the leading partial line was always dropped; it is now kept unless
that would discard more than half the window, as in the Modal backend. A
gateway drop during the not-ready retry escaped run_command as a private
exception. The client lock is module level so the backend deep-copies like
Modal's, results report applied_timeout, and the write_file docstring
states the real reason it chunks.
* Restore the blank line before the sandbox example paragraph
Without it the bullet list above runs into the paragraph and the docs build
fails on a warning treated as an error.
* Time the OpenShell command budget on elapsed time
The guest wrapper's timer counted iterations of `sleep 1`, and each
iteration forks under the sandbox's CPU quota, so on a busy sandbox the
delay grew with the budget. A 120 s budget fired after 147 s with 32
busy processes, and a 300 s one after 367 s, past the gRPC deadline of
budget plus 30 s; the backend then took the hang for a stalled gateway
and destroyed the sandbox and its files over an ordinary timeout.
The timer now fires once /proc/uptime has advanced by the budget. It is
monotonic, has centisecond resolution so the timer never fires early,
and is read with a shell builtin, so a starved timer is at most one
sleep late instead of one per second of budget.
* Fail an OpenShell command that could not be staged in /tmp
The wrapper ignored a failure to write the command to /tmp. With /tmp
nearly full it ran the truncated prefix that did fit, and with /tmp
full it ran an empty file; either way it reported that run's status,
usually 0, so the model was told a command succeeded that never ran
in full.
A failed mktemp or cat now removes the staging directory and exits 125
with a fixed stderr line, and run_command raises a recoverable
SandboxError for that pair. The stderr line is needed as well as the
status because commands such as docker run, env and nohup exit 125 on
their own errors, and those still reach the model as exit codes.
* Test OpenShell policy-read retries, middlewares and file hangs
Three behaviours of the OpenShell backend had no test that failed
without them: the policy read waiting out a restarting gateway, the
refusal of a policy with network middlewares, and a hung file
operation being a recoverable error that keeps the sandbox. Disabling
any of them left the suite green, and the only test that noticed a
missing retry did so by counting time.monotonic calls.
The new tests cover the retry on UNAVAILABLE and DEADLINE_EXCEEDED,
its bound at the recovery window, the middleware refusal and the hung
read, and each fails when its behaviour is removed.
* Correct the OpenShell sandbox docs
The rewrap of the teardown paragraph in the sandbox overview had
dropped the blank line before the "Why not KubernetesPodOperator or the
KubernetesExecutor?" title, so docutils folded the title into the
paragraph without a warning and the section vanished from the page.
The paragraph is back to its original wrapping with the blank line.
On the backends page, the Modal intro still counted two backends that
run on Kubernetes, and the OpenShell credentials paragraph gave a false
reason for having no connection (Airflow does mask extras whose keys
look sensitive) and compared the setup to ~/.modal.toml, which Modal
now only falls back to. The paragraph now says where the credentials
live and that the SDK reads them itself.
The timeout sweep was described as rescanning until no process is
left, in the docs, the backend docstring and the wrapper comment. It
stops at the first pass that kills nothing, after at most 50 passes,
so a command that keeps forking can outlast it. The docs also say that
a command that cannot be staged in /tmp is reported and not run.
* Sweep /proc without forking in the OpenShell timeout wrapper
The timeout sweep forked a `cat` for every /proc entry, under the same
CPU quota as the processes it was killing, so its cost grew faster than
linearly with their number. With a 5 s budget and 128 busy processes
on one CPU the wrapper returned 30 to 98 s after the budget, past the
gRPC deadline of budget plus 30 s, and the backend then destroyed the
sandbox over an ordinary timeout.
The sweep now reads each stat line with the `read` builtin and strips
the process name with the shortest match, falling back to the longest
match only when a name contains a parenthesis; on a 300-byte stat line
the longest match alone cost as much as the read. Measured in
python:3.12-slim with --cpus=1 and a reaping init, as the sandbox has:
128 busy processes and a 5 s budget now return 1.5 to 3.6 s after the
budget, and 256 with a 10 s budget 7 to 22 s, against 142 s before.
The starved-timer test bounded the time after SIGCONT at 1.5 s, which
also covered the forking sweep and failed with 300 processes in the
namespace. It now stops the wrapper for its whole 6 s budget, so a
timer that counts its own sleeps would be 5 s late, and allows 3 s;
with 600 processes and 8 busy loops on 2 CPUs it measured at most
1.1 s. A new test covers a process whose name contains ") ".
* Report an OpenShell staging failure without advice that misleads
The error for a command that could not be staged in /tmp told the model
to free space by overwriting a large file there with write_file. That
does not hold in the cases that matter: list_directory goes through the
same wrapper and fails the same way, so the model cannot find the file;
the space can be held by an unlinked spool file that a background
process from an earlier command is still writing; and /tmp usually
shares a filesystem with the workdir. The message also named /tmp twice.
It now says the command was not run and gives the cause the shell
printed, such as "No space left on device", and nothing else.
The failure was recognised by searching stderr as capped at
max_output_bytes, so under a cap shorter than the 50-byte staging line
it came back as a plain exit 125. The stderr buffer now keeps at least
that line's length for the check, while the text and the truncation
flag still follow the cap, and the check requires stderr to end with
the line.
The staging test is split and parametrized so that the staging line
with another exit status, and a cap below the line's length, are each
covered by a case that fails without the matching half of the check.
* Rewrap the OpenShell commands paragraph
Shortening the sweep sentence left "that keeps forking faster than the
sweep. If the gateway" as a short line in the middle of the paragraph.
The rest of the paragraph is rewrapped; no wording changes.
* Keep the real sleep in the OpenShell wrapper tests
The module's autouse fixture patches time.sleep through the backend
module, whose `time` is the global module, so every sleep in
TestRunWrapper returned at once. The CPU-starvation test then never
held the wrapper stopped, the wrapper ran its whole 6 s budget after
the resume, and the test failed on Linux in the full suite; the
settling waits in the two timeout tests did nothing either. The figure
of at most 1.1 s in the earlier sweep commit came from a copy of the
class that left the fixture out, so it does not describe this suite.
TestRunWrapper now overrides the fixture with one that leaves sleep
alone. In the Airflow CI image (python3.12) the whole file passes, and
the starvation test sees the wrapper exit 0.02 to 0.04 s after the
resume; with the old timer that counts its own sleeps it is 5.1 s and
the test fails.
* Test that the OpenShell staging line must end stderr
The staging check was tightened to require stderr to end with the
wrapper's staging line, but no test failed when the line only appeared
somewhere in stderr. A command can print that text and keep going, and
exit 125 on its own, so a status of 125 with the line followed by more
output has to stay the command's own result. That case is now in the
parametrized test, which fails if the check looks for the line anywhere
in stderr, and the test is renamed to match what its cases share.
* Kill OpenShell processes whose name holds a newline on timeout
The timeout sweep read only the first line of each /proc/<pid>/stat.
The kernel writes the process name there unescaped, so a process whose
name contains a newline, because it was run through such a file name
or renamed with prctl, left its session field on a line the sweep never
read, and it survived the timeout. The `cat` the sweep used before read
the whole file and killed it.
The sweep now joins every line of the file with the `read` builtin, so
it still forks nothing per entry. The wrapper test for a name with a
parenthesis is parametrized with a newline name, which survives the
single-line read. In python:3.12-slim with --cpus=1 and an init, 128
busy processes and a 5 s budget returned 1.4 to 2.6 s after the
budget, against 2.9 to 4.6 s with the single-line read, so the extra
read per entry costs nothing measurable.
* Use datetime.UTC in the sandbox system test
Main now targets Python 3.11, so ruff's UP017 asks for datetime.UTC
instead of timezone.utc and fails the static checks.
* Drop the Python 3.11 markers from the OpenShell extra
Every distribution now requires Python 3.11 (#74157), so the
python_version >= "3.11" markers on the openshell extra and dev
dependencies, their copies in uv.lock and the extras tables, and the
"needs Python 3.11 or later" notes in the docs and error message
no longer say anything.
* Fail the task when the OpenShell image has no setsid
Without setsid every command came back as exit 125 with a stderr line the
model read as its own failure, so it retried a command it could never fix
from inside a sandbox with no network. The wrapper now reports the missing
binary with its own status and a private last stderr line, the same way it
reports a staging failure, and run_command turns that into a terminal error
naming the image requirement.
* Explain a read-only OIDC registration for OpenShell
The SDK refreshes a stale OIDC token inside the call's auth interceptor and
writes it back next to the gateway registration. On a registration mounted
read-only, as a Kubernetes Secret or ConfigMap is, that write raises an
OSError out of every call, which the backend reported as an opaque failure
of whatever it was doing. Turning off the write-back would break refresh
token rotation across workers, so the backend now recognises the failed
write-back and fails the task with what to change, and the docs say that
an OIDC registration must be writable and per worker and that mTLS is the
supported setup for a read-only one.
* Type the OpenShell protobuf messages from the SDK stubs
The SDK ships .pyi stubs for its generated messages, but the backend typed
the sandbox config, the sandbox, the spec and the workspace selector as Any,
so mypy could not see a misspelled field on any of them. They now carry the
stub types, imported for type checking only.
* Refuse an OpenShell file path the gateway would reject
read_file and write_file pass the path as a command argument, and the
gateway rejects an argument with a NUL in it or over 32 KiB of UTF-8 with
INVALID_ARGUMENT, which the backend reports as terminal. A bad path is the
model's to correct, so both operations now refuse it as a recoverable error
before anything is sent. list_directory sends its path on stdin and is not
affected.
* Say what an abandoned OpenShell command leaves behind
The docs said the sandbox is destroyed when the gateway stops relaying a
command. The backend only asks that same gateway to delete it and logs a
failed delete, so the sandbox can survive and is left for the operator's
reaper. Both the commands section and the portability note now say so.
* Pin the OpenShell filesystem policy and exec count in tests
The filesystem-policy test compared the read-write paths with the constant
they come from and never looked at the read-only paths or include_workdir,
so widening the policy left it green. It now spells out every path. The
gateway-error test did not count exec calls, so retrying on any terminal
status, rather than only on a sandbox that is not ready, also passed it.
* Check stderr and the marker in the OpenShell system test
The exit-code step only looked for the exit status, so a backend that lost
stderr passed it, and the shell step only looked for 42, not the marker the
first step wrote. Each step can now require several strings, and these two
check what the OpenSandbox system test checks.
* Pin which OS errors read as a read-only OIDC registration
Only a failed write of the SDK's .oidc_token.* temp file means the
registration is read-only. Nothing kept the mapping from catching any
OSError, which would tell an operator with an unreadable mTLS key or a
full disk to switch to mTLS. These cases now must keep the generic
message.
* Stop a command from faking the OpenShell wrapper's failures
The wrapper reported a missing setsid as status 127 with a fixed last
stderr line, so a command that printed that line and exited 127 was
indistinguishable from it and failed the task; a faked staging line
likewise passed for an unrun command. run_command now sends a per-call
nonce as the first stdin line, the wrapper reads it with the read builtin
and appends it to both lines, and only a line carrying that nonce counts.
The nonce stays out of the command's argv, environment and staged script,
so printing the fixed text is no longer enough.
* Pin the OpenShell path limit from the accepted side
The refused-path test only covered paths over the limit, so an
off-by-one or a stricter threshold still passed: the error message
prints the constant, not the threshold that was checked. A path of
exactly 32 KiB of UTF-8, and one with control characters but no NUL,
must now reach the gateway for both read_file and write_file.
* Read only permission errors as a read-only OIDC registration
A full disk or a used-up quota fails the SDK's .oidc_token temp file on a
writable registration too, and telling the operator to switch to mTLS for
those would send them the wrong way.
* Narrow the OpenShell docs to what a live run measured
A live run against an OpenShell 0.1.2 gateway on the Docker driver showed
the workload keeps a loopback interface, and that a restarted gateway does
not always bring a sandbox back, in which case the task fails. Only the
Docker driver has been exercised, and on Kubernetes the cluster's CNI has
to enforce NetworkPolicy for egress to stay behind the supervisor.
* Scope the OpenShell loopback-only claim to the Docker driver
The live run measured a loopback-only interface on OpenShell's Docker
driver. On Kubernetes the Agent Sandbox controller creates a separate
workload Pod, so the claim is not known to hold there.
* Stop the OpenShell backend overstating what happened to a command
Three failure paths told the model something stronger than the backend
knew. The gateway answers DEADLINE_EXCEEDED after about 10 seconds when
its relay into the sandbox does not open, before the command starts, and
that was taken for a hung command, so a healthy sandbox was deleted and
the model saw a timeout for a command that never ran. A dropped
connection to a gateway that stays up leaves the command running until
the wrapper's timer fires, yet the model was told it had been stopped and
could run it again. And a widened sandbox whose delete failed was still
reported as destroyed, with nothing logged for the operator who has to
reap it, since OpenShell has no server-side lifetime.
The output-cap test now checks the buffer itself, because the returned
text is sliced to the cap anyway and passed with the trim removed.
Claude-Session: https://claude.ai/code/session_01W3CHee8ErP7UArBK3SKZUa
* Prove the OpenShell timeout kill in the sandbox that timed out
If the wrapper's kill on timeout broke, the backend would abandon the
command and destroy the sandbox, and the next step would count sleepers
in the replacement, where there are none, so the system test passed
either way. Reading the file written at the start of the run in the same
command shows the count comes from the sandbox that timed out.
Claude-Session: https://claude.ai/code/session_01W3CHee8ErP7UArBK3SKZUa
* Stop presenting OpenShell as verified for production and Kubernetes
The backend has only been verified on OpenShell's Docker driver, which
OpenShell itself scopes to local development and single-machine
gateways, but the overview listed it with Modal and OpenSandbox as a
production and Kubernetes backend, contradicting the backend page.
Claude-Session: https://claude.ai/code/session_01W3CHee8ErP7UArBK3SKZUa
* Treat only OpenShell's relay timeout as a command that did not run
An early DEADLINE_EXCEEDED can also come from a proxy capping stream duration after the command started, so only the gateway's own relay-open timeout is reported as not run; anything else is reported as possibly still running.
* Check the OpenShell policy around file operations and reserve wrapper variable names
read_file and write_file now run the same egress check as run_command before
and after the operation, and refuse a named pipe instead of blocking on it with
nothing to end the wait. The command wrapper's shell variables are renamed under
an _airflow_ prefix that SandboxSpec.env may not use, so an env key can no longer
share a name with the variable holding the per-call nonce and receive it.
list_directory fails the task when the sandbox was destroyed during the listing
and reports a timed-out listing as recoverable. write_file copies one chunk at a
time, and an abandoned command logs why its sandbox is being destroyed.
Docs stop recommending OpenShell for Kubernetes, cover it in the network
configuration section, and say when to pick it over OpenSandbox. Tests cover the
/proc rescan deterministically and bind the wait before a not-ready retry.
* Say which shells the OpenShell wrapper supports and tighten round-4 review gaps
BusyBox ash, the /bin/sh of Alpine images, never finishes the wrapper's timeout
sweep, so the docs and docstring now ask for dash or bash. The policy re-check
is described as running after each operation that completes, which is what the
code does, and a test covers a policy widened during a file operation. A test
checks every variable the wrapper assigns carries the reserved prefix. A listing
the gateway stopped relaying no longer claims its sandbox was destroyed, and the
docs drop the Linux-only gateway host claim and say a reaper is only needed when
the delete itself fails.
* Align OpenShell policy and timeout wording with what the backend does
The policy re-check runs after operations that succeed, not every one that ends,
a sandbox left behind by a BusyBox timeout is only asked to be deleted, and a
deny-all connection fails with Permission denied rather than being refused. Add
colima to the spelling wordlist.
---------
Co-authored-by: Kaxil Naik <ka...@gmail.com>
---
docs/spelling_wordlist.txt | 12 +
providers/common/ai/README.rst | 1 +
providers/common/ai/docs/index.rst | 1 +
providers/common/ai/docs/installation.rst | 7 +-
providers/common/ai/docs/sandbox/backends.rst | 221 ++-
providers/common/ai/docs/sandbox/configuration.rst | 21 +-
providers/common/ai/docs/sandbox/index.rst | 48 +-
providers/common/ai/provider.yaml | 3 +
providers/common/ai/pyproject.toml | 13 +
.../providers/common/ai/get_provider_info.py | 5 +
.../providers/common/ai/sandbox/__init__.py | 2 +
.../providers/common/ai/sandbox/openshell.py | 1139 +++++++++++++++
.../common/ai/tests/system/common/ai/README.md | 17 +
.../common/ai/example_sandbox_toolset_openshell.py | 122 ++
.../tests/unit/common/ai/sandbox/test_openshell.py | 1479 ++++++++++++++++++++
uv.lock | 33 +-
16 files changed, 3082 insertions(+), 42 deletions(-)
diff --git a/docs/spelling_wordlist.txt b/docs/spelling_wordlist.txt
index 00f4f3081b5..c41bd9013db 100644
--- a/docs/spelling_wordlist.txt
+++ b/docs/spelling_wordlist.txt
@@ -1,5 +1,6 @@
aarch
abc
+ABI
AbstractCapability
AbstractFileSystem
AbstractToolset
@@ -284,11 +285,13 @@ cmdlet
cmds
cncf
cnf
+CNI
cnt
codebase
codec
codecov
codepoints
+colima
Colour
colour
colours
@@ -371,6 +374,7 @@ Cypher
cypher
daemonize
daemonized
+daemonizing
dagbag
DagCallbackRequest
dagfile
@@ -687,6 +691,7 @@ filesystems
filetype
FilteredToolset
finalStatus
+findutils
fips
firebase
Firehose
@@ -1018,6 +1023,7 @@ Kylin
kylin
Kyverno
Lakehouse
+Landlock
langchain
Langfuse
LanguageServiceClient
@@ -1234,6 +1240,8 @@ openlineage
opensandbox
OpenSearch
opensearch
+OpenShell
+openshell
OpenTelemetry
oper
OperatorLineage
@@ -1298,6 +1306,7 @@ pinecone
Pinot
pinot
PipelineRun
+PKI
pkill
plaintext
pluggable
@@ -1305,6 +1314,7 @@ pluggy
plyvel
png
PoC
+Podman
PodManager
PodSpec
podSpec
@@ -1482,6 +1492,7 @@ requeues
requeuing
rerank
reranking
+rescanning
reserialize
Reserializing
resetdb
@@ -1553,6 +1564,7 @@ sdk
sdks
searchpath
SearchResultGenerator
+seccomp
secretmanager
SecretManagerClient
SecretManagerServiceClient
diff --git a/providers/common/ai/README.rst b/providers/common/ai/README.rst
index a7f249e9a81..c53022d38f0 100644
--- a/providers/common/ai/README.rst
+++ b/providers/common/ai/README.rst
@@ -96,6 +96,7 @@ Extra Dependencies
``modal`` ``apache-airflow-providers-modal``, ``modal>=1.5.2``
``opensandbox`` ``opensandbox>=1.1.0``
``boat`` ``boat-sdk>=1.9.3``
+``openshell`` ``openshell>=0.1.2,<0.2``, ``grpcio>=1.78.0``, ``protobuf>=6.31.1``
``code-mode`` ``pydantic-ai-harness[codemode]>=0.24.0``
``shields`` ``pydantic-ai-shields>=0.3.4``
``skills`` ``apache-airflow-providers-git>=0.4.0``, ``pydantic-ai-skills>=1.2.0``
diff --git a/providers/common/ai/docs/index.rst b/providers/common/ai/docs/index.rst
index 47ec2a7e043..33383e29213 100644
--- a/providers/common/ai/docs/index.rst
+++ b/providers/common/ai/docs/index.rst
@@ -234,6 +234,7 @@ Extra Dependencies
``modal`` ``apache-airflow-providers-modal``, ``modal>=1.5.2``
``opensandbox`` ``opensandbox>=1.1.0``
``boat`` ``boat-sdk>=1.9.3``
+``openshell`` ``openshell>=0.1.2,<0.2``, ``grpcio>=1.78.0``, ``protobuf>=6.31.1``
``code-mode`` ``pydantic-ai-harness[codemode]>=0.24.0``
``shields`` ``pydantic-ai-shields>=0.3.4``
``skills`` ``apache-airflow-providers-git>=0.4.0``, ``pydantic-ai-skills>=1.2.0``
diff --git a/providers/common/ai/docs/installation.rst b/providers/common/ai/docs/installation.rst
index d7ab8dd7417..7b72f0a9781 100644
--- a/providers/common/ai/docs/installation.rst
+++ b/providers/common/ai/docs/installation.rst
@@ -48,9 +48,10 @@ The provider's extras split into a few groups:
than these, each under its own extra name, so check the
`pydantic-ai install docs <https://ai.pydantic.dev/install/#slim-install>`__ for the full list.
* **Agent tooling** (``mcp``, ``skills``, ``code-mode``, ``shields``, ``modal``,
- ``opensandbox``, ``boat``, ``islo``): MCP servers, Agent Skills, code-mode tool
- execution, shield capabilities (input/output guards, tool guards, cost tracking), and
- the hosted Modal, Boat and Islo backends and the self-hosted OpenSandbox backend for
+ ``opensandbox``, ``openshell``, ``boat``, ``islo``): MCP servers, Agent Skills, code-mode
+ tool execution, shield capabilities (input/output guards, tool guards, cost tracking),
+ and the hosted Modal, Boat and Islo backends and the self-hosted OpenSandbox and
+ OpenShell backends for
:doc:`sandboxed execution <sandbox/index>`.
* **Document loading** (``pdf``, ``docx``, ``avro``, ``parquet``): file formats for
document pipelines.
diff --git a/providers/common/ai/docs/sandbox/backends.rst b/providers/common/ai/docs/sandbox/backends.rst
index 2ecc67a142e..4a6f34848b3 100644
--- a/providers/common/ai/docs/sandbox/backends.rst
+++ b/providers/common/ai/docs/sandbox/backends.rst
@@ -31,9 +31,9 @@ Modal (hosted)
:class:`~airflow.providers.common.ai.sandbox.modal.ModalSandboxBackend` runs each
sandbox in Modal, provisioned over the API. Of the backends that ship with the
provider, **this is the managed one to use in production**, and with
-:ref:`OpenSandbox <sandbox-backend-opensandbox>`, :ref:`Boat <sandbox-backend-boat>` and
-:ref:`Islo <sandbox-backend-islo>` one of the four that run on Kubernetes: nothing has to
-be installed on the worker, model-written code never executes on the worker host, and Modal reclaims
+:ref:`OpenSandbox <sandbox-backend-opensandbox>`, :ref:`OpenShell <sandbox-backend-openshell>`,
+:ref:`Boat <sandbox-backend-boat>` and :ref:`Islo <sandbox-backend-islo>` one of the five a
+worker on Kubernetes can use: nothing has to be installed on the worker, model-written code never executes on the worker host, and Modal reclaims
a sandbox at its own lifetime whether or not the worker survives. It needs the
``modal`` extra and Modal credentials, from a ``modal`` connection or the worker
environment, as under :ref:`Quick start <sandbox-quick-start>`.
@@ -343,6 +343,191 @@ The runtime remains a deployment choice. The default Docker runtime shares the
host kernel; choose a stronger runtime such as Kata when your threat model needs
a VM boundary.
+.. _sandbox-backend-openshell:
+
+OpenShell (self-hosted remote)
+------------------------------
+
+:class:`~airflow.providers.common.ai.sandbox.openshell.OpenShellSandboxBackend`
+runs sandboxes through an `NVIDIA OpenShell <https://github.com/NVIDIA/OpenShell>`__
+gateway, which runs each one on Docker, Podman or Kubernetes. Inside the
+container the workload is confined by Landlock and seccomp; on the Docker driver it
+has only a loopback network interface, and a per-sandbox supervisor opens every
+outbound connection on its behalf, against a policy the backend writes and reads back. Airflow workers
+only need gRPC access to the gateway.
+
+Choose it over :ref:`OpenSandbox <sandbox-backend-opensandbox>` when you want the
+sandbox runtime itself to enforce deny-all and the hostname allowlist, rather than
+an egress sidecar that has to run in the right mode. It is Apache-2.0 and runs on a
+single machine with no account, so it is also the self-hosted backend to try first
+on a laptop. For a production gateway on Kubernetes, prefer OpenSandbox or Modal:
+this backend has been verified on OpenShell's Docker driver only, which OpenShell
+scopes to local development and single-machine gateways.
+
+Install the SDK extra:
+
+.. code-block:: bash
+
+ pip install "apache-airflow-providers-common-ai[openshell]"
+
+**Credentials are ambient; there is no Airflow connection for this backend.**
+The gateway's endpoint, mTLS material and OIDC token live in the gateway
+registration that the ``openshell`` CLI keeps under
+``$XDG_CONFIG_HOME/openshell/gateways/<name>/`` (``~/.config`` by default), and
+the OpenShell SDK reads them from there itself: ``metadata.json`` with the
+endpoint, and ``mtls/ca.crt``, ``mtls/tls.crt`` and ``mtls/tls.key`` for mTLS,
+or the CLI's cached token for an OIDC gateway. The backend only names the
+registration. A worker without the CLI needs the same files, provisioned by the
+Deployment Manager:
+
+.. code-block:: json
+
+ {"name": "prod", "gateway_endpoint": "https://openshell.example.com:17670", "auth_mode": "mtls"}
+
+.. code-block:: python
+
+ from airflow.providers.common.ai.sandbox import OpenShellSandboxBackend
+ from airflow.providers.common.ai.toolsets import SandboxToolset
+
+ SandboxToolset(OpenShellSandboxBackend(gateway="prod"))
+
+An OIDC registration must be writable and per worker. The SDK refreshes an
+expiring OIDC token itself and writes the new one back into the registration,
+so a registration mounted read-only, from a Kubernetes Secret or ConfigMap for
+example, fails the task at the first refresh, and copies of one registration on
+several workers share a refresh token that the first refresh can invalidate for
+the others. mTLS is the supported setup for workers that mount the registration
+read-only.
+
+Whoever holds that credential is inside the trust boundary. On a gateway
+without OIDC, an mTLS client is a gateway-wide administrator: it can change the
+policy and settings of every sandbox. ``openshell-gateway generate-certs`` issues
+a single client identity whose certificate does not expire for practical
+purposes, so use your own PKI, with expiry and rotation, for a worker credential.
+
+**Network policy.** A default ``SandboxSpec()`` sends a policy with no egress
+rule, which OpenShell enforces as no egress: a connection fails with
+``EACCES``, and a name lookup is answered by the supervisor with a synthetic
+``198.18.0.0/15`` address rather than failing, so no query leaves the sandbox
+although the tool description's "including DNS" wording reads stricter than the
+lookup behaves. ``allow_egress_to`` becomes one rule admitting each listed host
+on port 443 for any program in the sandbox; ports are part of every rule, so
+plain HTTP and other ports stay closed. HTTPS to a listed host is terminated by
+the supervisor, which injects its CA through ``SSL_CERT_FILE``,
+``REQUESTS_CA_BUNDLE`` and similar variables; a client with its own trust store
+(a JVM, a statically linked binary) has to be pointed at it. A leading ``*.``
+label is accepted on a name of three labels or more.
+
+The backend verifies the policy rather than trusting the request. After create
+it reads the effective policy back and destroys the sandbox unless it admits
+exactly the requested hosts, comes from the sandbox rather than a gateway-wide
+policy, keeps Landlock at ``hard_requirement``, and has neither
+``proposal_approval_mode=auto`` nor agent policy proposals enabled at any scope.
+Auto-approval is refused because OpenShell turns a denied connection into a
+proposed allow rule and, in that mode, approves it without review. The same
+check runs before every command and file operation and after each one that
+succeeds, because an administrator, an approved
+draft or a gateway-wide policy can widen a running sandbox; a change destroys the
+sandbox and fails the task. That detects a widening, it does not prevent one: a
+command already running when the policy changes can use it. The check sees
+OpenShell's policy, not the network under the gateway: on Kubernetes, a sandbox
+pod is kept behind its supervisor by a NetworkPolicy, which only a CNI that
+enforces NetworkPolicy upholds. Every denied connection also becomes a draft
+proposal in the gateway's approval inbox, even in manual mode, so operators see
+which destinations an agent tried.
+
+Refused at create: ``block_network=False`` (every OpenShell rule names a host and
+its ports, so an open network cannot be expressed), ``allow_egress_to_cidrs``
+(address rules admit TCP on listed ports only), ``SandboxSpec.owner``, and
+``SandboxSpec.env`` keys the supervisor owns: it removes ``HTTP_PROXY`` and the
+other proxy variables from every command and replaces ``SSL_CERT_FILE``,
+``REQUESTS_CA_BUNDLE``, ``CURL_CA_BUNDLE``, ``GIT_SSL_CAINFO``,
+``NODE_EXTRA_CA_CERTS`` and ``DENO_CERT`` with its own, without an error, and
+reserves ``OPENSHELL_*``. Keys starting with ``_airflow_`` are refused too, because the
+command wrapper below names its own shell variables that way. Everything else in
+``SandboxSpec.env``, ``PATH`` included, reaches commands as given, since they run
+without a login shell.
+
+**Commands.** OpenShell's own exec timeout reports exit 124 and leaves the
+command running, so the backend runs each command through a small shell wrapper
+in the sandbox. The command arrives on stdin and runs in a session of its own,
+with the command and its output spooled to ``/tmp``; one that cannot be written
+there, because ``/tmp`` is full for example, is not run and is reported as an
+error. When the budget runs out the wrapper kills the processes in that session,
+scanning again until a pass finds none to kill, at most 50 times, and returns the
+tail of each stream. A process the command left in the background keeps running
+after a command that finishes in time, and one that started a session of its own
+(``setsid``, a daemonizing server) escapes the kill on timeout, as can a command
+that keeps forking faster than the sweep. If the gateway stops relaying the
+command for longer than its budget plus 30 seconds, ``sandbox_terminated`` is
+reported and the backend asks the gateway to delete the sandbox; if that call
+fails too, it logs a warning and the sandbox is left for the reaper described
+below. Nothing crosses the stream
+until the command ends, so a proxy or load balancer in front of the gateway
+needs an idle timeout longer than the longest command budget plus 30 seconds. A
+command the gateway could not start, because its relay into the sandbox did not
+open, is reported to the model as not run. A gateway restart stops every process
+in its sandboxes. If the gateway brings the sandbox back, its files are kept and a
+command in flight is reported to the model as having an unknown outcome rather
+than retried; if it does not, the task fails. A dropped connection to a gateway
+that stays up is reported the same way, but does not stop the command: it runs
+until its budget runs out, and the model is told it may still be running.
+
+**Files.** ``write_file`` sends content on stdin in chunks of 768 KiB, since the
+gateway limits one command argument to 32 KiB and one request to 1 MiB; a larger
+write is not atomic, and it follows a symlink like a shell redirect. ``read_file``
+transfers raw bytes, capped inside the guest. Both refuse a named pipe, since
+opening one waits for another process and nothing would end the wait in the
+sandbox; ``run_command`` with a timeout can still read or write one. The image needs ``setsid`` (from
+util-linux), GNU coreutils and findutils, and a ``/bin/sh`` that is dash or bash,
+which ``python:*-slim`` and other Debian or Ubuntu based images have; an image
+without ``setsid`` fails the task at its first command. BusyBox ``ash``, the
+``/bin/sh`` of Alpine images, is not supported: the timeout sweep does not finish
+under it, so a command that runs out of time is not killed, and after its budget
+plus 30 seconds the backend asks the gateway to delete the sandbox instead. The gateway's own default image has no ``python3``, which is
+why the backend defaults to ``python:3.12-slim``.
+
+**Cleanup.** OpenShell has no server-side sandbox lifetime. ``destroy`` deletes
+the sandbox, but one whose worker died is kept until someone deletes it, even
+after its main process has ended. Sandboxes are named ``airflow-`` plus 11 hex
+characters, within OpenShell's 19-character limit, and labeled
+``created-by=airflow`` and ``airflow-created-at=<unix seconds>``, so a reaper can
+delete by age:
+
+.. code-block:: python
+
+ import time
+
+ from openshell import SandboxClient
+
+ client = SandboxClient.from_active_cluster(cluster="prod")
+ cutoff = time.time() - 6 * 3600
+ for sandbox in client.list_all(workspace="default", label_selector="created-by=airflow"):
+ if int(sandbox.labels.get("airflow-created-at", "0")) < cutoff:
+ client.delete(sandbox.name, workspace="default", allow_missing=True)
+
+Constructor parameters:
+
+- ``gateway``: the CLI gateway registration to use. ``None`` (default) uses
+ ``$OPENSHELL_GATEWAY``, then the CLI's active gateway.
+- ``workspace``: OpenShell workspace. Default ``"default"``.
+- ``image``: image used by the gateway. Default ``"python:3.12-slim"``.
+- ``cpu`` and ``memory``: resource limits as Kubernetes quantities. Defaults
+ ``"1"`` and ``"2Gi"``. ``nproc`` in the sandbox still reports the host's CPUs.
+- ``ready_timeout``: seconds to wait for a new sandbox. Default ``120``; a first
+ pull of a large image can take longer, so pre-pull it on the gateway host.
+- ``request_timeout``: seconds for each gateway call other than a command.
+ Default ``30``.
+
+The kernel the sandboxes run on needs Landlock ABI 3 or later (Linux 6.2+) and
+seccomp user notification, and the gateway needs Docker 28 or later, Podman 5, or
+a Kubernetes cluster with the Agent Sandbox controller. On macOS the gateway runs
+against the Linux VM of a Docker runtime such as OrbStack or colima. OpenShell scopes its Docker driver to
+local development and single-machine gateways. This backend has been verified with
+the Docker driver only; on Kubernetes, egress enforcement also depends on the
+cluster's CNI enforcing NetworkPolicy. The gateway sends anonymous
+telemetry unless it runs with ``OPENSHELL_TELEMETRY_ENABLED=false``.
+
.. _sandbox-backend-boat:
Boat (hosted)
@@ -526,15 +711,19 @@ do not change. Four behaviours do, so read them before assuming the same Dag
behaves identically everywhere:
- **CPU.** ``sbx`` gives a sandbox every host CPU; Modal defaults to a request of
- 0.125 of one, so set ``cpu``; OpenSandbox takes ``cpu`` as a limit the server enforces;
- Boat sizes by ``machine_type``; Islo uses the server default unless you set ``vcpus``.
+ 0.125 of one, so set ``cpu``; OpenSandbox and OpenShell take ``cpu`` as a limit the
+ server enforces; Boat sizes by ``machine_type``; Islo uses the server default unless
+ you set ``vcpus``.
- **Egress allowlists.** ``sbx`` enforces ``allow_egress_to`` at the host policy
layer; Modal matches TLS handshake names, which is weaker and has to be opted
into; OpenSandbox enforces it in an egress sidecar, and the backend reads the
enforced policy and the sidecar's mode back rather than trusting the create
request, refusing a sidecar that filters only name resolution (``dns``) rather
- than connections (``dns+nft``). ``allow_egress_to_cidrs`` is enforced at the
- address layer on Modal and refused on ``sbx``, OpenSandbox and Islo. Islo has no
+ than connections (``dns+nft``); OpenShell enforces it in a per-sandbox supervisor
+ on port 443 only, and reads the effective policy back at create and around every
+ operation. ``allow_egress_to_cidrs`` is enforced at the address layer on Modal and
+ refused on ``sbx``, OpenSandbox, OpenShell and Islo; OpenShell's address rules admit
+ TCP on listed ports only, and it also refuses ``block_network=False``. Islo has no
per-host form and refuses the spec; Boat has no form of egress control at all and
refuses every network restriction. Both refuse rather than provision something
weaker.
@@ -542,15 +731,19 @@ behaves identically everywhere:
Modal, Boat and a server-enforced OpenSandbox timeout preserve the sandbox and
files. OpenSandbox destroys it only if the command event stream itself stalls
past the client-side grace period, and Boat only if its own deadline, set past
- the one in the guest, ends the call.
-- **Symlinks.** ``write_file`` through a symlink follows the link on ``sbx`` and
- Boat, which write the target and leave the link in place, and replaces the link
- with a regular file on Modal and OpenSandbox.
+ the one in the guest, ends the call. OpenShell kills the command's session and
+ keeps the sandbox; if the gateway stops relaying the command, the backend asks it
+ to delete the sandbox, and only a delete that fails too leaves it for the reaper
+ described under :ref:`OpenShell <sandbox-backend-openshell>`.
+- **Symlinks.** ``write_file`` through a symlink follows the link on ``sbx``, Boat
+ and OpenShell, which write the target and leave the link in place, and replaces
+ the link with a regular file on Modal and OpenSandbox.
- **Attaching.** A Modal sandbox can be provisioned by one task and used by an
agent in another (:ref:`sandbox-attach`). An ``sbx`` microVM lives on the worker
- that created it and cannot be reached from another task, and OpenSandbox, Boat and
- Islo have no per-sandbox metadata the ownership rules could be kept in, so all four
- refuse ``SandboxSpec.owner`` and the toolset refuses ``attach_to`` for them.
+ that created it and cannot be reached from another task, OpenSandbox, Boat and
+ Islo have no per-sandbox metadata the ownership rules could be kept in, and
+ OpenShell does not implement attaching yet, so all five refuse ``SandboxSpec.owner``
+ and the toolset refuses ``attach_to`` for them.
.. _sandbox-byo:
diff --git a/providers/common/ai/docs/sandbox/configuration.rst b/providers/common/ai/docs/sandbox/configuration.rst
index 49e868ba77e..948ea366cbe 100644
--- a/providers/common/ai/docs/sandbox/configuration.rst
+++ b/providers/common/ai/docs/sandbox/configuration.rst
@@ -58,7 +58,11 @@ an open sandbox. On OpenSandbox the default and ``allow_egress_to`` become a pol
for the server's egress sidecar, and the backend destroys the sandbox and fails the task unless that
sidecar runs in ``dns+nft`` mode, the one that drops connections to unlisted
addresses rather than only name resolution (see
-:ref:`the OpenSandbox backend <sandbox-backend-opensandbox>`). ``allow_egress_to``
+:ref:`the OpenSandbox backend <sandbox-backend-opensandbox>`). On OpenShell they become
+a policy the sandbox runtime enforces, which the backend reads back before every
+operation and after each one that succeeds; a name lookup there is answered with a synthetic address instead of
+failing, though no query leaves the sandbox (see
+:ref:`the OpenShell backend <sandbox-backend-openshell>`). ``allow_egress_to``
names hosts the sandbox may reach, and ``sbx`` and Modal refuse it until you have
said something that makes it enforceable:
``sbx`` applies it as a per-sandbox rule on top of that ``deny-all`` host policy,
@@ -68,15 +72,18 @@ opted into with ``egress_enforcement="sni"`` (see
:ref:`the Modal backend <sandbox-backend-modal>` for exactly what it does and does
not stop). ``allow_egress_to_cidrs`` names address ranges instead, and on Modal it
is enforced on the destination address for any port, so it needs no opt-in; it is
-the mode for one service at a fixed public IPv4 address, and ``sbx`` refuses it. A
+the mode for one service at a fixed public IPv4 address, and ``sbx`` and OpenShell refuse it. A
private address is unreachable from a hosted sandbox whether or not it is listed.
The two lists can be set together, and traffic matching either is allowed, which
weakens the address list on port 443.
``block_network=False`` opens outbound access; on Modal the cloud metadata
-endpoint and private address ranges stay unreachable even then.
+endpoint and private address ranges stay unreachable even then. OpenShell refuses
+it, because every egress rule there names a host and its ports.
**Packages.** A default sandbox has the Python standard library and no network,
-so an agent that reaches for ``pip install`` gets a DNS failure in a few seconds.
+so an agent that reaches for ``pip install`` gets a DNS failure in a few seconds
+(on OpenShell, where lookups get a synthetic answer, a connection that fails with
+"Permission denied").
Bake what it needs into the image, which needs no egress at all:
.. code-block:: python
@@ -360,9 +367,9 @@ toolset's own.
**Which backends.** Modal sandboxes can be found by id from any process, so
``ModalSandboxBackend`` supports this. ``sbx`` runs a microVM on the worker that
-created it and cannot be reached from another task, and ``OpenSandboxBackend`` and
-``BoatSandboxBackend`` have nowhere on a sandbox to keep the owner and holder, so all
-three refuse
+created it and cannot be reached from another task, ``OpenSandboxBackend`` and
+``BoatSandboxBackend`` have nowhere on a sandbox to keep the owner and holder, and
+``OpenShellSandboxBackend`` does not implement attaching yet, so all four refuse
``SandboxSpec.owner`` and ``attach_to``. A backend of your own opts in by
implementing :class:`~airflow.providers.common.ai.sandbox.AttachableSandboxBackend`
(:ref:`sandbox-byo`).
diff --git a/providers/common/ai/docs/sandbox/index.rst b/providers/common/ai/docs/sandbox/index.rst
index ce093fe5735..318d0c03338 100644
--- a/providers/common/ai/docs/sandbox/index.rst
+++ b/providers/common/ai/docs/sandbox/index.rst
@@ -73,7 +73,7 @@ stays where the task runs:
end
sbx["sbx microVM<br/>local development only"]
end
- remote["Modal, OpenSandbox, Boat or Islo sandbox<br/>run_command, read_file,<br/>write_file, list_directory"]
+ remote["Modal, OpenSandbox, OpenShell, Boat or Islo sandbox<br/>run_command, read_file,<br/>write_file, list_directory"]
model <--> loop
db <--> other
loop <--> other
@@ -89,15 +89,16 @@ agent loop, the model calls and every other toolset keep the task's authority, s
sandbox contains model-written code, not the agent (:ref:`sandbox-security`). Under
``KubernetesExecutor`` the outer box is the task's pod: the pod protects the cluster
from the task, and the sandbox protects the task from its model. An ``sbx`` microVM
-cannot run inside an unprivileged pod, so that combination needs Modal, OpenSandbox, Boat
-or Islo.
+cannot run inside an unprivileged pod, so that combination needs Modal, OpenSandbox,
+OpenShell, Boat or Islo.
The credential that provisions the sandbox depends on the backend: a ``modal``
connection (``modal_conn_id``; see the :ref:`Modal connection page <howto/connection:modal>`)
for Modal; ``boat_conn_id``, or ``BOAT_API_KEY`` and optional ``BOAT_BASE_URL``, for Boat;
``ISLO_API_KEY`` (and optional ``ISLO_BASE_URL`` and ``ISLO_COMPUTE_URL``) for Islo;
``opensandbox_conn_id``, or ``OPEN_SANDBOX_DOMAIN`` and ``OPEN_SANDBOX_API_KEY``,
-for OpenSandbox; and the host's ``sbx login`` for ``sbx``.
+for OpenSandbox; the worker's ``openshell`` CLI gateway registration for OpenShell;
+and the host's ``sbx login`` for ``sbx``.
The four tools:
@@ -120,11 +121,13 @@ The four tools:
- Lists a directory. Directories are shown with a trailing ``/``.
A :class:`~airflow.providers.common.ai.sandbox.SandboxBackend` provisions the sandbox
-on the model's first tool call and tears it down when the agent run ends. Five
+on the model's first tool call and tears it down when the agent run ends. Six
backends ship: `Modal <https://modal.com/docs/guide/sandbox>`__,
`Boat <https://docs.boat.dev/quickstart>`__ and `Islo <https://islo.dev>`__ (hosted) and
`OpenSandbox <https://open-sandbox.ai/>`__ (self-hosted) for production and
-Kubernetes, and `Docker Sandboxes <https://docs.docker.com/ai/sandboxes/>`__ (``sbx``,
+Kubernetes, `NVIDIA OpenShell <https://github.com/NVIDIA/OpenShell>`__ (self-hosted,
+verified on its Docker driver only; see :ref:`OpenShell <sandbox-backend-openshell>`),
+and `Docker Sandboxes <https://docs.docker.com/ai/sandboxes/>`__ (``sbx``,
a microVM on the worker host) for local development. The tool names are the ones
pydantic-ai's `Shell <https://pydantic.dev/docs/ai/harness/shell/>`__ and
`FileSystem <https://pydantic.dev/docs/ai/harness/filesystem/>`__ capabilities use.
@@ -351,6 +354,14 @@ leaves its contents alone: the same four tools run in the sandbox.
Kubernetes.
- Your OpenSandbox server's Docker host or Kubernetes cluster, off the worker.
- Ended by the OpenSandbox server at ``sandbox_timeout``.
+ * - ``OpenShellSandboxBackend``
+ - A container confined by Landlock and seccomp; egress goes through a
+ per-sandbox supervisor (on the Docker driver, the workload has only a
+ loopback network interface).
+ - Your OpenShell gateway's Docker or Podman host or Kubernetes cluster, off the
+ worker. Verified with the Docker driver only.
+ - Left running. No server-side lifetime; sandboxes are labeled
+ ``created-by=airflow`` so an operator can reap them.
* - ``BoatSandboxBackend``
- A full Linux VM that Boat provisions per sandbox.
- Boat's infrastructure, off the worker.
@@ -363,9 +374,9 @@ leaves its contents alone: the same four tools run in the sandbox.
disables that backstop.
When a run ends normally, the task calls the backend's ``destroy``. ``sbx`` runs its
-removal command and waits up to two minutes for it; Modal, OpenSandbox, Boat and Islo each
-send a
-termination request and return without waiting for the sandbox to stop. Any of them
+removal command and waits up to two minutes for it; Modal, OpenSandbox, OpenShell, Boat
+and Islo each send a termination request and return without waiting for the sandbox to
+stop. Any of them
can return with the sandbox still present, and none of those cases fails the task. A
SIGKILL, an out-of-memory kill or a lost node skips that teardown entirely, and then
only the last column applies.
@@ -431,7 +442,7 @@ task, a script the model writes, runs, and fixes from its own traceback.
:ref:`sandbox-security`.
- **Nothing survives the run** in a sandbox the toolset provisions itself,
including across task retries. On Modal, a sandbox a task provisions and the agent
- attaches to does; ``sbx``, OpenSandbox, Boat and Islo cannot be attached to.
+ attaches to does; ``sbx``, OpenSandbox, OpenShell, Boat and Islo cannot be attached to.
:ref:`Lifecycle <sandbox-lifecycle>`,
:ref:`A sandbox another task owns <sandbox-attach>`.
- **A file the agent built leaves through** ``exports`` **or a task**, never through
@@ -452,12 +463,15 @@ task, a script the model writes, runs, and fixes from its own traceback.
``SandboxSpec()`` is refused on ``sbx`` until the host policy is declared
``deny-all``. Modal's hostname allowlist is a weak control and has to be opted into;
its address allowlist is enforced properly but cannot serve a package registry whose
- addresses rotate. Boat cannot restrict egress at all, so it refuses every network
+ addresses rotate. OpenShell enforces a hostname allowlist on port 443 in its
+ per-sandbox supervisor and refuses ``allow_egress_to_cidrs`` and an open network.
+ Boat cannot restrict egress at all, so it refuses every network
restriction, including the default ``block_network=True``: pass
``SandboxSpec(block_network=False)`` to accept open egress there. Islo turns outbound
access on or off and refuses hostname and CIDR allowlists.
:ref:`Configuring a sandbox <sandbox-configuring>`,
:ref:`Modal <sandbox-backend-modal>`,
+ :ref:`OpenShell <sandbox-backend-openshell>`,
:ref:`Boat <sandbox-backend-boat>`,
:ref:`Islo <sandbox-backend-islo>`.
- **On Modal, commands run as root and** ``workdir`` **is not a jail.**
@@ -468,8 +482,10 @@ task, a script the model writes, runs, and fixes from its own traceback.
- **The** ``sbx`` **backend is for local development.** It needs the ``sbx`` binary,
a Docker login and, on Linux, KVM or nested virtualization, so it does not run on
unprivileged Kubernetes, and a worker killed outright leaves its microVM behind.
- Production and Kubernetes use Modal, OpenSandbox, Boat or Islo instead.
- :ref:`sbx <sandbox-backend-sbx>`.
+ Production and Kubernetes use Modal, OpenSandbox, Boat or Islo instead. OpenShell
+ has been verified on its Docker driver only, and has no server-side lifetime either;
+ its sandboxes are labeled ``created-by=airflow`` so an operator can reap them.
+ :ref:`sbx <sandbox-backend-sbx>`, :ref:`OpenShell <sandbox-backend-openshell>`.
- **A failed teardown is logged, not raised**, so a teardown blip cannot fail a
finished run; reclaiming the sandbox is then the backend's lifetime or an
operator's sweep. :ref:`Cost and operations <sandbox-cost>`.
@@ -479,15 +495,15 @@ has the two agents in the quick start, a ``@task`` producing a file through a
sandbox, an agent attaching to a task-owned sandbox, and an agent exporting a file,
shown on this page and in :doc:`configuration`. The system tests
``example_sandbox_toolset_sbx.py``, ``example_sandbox_toolset_modal.py``,
-``example_sandbox_toolset_opensandbox.py``, ``example_sandbox_toolset_boat.py`` and
-``example_sandbox_toolset_islo.py``
+``example_sandbox_toolset_opensandbox.py``, ``example_sandbox_toolset_openshell.py``,
+``example_sandbox_toolset_boat.py`` and ``example_sandbox_toolset_islo.py``
run against a real backend and are
reachable from the System Tests entry in the sidebar.
**Credentials and where it runs.** Only what ``SandboxSpec.env`` names enters the
sandbox; the provisioning credential for each backend is listed under
:ref:`sandbox-placement`. Work runs in Modal's, Boat's or Islo's infrastructure, on the
-OpenSandbox server's runtime (off the worker unless you run the server there), or in a per-run
+OpenSandbox or OpenShell server's runtime (off the worker unless you run the server there), or in a per-run
microVM on the worker host with ``sbx``. Its tool calls act as barriers, as they do
for the other routes that build their own tools; see :ref:`toolset-call-barriers`.
diff --git a/providers/common/ai/provider.yaml b/providers/common/ai/provider.yaml
index a65b35c3a2a..ee1e188fc1b 100644
--- a/providers/common/ai/provider.yaml
+++ b/providers/common/ai/provider.yaml
@@ -83,6 +83,9 @@ integrations:
- integration-name: Boat
external-doc-url: https://docs.boat.dev/
tags: [service]
+ - integration-name: NVIDIA OpenShell
+ external-doc-url: https://github.com/NVIDIA/OpenShell
+ tags: [software]
hooks:
- integration-name: Pydantic AI
diff --git a/providers/common/ai/pyproject.toml b/providers/common/ai/pyproject.toml
index 0b93658996e..9aebf776494 100644
--- a/providers/common/ai/pyproject.toml
+++ b/providers/common/ai/pyproject.toml
@@ -112,6 +112,16 @@ dependencies = [
]
"opensandbox" = ["opensandbox>=1.1.0"]
"boat" = ["boat-sdk>=1.9.3"]
+# NVIDIA OpenShell gateways for SandboxToolset. The SDK wheel declares protobuf>=4.25 and
+# grpcio>=1.60, but its generated code imports only with protobuf>=6.31.1 and grpcio>=1.78.0,
+# so those floors are stated here. Capped below 0.2 because a minor release may break the
+# Stable protos (OpenShell RFC 0014), and the backend builds requests from the SDK's
+# generated messages.
+"openshell" = [
+ "openshell>=0.1.2,<0.2",
+ "grpcio>=1.78.0",
+ "protobuf>=6.31.1",
+]
# Code mode: collapse tool calls into a single `run_code` tool that the model
# drives by writing Python, executed in the Monty sandbox (pydantic-monty).
# Provides the CodeMode capability for AgentOperator(capabilities=[CodeMode()]).
@@ -190,6 +200,9 @@ dev = [
"boat-sdk>=1.9.3",
"islo>=0.3.19",
"opensandbox>=1.1.0",
+ "openshell>=0.1.2,<0.2",
+ "grpcio>=1.78.0",
+ "protobuf>=6.31.1",
]
# To build docs:
diff --git a/providers/common/ai/src/airflow/providers/common/ai/get_provider_info.py b/providers/common/ai/src/airflow/providers/common/ai/get_provider_info.py
index 100f4f067ef..07bba3befff 100644
--- a/providers/common/ai/src/airflow/providers/common/ai/get_provider_info.py
+++ b/providers/common/ai/src/airflow/providers/common/ai/get_provider_info.py
@@ -83,6 +83,11 @@ def get_provider_info():
"tags": ["software"],
},
{"integration-name": "Boat", "external-doc-url": "https://docs.boat.dev/", "tags": ["service"]},
+ {
+ "integration-name": "NVIDIA OpenShell",
+ "external-doc-url": "https://github.com/NVIDIA/OpenShell",
+ "tags": ["software"],
+ },
],
"hooks": [
{
diff --git a/providers/common/ai/src/airflow/providers/common/ai/sandbox/__init__.py b/providers/common/ai/src/airflow/providers/common/ai/sandbox/__init__.py
index 3e770ed26b7..271d20633c8 100644
--- a/providers/common/ai/src/airflow/providers/common/ai/sandbox/__init__.py
+++ b/providers/common/ai/src/airflow/providers/common/ai/sandbox/__init__.py
@@ -35,6 +35,7 @@ from airflow.providers.common.ai.sandbox.boat import BoatSandboxBackend
# actionable error from the backend instead.
from airflow.providers.common.ai.sandbox.islo import IsloSandboxBackend
from airflow.providers.common.ai.sandbox.opensandbox import OpenSandboxBackend
+from airflow.providers.common.ai.sandbox.openshell import OpenShellSandboxBackend
from airflow.providers.common.ai.sandbox.sbx import SbxSandboxBackend
__all__ = [
@@ -42,6 +43,7 @@ __all__ = [
"BoatSandboxBackend",
"IsloSandboxBackend",
"OpenSandboxBackend",
+ "OpenShellSandboxBackend",
"ModalSandboxBackend",
"SandboxBackend",
"SandboxError",
diff --git a/providers/common/ai/src/airflow/providers/common/ai/sandbox/openshell.py b/providers/common/ai/src/airflow/providers/common/ai/sandbox/openshell.py
new file mode 100644
index 00000000000..9a5156fa143
--- /dev/null
+++ b/providers/common/ai/src/airflow/providers/common/ai/sandbox/openshell.py
@@ -0,0 +1,1139 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""NVIDIA OpenShell backend for :class:`~airflow.providers.common.ai.toolsets.sandbox.SandboxToolset`."""
+
+from __future__ import annotations
+
+import errno
+import logging
+import math
+import os
+import secrets
+import shlex
+import threading
+import time
+import uuid
+from contextlib import contextmanager, suppress
+from dataclasses import dataclass
+from typing import TYPE_CHECKING
+
+from airflow.providers.common.ai.sandbox.base import (
+ _FILE_OP_OUTPUT_CAP,
+ _FILE_OP_TIMEOUT,
+ SandboxBackend,
+ SandboxError,
+ SandboxExecResult,
+ SandboxFileTooLargeError,
+ SandboxTerminalError,
+ _validate_positive_finite,
+)
+
+# The OpenShell SDK and its protobufs are imported inside the methods that use them:
+# sandbox/__init__.py imports this module eagerly, and ``openshell`` is an optional extra.
+if TYPE_CHECKING:
+ from collections.abc import Iterator, Mapping, Sequence
+
+ from openshell import SandboxClient
+ from openshell._proto import datamodel_pb2, openshell_pb2, sandbox_pb2
+
+ from airflow.providers.common.ai.sandbox.base import SandboxSpec
+
+log = logging.getLogger(__name__)
+
+_CLIENT_BUILD_LOCK = threading.Lock()
+
+# Wall-clock allowance past the per-command budget before the exec stream is
+# treated as hung. The guest wrapper enforces the budget itself; this covers a
+# gateway or supervisor that stops delivering events at all.
+_EXEC_GRACE = 30.0
+# A DEADLINE_EXCEEDED arriving at least this long before the call's own deadline
+# came from the gateway or a proxy, not the client.
+_EARLY_DEADLINE_MARGIN = 1.0
+# OpenShell 0.1.2's gateway ends an exec with DEADLINE_EXCEEDED and exactly this detail
+# when the relay into the sandbox does not open within 10 s, before the command starts
+# (openshell-server grpc/sandbox.rs await_relay_stream).
+_RELAY_OPEN_TIMED_OUT = "relay open timed out"
+# How long a command or policy read waits for a restarting gateway to come back
+# before the failure is reported. A restart measured at 6-11 s end to end.
+_GATEWAY_RECOVERY = 60.0
+_POLL_INTERVAL = 0.25
+# The gateway decodes at most 1 MiB per request, and the command or a file chunk
+# travels as stdin inside that request, so both stay under it with headroom.
+_MAX_STDIN_BYTES = 1_000_000
+_WRITE_CHUNK_BYTES = 768 * 1024
+# The gateway refuses a command argument over 32 KiB of UTF-8 or containing NUL
+# with INVALID_ARGUMENT, which is terminal, and file paths travel as arguments.
+_MAX_ARG_BYTES = 32 * 1024
+# The gateway rejects a sandbox name longer than 19 characters, which
+# _new_sandbox_name's ``airflow-sandbox-<12 hex>`` is, so the name is shorter
+# here. Labels carry the attribution instead, and the creation time so an
+# operator can reap by age: OpenShell has no server-side lifetime.
+_NAME_PREFIX = "airflow-"
+_CREATED_BY_LABEL = ("created-by", "airflow")
+_CREATED_AT_LABEL = "airflow-created-at"
+_EGRESS_RULE = "airflow-egress"
+_HTTPS_PORT = 443
+_ANY_BINARY = "/**"
+# OpenShell's restrictive default filesystem policy, plus /dev/shm so Python's
+# multiprocessing and anything else using POSIX shared memory works.
+_READ_ONLY_PATHS = ("/bin", "/usr", "/lib", "/proc", "/dev/urandom", "/etc", "/var/log")
+_READ_WRITE_PATHS = ("/tmp", "/dev/null", "/dev/shm")
+# The sandbox supervisor removes the proxy variables from every command it runs
+# and overwrites the CA-bundle variables with its own TLS-terminating CA
+# (openshell-sandbox process.rs PROXY_ENV_VARS and child_env.rs tls_env_vars in
+# 0.1.2), in both cases without an error. A spec naming them would be silently
+# changed, so it is refused instead.
+_SUPERVISOR_OWNED_ENV = frozenset(
+ {
+ "ALL_PROXY",
+ "HTTP_PROXY",
+ "HTTPS_PROXY",
+ "NO_PROXY",
+ "all_proxy",
+ "http_proxy",
+ "https_proxy",
+ "no_proxy",
+ "grpc_proxy",
+ "NODE_USE_ENV_PROXY",
+ "SSL_CERT_FILE",
+ "REQUESTS_CA_BUNDLE",
+ "CURL_CA_BUNDLE",
+ "GIT_SSL_CAINFO",
+ "NODE_EXTRA_CA_CERTS",
+ "DENO_CERT",
+ }
+)
+_PROPOSAL_APPROVAL_MODE = "proposal_approval_mode"
+_AGENT_POLICY_PROPOSALS = "agent_policy_proposals_enabled"
+# Wrapper-private exit status for "the budget ran out". The wrapper only uses it
+# after its own timer fired, and the caller also checks the elapsed time, so a
+# command that exits 124 on its own inside the budget is not a timeout.
+_TIMEOUT_STATUS = 124
+# Wrapper-private exit status and stderr line for "the command could not be
+# staged in /tmp, so it did not run". The status alone is not enough, because a
+# command can exit 125 itself, as docker run, env and nohup do on their own errors,
+# and neither is the text, because a command can print it. The wrapper ends the
+# line with a per-call nonce it reads from the first line of stdin, which reaches
+# neither the command's argv, its environment nor the staged script.
+_STAGING_STATUS = 125
+# Every variable the wrapper sets starts with this, and SandboxSpec.env may not:
+# a shell keeps a variable it inherited exported when it assigns it, so a spec key
+# naming a wrapper variable would hand its value, the nonce included, to the command.
+_WRAPPER_VAR_PREFIX = "_airflow_"
+_STAGING_FAILED = "airflow-exec: could not stage the command in /tmp"
+# The same for "the image has no setsid", which no command can fix from inside.
+_NO_SETSID_STATUS = 127
+_NO_SETSID = "airflow-exec: setsid is not installed in the sandbox image"
+_SYSTEM_PATH = "PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin"
+
+# Runs one command for run_command. OpenShell's own exec timeout returns a
+# synthetic 124 and leaves the process running, and a backgrounded child that
+# inherits stdout holds the call open until the supervisor gives up on it, so
+# the wrapper owns both problems:
+#
+# * The command arrives on stdin after the nonce line, which avoids the gateway's
+# 32 KiB per-argument cap, and runs in its own session via setsid, with stdout
+# and stderr going to files, so a background child holds a file rather than
+# the exec stream.
+# * A timer reading the monotonic /proc/uptime signals the wrapper when the
+# budget is spent; counting its own one-second sleeps instead drifts late on a
+# CPU-starved sandbox, past _EXEC_GRACE. The wrapper then SIGKILLs the
+# processes in the command's session, rescanning /proc until a pass kills
+# nothing, at most 50 times, so children forked mid-sweep are usually caught
+# too. A negative-pid kill of the group is blocked by the sandbox's seccomp
+# filter. The sweep shares the CPU with the processes it is killing, so it
+# reads /proc with builtins, joining the lines of a stat file whose process
+# name holds a newline, and strips the name with the cheap shortest match,
+# taking the longest only for a name that contains ")": a `cat` per entry let
+# one sweep of 128 busy processes on one CPU run past _EXEC_GRACE.
+# * The exit status is the wrapper's own process status, which the supervisor
+# reports out of band, so nothing the command prints can change it; a command
+# that signals the wrapper only ends its own run early. Output is sent last as
+# one byte more than the caller's cap, so an over-cap stream is detected by
+# counting what arrives rather than trusting a trailer.
+#
+# A command that starts its own session (setsid, a daemonizing server) leaves
+# this one and is not killed on timeout, and one that keeps forking faster than
+# the sweep can outlast it.
+_RUN_WRAPPER = rf"""_airflow_t=$1 _airflow_c=$2 _airflow_o=$PATH
+read -r _airflow_q
+{_SYSTEM_PATH}
+_airflow_d=$(mktemp -d /tmp/.airflow-exec.XXXXXX) && cat >"$_airflow_d/c" || {{ rm -rf "$_airflow_d"; echo "{_STAGING_FAILED} $_airflow_q" >&2; exit {_STAGING_STATUS}; }}
+_airflow_f=0
+trap _airflow_f=1 ALRM
+trap _airflow_f=2 HUP INT TERM
+_airflow_z=$(command -v setsid) || {{ rm -rf "$_airflow_d"; echo "{_NO_SETSID} $_airflow_q" >&2; exit {_NO_SETSID_STATUS}; }}
+PATH=$_airflow_o "$_airflow_z" /bin/sh "$_airflow_d/c" </dev/null >"$_airflow_d/o" 2>"$_airflow_d/e" &
+_airflow_p=$!
+(
+ read _airflow_u _ </proc/uptime
+ _airflow_e=$((${{_airflow_u%.*}}${{_airflow_u#*.}} + _airflow_t * 100))
+ while [ "${{_airflow_u%.*}}${{_airflow_u#*.}}" -lt "$_airflow_e" ]; do
+ sleep 1
+ [ -d "/proc/$$" ] || exit 0
+ read _airflow_u _ </proc/uptime
+ done
+ kill -ALRM $$
+) </dev/null >/dev/null 2>&1 &
+_airflow_w=$!
+wait "$_airflow_p"
+_airflow_r=$?
+[ "$_airflow_f" = 0 ] && trap '' ALRM
+if [ "$_airflow_f" != 0 ]; then
+ _airflow_n=0
+ while [ "$_airflow_n" -lt 50 ]; do
+ _airflow_k=0
+ for _airflow_x in /proc/[0-9]*/stat; do
+ _airflow_s=
+ while read -r _airflow_l; do _airflow_s="$_airflow_s $_airflow_l"; done 2>/dev/null <"$_airflow_x" || continue
+ _airflow_s=${{_airflow_s#*) }}
+ case $_airflow_s in *")"*) _airflow_s=${{_airflow_s##*) }} ;; esac
+ set -- $_airflow_s
+ [ "$4" = "$_airflow_p" ] || continue
+ _airflow_x=${{_airflow_x#/proc/}}
+ kill -KILL "${{_airflow_x%/stat}}" 2>/dev/null && _airflow_k=1
+ done
+ [ "$_airflow_k" = 0 ] && break
+ _airflow_n=$((_airflow_n + 1))
+ done
+ wait "$_airflow_p" 2>/dev/null
+fi
+kill -KILL "$_airflow_w" 2>/dev/null
+tail -c "$((_airflow_c + 1))" "$_airflow_d/o"
+tail -c "$((_airflow_c + 1))" "$_airflow_d/e" >&2
+rm -rf "$_airflow_d"
+[ "$_airflow_f" = 1 ] && exit {_TIMEOUT_STATUS}
+[ "$_airflow_f" = 2 ] && exit 143
+exit "$_airflow_r"
+"""
+# File operations run without the command wrapper, so nothing kills them in the
+# guest when the call times out. Opening a named pipe blocks until another process
+# opens its other end, which would leave the operation waiting forever, so the
+# scripts refuse one before opening it.
+_NAMED_PIPE_STATUS = SandboxBackend._NOT_REGULAR_FILE_STATUS
+_REFUSE_NAMED_PIPE = f'[ -p "$1" ] && exit {_NAMED_PIPE_STATUS}'
+# read_file: the base class's script with the content sent raw instead of as
+# base64, so a binary file arrives intact and the transfer is half the size.
+_READ_SCRIPT = f"""{_SYSTEM_PATH}
+sz=$(stat -Lc %s -- "$1" 2>/dev/null) || exit {SandboxBackend._MISSING_PATH_STATUS}
+[ -d "$1" ] && exit {SandboxBackend._IS_DIRECTORY_STATUS}
+{_REFUSE_NAMED_PIPE}
+printf '%s\\n' "$sz"
+exec head -c "$2" -- "$1"
+"""
+_WRITE_FIRST_SCRIPT = f'{_SYSTEM_PATH}\n{_REFUSE_NAMED_PIPE}\nmkdir -p -- "$(dirname -- "$1")" && cat >"$1"\n'
+_WRITE_NEXT_SCRIPT = f'{_SYSTEM_PATH}\n{_REFUSE_NAMED_PIPE}\ncat >>"$1"\n'
+
+
+def _wrapper_line(message: str, nonce: str) -> bytes:
+ """Return the stderr line only the guest wrapper writes: ``message`` and the nonce of the call."""
+ return f"{message} {nonce}\n".encode()
+
+
+def _new_openshell_name() -> str:
+ return f"{_NAME_PREFIX}{uuid.uuid4().hex[:11]}"
+
+
+def _status_name(error: BaseException) -> str | None:
+ code = getattr(error, "code", None)
+ if not callable(code):
+ return None
+ try:
+ status = code()
+ except Exception:
+ return None
+ return getattr(status, "name", None)
+
+
+def _status_details(error: BaseException) -> str:
+ details = getattr(error, "details", None)
+ return (details() or "") if callable(details) else ""
+
+
+def _describe_error(error: BaseException) -> str:
+ status = _status_name(error)
+ if status is None:
+ return f"{type(error).__name__}: {error}"
+ text = _status_details(error)
+ return f"{status}: {text}" if text else status
+
+
+_READ_ONLY_ERRNOS = frozenset({errno.EACCES, errno.EPERM, errno.EROFS})
+
+
+def _is_token_write_back_failure(error: BaseException) -> bool:
+ # A stale OIDC token is refreshed inside the call's auth interceptor, which
+ # writes the new one next to the registration through a ".oidc_token.*" temp
+ # file (openshell 0.1.2 _OidcRefresher._write_to_disk) and raises its OSError.
+ # Only a permission error says the registration is read-only; a full disk or a used-up
+ # quota on a writable one fails the same temp file with ENOSPC or EDQUOT.
+ filename = getattr(error, "filename", None)
+ return (
+ isinstance(error, OSError)
+ and error.errno in _READ_ONLY_ERRNOS
+ and isinstance(filename, str)
+ and os.path.basename(filename).startswith(".oidc_token.")
+ )
+
+
+@contextmanager
+def _translate_openshell_errors(
+ operation: str, *, recoverable_statuses: frozenset[str] = frozenset()
+) -> Iterator[None]:
+ try:
+ yield
+ except SandboxError:
+ raise
+ except ImportError as e:
+ raise SandboxTerminalError(
+ 'The OpenShell SDK is not installed. Install "apache-airflow-providers-common-ai[openshell]".'
+ ) from e
+ except Exception as e:
+ if _is_token_write_back_failure(e):
+ raise SandboxTerminalError(
+ f"OpenShell could not {operation}: the SDK refreshed the gateway's OIDC token and could not "
+ f"write it back, because the gateway registration is read-only ({e}). An OIDC gateway needs "
+ "a writable registration of its own on each worker; mTLS is the supported setup for a "
+ "worker that mounts the registration read-only."
+ ) from e
+ message = f"OpenShell could not {operation} ({_describe_error(e)})."
+ if _status_name(e) in recoverable_statuses:
+ raise SandboxError(message) from e
+ raise SandboxTerminalError(message) from e
+
+
+def _is_hostname(value: object) -> bool:
+ """Whether ``value`` is a hostname OpenShell's policy can match, optionally with one leading ``*.``."""
+ if not isinstance(value, str) or not value or len(value) > 253:
+ return False
+ labels = value.removeprefix("*.").split(".")
+ # A single label ('localhost', or '*.com' once the wildcard is removed) is
+ # refused by the gateway for wildcards and cannot name a public endpoint.
+ if len(labels) < 2 or labels[-1].isdigit():
+ return False
+ return all(
+ 0 < len(label) <= 63
+ and label[0] != "-"
+ and label[-1] != "-"
+ and all(ch.isascii() and (ch.isalnum() or ch == "-") for ch in label)
+ for label in labels
+ )
+
+
+class _BoundedTail:
+ """
+ The last ``max_bytes`` of a stream, and how many bytes the stream carried in total.
+
+ ``min_window`` keeps that many trailing bytes even under a smaller ``max_bytes``,
+ for :meth:`ends_with` only; the text and the truncation flag still follow ``max_bytes``.
+ """
+
+ def __init__(self, max_bytes: int, *, min_window: int = 0) -> None:
+ self._max_bytes = max_bytes
+ self._window = max(max_bytes, min_window)
+ self._data = bytearray()
+ self.received = 0
+
+ def add(self, chunk: bytes) -> None:
+ self.received += len(chunk)
+ self._data.extend(chunk)
+ if len(self._data) > self._window:
+ del self._data[: len(self._data) - self._window]
+
+ @property
+ def truncated(self) -> bool:
+ return self.received > self._max_bytes
+
+ def ends_with(self, suffix: bytes) -> bool:
+ return self._data.endswith(suffix)
+
+ def get_text(self) -> str:
+ data = bytes(self._data[max(0, len(self._data) - self._max_bytes) :])
+ if self.truncated:
+ # Drop the leading partial line so no fragment reads as a whole record, unless that
+ # would throw away most of the window: one line longer than the cap has its newline
+ # at the very end, which would leave "(no output)" for a command that wrote megabytes.
+ newline = data.find(b"\n")
+ if newline != -1 and len(data) - (newline + 1) >= self._max_bytes // 2:
+ data = data[newline + 1 :]
+ return data.decode("utf-8", errors="replace")
+
+
+class _BoundedHead:
+ """The first ``max_bytes`` of a stream; anything past that is dropped."""
+
+ def __init__(self, max_bytes: int) -> None:
+ self._max_bytes = max_bytes
+ self._data = bytearray()
+
+ def add(self, chunk: bytes) -> None:
+ room = self._max_bytes - len(self._data)
+ if room > 0:
+ self._data.extend(chunk[:room])
+
+ def get_bytes(self) -> bytes:
+ return bytes(self._data)
+
+
+@dataclass(frozen=True)
+class _ExecOutcome:
+ exit_code: int
+ stdout: _BoundedTail | _BoundedHead
+ stderr: _BoundedTail
+ elapsed: float
+
+
+class _ExecHung(Exception):
+ """The exec stream outlived its deadline without reporting an exit status."""
+
+
+class _GatewayDropped(Exception):
+ """The gateway went away after the exec request was sent; the command's outcome is unknown."""
+
+
+@dataclass(frozen=True)
+class _EgressPolicy:
+ """What an OpenShell policy lets a sandbox reach, in a form two policies can be compared by."""
+
+ destinations: frozenset[tuple[str, int, frozenset[str]]]
+ problems: tuple[str, ...]
+
+ @classmethod
+ def requested(cls, hosts: Sequence[str]) -> _EgressPolicy:
+ return cls(
+ destinations=frozenset((host, _HTTPS_PORT, frozenset({_ANY_BINARY})) for host in hosts),
+ problems=(),
+ )
+
+ @classmethod
+ def effective(cls, config: sandbox_pb2.GetSandboxConfigResponse) -> _EgressPolicy:
+ """Read the egress a ``GetSandboxConfigResponse`` says is in force."""
+ from openshell._proto import sandbox_pb2
+
+ destinations: set[tuple[str, int, frozenset[str]]] = set()
+ problems: list[str] = []
+ if config.policy_source != sandbox_pb2.POLICY_SOURCE_SANDBOX:
+ problems.append(
+ "a gateway-wide policy is in force instead of the sandbox's own "
+ f"(policy_source {sandbox_pb2.PolicySource.Name(config.policy_source)})"
+ )
+ if not config.configuration_admitted:
+ problems.append(
+ f"the gateway has not admitted the sandbox's configuration ({config.configuration_error or 'no reason given'})"
+ )
+ if config.policy.landlock.compatibility != "hard_requirement":
+ problems.append(
+ f"Landlock enforcement is {config.policy.landlock.compatibility or 'unset'!r}, not 'hard_requirement'"
+ )
+ if config.policy.network_middlewares:
+ problems.append(
+ f"network middlewares are configured: {sorted(config.policy.network_middlewares)}"
+ )
+ for rule in config.policy.network_policies.values():
+ binaries = frozenset(binary.path for binary in rule.binaries)
+ for endpoint in rule.endpoints:
+ if endpoint.allowed_ips:
+ problems.append(
+ f"{endpoint.host or 'a rule with no host'} admits addresses {list(endpoint.allowed_ips)}"
+ )
+ ports = set(endpoint.ports)
+ if endpoint.port:
+ ports.add(endpoint.port)
+ destinations.update((endpoint.host.lower(), port, binaries) for port in ports)
+ if _PROPOSAL_APPROVAL_MODE in config.settings:
+ setting = config.settings[_PROPOSAL_APPROVAL_MODE]
+ if setting.value.string_value == "auto":
+ problems.append(
+ "proposal_approval_mode is 'auto' "
+ f"({sandbox_pb2.SettingScope.Name(setting.scope)}), so a connection the sandbox "
+ "is denied can become an allow rule without review"
+ )
+ if _AGENT_POLICY_PROPOSALS in config.settings:
+ if config.settings[_AGENT_POLICY_PROPOSALS].value.bool_value:
+ problems.append(
+ "agent_policy_proposals_enabled is on, so the agent can propose its own egress"
+ )
+ return cls(destinations=frozenset(destinations), problems=tuple(problems))
+
+ def describe(self) -> list[str]:
+ return sorted(f"{host}:{port}" for host, port, _ in self.destinations)
+
+ def mismatch(self, wanted: _EgressPolicy) -> str | None:
+ """Why this policy is not ``wanted``, or ``None`` when it is exactly that."""
+ problems = list(self.problems)
+ if self.destinations != wanted.destinations:
+ problems.append(
+ f"it admits {self.describe() or 'nothing'} where {wanted.describe() or 'nothing'} was asked for"
+ )
+ return "; ".join(problems) or None
+
+
+class OpenShellSandboxBackend(SandboxBackend):
+ """
+ Run sandbox tools through an NVIDIA OpenShell gateway.
+
+ .. note::
+
+ Experimental: this can change or be removed in a minor release of this provider.
+ See :ref:`howto/stability`.
+
+ `OpenShell <https://github.com/NVIDIA/OpenShell>`__ is a self-hosted gateway
+ that runs each sandbox on Docker, Podman or Kubernetes with Landlock and
+ seccomp inside the container and a per-sandbox supervisor that mediates
+ every outbound connection. Airflow workers only need gRPC access to the
+ gateway, with the ``openshell`` extra installed.
+
+ **Credentials are ambient.** The backend reads the gateway registration the
+ ``openshell`` CLI keeps under ``$XDG_CONFIG_HOME/openshell/gateways/<name>/``:
+ its endpoint and either mTLS material or an OIDC token. There is no Airflow
+ connection type for it; see the backend page for how to provision a worker.
+ An OIDC registration must be writable and per worker, because the SDK
+ refreshes the token and writes it back into the registration; mTLS is the
+ supported setup for a worker that mounts the registration read-only.
+
+ **Egress is deny-all unless listed, and verified.** A spec's hostnames become
+ one policy rule admitting each host on port 443 only, and a default spec
+ sends no rule at all, which OpenShell enforces as no egress; name resolution
+ is answered by the supervisor without leaving the sandbox. After create the
+ backend reads the effective policy back and destroys the sandbox unless it
+ is exactly what was asked for, comes from the sandbox rather than a
+ gateway-wide override, and cannot be widened by auto-approved proposals.
+ The same check runs before every command and file operation and after each
+ one that succeeds, because a gateway administrator, an approved draft or a
+ global policy can widen a running sandbox; a change fails the task. What it cannot see is the network the
+ gateway runs on: on Kubernetes the NetworkPolicy that keeps a sandbox pod
+ behind its supervisor is the cluster's to enforce.
+
+ **Commands run through a guest wrapper.** OpenShell's own exec timeout
+ leaves the command running, so the wrapper enforces the budget itself and
+ kills the processes in the command's session when it runs out. Output is
+ spooled to the sandbox's ``/tmp`` and each stream is capped to
+ ``max_output_bytes`` before it leaves the sandbox. ``allow_egress_to_cidrs``,
+ an open network and ``SandboxSpec.owner`` are refused. The image needs
+ ``setsid``, GNU coreutils and findutils, and a ``/bin/sh`` that is dash or bash;
+ ``python:*-slim`` has them. BusyBox ``ash``, as in Alpine, is not supported:
+ the timeout sweep does not finish under it.
+
+ :param gateway: Name of the OpenShell CLI gateway registration to use.
+ ``None`` uses ``$OPENSHELL_GATEWAY``, then the CLI's active gateway.
+ :param workspace: OpenShell workspace the sandboxes are created in.
+ :param image: Container image used for each sandbox.
+ :param cpu: CPU limit, as a Kubernetes quantity.
+ :param memory: Memory limit, as a Kubernetes quantity.
+ :param ready_timeout: Seconds to wait for a new sandbox to become ready.
+ :param request_timeout: Seconds allowed for each gateway call other than a command.
+ """
+
+ name = "openshell"
+
+ def __init__(
+ self,
+ *,
+ gateway: str | None = None,
+ workspace: str = "default",
+ image: str = "python:3.12-slim",
+ cpu: str = "1",
+ memory: str = "2Gi",
+ ready_timeout: float = 120.0,
+ request_timeout: float = 30.0,
+ ) -> None:
+ if gateway is not None and not gateway:
+ raise ValueError("gateway must not be empty; pass None to use the CLI's active gateway.")
+ if not workspace:
+ raise ValueError("workspace must not be empty.")
+ if not image:
+ raise ValueError("image must not be empty.")
+ if not cpu:
+ raise ValueError("cpu must not be empty.")
+ if not memory:
+ raise ValueError("memory must not be empty.")
+ _validate_positive_finite(ready_timeout, "ready_timeout")
+ _validate_positive_finite(request_timeout, "request_timeout")
+ self._gateway = gateway
+ self._workspace = workspace
+ self._image = image
+ self._resources = {"limits": {"cpu": cpu, "memory": memory}}
+ self._ready_timeout = ready_timeout
+ self._request_timeout = request_timeout
+ self._client: SandboxClient | None = None
+ # The egress each sandbox was created with, compared against the effective
+ # policy before every command and file operation and after each one that succeeds.
+ self._egress: dict[str, _EgressPolicy] = {}
+
+ def _get_client(self) -> SandboxClient:
+ with _CLIENT_BUILD_LOCK:
+ if self._client is None:
+ with _translate_openshell_errors("load the gateway registration"):
+ from openshell import SandboxClient
+
+ self._client = SandboxClient.from_active_cluster(
+ cluster=self._gateway, timeout=self._request_timeout
+ )
+ return self._client
+
+ def _workspace_scope(self) -> datamodel_pb2.WorkspaceSelector:
+ from openshell._proto import datamodel_pb2
+
+ return datamodel_pb2.WorkspaceSelector(workspace=self._workspace)
+
+ @staticmethod
+ def _check_spec(spec: SandboxSpec | None) -> tuple[list[str], dict[str, str]]:
+ """
+ Return the hosts and environment to provision, or refuse a spec OpenShell cannot enforce.
+
+ ``None`` gets the same deny-all policy as a default spec: OpenShell has no
+ policy-free mode that is not the image's own, which may be permissive.
+ """
+ if spec is None:
+ return [], {}
+ if spec.owner is not None:
+ # An owner exists so that a later task can attach to the sandbox. OpenShell
+ # labels are fixed at create and its annotations change only through an
+ # admin-scoped config mutation, so this version does not attach.
+ raise SandboxTerminalError(
+ "SandboxSpec names an owner, but this backend does not implement attaching, so a "
+ "sandbox created here cannot be attached to from another task. Drop owner, or "
+ "provision the sandbox on a backend that supports attaching, such as ModalSandboxBackend."
+ )
+ if spec.allow_egress_to_cidrs:
+ raise SandboxTerminalError(
+ "SandboxSpec names allow_egress_to_cidrs, which this backend cannot enforce as "
+ "specified: OpenShell address rules admit TCP on listed ports only, never UDP or any "
+ "port. Use allow_egress_to with hostnames, or a backend with an address-layer allowlist."
+ )
+ if not spec.block_network:
+ raise SandboxTerminalError(
+ "SandboxSpec asks for an open network, which OpenShell cannot provide: every egress "
+ "rule names a host and its ports. Set block_network=True and list the hosts the "
+ "sandbox needs in allow_egress_to."
+ )
+ hosts: list[str] = []
+ if spec.allow_egress_to:
+ if isinstance(spec.allow_egress_to, str):
+ raise SandboxTerminalError(
+ "SandboxSpec.allow_egress_to must be a sequence of hostnames, not one string: "
+ f"{spec.allow_egress_to!r} would be read a character at a time. Wrap it in a list."
+ )
+ rejected = [host for host in spec.allow_egress_to if not _is_hostname(host)]
+ if rejected:
+ raise SandboxTerminalError(
+ "SandboxSpec.allow_egress_to must contain bare hostnames, optionally with a "
+ f"leading '*.' label, and these entries are not: {rejected}. Write 'pypi.org' or "
+ "'*.pythonhosted.org', not a URL, a host:port, an address, or a single label."
+ )
+ hosts = sorted({host.lower() for host in spec.allow_egress_to})
+ env: dict[str, str] = {}
+ for key, value in (spec.env or {}).items():
+ if not isinstance(key, str) or not isinstance(value, str):
+ raise SandboxTerminalError(
+ f"SandboxSpec.env must map strings to strings; {key!r} is set to {type(value).__name__}."
+ )
+ if key in _SUPERVISOR_OWNED_ENV or key.startswith("OPENSHELL_"):
+ raise SandboxTerminalError(
+ f"SandboxSpec.env sets {key}, which the OpenShell supervisor owns: it removes the "
+ "proxy variables and replaces the CA bundle variables with its own TLS-terminating "
+ "CA in every command, and reserves OPENSHELL_*. Remove it from the spec."
+ )
+ if key.startswith(_WRAPPER_VAR_PREFIX):
+ raise SandboxTerminalError(
+ f"SandboxSpec.env sets {key}, but names starting with {_WRAPPER_VAR_PREFIX} are "
+ "reserved for the shell wrapper the backend runs every command in. Rename it."
+ )
+ env[key] = value
+ return hosts, env
+
+ def _build_spec(self, hosts: Sequence[str], env: Mapping[str, str]) -> openshell_pb2.SandboxSpec:
+ from google.protobuf import struct_pb2
+ from openshell._proto import openshell_pb2, sandbox_pb2
+
+ policy = sandbox_pb2.SandboxPolicy(
+ version=1,
+ filesystem=sandbox_pb2.FilesystemPolicy(
+ include_workdir=True, read_only=list(_READ_ONLY_PATHS), read_write=list(_READ_WRITE_PATHS)
+ ),
+ # Refuse to start at all rather than run without the filesystem confinement.
+ landlock=sandbox_pb2.LandlockPolicy(compatibility="hard_requirement"),
+ )
+ if hosts:
+ policy.network_policies[_EGRESS_RULE].CopyFrom(
+ sandbox_pb2.NetworkPolicyRule(
+ name=_EGRESS_RULE,
+ endpoints=[sandbox_pb2.NetworkEndpoint(host=host, ports=[_HTTPS_PORT]) for host in hosts],
+ binaries=[sandbox_pb2.NetworkBinary(path=_ANY_BINARY)],
+ )
+ )
+ spec = openshell_pb2.SandboxSpec(environment=dict(env), policy=policy)
+ spec.template.image = self._image
+ resources = struct_pb2.Struct()
+ resources.update(self._resources)
+ spec.template.resources.CopyFrom(resources)
+ return spec
+
+ def create(self, *, spec: SandboxSpec | None = None) -> str:
+ hosts, env = self._check_spec(spec)
+ client = self._get_client()
+ request = self._build_spec(hosts, env)
+ name = _new_openshell_name()
+ try:
+ with _translate_openshell_errors("create a sandbox"):
+ client.create(
+ workspace=self._workspace,
+ spec=request,
+ name=name,
+ labels={
+ _CREATED_BY_LABEL[0]: _CREATED_BY_LABEL[1],
+ _CREATED_AT_LABEL: str(int(time.time())),
+ },
+ )
+ self._wait_until_ready(name, deadline=time.monotonic() + self._ready_timeout)
+ wanted = _EgressPolicy.requested(hosts)
+ mismatch = self._read_egress(name).mismatch(wanted)
+ if mismatch is not None:
+ raise SandboxTerminalError(
+ f"OpenShell did not apply the requested network policy to sandbox {name} ({mismatch}), "
+ "so the sandbox is not used and is being destroyed."
+ )
+ except BaseException:
+ # Bound before the call, so a create that fails half way -- after the
+ # gateway accepted it, or at the policy check -- still gets deleted.
+ self._destroy_or_warn(name, after="a failed create")
+ raise
+ self._egress[name] = wanted
+ return name
+
+ def _get_sandbox(self, name: str) -> openshell_pb2.Sandbox:
+ from openshell._proto import openshell_pb2
+
+ return (
+ self._get_client()
+ ._stub.GetSandbox(
+ openshell_pb2.GetSandboxRequest(workspace_scope=self._workspace_scope(), name=name),
+ timeout=self._request_timeout,
+ )
+ .sandbox
+ )
+
+ def _wait_until_ready(self, name: str, *, deadline: float) -> None:
+ """
+ Poll until the sandbox is ready, riding out a gateway that is restarting.
+
+ Raises :class:`SandboxTerminalError` for a sandbox that is gone, has failed or
+ has stopped, or that is not ready by ``deadline``.
+ """
+ from openshell._proto import openshell_pb2
+
+ last = "it did not report a phase"
+ while True:
+ try:
+ sandbox = self._get_sandbox(name)
+ except Exception as e:
+ if _status_name(e) not in {"UNAVAILABLE", "DEADLINE_EXCEEDED"}:
+ with _translate_openshell_errors(f"check the status of sandbox {name}"):
+ raise
+ last = f"the gateway did not answer ({_describe_error(e)})"
+ else:
+ phase = sandbox.status.phase
+ if phase == openshell_pb2.SANDBOX_PHASE_READY:
+ return
+ reasons = "; ".join(
+ f"{condition.reason}: {condition.message}"
+ for condition in sandbox.status.conditions
+ if condition.status != "True" and (condition.reason or condition.message)
+ )
+ last = f"it is {openshell_pb2.SandboxPhase.Name(phase)}" + (
+ f" ({reasons})" if reasons else ""
+ )
+ if phase in {
+ openshell_pb2.SANDBOX_PHASE_ERROR,
+ openshell_pb2.SANDBOX_PHASE_STOPPING,
+ openshell_pb2.SANDBOX_PHASE_STOPPED,
+ openshell_pb2.SANDBOX_PHASE_COMPLETED,
+ openshell_pb2.SANDBOX_PHASE_DELETING,
+ }:
+ raise SandboxTerminalError(f"OpenShell sandbox {name} cannot run commands: {last}.")
+ if time.monotonic() >= deadline:
+ raise SandboxTerminalError(f"OpenShell sandbox {name} is not ready: {last}.")
+ time.sleep(_POLL_INTERVAL)
+
+ def _read_egress(self, name: str) -> _EgressPolicy:
+ """Read the sandbox's effective egress, waiting out a restarting gateway for a bounded time."""
+ from openshell._proto import sandbox_pb2
+
+ request = sandbox_pb2.GetSandboxConfigRequest(workspace_scope=self._workspace_scope(), name=name)
+ deadline = time.monotonic() + _GATEWAY_RECOVERY
+ while True:
+ try:
+ config = self._get_client()._stub.GetSandboxConfig(request, timeout=self._request_timeout)
+ except Exception as e:
+ if (
+ _status_name(e) not in {"UNAVAILABLE", "DEADLINE_EXCEEDED"}
+ or time.monotonic() >= deadline
+ ):
+ with _translate_openshell_errors(f"read the network policy of sandbox {name}"):
+ raise
+ time.sleep(_POLL_INTERVAL)
+ continue
+ return _EgressPolicy.effective(config)
+
+ def _check_egress(self, name: str) -> None:
+ """Destroy the sandbox and fail the task if its egress is no longer what it was created with."""
+ effective = self._read_egress(name)
+ # A handle this instance did not create is held to what it admits the first
+ # time it is seen, on top of the same invariants.
+ wanted = self._egress.setdefault(name, _EgressPolicy(effective.destinations, ()))
+ mismatch = effective.mismatch(wanted)
+ if mismatch is None:
+ return
+ if self._destroy_or_warn(name, after="its network policy changed"):
+ outcome = "so the sandbox was destroyed"
+ else:
+ outcome = "and destroying the sandbox failed, so it may still be running"
+ raise SandboxTerminalError(
+ f"The network policy of OpenShell sandbox {name} changed after it was created ({mismatch}), "
+ f"{outcome}. A gateway administrator, an approved policy draft or a gateway-wide policy can "
+ "widen a running sandbox; commands already run may have used it."
+ )
+
+ def _exec(
+ self,
+ sandbox: str,
+ argv: list[str],
+ *,
+ stdin: bytes,
+ timeout: float,
+ stdout: _BoundedTail | _BoundedHead,
+ stderr: _BoundedTail,
+ ) -> _ExecOutcome:
+ """
+ Run ``argv`` once, streaming both outputs into the given bounded buffers.
+
+ Streams the gateway's events directly: the SDK's ``exec_stream`` keeps every
+ chunk in memory until the command exits. Raises :class:`_ExecHung` after
+ ``timeout`` seconds, :class:`SandboxError` when the gateway could not open its relay
+ to start the command, and :class:`_GatewayDropped` when the stream ends early
+ after the request left, where the command may or may not have run.
+ """
+ from openshell._proto import openshell_pb2
+
+ client = self._get_client()
+ request = openshell_pb2.ExecSandboxRequest(
+ workspace_scope=self._workspace_scope(),
+ sandbox=sandbox,
+ command=argv,
+ stdin=stdin,
+ # A login shell would source profiles that can override SandboxSpec.env.
+ no_login_shell=True,
+ request_id=str(uuid.uuid4()),
+ )
+ exit_code: int | None = None
+ started = time.monotonic()
+ stream = None
+ try:
+ stream = client._stub.ExecSandbox(request, timeout=timeout)
+ for event in stream:
+ kind = event.WhichOneof("payload")
+ if kind == "stdout":
+ stdout.add(event.stdout.data)
+ elif kind == "stderr":
+ stderr.add(event.stderr.data)
+ elif kind == "exit":
+ exit_code = event.exit.exit_code
+ except Exception as e:
+ status = _status_name(e)
+ if status == "DEADLINE_EXCEEDED":
+ if time.monotonic() - started >= timeout - _EARLY_DEADLINE_MARGIN:
+ raise _ExecHung from e
+ if _status_details(e) == _RELAY_OPEN_TIMED_OUT:
+ raise SandboxError(
+ f"The command was not run: the OpenShell gateway could not reach sandbox {sandbox} "
+ f"to start it ({_describe_error(e)}). Run it again."
+ ) from e
+ # Any other early end, a proxy's cap on stream duration for example, can cut
+ # off a command that has started and keeps running.
+ raise _GatewayDropped(_describe_error(e)) from e
+ if status == "UNAVAILABLE":
+ raise _GatewayDropped(_describe_error(e)) from e
+ with _translate_openshell_errors(
+ f"run a command in sandbox {sandbox}",
+ recoverable_statuses=frozenset({"OUT_OF_RANGE", "RESOURCE_EXHAUSTED"}),
+ ):
+ raise
+ finally:
+ if stream is not None:
+ with suppress(Exception):
+ stream.cancel()
+ if exit_code is None:
+ raise _GatewayDropped("the exec stream ended without an exit status")
+ return _ExecOutcome(exit_code, stdout, stderr, time.monotonic() - started)
+
+ def _exec_with_recovery(
+ self,
+ sandbox: str,
+ argv: list[str],
+ *,
+ stdin: bytes,
+ timeout: float,
+ stdout: _BoundedTail | _BoundedHead,
+ stderr: _BoundedTail,
+ ) -> _ExecOutcome:
+ """
+ :meth:`_exec`, with a gateway restart turned into an honest outcome.
+
+ A sandbox that is not ready yet, which is how it looks for a few seconds after a
+ gateway restart, has not launched anything, so the command is sent again once it
+ is. An exec dropped in flight has an unknown outcome, so that is reported to the
+ model rather than retried or guessed: a gateway restart kills the command with
+ every other process in the sandbox, but a dropped connection to a gateway that
+ stays up does not stop it, and it runs on until the wrapper's timer ends it.
+ """
+ try:
+ try:
+ return self._exec(sandbox, argv, stdin=stdin, timeout=timeout, stdout=stdout, stderr=stderr)
+ except SandboxTerminalError as e:
+ if _status_name(e.__cause__ or e) != "FAILED_PRECONDITION":
+ raise
+ self._wait_until_ready(sandbox, deadline=time.monotonic() + self._ready_timeout)
+ return self._exec(sandbox, argv, stdin=stdin, timeout=timeout, stdout=stdout, stderr=stderr)
+ except _GatewayDropped as e:
+ self._wait_until_ready(sandbox, deadline=time.monotonic() + _GATEWAY_RECOVERY)
+ raise SandboxError(
+ f"The OpenShell gateway dropped the command before it reported an exit status ({e}). The "
+ "command may or may not have run. If only the connection dropped, it may still be running "
+ "until its timeout runs out; if the gateway restarted, the restart stopped it along with "
+ "processes that earlier commands left running. Files are kept. Check what state you need "
+ "before running the command again."
+ ) from e
+
+ def run_command(
+ self, sandbox: str, command: str, *, timeout: float, max_output_bytes: int
+ ) -> SandboxExecResult:
+ _validate_positive_finite(timeout, "timeout")
+ _validate_positive_finite(max_output_bytes, "max_output_bytes")
+ # Whole seconds, at least one: the wrapper takes its budget as an integer.
+ seconds = max(1, math.ceil(timeout))
+ payload = command.encode("utf-8")
+ if len(payload) > _MAX_STDIN_BYTES:
+ raise SandboxError(
+ f"The command is {len(payload)} bytes, over the {_MAX_STDIN_BYTES} bytes OpenShell carries "
+ "in one request. Write the script to a file with write_file and run the file."
+ )
+ self._check_egress(sandbox)
+ nonce = secrets.token_hex(16)
+ staging_line = _wrapper_line(_STAGING_FAILED, nonce)
+ no_setsid_line = _wrapper_line(_NO_SETSID, nonce)
+ stdout = _BoundedTail(int(max_output_bytes))
+ stderr = _BoundedTail(int(max_output_bytes), min_window=max(len(staging_line), len(no_setsid_line)))
+ try:
+ outcome = self._exec_with_recovery(
+ sandbox,
+ ["/bin/sh", "-c", _RUN_WRAPPER, "airflow-exec", str(seconds), str(int(max_output_bytes))],
+ stdin=f"{nonce}\n".encode() + payload,
+ timeout=seconds + _EXEC_GRACE,
+ stdout=stdout,
+ stderr=stderr,
+ )
+ except _ExecHung:
+ return self._abandon_command(sandbox, stdout, stderr, seconds=seconds)
+ self._check_egress(sandbox)
+ if outcome.exit_code == _NO_SETSID_STATUS and stderr.ends_with(no_setsid_line):
+ raise SandboxTerminalError(
+ f"OpenShell sandbox {sandbox} cannot run commands: its image has no setsid, which the "
+ "backend runs every command under. Use an image with setsid from util-linux; "
+ "python:*-slim has it."
+ )
+ if outcome.exit_code == _STAGING_STATUS and stderr.ends_with(staging_line):
+ cause = stderr.get_text().rpartition(_STAGING_FAILED)[0].strip()
+ raise SandboxError(
+ "The command was not run: the sandbox could not write it to /tmp"
+ + (f" ({cause})." if cause else ".")
+ )
+ return SandboxExecResult(
+ exit_code=outcome.exit_code,
+ stdout=stdout.get_text(),
+ stderr=stderr.get_text(),
+ timed_out=outcome.exit_code == _TIMEOUT_STATUS and outcome.elapsed >= seconds,
+ stdout_truncated=stdout.truncated,
+ stderr_truncated=stderr.truncated,
+ applied_timeout=float(seconds),
+ )
+
+ def _abandon_command(
+ self, sandbox: str, stdout: _BoundedTail, stderr: _BoundedTail, *, seconds: int
+ ) -> SandboxExecResult:
+ # The wrapper reports within its budget unless the command disabled it or the
+ # gateway stopped relaying; either way the command may still be running, and
+ # OpenShell does not stop an exec whose caller went away, so only destroying
+ # the sandbox ends it.
+ log.warning(
+ "OpenShell sandbox %s reported nothing for %gs, its command budget of %ds plus %gs; "
+ "destroying it to stop the command",
+ sandbox,
+ seconds + _EXEC_GRACE,
+ seconds,
+ _EXEC_GRACE,
+ )
+ self._destroy_or_warn(sandbox, after="a command timed out")
+ return SandboxExecResult(
+ exit_code=-1,
+ stdout=stdout.get_text(),
+ stderr=stderr.get_text(),
+ timed_out=True,
+ stdout_truncated=stdout.truncated,
+ stderr_truncated=stderr.truncated,
+ sandbox_terminated=True,
+ applied_timeout=float(seconds),
+ )
+
+ @staticmethod
+ def _check_path(path: str) -> None:
+ if "\0" in path:
+ raise SandboxError("The path contains a NUL character, which OpenShell does not accept.")
+ size = len(path.encode("utf-8"))
+ if size > _MAX_ARG_BYTES:
+ raise SandboxError(
+ f"The path is {size} bytes, over the {_MAX_ARG_BYTES} bytes OpenShell accepts. Use a shorter path."
+ )
+
+ def _run_file_op(
+ self, sandbox: str, script: str, *args: str, stdin: bytes = b"", stdout_cap: int = _FILE_OP_OUTPUT_CAP
+ ) -> tuple[_ExecOutcome, bytes]:
+ head = _BoundedHead(stdout_cap)
+ try:
+ outcome = self._exec_with_recovery(
+ sandbox,
+ ["/bin/sh", "-c", script, "airflow-file", *args],
+ stdin=stdin,
+ timeout=_FILE_OP_TIMEOUT,
+ stdout=head,
+ stderr=_BoundedTail(_FILE_OP_OUTPUT_CAP),
+ )
+ except _ExecHung as e:
+ raise SandboxError(
+ f"The file operation did not finish within {_FILE_OP_TIMEOUT:g} seconds."
+ ) from e
+ if outcome.exit_code == _NAMED_PIPE_STATUS:
+ raise SandboxError(
+ f"{args[0]!r} is a named pipe, which would block until another process opens its "
+ "other end. Use run_command to read or write it."
+ )
+ return outcome, head.get_bytes()
+
+ def read_file(self, sandbox: str, path: str, *, max_bytes: int) -> bytes:
+ """Read a file as raw bytes, stopping one byte past ``max_bytes`` inside the guest."""
+ _validate_positive_finite(max_bytes, "max_bytes")
+ self._check_path(path)
+ self._check_egress(sandbox)
+ # The size line is at most 20 digits and a newline; the rest is the content.
+ outcome, raw = self._run_file_op(
+ sandbox, _READ_SCRIPT, path, str(max_bytes + 1), stdout_cap=max_bytes + 1 + 32
+ )
+ self._check_egress(sandbox)
+ if outcome.exit_code == self._MISSING_PATH_STATUS:
+ raise SandboxError(f"{path!r} does not exist in the sandbox, or is not readable.")
+ if outcome.exit_code == self._IS_DIRECTORY_STATUS:
+ raise SandboxError(f"{path!r} is a directory. Use list_directory to see what is in it.")
+ if outcome.exit_code:
+ raise SandboxError(outcome.stderr.get_text().strip() or f"Could not read {path!r}.")
+ reported, _, data = raw.partition(b"\n")
+ if len(data) > max_bytes:
+ try:
+ size = int(reported.strip())
+ except ValueError:
+ size = 0
+ # A streaming source reports 0; the true size is then unknown but over.
+ raise SandboxFileTooLargeError(path, max(size, len(data)), max_bytes)
+ return data
+
+ def write_file(self, sandbox: str, path: str, content: bytes) -> None:
+ """
+ Write ``content`` over stdin, creating parent directories.
+
+ The base class's in-command payload is base64 inside the one request that
+ carries the command, so it stops at about 750 KB of content. Raw stdin in
+ chunks has no such ceiling. A file larger than one chunk is not written
+ atomically: a failure part way leaves the chunks written so far.
+ """
+ self._check_path(path)
+ self._check_egress(sandbox)
+ # One empty chunk for empty content, so the file is still created. Sliced as it
+ # is sent, so only one chunk is copied out of ``content`` at a time.
+ for start in range(0, max(len(content), 1), _WRITE_CHUNK_BYTES):
+ script = _WRITE_FIRST_SCRIPT if start == 0 else _WRITE_NEXT_SCRIPT
+ chunk = content[start : start + _WRITE_CHUNK_BYTES]
+ outcome, _ = self._run_file_op(sandbox, script, path, stdin=chunk)
+ if outcome.exit_code:
+ raise SandboxError(outcome.stderr.get_text().strip() or f"Could not write {path!r}.")
+ self._check_egress(sandbox)
+
+ def list_directory(self, sandbox: str, path: str) -> list[tuple[str, bool]]:
+ """List as the base class does, but refuse a listing cut short at the output cap."""
+ result = self.run_command(
+ sandbox,
+ f"find -- {shlex.quote(path)} -maxdepth 1 -mindepth 1 -printf '%y %f\\0'",
+ timeout=_FILE_OP_TIMEOUT,
+ max_output_bytes=_FILE_OP_OUTPUT_CAP,
+ )
+ if result.sandbox_terminated:
+ # The toolset only replaces a sandbox after a command, so a listing that lost
+ # it has to fail the task rather than leave the toolset holding a deleted one.
+ raise SandboxTerminalError(
+ f"The OpenShell gateway stopped answering while listing {path!r}, so the backend "
+ f"asked it to destroy sandbox {sandbox} to stop the listing."
+ )
+ if result.timed_out:
+ raise SandboxError(f"Listing {path!r} did not finish within {_FILE_OP_TIMEOUT:g} seconds.")
+ if result.exit_code:
+ raise SandboxError(result.stderr.strip() or f"Could not list {path!r}.")
+ if result.stdout_truncated:
+ raise SandboxError(
+ f"{path!r} has too many entries to list; list a subdirectory instead, or use a shell "
+ "command such as `ls | head`."
+ )
+ entries: list[tuple[str, bool]] = []
+ for record in result.stdout.split("\0"):
+ kind, _, name = record.partition(" ")
+ if name:
+ entries.append((name, kind == "d"))
+ return entries
+
+ def _destroy_or_warn(self, sandbox: str, *, after: str) -> bool:
+ """Destroy ``sandbox``, logging a failure instead of raising it; return whether it was destroyed."""
+ try:
+ self.destroy(sandbox)
+ except Exception:
+ # OpenShell has no server-side lifetime to fall back on.
+ log.warning(
+ "Could not destroy OpenShell sandbox %s after %s; it will need manual cleanup",
+ sandbox,
+ after,
+ exc_info=True,
+ )
+ return False
+ return True
+
+ def destroy(self, sandbox: str) -> None:
+ self._egress.pop(sandbox, None)
+ client = self._get_client()
+ try:
+ # Accepted asynchronously; waiting for the containers to go takes seconds
+ # and changes nothing for the caller.
+ client.delete(sandbox, workspace=self._workspace, allow_missing=True)
+ except Exception as e:
+ if _status_name(e) == "NOT_FOUND":
+ return
+ with _translate_openshell_errors(f"destroy sandbox {sandbox}"):
+ raise
diff --git a/providers/common/ai/tests/system/common/ai/README.md b/providers/common/ai/tests/system/common/ai/README.md
index bf3a225a59b..6e333ee97e3 100644
--- a/providers/common/ai/tests/system/common/ai/README.md
+++ b/providers/common/ai/tests/system/common/ai/README.md
@@ -99,3 +99,20 @@ the one sandbox when the run ends, and requests open egress
A 15-minute server-side TTL stops the sandbox if worker-side teardown cannot run; the
backend creates it without snapshots or monitoring, so stopping it erases its disk and Boat
keeps no history of its commands.
+
+## OpenShell
+
+Install the OpenShell extra, register the gateway the way the
+`openshell` CLI does, under `$XDG_CONFIG_HOME/openshell/gateways/<name>/` (`metadata.json`
+plus `mtls/ca.crt`, `mtls/tls.crt` and `mtls/tls.key`), and run:
+
+```console
+pip install "apache-airflow-providers-common-ai[openshell]"
+export OPENSHELL_GATEWAY="<name>"
+pytest --system providers/common/ai/tests/system/common/ai/example_sandbox_toolset_openshell.py
+```
+
+The test uses the default deny-all policy and a `python:3.12-slim` sandbox, and checks
+that a timed-out command leaves no process behind and that egress is refused. OpenShell
+has no server-side sandbox lifetime, so a test killed mid-run leaves its sandbox, labelled
+`created-by=airflow`, for you to delete.
diff --git a/providers/common/ai/tests/system/common/ai/example_sandbox_toolset_openshell.py b/providers/common/ai/tests/system/common/ai/example_sandbox_toolset_openshell.py
new file mode 100644
index 00000000000..fc7b63073be
--- /dev/null
+++ b/providers/common/ai/tests/system/common/ai/example_sandbox_toolset_openshell.py
@@ -0,0 +1,122 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""End-to-end system test for SandboxToolset with NVIDIA OpenShell."""
+
+from __future__ import annotations
+
+import os
+from datetime import UTC, datetime
+
+from airflow.providers.common.compat.sdk import dag as airflow_dag, task
+
+ENV_ID = os.environ.get("SYSTEM_TESTS_ENV_ID")
+DAG_ID = f"common_ai_sandbox_toolset_openshell_{ENV_ID}" if ENV_ID else "common_ai_sandbox_toolset_openshell"
+
+MARKER = "boundary-ok"
+STATE_PATH = "/tmp/airflow_sandbox_e2e"
+# Counts processes whose command line is exactly "sleep 300", without needing ps in the image.
+COUNT_SLEEPERS = (
+ "n=0; for f in /proc/[0-9]*/cmdline; do "
+ "[ \"$(tr '\\0' ' ' < \"$f\" 2>/dev/null)\" = 'sleep 300 ' ] && n=$((n + 1)); done; echo sleepers=$n"
+)
+CONNECT = (
+ 'python3 -c "import socket\n'
+ "try:\n socket.create_connection(('1.1.1.1', 443), 5)\n print('egress=open')\n"
+ "except OSError as e:\n print('egress=denied', e.errno)\""
+)
+
+
+@airflow_dag(
+ dag_id=DAG_ID,
+ schedule="@once",
+ start_date=datetime(2024, 1, 1, tzinfo=UTC),
+ catchup=False,
+ tags=["common.ai", "sandbox", "openshell", "system_test"],
+)
+def example_sandbox_toolset_openshell():
+ @task
+ def run_sandbox_agent() -> str:
+ # Keep task-only dependencies out of the Dag-parsing process.
+ from pydantic_ai import Agent
+ from pydantic_ai.messages import ModelMessage, ModelResponse, TextPart, ToolCallPart
+ from pydantic_ai.models.function import AgentInfo, FunctionModel
+
+ from airflow.providers.common.ai.sandbox import OpenShellSandboxBackend
+ from airflow.providers.common.ai.toolsets import SandboxToolset
+
+ # Each step is a tool call and the text its result must contain.
+ steps: list[tuple[str, dict, tuple[str, ...]]] = [
+ ("write_file", {"path": STATE_PATH, "content": MARKER}, ("Wrote",)),
+ (
+ "run_command",
+ {"command": f"cat {STATE_PATH} && python3 -c 'print(6 * 7)'"},
+ (MARKER, "42"),
+ ),
+ ("read_file", {"path": STATE_PATH}, (MARKER,)),
+ ("run_command", {"command": "echo to-stderr >&2; exit 3"}, ("[exit code: 3]", "to-stderr")),
+ ("list_directory", {"path": "/tmp"}, ("airflow_sandbox_e2e",)),
+ (
+ "run_command",
+ {"command": "sleep 300 & sleep 300", "timeout_seconds": 3},
+ ("[timed out after 3s]",),
+ ),
+ # The marker proves the sleepers were counted in the sandbox that timed out,
+ # not in one provisioned after it was destroyed.
+ ("run_command", {"command": f"cat {STATE_PATH}; {COUNT_SLEEPERS}"}, (MARKER, "sleepers=0")),
+ ("run_command", {"command": CONNECT}, ("egress=denied",)),
+ ]
+
+ def model_function(messages: list[ModelMessage], _info: AgentInfo) -> ModelResponse:
+ returns = [
+ part.content
+ for message in messages
+ for part in message.parts
+ if part.part_kind == "tool-return"
+ ]
+ for (_, _, expected), returned in zip(steps, returns):
+ missing = [text for text in expected if text not in str(returned)]
+ if missing:
+ raise RuntimeError(f"Expected {missing!r} in {returned!r}")
+ if len(returns) < len(steps):
+ tool_name, args, _ = steps[len(returns)]
+ return ModelResponse(
+ parts=[ToolCallPart(tool_name=tool_name, args=args, tool_call_id=f"step-{len(returns)}")]
+ )
+ return ModelResponse(parts=[TextPart(content="sandbox boundary e2e passed")])
+
+ agent = Agent(
+ FunctionModel(model_function),
+ instructions="Use the sandbox tools as requested.",
+ toolsets=[
+ SandboxToolset(
+ OpenShellSandboxBackend(), default_command_timeout=30.0, max_command_timeout=30.0
+ )
+ ],
+ )
+ result = agent.run_sync("Run the sandbox boundary system test.")
+ if result.output != "sandbox boundary e2e passed":
+ raise RuntimeError(f"Unexpected agent output: {result.output!r}")
+ return result.output
+
+ run_sandbox_agent()
+
+
+dag = example_sandbox_toolset_openshell()
+
+from tests_common.test_utils.system_tests import get_test_run # noqa: E402
+
+test_run = get_test_run(dag)
diff --git a/providers/common/ai/tests/unit/common/ai/sandbox/test_openshell.py b/providers/common/ai/tests/unit/common/ai/sandbox/test_openshell.py
new file mode 100644
index 00000000000..9dd4e6c8917
--- /dev/null
+++ b/providers/common/ai/tests/unit/common/ai/sandbox/test_openshell.py
@@ -0,0 +1,1479 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import builtins
+import contextlib
+import copy
+import errno
+import os
+import re
+import shlex
+import shutil
+import signal
+import subprocess
+import sys
+import time
+from collections.abc import Callable
+from unittest import mock
+
+import pytest
+
+pytest.importorskip("openshell")
+
+import grpc
+from openshell import SandboxClient, SandboxError as SdkError
+from openshell._proto import openshell_pb2, sandbox_pb2
+
+from airflow.providers.common.ai.sandbox.base import (
+ SandboxError,
+ SandboxFileTooLargeError,
+ SandboxSpec,
+ SandboxTerminalError,
+)
+from airflow.providers.common.ai.sandbox.openshell import (
+ _EXEC_GRACE,
+ _NO_SETSID,
+ _NO_SETSID_STATUS,
+ _RUN_WRAPPER,
+ _STAGING_FAILED,
+ _SYSTEM_PATH,
+ _WRAPPER_VAR_PREFIX,
+ OpenShellSandboxBackend,
+ _BoundedTail,
+ _wrapper_line,
+)
+
+_MODULE = "airflow.providers.common.ai.sandbox.openshell"
+_MONOTONIC_PATH = f"{_MODULE}.time.monotonic"
+_GATEWAY_DIR = "/home/airflow/.config/openshell/gateways/prod"
+# The temp file the SDK writes a refreshed OIDC token through, next to the registration.
+_TOKEN_TEMP_FILE = f"{_GATEWAY_DIR}/.oidc_token.k2x9w1qz.tmp"
+
+
+class _RpcError(grpc.RpcError):
+ def __init__(self, code: grpc.StatusCode, details: str = "") -> None:
+ super().__init__(details)
+ self._code = code
+ self._details = details
+
+ def code(self) -> grpc.StatusCode:
+ return self._code
+
+ def details(self) -> str:
+ return self._details
+
+
+class _Stream:
+ """An ExecSandbox response stream: events, optionally ending in an error."""
+
+ def __init__(self, events, error: Exception | None = None) -> None:
+ self._events = iter(events)
+ self._error = error
+ self.cancel = mock.Mock()
+
+ def __iter__(self):
+ return self
+
+ def __next__(self):
+ try:
+ return next(self._events)
+ except StopIteration:
+ if self._error is not None:
+ raise self._error from None
+ raise
+
+
+def _stdout(data: bytes):
+ return openshell_pb2.ExecSandboxEvent(stdout=openshell_pb2.ExecSandboxStdout(data=data))
+
+
+def _stderr(data: bytes):
+ return openshell_pb2.ExecSandboxEvent(stderr=openshell_pb2.ExecSandboxStderr(data=data))
+
+
+def _exit(code: int):
+ return openshell_pb2.ExecSandboxEvent(exit=openshell_pb2.ExecSandboxExit(exit_code=code))
+
+
+def _result(code: int = 0, out: bytes = b"", err: bytes = b"") -> _Stream:
+ events = []
+ if out:
+ events.append(_stdout(out))
+ if err:
+ events.append(_stderr(err))
+ events.append(_exit(code))
+ return _Stream(events)
+
+
+def _sandbox(phase: openshell_pb2.SandboxPhase = openshell_pb2.SANDBOX_PHASE_READY, *conditions):
+ response = openshell_pb2.SandboxResponse()
+ response.sandbox.status.phase = phase
+ for reason, message in conditions:
+ response.sandbox.status.conditions.add(type="Ready", status="False", reason=reason, message=message)
+ return response
+
+
+def _config(
+ hosts=(),
+ *,
+ source: sandbox_pb2.PolicySource = sandbox_pb2.POLICY_SOURCE_SANDBOX,
+ admitted: bool = True,
+ landlock: str = "hard_requirement",
+ approval_mode: str | None = None,
+ proposals: bool = False,
+ extra_rule: tuple[str, int] | None = None,
+ allowed_ips=(),
+ binaries=("/**",),
+ middlewares=(),
+):
+ config = sandbox_pb2.GetSandboxConfigResponse(
+ version=1, policy_source=source, configuration_admitted=admitted
+ )
+ config.policy.landlock.compatibility = landlock
+ for name in middlewares:
+ # Reading a missing key of a message map adds it.
+ config.policy.network_middlewares[name]
+ if hosts or allowed_ips:
+ rule = config.policy.network_policies["airflow-egress"]
+ rule.name = "airflow-egress"
+ for host in hosts:
+ rule.endpoints.add(host=host, ports=[443], allowed_ips=list(allowed_ips))
+ if not hosts:
+ rule.endpoints.add(ports=[443], allowed_ips=list(allowed_ips))
+ rule.binaries.extend(sandbox_pb2.NetworkBinary(path=path) for path in binaries)
+ if extra_rule is not None:
+ rule = config.policy.network_policies["allow_extra"]
+ rule.endpoints.add(host=extra_rule[0], port=extra_rule[1])
+ rule.binaries.add(path="/usr/local/bin/python3.12")
+ if approval_mode is not None:
+ config.settings["proposal_approval_mode"].value.string_value = approval_mode
+ config.settings["proposal_approval_mode"].scope = sandbox_pb2.SETTING_SCOPE_GLOBAL
+ if proposals:
+ config.settings["agent_policy_proposals_enabled"].value.bool_value = True
+ return config
+
+
+def _backend(**kwargs) -> tuple[OpenShellSandboxBackend, mock.MagicMock]:
+ backend = OpenShellSandboxBackend(gateway="test-gateway", **kwargs)
+ client = mock.MagicMock(spec=SandboxClient)
+ client._stub = mock.MagicMock(spec=["GetSandbox", "GetSandboxConfig", "ExecSandbox"])
+ client._stub.GetSandbox.return_value = _sandbox()
+ client._stub.GetSandboxConfig.return_value = _config()
+ client._stub.ExecSandbox.return_value = _result()
+ backend._client = client
+ return backend, client
+
+
+def _exec_request(client, call: int = -1):
+ return client._stub.ExecSandbox.call_args_list[call].args[0]
+
+
+def _nonce(request) -> str:
+ return request.stdin.partition(b"\n")[0].decode()
+
+
+def _reply(code: int, stderr: Callable[[str], bytes]):
+ """An ExecSandbox side effect whose stderr is built from the nonce the call sent."""
+
+ def reply(request, timeout):
+ return _result(code, err=stderr(_nonce(request)))
+
+ return reply
+
+
+class _Clock:
+ def __init__(self) -> None:
+ self.now = 0.0
+
+ def __call__(self) -> float:
+ return self.now
+
+
+def _deadline_exceeded(clock: _Clock, *, after: float | None = None, details: str = ""):
+ """An ExecSandbox side effect failing with DEADLINE_EXCEEDED ``after`` seconds in, by default at its deadline."""
+
+ def reply(request, timeout):
+ clock.now += timeout if after is None else after
+ return _Stream([], _RpcError(grpc.StatusCode.DEADLINE_EXCEEDED, details))
+
+ return reply
+
+
+@pytest.fixture(autouse=True)
+def _no_sleep():
+ with mock.patch(f"{_MODULE}.time.sleep", autospec=True):
+ yield
+
+
+@pytest.fixture
+def clock():
+ clock = _Clock()
+ with mock.patch(_MONOTONIC_PATH, side_effect=clock):
+ yield clock
+
+
+def test_missing_sdk_error_is_actionable():
+ real_import = builtins.__import__
+
+ def blocked_import(name, *args, **kwargs):
+ if name.startswith("openshell"):
+ raise ImportError("blocked for test")
+ return real_import(name, *args, **kwargs)
+
+ backend = OpenShellSandboxBackend()
+ with mock.patch("builtins.__import__", side_effect=blocked_import):
+ with pytest.raises(SandboxTerminalError, match=r"\[openshell\]"):
+ backend.create()
+
+
+@pytest.mark.parametrize(
+ ("kwargs", "message"),
+ [
+ ({"gateway": ""}, "gateway"),
+ ({"workspace": ""}, "workspace"),
+ ({"image": ""}, "image"),
+ ({"cpu": ""}, "cpu"),
+ ({"memory": ""}, "memory"),
+ ({"ready_timeout": 0}, "ready_timeout"),
+ ({"request_timeout": float("inf")}, "request_timeout"),
+ ],
+)
+def test_constructor_rejects_invalid_values(kwargs, message):
+ with pytest.raises(ValueError, match=message):
+ OpenShellSandboxBackend(**kwargs)
+
+
+class TestClient:
+ @mock.patch("openshell.SandboxClient.from_active_cluster", autospec=True)
+ def test_construction_reads_no_gateway_registration(self, from_active_cluster):
+ OpenShellSandboxBackend(gateway="prod")
+
+ from_active_cluster.assert_not_called()
+
+ def test_the_backend_can_be_deep_copied(self):
+ backend = OpenShellSandboxBackend(gateway="prod")
+
+ assert copy.deepcopy(backend)._gateway == "prod"
+
+ @mock.patch("openshell.SandboxClient.from_active_cluster", autospec=True)
+ def test_the_cli_gateway_registration_is_loaded_once(self, from_active_cluster):
+ backend = OpenShellSandboxBackend(gateway="prod", request_timeout=12)
+
+ assert backend._get_client() is backend._get_client()
+
+ from_active_cluster.assert_called_once_with(cluster="prod", timeout=12)
+
+ @mock.patch("openshell.SandboxClient.from_active_cluster", autospec=True)
+ def test_a_missing_registration_is_terminal(self, from_active_cluster):
+ from_active_cluster.side_effect = SdkError("gateway 'prod' not found")
+
+ with pytest.raises(SandboxTerminalError, match="gateway 'prod' not found"):
+ OpenShellSandboxBackend(gateway="prod")._get_client()
+
+ @pytest.mark.parametrize(
+ "error",
+ [
+ pytest.param(
+ PermissionError(errno.EACCES, "Permission denied", _TOKEN_TEMP_FILE), id="not-writable"
+ ),
+ pytest.param(
+ OSError(errno.EROFS, "Read-only file system", _TOKEN_TEMP_FILE), id="read-only-mount"
+ ),
+ ],
+ )
+ def test_an_oidc_token_that_cannot_be_written_back_is_terminal(self, error):
+ backend, client = _backend()
+ client._stub.GetSandboxConfig.side_effect = error
+
+ with pytest.raises(SandboxTerminalError, match="registration is read-only.*mTLS is the supported"):
+ backend.run_command("box", "true", timeout=5, max_output_bytes=100)
+
+ client._stub.ExecSandbox.assert_not_called()
+
+ @pytest.mark.parametrize(
+ "error",
+ [
+ pytest.param(
+ PermissionError(errno.EACCES, "Permission denied", f"{_GATEWAY_DIR}/mtls/tls.key"),
+ id="an-unreadable-mtls-key",
+ ),
+ pytest.param(
+ PermissionError(errno.EACCES, "Permission denied", f"{_GATEWAY_DIR}/oidc_token.json"),
+ id="the-token-file-itself",
+ ),
+ pytest.param(OSError(errno.ENOSPC, "No space left on device"), id="no-file-named"),
+ pytest.param(OSError(errno.ENOSPC, "No space left on device", _TOKEN_TEMP_FILE), id="disk-full"),
+ pytest.param(OSError(errno.EDQUOT, "Disk quota exceeded", _TOKEN_TEMP_FILE), id="quota-used-up"),
+ ],
+ )
+ def test_other_os_errors_are_not_reported_as_a_read_only_registration(self, error):
+ backend, client = _backend()
+ client._stub.GetSandboxConfig.side_effect = error
+
+ with pytest.raises(SandboxTerminalError) as raised:
+ backend.run_command("box", "true", timeout=5, max_output_bytes=100)
+
+ assert str(raised.value) == (
+ f"OpenShell could not read the network policy of sandbox box ({type(error).__name__}: {error})."
+ )
+
+
+class TestSpecRefusals:
+ @pytest.mark.parametrize(
+ ("spec", "message"),
+ [
+ (SandboxSpec(owner="dag/run"), "owner"),
+ (SandboxSpec(allow_egress_to_cidrs=["203.0.113.0/24"]), "allow_egress_to_cidrs"),
+ (SandboxSpec(block_network=False), "open network"),
+ (SandboxSpec(block_network=False, allow_egress_to=["pypi.org"]), "open network"),
+ (SandboxSpec(allow_egress_to="pypi.org"), "not one string"),
+ (SandboxSpec(env={"FOO": 1}), "strings to strings"), # type: ignore[dict-item]
+ ],
+ )
+ def test_unenforceable_specs_are_refused_before_the_gateway_is_asked(self, spec, message):
+ backend, client = _backend()
+
+ with pytest.raises(SandboxTerminalError, match=message):
+ backend.create(spec=spec)
+
+ client.create.assert_not_called()
+
+ @pytest.mark.parametrize(
+ "host",
+ ["https://pypi.org", "pypi.org:443", "203.0.113.7", "localhost", "*.com", "*", "pypi.org/simple", ""],
+ )
+ def test_entries_that_are_not_hostnames_are_refused(self, host):
+ backend, client = _backend()
+
+ with pytest.raises(SandboxTerminalError, match="bare hostnames"):
+ backend.create(spec=SandboxSpec(allow_egress_to=[host]))
+
+ client.create.assert_not_called()
+
+ @pytest.mark.parametrize(
+ "key", ["HTTPS_PROXY", "no_proxy", "SSL_CERT_FILE", "REQUESTS_CA_BUNDLE", "OPENSHELL_SANDBOX"]
+ )
+ def test_env_the_supervisor_would_silently_change_is_refused(self, key):
+ backend, client = _backend()
+
+ with pytest.raises(SandboxTerminalError, match=f"sets {key}"):
+ backend.create(spec=SandboxSpec(env={key: "value"}))
+
+ client.create.assert_not_called()
+
+ def test_every_variable_the_wrapper_sets_has_the_reserved_prefix(self):
+ # A wrapper variable outside the prefix could share a name with a spec's env key,
+ # and the shell would hand its value to the command. PATH holds no secret, and _
+ # is read's throwaway field.
+ assigned = set(re.findall(r"\b([A-Za-z_]\w*)=", _RUN_WRAPPER))
+ assigned |= set(re.findall(r"\bfor (\w+) in\b", _RUN_WRAPPER))
+ for names in re.findall(r"\bread (?:-r )?(\w+(?: \w+)*)", _RUN_WRAPPER):
+ assigned |= set(names.split())
+ assert f"{_WRAPPER_VAR_PREFIX}q" in assigned # the nonce, read from stdin
+
+ assert {name for name in assigned if not name.startswith(_WRAPPER_VAR_PREFIX)} == {"PATH", "_"}
+
+ def test_env_named_like_a_wrapper_variable_is_refused(self):
+ backend, client = _backend()
+
+ with pytest.raises(SandboxTerminalError, match="sets _airflow_q.*reserved for the shell wrapper"):
+ backend.create(spec=SandboxSpec(env={"_airflow_q": "value"}))
+
+ client.create.assert_not_called()
+
+
+class TestCreate:
+ def test_default_spec_is_deny_all_under_hard_landlock(self):
+ backend, client = _backend(image="python:3.13-slim", cpu="2", memory="4Gi")
+
+ with mock.patch(f"{_MODULE}.time.time", return_value=1_700_000_000.4):
+ name = backend.create(spec=SandboxSpec(env={"TOKEN": "value"}))
+
+ kwargs = client.create.call_args.kwargs
+ assert kwargs["name"] == name
+ assert name.startswith("airflow-")
+ assert len(name) <= 19
+ assert kwargs["workspace"] == "default"
+ assert kwargs["labels"] == {"created-by": "airflow", "airflow-created-at": "1700000000"}
+ spec = kwargs["spec"]
+ assert dict(spec.environment) == {"TOKEN": "value"}
+ assert spec.template.image == "python:3.13-slim"
+ assert dict(spec.template.resources)["limits"] == {"cpu": "2", "memory": "4Gi"}
+ assert len(spec.policy.network_policies) == 0
+ assert spec.policy.landlock.compatibility == "hard_requirement"
+ assert spec.policy.filesystem.include_workdir is True
+ assert list(spec.policy.filesystem.read_only) == [
+ "/bin",
+ "/usr",
+ "/lib",
+ "/proc",
+ "/dev/urandom",
+ "/etc",
+ "/var/log",
+ ]
+ assert list(spec.policy.filesystem.read_write) == ["/tmp", "/dev/null", "/dev/shm"]
+ client.delete.assert_not_called()
+
+ def test_none_spec_still_gets_an_explicit_deny_all_policy(self):
+ # Without a policy OpenShell falls back to one baked into the image, which may be open.
+ backend, client = _backend()
+
+ backend.create()
+
+ spec = client.create.call_args.kwargs["spec"]
+ assert spec.HasField("policy")
+ assert len(spec.policy.network_policies) == 0
+
+ def test_hostnames_become_one_rule_on_port_443_for_any_binary(self):
+ backend, client = _backend()
+ client._stub.GetSandboxConfig.return_value = _config(["files.pythonhosted.org", "pypi.org"])
+
+ backend.create(spec=SandboxSpec(allow_egress_to=["PyPI.org", "files.pythonhosted.org", "pypi.org"]))
+
+ rules = client.create.call_args.kwargs["spec"].policy.network_policies
+ assert list(rules) == ["airflow-egress"]
+ assert [(endpoint.host, list(endpoint.ports)) for endpoint in rules["airflow-egress"].endpoints] == [
+ ("files.pythonhosted.org", [443]),
+ ("pypi.org", [443]),
+ ]
+ assert [binary.path for binary in rules["airflow-egress"].binaries] == ["/**"]
+ client.delete.assert_not_called()
+
+ def test_waits_for_the_sandbox_to_become_ready(self):
+ backend, client = _backend()
+ client._stub.GetSandbox.side_effect = [
+ _sandbox(openshell_pb2.SANDBOX_PHASE_PROVISIONING),
+ _RpcError(grpc.StatusCode.UNAVAILABLE, "restarting"),
+ _sandbox(),
+ ]
+
+ backend.create(spec=SandboxSpec())
+
+ assert client._stub.GetSandbox.call_count == 3
+
+ def test_a_sandbox_that_fails_to_start_is_deleted_and_terminal(self):
+ backend, client = _backend()
+ client._stub.GetSandbox.return_value = _sandbox(
+ openshell_pb2.SANDBOX_PHASE_ERROR, ("IdentityResolutionFailed", "no passwd entry")
+ )
+
+ with pytest.raises(SandboxTerminalError, match="IdentityResolutionFailed: no passwd entry"):
+ backend.create(spec=SandboxSpec())
+
+ name = client.create.call_args.kwargs["name"]
+ client.delete.assert_called_once_with(name, workspace="default", allow_missing=True)
+
+ def test_a_sandbox_not_ready_in_time_is_deleted_and_terminal(self):
+ backend, client = _backend(ready_timeout=5)
+ client._stub.GetSandbox.return_value = _sandbox(openshell_pb2.SANDBOX_PHASE_PROVISIONING)
+
+ with mock.patch(_MONOTONIC_PATH, side_effect=[0.0, 1.0, 6.0]):
+ with pytest.raises(SandboxTerminalError, match="not ready: it is SANDBOX_PHASE_PROVISIONING"):
+ backend.create(spec=SandboxSpec())
+
+ client.delete.assert_called_once()
+
+ def test_a_create_the_gateway_rejects_is_terminal_and_cleaned_up(self):
+ backend, client = _backend()
+ client.create.side_effect = _RpcError(grpc.StatusCode.INVALID_ARGUMENT, "name exceeds maximum length")
+
+ with pytest.raises(SandboxTerminalError, match="INVALID_ARGUMENT: name exceeds maximum length"):
+ backend.create(spec=SandboxSpec())
+
+ client.delete.assert_called_once()
+
+ @pytest.mark.parametrize(
+ ("config", "message"),
+ [
+ (_config(source=sandbox_pb2.POLICY_SOURCE_GLOBAL), "gateway-wide policy"),
+ (_config(["pypi.org", "example.com"]), "admits"),
+ (_config(), "admits nothing where"),
+ (_config(extra_rule=("www.google.com", 443)), "www.google.com:443"),
+ (_config(["pypi.org"], binaries=("/usr/bin/curl",)), "admits"),
+ (_config(allowed_ips=["0.0.0.0/1"]), "admits addresses"),
+ (_config(["pypi.org"], approval_mode="auto"), "proposal_approval_mode is 'auto'"),
+ (_config(["pypi.org"], proposals=True), "agent_policy_proposals_enabled"),
+ (_config(["pypi.org"], landlock="best_effort"), "Landlock"),
+ (_config(["pypi.org"], admitted=False), "not admitted"),
+ (_config(["pypi.org"], middlewares=["inspect"]), "network middlewares"),
+ (_RpcError(grpc.StatusCode.PERMISSION_DENIED, "config:read"), "PERMISSION_DENIED"),
+ ],
+ ids=[
+ "global-override",
+ "wider-allowlist",
+ "missing-rule",
+ "auto-approved-rule",
+ "narrower-binaries",
+ "address-rule",
+ "auto-approval",
+ "agent-proposals",
+ "landlock-best-effort",
+ "not-admitted",
+ "network-middlewares",
+ "unreadable",
+ ],
+ )
+ def test_a_policy_other_than_the_requested_one_destroys_the_sandbox(self, config, message):
+ backend, client = _backend()
+ if isinstance(config, Exception):
+ client._stub.GetSandboxConfig.side_effect = config
+ else:
+ client._stub.GetSandboxConfig.return_value = config
+
+ with pytest.raises(SandboxTerminalError, match=message):
+ backend.create(spec=SandboxSpec(allow_egress_to=["pypi.org"]))
+
+ client.delete.assert_called_once()
+ assert backend._egress == {}
+
+ def test_a_sandbox_that_cannot_be_deleted_after_a_failed_create_is_logged(self, caplog):
+ backend, client = _backend()
+ client._stub.GetSandboxConfig.return_value = _config(extra_rule=("www.google.com", 443))
+ client.delete.side_effect = _RpcError(grpc.StatusCode.UNAVAILABLE, "down")
+
+ with pytest.raises(SandboxTerminalError, match="www.google.com:443") as error:
+ backend.create(spec=SandboxSpec())
+
+ name = client.create.call_args.kwargs["name"]
+ assert "was destroyed" not in str(error.value)
+ assert (
+ f"Could not destroy OpenShell sandbox {name} after a failed create; it will need manual cleanup"
+ ) in caplog.messages
+
+ @pytest.mark.parametrize("code", [grpc.StatusCode.UNAVAILABLE, grpc.StatusCode.DEADLINE_EXCEEDED])
+ def test_a_policy_read_rides_out_a_restarting_gateway(self, code):
+ backend, client = _backend()
+ client._stub.GetSandboxConfig.side_effect = [_RpcError(code, "restarting"), _config()]
+
+ backend.create(spec=SandboxSpec())
+
+ assert client._stub.GetSandboxConfig.call_count == 2
+ client.delete.assert_not_called()
+
+ def test_manual_approval_mode_is_accepted(self):
+ backend, client = _backend()
+ client._stub.GetSandboxConfig.return_value = _config(approval_mode="manual")
+
+ backend.create(spec=SandboxSpec())
+
+ client.delete.assert_not_called()
+
+
+class TestRunCommand:
+ def test_the_command_travels_on_stdin_to_the_guest_wrapper(self):
+ backend, client = _backend()
+ client._stub.ExecSandbox.return_value = _result(3, b"out\n", b"err\n")
+
+ result = backend.run_command(
+ "box", "echo out; echo err >&2; exit 3", timeout=2.5, max_output_bytes=100
+ )
+
+ request = _exec_request(client)
+ assert request.sandbox == "box"
+ assert list(request.command) == ["/bin/sh", "-c", _RUN_WRAPPER, "airflow-exec", "3", "100"]
+ nonce, _, command = request.stdin.partition(b"\n")
+ assert re.fullmatch(rb"[0-9a-f]{32}", nonce)
+ assert command == b"echo out; echo err >&2; exit 3"
+ assert request.no_login_shell is True
+ assert request.request_id
+ assert not request.HasField("execution_timeout")
+ assert client._stub.ExecSandbox.call_args.kwargs["timeout"] == 3 + _EXEC_GRACE
+ assert (result.exit_code, result.stdout, result.stderr) == (3, "out\n", "err\n")
+ assert not result.timed_out
+ assert not result.sandbox_terminated
+
+ @pytest.mark.parametrize(
+ ("exit_code", "elapsed", "timed_out"),
+ [(124, 3.1, True), (124, 0.2, False), (137, 3.1, False), (0, 3.1, False)],
+ )
+ def test_only_the_wrappers_status_after_the_budget_is_a_timeout(self, exit_code, elapsed, timed_out):
+ backend, client = _backend()
+ client._stub.ExecSandbox.return_value = _result(exit_code)
+
+ # The policy read before and after the command each take a deadline reading too.
+ with mock.patch(_MONOTONIC_PATH, side_effect=[0.0, 100.0, 100.0 + elapsed, 0.0]):
+ result = backend.run_command("box", "sleep 300", timeout=3, max_output_bytes=100)
+
+ assert result.timed_out is timed_out
+
+ def test_each_command_carries_a_fresh_nonce(self):
+ backend, client = _backend()
+ client._stub.ExecSandbox.side_effect = _reply(0, lambda nonce: b"")
+
+ backend.run_command("box", "true", timeout=5, max_output_bytes=100)
+ backend.run_command("box", "true", timeout=5, max_output_bytes=100)
+
+ assert _nonce(_exec_request(client, 0)) != _nonce(_exec_request(client, 1))
+
+ @pytest.mark.parametrize(
+ ("exit_code", "stderr"),
+ [
+ pytest.param(
+ 125, lambda nonce: b"docker: invalid reference format\n", id="a-command-exiting-125"
+ ),
+ pytest.param(
+ 1,
+ lambda nonce: _wrapper_line(_STAGING_FAILED, nonce),
+ id="the-staging-line-with-another-status",
+ ),
+ pytest.param(
+ 125,
+ lambda nonce: _wrapper_line(_STAGING_FAILED, nonce) + b"more\n",
+ id="the-staging-line-not-last",
+ ),
+ pytest.param(
+ 125,
+ lambda nonce: f"{_STAGING_FAILED}\n".encode(),
+ id="the-staging-text-printed-by-the-command",
+ ),
+ pytest.param(
+ 125,
+ lambda nonce: _wrapper_line(_STAGING_FAILED, "0" * 32),
+ id="the-staging-line-with-another-nonce",
+ ),
+ pytest.param(127, lambda nonce: b"sh: 1: setsid: not found\n", id="a-command-exiting-127"),
+ pytest.param(
+ 1, lambda nonce: _wrapper_line(_NO_SETSID, nonce), id="the-setsid-line-with-another-status"
+ ),
+ pytest.param(
+ 127, lambda nonce: _wrapper_line(_NO_SETSID, nonce) + b"more\n", id="the-setsid-line-not-last"
+ ),
+ pytest.param(
+ 127, lambda nonce: f"{_NO_SETSID}\n".encode(), id="the-setsid-text-printed-by-the-command"
+ ),
+ pytest.param(
+ 127,
+ lambda nonce: _wrapper_line(_NO_SETSID, "0" * 32),
+ id="the-setsid-line-with-another-nonce",
+ ),
+ ],
+ )
+ def test_a_result_unlike_a_wrapper_failure_is_the_commands_own(self, exit_code, stderr):
+ backend, client = _backend()
+ client._stub.ExecSandbox.side_effect = _reply(exit_code, stderr)
+
+ result = backend.run_command("box", "cmd", timeout=5, max_output_bytes=100)
+
+ expected = stderr(_nonce(_exec_request(client))).decode()
+ assert (result.exit_code, result.stderr) == (exit_code, expected)
+
+ @pytest.mark.parametrize(
+ ("max_output_bytes", "message"),
+ [
+ pytest.param(
+ 200,
+ "The command was not run: the sandbox could not write it to /tmp "
+ "(cat: write error: No space left on device).",
+ id="with-the-cause",
+ ),
+ pytest.param(
+ 10,
+ "The command was not run: the sandbox could not write it to /tmp.",
+ id="under-a-cap-shorter-than-the-staging-line",
+ ),
+ ],
+ )
+ def test_the_wrappers_own_staging_failure_is_a_recoverable_error(self, max_output_bytes, message):
+ backend, client = _backend()
+ client._stub.ExecSandbox.side_effect = _reply(
+ 125,
+ lambda nonce: (
+ b"cat: write error: No space left on device\n" + _wrapper_line(_STAGING_FAILED, nonce)
+ ),
+ )
+
+ with pytest.raises(SandboxError) as error:
+ backend.run_command("box", "make", timeout=5, max_output_bytes=max_output_bytes)
+
+ assert str(error.value) == message
+ assert not isinstance(error.value, SandboxTerminalError)
+
+ def test_an_image_without_setsid_is_terminal(self):
+ backend, client = _backend()
+ client._stub.ExecSandbox.side_effect = _reply(
+ _NO_SETSID_STATUS, lambda nonce: _wrapper_line(_NO_SETSID, nonce)
+ )
+
+ # A cap shorter than the line, so the line is still recognised past it.
+ with pytest.raises(SandboxTerminalError, match="no setsid.*util-linux"):
+ backend.run_command("box", "true", timeout=5, max_output_bytes=10)
+
+ def test_stderr_stays_capped_below_the_length_of_the_staging_line(self):
+ backend, client = _backend()
+ client._stub.ExecSandbox.return_value = _result(1, err=b"0123456789" * 3)
+
+ result = backend.run_command("box", "cmd", timeout=5, max_output_bytes=20)
+
+ assert (result.stderr, result.stderr_truncated) == ("0123456789" * 2, True)
+
+ def test_each_stream_is_capped_and_flagged_on_its_own(self):
+ backend, client = _backend()
+ # The wrapper sends one byte past the cap for a stream it had to cut.
+ client._stub.ExecSandbox.return_value = _Stream(
+ [_stdout(b"partial line\nkept 1\n"), _stdout(b"kept 2\n"), _stderr(b"short\n"), _exit(0)]
+ )
+
+ result = backend.run_command("box", "cmd", timeout=5, max_output_bytes=20)
+
+ assert result.stdout == "kept 1\nkept 2\n"
+ assert result.stdout_truncated
+ assert result.stderr == "short\n"
+ assert not result.stderr_truncated
+
+ def test_one_line_longer_than_the_budget_still_reaches_the_model(self):
+ """Dropping the leading partial line must not empty the window: the toolset prints "(no output)" for it."""
+ backend, client = _backend()
+ client._stub.ExecSandbox.return_value = _Stream([_stdout(b"x" * 60000 + b"\n"), _exit(0)])
+
+ result = backend.run_command("box", "cat big.json", timeout=5, max_output_bytes=51200)
+
+ assert result.stdout != ""
+ assert result.stdout_truncated
+ assert len(result.stdout.encode()) <= 51200
+
+ def test_a_long_line_followed_by_a_short_one_keeps_the_window(self):
+ backend, client = _backend()
+ client._stub.ExecSandbox.return_value = _Stream(
+ [_stdout(b"L" * 200_000 + b"\nshort tail\n"), _exit(0)]
+ )
+
+ result = backend.run_command("box", "spew", timeout=5, max_output_bytes=51200)
+
+ assert len(result.stdout.encode()) > 51200 // 2
+ assert result.stdout.endswith("short tail\n")
+
+ def test_the_whole_second_deadline_the_command_got_is_reported(self):
+ backend, client = _backend()
+
+ result = backend.run_command("box", "true", timeout=2.2, max_output_bytes=100)
+
+ assert result.applied_timeout == 3.0
+
+ def test_the_whole_second_deadline_is_reported_for_an_abandoned_command_too(self, clock):
+ backend, client = _backend()
+ client._stub.ExecSandbox.side_effect = _deadline_exceeded(clock)
+
+ result = backend.run_command("box", "sleep 300", timeout=2.2, max_output_bytes=100)
+
+ assert (result.sandbox_terminated, result.applied_timeout) == (True, 3.0)
+
+ def test_output_injected_past_the_wrapper_stays_bounded_in_worker_memory(self):
+ backend, client = _backend()
+ chunk = b"y\n" * 32768
+ client._stub.ExecSandbox.return_value = _Stream([_stdout(chunk) for _ in range(200)] + [_exit(0)])
+ buffered: list[int] = []
+ real_add = _BoundedTail.add
+
+ def add(tail, data):
+ real_add(tail, data)
+ buffered.append(len(tail._data))
+
+ with mock.patch.object(_BoundedTail, "add", autospec=True, side_effect=add):
+ result = backend.run_command("box", "yes > /proc/$PPID/fd/1", timeout=5, max_output_bytes=1024)
+
+ assert len(buffered) == 200
+ assert max(buffered) == 1024
+ assert len(result.stdout.encode()) <= 1024
+ assert result.stdout_truncated
+
+ def test_undecodable_output_is_replaced_not_raised(self):
+ backend, client = _backend()
+ client._stub.ExecSandbox.return_value = _result(0, b"\xff\xfeok\n")
+
+ assert backend.run_command("box", "cmd", timeout=5, max_output_bytes=100).stdout == "��ok\n"
+
+ def test_a_command_over_the_request_limit_is_recoverable_and_not_sent(self):
+ backend, client = _backend()
+
+ with pytest.raises(SandboxError, match="write_file") as error:
+ backend.run_command("box", "x" * 1_000_001, timeout=5, max_output_bytes=100)
+
+ assert not isinstance(error.value, SandboxTerminalError)
+ client._stub.ExecSandbox.assert_not_called()
+
+ @pytest.mark.parametrize(
+ "after", [None, 3 + _EXEC_GRACE - 0.5], ids=["at-the-deadline", "just-before-the-deadline"]
+ )
+ def test_a_hung_exec_destroys_the_sandbox(self, clock, after):
+ backend, client = _backend()
+ client._stub.ExecSandbox.side_effect = _deadline_exceeded(clock, after=after)
+
+ result = backend.run_command("box", "sleep 300", timeout=3, max_output_bytes=100)
+
+ assert (result.exit_code, result.timed_out, result.sandbox_terminated) == (-1, True, True)
+ client.delete.assert_called_once_with("box", workspace="default", allow_missing=True)
+
+ def test_a_deadline_the_gateway_reports_early_is_a_command_that_did_not_run(self, clock):
+ backend, client = _backend()
+ client._stub.ExecSandbox.side_effect = _deadline_exceeded(
+ clock, after=10.0, details="relay open timed out"
+ )
+
+ with pytest.raises(SandboxError, match="was not run.*relay open timed out") as error:
+ backend.run_command("box", "sleep 300", timeout=3, max_output_bytes=100)
+
+ assert not isinstance(error.value, SandboxTerminalError)
+ client.delete.assert_not_called()
+
+ @pytest.mark.parametrize("details", ["upstream max stream duration reached", ""])
+ def test_any_other_early_deadline_may_have_left_the_command_running(self, clock, details):
+ backend, client = _backend()
+ client._stub.ExecSandbox.side_effect = _deadline_exceeded(clock, after=10.0, details=details)
+
+ with pytest.raises(SandboxError, match="may or may not have run.*may still be running") as error:
+ backend.run_command("box", "make install", timeout=60, max_output_bytes=100)
+
+ assert not isinstance(error.value, SandboxTerminalError)
+ client.delete.assert_not_called()
+
+ def test_a_sandbox_not_ready_after_a_gateway_restart_is_waited_for_and_retried(self):
+ backend, client = _backend()
+ client._stub.ExecSandbox.side_effect = [
+ _RpcError(grpc.StatusCode.FAILED_PRECONDITION, "sandbox is not ready"),
+ _result(0, b"ok\n"),
+ ]
+ client._stub.GetSandbox.side_effect = [_sandbox(openshell_pb2.SANDBOX_PHASE_PROVISIONING), _sandbox()]
+
+ result = backend.run_command("box", "echo ok", timeout=5, max_output_bytes=100)
+
+ assert result.stdout == "ok\n"
+ assert client._stub.ExecSandbox.call_count == 2
+ # Polled until ready before the retry: once while provisioning, once ready.
+ assert client._stub.GetSandbox.call_count == 2
+ assert _exec_request(client, 0).request_id != _exec_request(client, 1).request_id
+
+ def test_a_gateway_drop_during_the_not_ready_retry_is_reported_not_leaked(self):
+ backend, client = _backend()
+ client._stub.ExecSandbox.side_effect = [
+ _RpcError(grpc.StatusCode.FAILED_PRECONDITION, "sandbox is not ready"),
+ _Stream([], _RpcError(grpc.StatusCode.UNAVAILABLE, "exec relay closed")),
+ ]
+
+ with pytest.raises(SandboxError, match="may or may not have run") as error:
+ backend.run_command("box", "make install", timeout=5, max_output_bytes=100)
+
+ assert not isinstance(error.value, SandboxTerminalError)
+ assert client._stub.ExecSandbox.call_count == 2
+
+ def test_an_exec_the_gateway_drops_is_reported_not_retried(self):
+ backend, client = _backend()
+ client._stub.ExecSandbox.return_value = _Stream(
+ [],
+ _RpcError(
+ grpc.StatusCode.UNAVAILABLE, "exec relay closed before the command reported an exit status"
+ ),
+ )
+ client._stub.GetSandbox.side_effect = [_RpcError(grpc.StatusCode.UNAVAILABLE), _sandbox()]
+
+ with pytest.raises(SandboxError, match="may or may not have run.*may still be running") as error:
+ backend.run_command("box", "make install", timeout=5, max_output_bytes=100)
+
+ assert not isinstance(error.value, SandboxTerminalError)
+ client._stub.ExecSandbox.assert_called_once()
+
+ def test_a_dropped_exec_on_a_sandbox_that_does_not_come_back_is_terminal(self):
+ backend, client = _backend()
+ client._stub.ExecSandbox.return_value = _Stream([_stdout(b"x")])
+ client._stub.GetSandbox.return_value = _sandbox(openshell_pb2.SANDBOX_PHASE_ERROR)
+
+ with pytest.raises(SandboxTerminalError, match="SANDBOX_PHASE_ERROR"):
+ backend.run_command("box", "cmd", timeout=5, max_output_bytes=100)
+
+ @pytest.mark.parametrize(
+ ("code", "terminal"),
+ [
+ (grpc.StatusCode.NOT_FOUND, True),
+ (grpc.StatusCode.UNAUTHENTICATED, True),
+ (grpc.StatusCode.PERMISSION_DENIED, True),
+ (grpc.StatusCode.OUT_OF_RANGE, False),
+ (grpc.StatusCode.RESOURCE_EXHAUSTED, False),
+ ],
+ )
+ def test_gateway_errors_are_classified(self, code, terminal):
+ backend, client = _backend()
+ client._stub.ExecSandbox.side_effect = _RpcError(code, "detail")
+
+ with pytest.raises(SandboxError, match=f"{code.name}: detail") as error:
+ backend.run_command("box", "cmd", timeout=5, max_output_bytes=100)
+
+ assert isinstance(error.value, SandboxTerminalError) is terminal
+ assert client._stub.ExecSandbox.call_count == 1
+
+ def test_egress_widened_before_the_command_is_terminal_and_nothing_runs(self):
+ backend, client = _backend()
+ backend.create(spec=SandboxSpec())
+ name = client.create.call_args.kwargs["name"]
+ client._stub.GetSandboxConfig.return_value = _config(extra_rule=("www.google.com", 443))
+
+ with pytest.raises(SandboxTerminalError, match="changed after it was created"):
+ backend.run_command(name, "curl https://www.google.com", timeout=5, max_output_bytes=100)
+
+ client._stub.ExecSandbox.assert_not_called()
+ client.delete.assert_called_once_with(name, workspace="default", allow_missing=True)
+
+ def test_widened_egress_on_a_sandbox_that_cannot_be_destroyed_is_not_reported_as_destroyed(self, caplog):
+ backend, client = _backend()
+ backend.create(spec=SandboxSpec())
+ name = client.create.call_args.kwargs["name"]
+ client._stub.GetSandboxConfig.return_value = _config(extra_rule=("www.google.com", 443))
+ client.delete.side_effect = _RpcError(grpc.StatusCode.UNAVAILABLE, "down")
+
+ with pytest.raises(SandboxTerminalError, match="destroying the sandbox failed") as error:
+ backend.run_command(name, "true", timeout=5, max_output_bytes=100)
+
+ assert "was destroyed" not in str(error.value)
+ assert (
+ f"Could not destroy OpenShell sandbox {name} after its network policy changed; "
+ "it will need manual cleanup"
+ ) in caplog.messages
+
+ def test_egress_widened_during_the_command_is_terminal(self):
+ backend, client = _backend()
+ backend.create(spec=SandboxSpec())
+ name = client.create.call_args.kwargs["name"]
+ client._stub.GetSandboxConfig.side_effect = [
+ _config(),
+ _config(source=sandbox_pb2.POLICY_SOURCE_GLOBAL),
+ ]
+
+ with pytest.raises(SandboxTerminalError, match="gateway-wide policy"):
+ backend.run_command(name, "sleep 20", timeout=30, max_output_bytes=100)
+
+ client._stub.ExecSandbox.assert_called_once()
+
+ @mock.patch(_MONOTONIC_PATH, side_effect=[0.0, 59.0, 61.0])
+ def test_a_gateway_down_past_the_recovery_window_is_terminal_and_nothing_runs(self, _):
+ backend, client = _backend()
+ client._stub.GetSandboxConfig.side_effect = [_RpcError(grpc.StatusCode.UNAVAILABLE, "down")] * 3
+
+ with pytest.raises(SandboxTerminalError, match="UNAVAILABLE: down"):
+ backend.run_command("box", "true", timeout=5, max_output_bytes=100)
+
+ assert client._stub.GetSandboxConfig.call_count == 2
+ client._stub.ExecSandbox.assert_not_called()
+
+ def test_a_handle_created_elsewhere_is_held_to_what_it_admits_when_first_seen(self):
+ backend, client = _backend()
+ client._stub.GetSandboxConfig.side_effect = [
+ _config(["pypi.org"]),
+ _config(["pypi.org"]),
+ _config(["pypi.org", "evil.example.com"]),
+ ]
+
+ backend.run_command("other", "true", timeout=5, max_output_bytes=100)
+ with pytest.raises(SandboxTerminalError, match="evil.example.com"):
+ backend.run_command("other", "true", timeout=5, max_output_bytes=100)
+
+ def test_a_stream_without_an_exit_status_is_not_a_success(self):
+ backend, client = _backend()
+ client._stub.ExecSandbox.return_value = _Stream([_stdout(b"half")])
+
+ with pytest.raises(SandboxError, match="without an exit status"):
+ backend.run_command("box", "cmd", timeout=5, max_output_bytes=100)
+
+
+_PATH_OPERATIONS = [
+ pytest.param(lambda backend, path: backend.read_file("box", path, max_bytes=10), id="read"),
+ pytest.param(lambda backend, path: backend.write_file("box", path, b"data"), id="write"),
+]
+
+
+class TestFiles:
+ def test_read_returns_raw_bytes_after_the_size_line(self):
+ backend, client = _backend()
+ payload = bytes(range(256))
+ client._stub.ExecSandbox.return_value = _result(0, b"256\n" + payload)
+
+ assert backend.read_file("box", "/tmp/data.bin", max_bytes=1000) == payload
+
+ request = _exec_request(client)
+ assert list(request.command[-2:]) == ["/tmp/data.bin", "1001"]
+ assert "head -c" in request.command[2]
+
+ def test_an_oversized_read_reports_the_real_size(self):
+ backend, client = _backend()
+ client._stub.ExecSandbox.return_value = _result(0, b"10000000\n" + b"x" * 11)
+
+ with pytest.raises(SandboxFileTooLargeError) as error:
+ backend.read_file("box", "/tmp/big", max_bytes=10)
+
+ assert (error.value.size_bytes, error.value.max_bytes) == (10_000_000, 10)
+
+ def test_a_stream_reporting_no_size_is_still_over_budget(self):
+ backend, client = _backend()
+ client._stub.ExecSandbox.return_value = _result(0, b"0\n" + b"z" * 11)
+
+ with pytest.raises(SandboxFileTooLargeError) as error:
+ backend.read_file("box", "/proc/self/fd/0", max_bytes=10)
+
+ assert error.value.size_bytes == 11
+
+ @pytest.mark.parametrize(
+ ("exit_code", "stderr", "message"),
+ [
+ (66, b"", "does not exist"),
+ (67, b"", "is a directory"),
+ (68, b"", "is a named pipe"),
+ (1, b"head: cannot open '/dev/zero' for reading: Permission denied\n", "Permission denied"),
+ ],
+ )
+ def test_read_failures_are_recoverable(self, exit_code, stderr, message):
+ backend, client = _backend()
+ client._stub.ExecSandbox.return_value = _result(exit_code, err=stderr)
+
+ with pytest.raises(SandboxError, match=message) as error:
+ backend.read_file("box", "/x", max_bytes=10)
+
+ assert not isinstance(error.value, SandboxTerminalError)
+
+ def test_a_hung_file_operation_is_recoverable_and_keeps_the_sandbox(self, clock):
+ backend, client = _backend()
+ client._stub.ExecSandbox.side_effect = _deadline_exceeded(clock)
+
+ with pytest.raises(SandboxError, match="did not finish") as error:
+ backend.read_file("box", "/x", max_bytes=10)
+
+ assert not isinstance(error.value, SandboxTerminalError)
+ client.delete.assert_not_called()
+
+ def test_write_streams_content_on_stdin_in_chunks(self):
+ backend, client = _backend()
+ client._stub.ExecSandbox.side_effect = lambda *args, **kwargs: _result(0)
+ content = bytes(range(256)) * 5
+
+ with mock.patch(f"{_MODULE}._WRITE_CHUNK_BYTES", 512):
+ backend.write_file("box", "/sandbox/dir/file.bin", content)
+
+ requests = [call.args[0] for call in client._stub.ExecSandbox.call_args_list]
+ assert b"".join(request.stdin for request in requests) == content
+ assert [len(request.stdin) for request in requests] == [512, 512, 256]
+ assert "mkdir -p" in requests[0].command[2]
+ assert 'cat >"$1"' in requests[0].command[2]
+ assert all('cat >>"$1"' in request.command[2] for request in requests[1:])
+ assert all(request.command[-1] == "/sandbox/dir/file.bin" for request in requests)
+
+ def test_writing_empty_content_still_creates_the_file(self):
+ backend, client = _backend()
+
+ backend.write_file("box", "/tmp/empty", b"")
+
+ client._stub.ExecSandbox.assert_called_once()
+ assert _exec_request(client).stdin == b""
+
+ def test_a_failed_write_is_recoverable(self):
+ backend, client = _backend()
+ client._stub.ExecSandbox.return_value = _result(1, err=b"cat: /opt/x: Permission denied\n")
+
+ with pytest.raises(SandboxError, match="Permission denied") as error:
+ backend.write_file("box", "/opt/x", b"data")
+
+ assert not isinstance(error.value, SandboxTerminalError)
+
+ @pytest.mark.parametrize("operation", _PATH_OPERATIONS)
+ def test_file_operations_check_the_policy_before_and_after(self, operation):
+ backend, client = _backend()
+ client._stub.ExecSandbox.return_value = _result(0, b"4\ndata")
+
+ operation(backend, "/tmp/x")
+
+ assert client._stub.GetSandboxConfig.call_count == 2
+
+ @pytest.mark.parametrize("operation", _PATH_OPERATIONS)
+ def test_a_policy_widened_during_a_file_operation_is_terminal(self, operation):
+ backend, client = _backend()
+ client._stub.ExecSandbox.return_value = _result(0, b"4\ndata")
+ client._stub.GetSandboxConfig.side_effect = [
+ _config(),
+ _config(source=sandbox_pb2.POLICY_SOURCE_GLOBAL),
+ ]
+
+ with pytest.raises(SandboxTerminalError, match="gateway-wide policy"):
+ operation(backend, "/tmp/x")
+
+ client._stub.ExecSandbox.assert_called_once()
+ client.delete.assert_called_once_with("box", workspace="default", allow_missing=True)
+
+ @pytest.mark.parametrize("operation", _PATH_OPERATIONS)
+ def test_a_gateway_wide_policy_before_a_file_operation_is_terminal_and_nothing_runs(self, operation):
+ backend, client = _backend()
+ client._stub.GetSandboxConfig.return_value = _config(source=sandbox_pb2.POLICY_SOURCE_GLOBAL)
+
+ with pytest.raises(SandboxTerminalError, match="gateway-wide policy"):
+ operation(backend, "/tmp/x")
+
+ client._stub.ExecSandbox.assert_not_called()
+ client.delete.assert_called_once_with("box", workspace="default", allow_missing=True)
+
+ @pytest.mark.parametrize(
+ ("path", "message"),
+ [
+ pytest.param("/tmp/a\0b", "contains a NUL", id="nul"),
+ # 16385 characters, but 32769 bytes of UTF-8.
+ pytest.param("/" + "é" * 16384, "is 32769 bytes, over the 32768", id="over-32-kib"),
+ ],
+ )
+ @pytest.mark.parametrize("operation", _PATH_OPERATIONS)
+ def test_a_path_the_gateway_refuses_is_recoverable_and_not_sent(self, operation, path, message):
+ backend, client = _backend()
+
+ with pytest.raises(SandboxError, match=message) as error:
+ operation(backend, path)
+
+ assert not isinstance(error.value, SandboxTerminalError)
+ client._stub.ExecSandbox.assert_not_called()
+
+ @pytest.mark.parametrize(
+ "path",
+ [
+ # 16385 characters, exactly 32768 bytes of UTF-8.
+ pytest.param("/" + "é" * 16383 + "x", id="at-32-kib"),
+ pytest.param("/tmp/a\tb\nc\x7f", id="control-characters"),
+ ],
+ )
+ @pytest.mark.parametrize("operation", _PATH_OPERATIONS)
+ def test_a_path_the_gateway_accepts_is_sent(self, operation, path):
+ backend, client = _backend()
+ client._stub.ExecSandbox.return_value = _result(0, b"4\ndata")
+
+ operation(backend, path)
+
+ client._stub.ExecSandbox.assert_called_once()
+ assert path in _exec_request(client).command
+
+ def test_listing_parses_nul_separated_records(self):
+ backend, client = _backend()
+ client._stub.ExecSandbox.return_value = _result(0, b"d sub\0f a file\0f new\nline\0")
+
+ assert backend.list_directory("box", "/sandbox") == [
+ ("sub", True),
+ ("a file", False),
+ ("new\nline", False),
+ ]
+ assert "find -- /sandbox" in _exec_request(client).stdin.decode()
+
+ def test_a_listing_cut_at_the_cap_is_refused_rather_than_returned_partial(self):
+ backend, client = _backend()
+ client._stub.ExecSandbox.return_value = _result(0, b"f x\0" * 300_000)
+
+ with pytest.raises(SandboxError, match="too many entries"):
+ backend.list_directory("box", "/sandbox")
+
+ def test_a_listing_the_gateway_stops_relaying_is_terminal(self, clock, caplog):
+ backend, client = _backend()
+ client._stub.ExecSandbox.side_effect = _deadline_exceeded(clock)
+
+ with pytest.raises(SandboxTerminalError, match="stopped answering while listing '/sandbox'"):
+ backend.list_directory("box", "/sandbox")
+
+ client.delete.assert_called_once_with("box", workspace="default", allow_missing=True)
+ assert (
+ "OpenShell sandbox box reported nothing for 150s, its command budget of 120s plus 30s; "
+ "destroying it to stop the command"
+ ) in caplog.messages
+
+ def test_a_listing_that_runs_out_of_time_is_recoverable(self, clock):
+ backend, client = _backend()
+
+ def timed_out(request, timeout):
+ clock.now += 120.5
+ return _result(124)
+
+ client._stub.ExecSandbox.side_effect = timed_out
+
+ with pytest.raises(SandboxError, match="did not finish within 120 seconds") as error:
+ backend.list_directory("box", "/sandbox")
+
+ assert not isinstance(error.value, SandboxTerminalError)
+ client.delete.assert_not_called()
+
+
+class TestDestroy:
+ def test_destroy_deletes_without_waiting_and_forgets_the_policy(self):
+ backend, client = _backend()
+ name = backend.create(spec=SandboxSpec())
+
+ backend.destroy(name)
+
+ client.delete.assert_called_once_with(name, workspace="default", allow_missing=True)
+ client.wait_deleted.assert_not_called()
+ assert name not in backend._egress
+
+ def test_destroying_a_missing_sandbox_is_not_an_error(self):
+ backend, client = _backend()
+ client.delete.side_effect = _RpcError(grpc.StatusCode.NOT_FOUND)
+
+ backend.destroy("gone")
+
+ def test_other_destroy_failures_are_raised_for_the_toolset_to_log(self):
+ backend, client = _backend()
+ client.delete.side_effect = _RpcError(grpc.StatusCode.UNAVAILABLE, "gateway down")
+
+ with pytest.raises(SandboxError, match="gateway down"):
+ backend.destroy("box")
+
+
+_HAS_PROC_AND_SETSID = sys.platform.startswith("linux") and os.path.isdir("/proc") and shutil.which("setsid")
+
+
+@pytest.mark.skipif(not _HAS_PROC_AND_SETSID, reason="the guest wrapper needs Linux /proc and setsid")
+class TestRunWrapper:
+ """Run the guest wrapper under the local /bin/sh, the way the sandbox runs it."""
+
+ _NONCE = "4f3c2b1a0e9d8c7b6a5f4e3d2c1b0a99"
+
+ @pytest.fixture(autouse=True)
+ def _no_sleep(self):
+ """Keep the real time.sleep: the module's fixture patches it everywhere, not only in the backend."""
+
+ @classmethod
+ def _run(
+ cls, command: str, seconds: int = 5, cap: int = 1000, wrapper: str = _RUN_WRAPPER
+ ) -> subprocess.CompletedProcess[bytes]:
+ return subprocess.run(
+ ["/bin/sh", "-c", wrapper, "airflow-exec", str(seconds), str(cap)],
+ input=f"{cls._NONCE}\n{command}".encode(),
+ capture_output=True,
+ timeout=seconds + 20,
+ check=False,
+ )
+
+ @staticmethod
+ def _path_without_setsid(tmp_path) -> str:
+ for tool in ("mktemp", "cat", "rm"):
+ (tmp_path / tool).symlink_to(shutil.which(tool))
+ return f"PATH={tmp_path}"
+
+ @staticmethod
+ def _backend_on_local_shell(system_path: str = _SYSTEM_PATH) -> OpenShellSandboxBackend:
+ """A backend whose gateway runs each exec request under the local /bin/sh."""
+ backend, client = _backend()
+
+ def exec_locally(request, timeout):
+ argv = list(request.command)
+ argv[2] = argv[2].replace(_SYSTEM_PATH, system_path)
+ done = subprocess.run(
+ argv, input=request.stdin, capture_output=True, timeout=timeout, check=False
+ )
+ return _result(done.returncode, done.stdout, done.stderr)
+
+ client._stub.ExecSandbox.side_effect = exec_locally
+ return backend
+
+ @staticmethod
+ def _pids(marker: str) -> list[int]:
+ pids = []
+ for entry in os.listdir("/proc"):
+ if entry.isdigit():
+ try:
+ with open(f"/proc/{entry}/cmdline", "rb") as f:
+ if f.read().replace(b"\0", b" ").strip() == marker.encode():
+ pids.append(int(entry))
+ except OSError:
+ continue
+ return pids
+
+ def test_status_and_streams_pass_through(self):
+ result = self._run("echo out; echo err >&2; exit 7")
+
+ assert (result.returncode, result.stdout, result.stderr) == (7, b"out\n", b"err\n")
+
+ def test_the_timeout_kills_the_command_and_everything_it_started(self):
+ started = time.monotonic()
+ result = self._run("sleep 301 & (sleep 302; echo x) & sleep 303", seconds=2)
+
+ elapsed = time.monotonic() - started
+ time.sleep(0.5)
+ survivors = [pid for n in (301, 302, 303) for pid in self._pids(f"sleep {n}")]
+ for pid in survivors:
+ os.kill(pid, signal.SIGKILL)
+ assert result.returncode == 124
+ assert elapsed < 10
+ assert not survivors
+
+ def test_the_timeout_rescans_for_children_forked_during_the_sweep(self):
+ # One pass over /proc misses a child forked after the pass listed it. Each watcher
+ # forks one sleep as soon as the canary, started first, is killed, while the pass is
+ # still working through the 200 fillers started before the watchers; only a sweep that
+ # scans again finds those sleeps. Each watcher forks once, so the rescan always ends.
+ command = (
+ "sleep 312 & c=$!\n"
+ "for i in $(seq 200); do sleep 313 & done\n"
+ "for i in $(seq 8); do (\n"
+ ' while read -r s </proc/$c/stat; do case $s in *") Z "*) break ;; esac; done 2>/dev/null\n'
+ " sleep 311 &\n"
+ ") & done\n"
+ "wait"
+ )
+ result = self._run(command, seconds=2)
+
+ time.sleep(0.5)
+ survivors = [pid for n in (311, 312, 313) for pid in self._pids(f"sleep {n}")]
+ for pid in survivors:
+ with contextlib.suppress(ProcessLookupError):
+ os.kill(pid, signal.SIGKILL)
+ assert result.returncode == 124
+ assert not survivors
+
+ @pytest.mark.parametrize(
+ "name", [pytest.param("a) b c", id="a-parenthesis"), pytest.param("a\nb", id="a-newline")]
+ )
+ def test_the_timeout_kills_a_process_with_a_parenthesis_or_newline_in_its_name(self, tmp_path, name):
+ # The kernel names a process after the file it executes, and /proc/<pid>/stat shows that
+ # name unescaped, in parentheses. Under a subshell, the sleep's parent is not the session leader.
+ named = tmp_path / name
+ named.symlink_to(shutil.which("sleep"))
+
+ result = self._run(f"({shlex.quote(str(named))} 309; :) & sleep 310", seconds=2)
+
+ time.sleep(0.5)
+ survivors = self._pids(f"{named} 309")
+ for pid in survivors:
+ os.kill(pid, signal.SIGKILL)
+ assert result.returncode == 124
+ assert not survivors
+
+ def test_a_timer_starved_of_cpu_fires_on_elapsed_time(self):
+ # Stopping the wrapper's process group, which the setsid'd command has left, stands in for
+ # CPU contention; a timer counting its own one-second sleeps would fire five seconds late.
+ stdin, writer = os.pipe()
+ os.write(writer, f"{self._NONCE}\nsleep 306".encode())
+ os.close(writer)
+ with subprocess.Popen(
+ ["/bin/sh", "-c", _RUN_WRAPPER, "airflow-exec", "6", "1000"],
+ stdin=stdin,
+ stdout=subprocess.DEVNULL,
+ stderr=subprocess.DEVNULL,
+ start_new_session=True,
+ ) as wrapper:
+ os.close(stdin)
+ for _ in range(100):
+ if self._pids("sleep 306"):
+ break
+ time.sleep(0.05)
+ time.sleep(0.3)
+ os.killpg(wrapper.pid, signal.SIGSTOP)
+ time.sleep(6)
+ os.killpg(wrapper.pid, signal.SIGCONT)
+ resumed = time.monotonic()
+
+ assert wrapper.wait(timeout=10) == 124
+ assert time.monotonic() - resumed < 3
+
+ def test_a_background_child_does_not_hold_the_call(self):
+ started = time.monotonic()
+ result = self._run("sleep 304 & echo started", seconds=30)
+
+ survivors = self._pids("sleep 304")
+ for pid in survivors:
+ os.kill(pid, signal.SIGKILL)
+ assert (result.returncode, result.stdout) == (0, b"started\n")
+ assert time.monotonic() - started < 5
+ assert survivors
+
+ def test_a_failure_to_stage_the_command_has_its_own_status(self, tmp_path):
+ # A missing directory fails the staging the way a read-only or full /tmp does.
+ wrapper = _RUN_WRAPPER.replace("mktemp -d /tmp/", f"mktemp -d {tmp_path}/missing/")
+ assert wrapper != _RUN_WRAPPER
+
+ result = self._run("echo ran", wrapper=wrapper)
+
+ assert (result.returncode, result.stdout) == (125, b"")
+ assert result.stderr.endswith(_wrapper_line(_STAGING_FAILED, self._NONCE))
+
+ def test_a_missing_setsid_has_its_own_status(self, tmp_path):
+ wrapper = _RUN_WRAPPER.replace(_SYSTEM_PATH, self._path_without_setsid(tmp_path))
+ assert wrapper != _RUN_WRAPPER
+
+ result = self._run("echo ran", wrapper=wrapper)
+
+ assert (result.returncode, result.stdout, result.stderr) == (
+ _NO_SETSID_STATUS,
+ b"",
+ _wrapper_line(_NO_SETSID, self._NONCE),
+ )
+
+ def test_the_command_cannot_read_the_nonce(self):
+ result = self._run('cat /proc/[0-9]*/cmdline /proc/[0-9]*/environ "$0" 2>/dev/null; env', cap=10**7)
+
+ # The wrapper's own argv was read, so its nonce would have been too.
+ assert _STAGING_FAILED.encode() in result.stdout
+ assert self._NONCE.encode() not in result.stdout
+
+ def test_spec_env_named_like_a_shell_variable_does_not_receive_the_nonce(self):
+ # A shell keeps an inherited variable exported when it assigns it, so a wrapper
+ # variable sharing a name with the spec's env would pass its value to the command.
+ env = {**os.environ, **{name: "from-the-spec" for name in "cdefklnopqrstuwxz"}}
+
+ result = subprocess.run(
+ ["/bin/sh", "-c", _RUN_WRAPPER, "airflow-exec", "5", "100000"],
+ input=f"{self._NONCE}\nenv".encode(),
+ capture_output=True,
+ env=env,
+ timeout=25,
+ check=False,
+ )
+
+ assert result.returncode == 0
+ assert b"q=from-the-spec" in result.stdout
+ assert self._NONCE.encode() not in result.stdout
+
+ @pytest.mark.parametrize("operation", _PATH_OPERATIONS)
+ def test_a_named_pipe_is_refused_rather_than_waited_on(self, tmp_path, operation):
+ fifo = tmp_path / "pipe"
+ os.mkfifo(fifo)
+ backend = self._backend_on_local_shell()
+ started = time.monotonic()
+
+ with pytest.raises(SandboxError, match="is a named pipe") as error:
+ operation(backend, str(fifo))
+
+ assert not isinstance(error.value, SandboxTerminalError)
+ assert time.monotonic() - started < 5
+
+ @pytest.mark.parametrize(
+ ("line", "status"),
+ [
+ pytest.param(_STAGING_FAILED, 125, id="staging"),
+ pytest.param(_NO_SETSID, _NO_SETSID_STATUS, id="setsid"),
+ ],
+ )
+ def test_a_command_printing_a_wrapper_line_gets_its_own_result(self, line, status):
+ backend = self._backend_on_local_shell()
+
+ result = backend.run_command(
+ "box", f"echo '{line}' >&2; exit {status}", timeout=5, max_output_bytes=1000
+ )
+
+ assert (result.exit_code, result.stdout, result.stderr) == (status, "", f"{line}\n")
+
+ def test_an_image_without_setsid_fails_the_task(self, tmp_path):
+ backend = self._backend_on_local_shell(self._path_without_setsid(tmp_path))
+
+ with pytest.raises(SandboxTerminalError, match="no setsid"):
+ backend.run_command("box", "echo ran", timeout=5, max_output_bytes=1000)
+
+ def test_each_stream_is_cut_to_one_byte_past_the_cap(self):
+ result = self._run("head -c 5000 /dev/zero; head -c 10 /dev/zero >&2", cap=100)
+
+ assert (len(result.stdout), len(result.stderr)) == (101, 10)
+
+ def test_a_status_of_124_from_the_command_itself_passes_through_fast(self):
+ started = time.monotonic()
+
+ assert self._run("exit 124", seconds=30).returncode == 124
+ assert time.monotonic() - started < 5
diff --git a/uv.lock b/uv.lock
index 61072416bf8..79c86d58b35 100644
--- a/uv.lock
+++ b/uv.lock
@@ -1,5 +1,5 @@
version = 1
-revision = 3
+revision = 5
requires-python = ">=3.11"
resolution-markers = [
"python_full_version >= '3.15' and sys_platform == 'win32'",
@@ -4623,6 +4623,11 @@ openai = [
opensandbox = [
{ name = "opensandbox" },
]
+openshell = [
+ { name = "grpcio" },
+ { name = "openshell" },
+ { name = "protobuf" },
+]
parquet = [
{ name = "pyarrow" },
]
@@ -4656,6 +4661,7 @@ dev = [
{ name = "apache-airflow-providers-standard" },
{ name = "apache-airflow-task-sdk" },
{ name = "boat-sdk" },
+ { name = "grpcio" },
{ name = "islo" },
{ name = "langchain" },
{ name = "llama-index-core" },
@@ -4663,6 +4669,8 @@ dev = [
{ name = "llama-index-llms-openai" },
{ name = "openai" },
{ name = "opensandbox" },
+ { name = "openshell" },
+ { name = "protobuf" },
{ name = "pydantic-ai-skills" },
{ name = "pydantic-ai-slim", extra = ["mcp"] },
{ name = "sqlglot" },
@@ -4686,6 +4694,7 @@ requires-dist = [
{ name = "dataclasses-json", marker = "extra == 'llamaindex'", specifier = ">=0.6.7" },
{ name = "fastavro", marker = "python_full_version >= '3.14' and extra == 'avro'", specifier = ">=1.12.1" },
{ name = "fastavro", marker = "python_full_version < '3.14' and extra == 'avro'", specifier = ">=1.10.0" },
+ { name = "grpcio", marker = "extra == 'openshell'", specifier = ">=1.78.0" },
{ name = "islo", marker = "extra == 'islo'", specifier = ">=0.3.19" },
{ name = "langchain", marker = "extra == 'langchain'", specifier = ">=1.0.0" },
{ name = "llama-index-core", marker = "extra == 'llamaindex'", specifier = ">=0.14.5" },
@@ -4694,6 +4703,8 @@ requires-dist = [
{ name = "modal", marker = "extra == 'modal'", specifier = ">=1.5.2" },
{ name = "openai", marker = "extra == 'openai'", specifier = ">=2.47.0" },
{ name = "opensandbox", marker = "extra == 'opensandbox'", specifier = ">=1.1.0" },
+ { name = "openshell", marker = "extra == 'openshell'", specifier = ">=0.1.2,<0.2" },
+ { name = "protobuf", marker = "extra == 'openshell'", specifier = ">=6.31.1" },
{ name = "pyarrow", marker = "python_full_version >= '3.14' and extra == 'parquet'", specifier = ">=22.0.0" },
{ name = "pyarrow", marker = "python_full_version < '3.14' and extra == 'parquet'", specifier = ">=18.0.0" },
{ name = "pydantic-ai-harness", extras = ["codemode"], marker = "extra == 'code-mode'", specifier = ">=0.24.0" },
@@ -4711,7 +4722,7 @@ requires-dist = [
{ name = "structlog", specifier = ">=24.2.0" },
{ name = "typesafe-sdk", marker = "extra == 'typesafe'", specifier = ">=0.6.0" },
]
-provides-extras = ["anthropic", "bedrock", "google", "openai", "typesafe", "mcp", "modal", "opensandbox", "boat", "code-mode", "shields", "skills", "islo", "avro", "parquet", "sql", "common-sql", "langchain", "llamaindex", "pdf", "docx", "git"]
+provides-extras = ["anthropic", "bedrock", "google", "openai", "typesafe", "mcp", "modal", "opensandbox", "boat", "openshell", "code-mode", "shields", "skills", "islo", "avro", "parquet", "sql", "common-sql", "langchain", "llamaindex", "pdf", "docx", "git"]
[package.metadata.requires-dev]
dev = [
@@ -4726,6 +4737,7 @@ dev = [
{ name = "apache-airflow-providers-standard", editable = "providers/standard" },
{ name = "apache-airflow-task-sdk", editable = "task-sdk" },
{ name = "boat-sdk", specifier = ">=1.9.3" },
+ { name = "grpcio", specifier = ">=1.78.0" },
{ name = "islo", specifier = ">=0.3.19" },
{ name = "langchain", specifier = ">=1.0.0" },
{ name = "llama-index-core", specifier = ">=0.14.5" },
@@ -4733,6 +4745,8 @@ dev = [
{ name = "llama-index-llms-openai", specifier = ">=0.6.8" },
{ name = "openai", specifier = ">=2.47.0" },
{ name = "opensandbox", specifier = ">=1.1.0" },
+ { name = "openshell", specifier = ">=0.1.2,<0.2" },
+ { name = "protobuf", specifier = ">=6.31.1" },
{ name = "pydantic-ai-skills", specifier = ">=1.2.0" },
{ name = "pydantic-ai-slim", extras = ["mcp"], specifier = ">=2.36.0" },
{ name = "sqlglot", specifier = ">=30.0.0" },
@@ -18386,6 +18400,21 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/ef/63/7abb96bf2e3619acbd27de99e60619bfacfb7c55b68c4792a258e6d92871/opensearch_py-3.2.0-py3-none-any.whl", hash = "sha256:721a0d3b13fbed9e82278aed748285cf63a1855354ab7e73e3d4992d1b93418b", size = 387286, upload-time = "2026-04-27T18:17:48.658Z" },
]
+[[package]]
+name = "openshell"
+version = "0.1.3"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+ { name = "cloudpickle" },
+ { name = "googleapis-common-protos" },
+ { name = "grpcio" },
+ { name = "httpx" },
+ { name = "protobuf" },
+]
+wheels = [
+ { url = "https://files.pythonhosted.org/packages/eb/32/abced5a8175714c36c93bdd0937e019989301f2cb3a9e51c42422bfae7b8/openshell-0.1.3-py3-none-any.whl", hash = "sha256:d1eaa6abcd714e2f19d4a8ecbc2079bebeb7b56d81e9f8d3f5a9b5165ed1956c", size = 131882, upload-time = "2026-10-09T15:23:07.155Z" },
+]
+
[[package]]
name = "opentelemetry-api"
version = "1.44.0"