wehub-resource-sync
1087 lines
45 KiB
Python
1087 lines
45 KiB
Python
"""
|
||
Linux Bubblewrap sandbox backend.
|
||
|
||
Builds a hermetic mount-namespace view via the ``bwrap`` launcher and
|
||
layers a hardened seccomp profile on top once the helper has exec'd
|
||
inside the namespace. Selected as the Linux platform default when
|
||
the ``bwrap`` binary is on ``PATH`` (see
|
||
:func:`omnigent.inner.sandbox._default_sandbox_for_platform`); spec
|
||
authors can also pin it explicitly via ``os_env.sandbox.type:
|
||
linux_bwrap`` in YAML.
|
||
|
||
Default view inside the sandbox:
|
||
|
||
- ``/usr``, ``/lib*``, ``/bin``, ``/sbin`` mounted read-only.
|
||
- A small allow-list of ``/etc`` files needed for libc / DNS / TLS
|
||
(``/etc/resolv.conf``, ``/etc/hosts``, ``/etc/nsswitch.conf``,
|
||
``/etc/passwd``, ``/etc/group``, ``/etc/localtime``,
|
||
``/etc/ld.so.cache``, ``/etc/ld.so.conf``,
|
||
``/etc/ld.so.conf.d``, ``/etc/ssl``, ``/etc/ca-certificates``,
|
||
``/etc/pki``) bound read-only via ``--ro-bind-try``.
|
||
- Fresh ``/proc``, ``/dev``, and ``/tmp``.
|
||
- Cwd bind-mounted read-only by default; explicit
|
||
``write_paths: ["."]`` flips it to read-write. Top-level
|
||
dotfiles / dotdirs in cwd are tmpfs-masked unless their name is
|
||
in :data:`_DEFAULT_CWD_ALLOW_HIDDEN` (``.venv`` by default) or
|
||
:attr:`OSEnvSandboxSpec.cwd_allow_hidden`.
|
||
- The per-helper scratch tmpdir created by
|
||
:func:`omnigent.inner.sandbox.create_private_tmpdir` is
|
||
bind-mounted read-write and surfaced via ``$TMPDIR``.
|
||
- ``$HOME`` is **never** mounted. ``/root``, ``/var``, and host
|
||
dotfiles never leak.
|
||
|
||
Hardened seccomp profile applied by :meth:`BwrapSandboxBackend.activate`
|
||
inside the helper, on top of ``PR_SET_NO_NEW_PRIVS``:
|
||
|
||
1. :data:`omnigent.inner._seccomp.BASELINE_DENYLIST_SYSCALLS` — the
|
||
shared k8s/containerd-derived denylist (mount/pivot_root family,
|
||
kernel-module loaders, BPF, kernel keyring, time-setters,
|
||
namespace primitives, plus our own ``ptrace``-family hardening).
|
||
2. Argument-filtered ``clone(flags, ...)``: any ``CLONE_NEW*`` bit
|
||
set in arg 0 returns ``EPERM``. ``fork()`` /
|
||
``pthread_create()`` keep working (they don't set ``CLONE_NEW*``).
|
||
3. ``clone3`` denied with ``ENOSYS`` (not ``EPERM``) — the
|
||
``struct clone_args`` flags live in user memory that seccomp
|
||
can't dereference, so we can't arg-filter ``clone3``. We return
|
||
``ENOSYS`` because glibc's ``clone_internal`` only falls back to
|
||
the legacy ``clone`` syscall on ``ENOSYS``; returning ``EPERM``
|
||
here makes ``pthread_create`` fail outright on Ubuntu 22.04 +
|
||
glibc 2.34+ (which use ``clone3`` for thread creation), which
|
||
would break the in-helper egress relay thread. Same trade-off
|
||
Flatpak / Snap / Firejail accept.
|
||
4. Argument-filtered ``socket(domain, ...)``: an allowlist of
|
||
``AF_UNIX`` (1), ``AF_INET`` (2), ``AF_INET6`` (10). Every other
|
||
socket family is denied with ``EPERM``. This is future-proof —
|
||
new kernel socket families are blocked by default without code
|
||
changes.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import ctypes
|
||
import errno
|
||
import logging
|
||
import os
|
||
import shutil
|
||
import sys
|
||
from collections.abc import Sequence
|
||
from pathlib import Path
|
||
|
||
from ._cwd_scan import scan_cwd_mask_entries
|
||
from ._seccomp import (
|
||
SCMP_CMP_EQ,
|
||
SCMP_CMP_GE,
|
||
SCMP_CMP_MASKED_EQ,
|
||
SeccompArgFilter,
|
||
SeccompRule,
|
||
apply_baseline_denylist,
|
||
apply_seccomp_filter,
|
||
scmp_act_errno,
|
||
)
|
||
from .datamodel import OSEnvSandboxSpec, OSEnvSpec
|
||
from .sandbox import (
|
||
SandboxBackend,
|
||
SandboxPolicy,
|
||
register_backend,
|
||
)
|
||
|
||
_LOGGER = logging.getLogger(__name__)
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Defaults
|
||
# ---------------------------------------------------------------------------
|
||
|
||
_PR_SET_NO_NEW_PRIVS = 38
|
||
|
||
# Top-level cwd dotfiles allowed through by default when the spec
|
||
# doesn't override ``cwd_allow_hidden``. ``.venv`` is whitelisted so the
|
||
# common Python project layout (per the project's CLAUDE.md guidance to
|
||
# use ``.venv`` with ``uv``) keeps working out of the box.
|
||
_DEFAULT_CWD_ALLOW_HIDDEN = (".venv",)
|
||
|
||
# Read-only directory binds that make the standard Linux toolchain
|
||
# usable inside the sandbox (libc, dynamic linker, system Python and
|
||
# friends). Each is emitted as ``--ro-bind-try`` so missing entries
|
||
# (e.g. /sbin on systemd merged-/usr distros) don't fail the spawn.
|
||
_DEFAULT_RO_DIRS = (
|
||
"/usr",
|
||
"/lib",
|
||
"/lib64",
|
||
"/lib32",
|
||
"/bin",
|
||
"/sbin",
|
||
)
|
||
|
||
# Read-only file binds for the minimal /etc files libc, DNS, the
|
||
# dynamic linker, and TLS clients reach for. Bind individual files
|
||
# (not /etc/) so /etc/shadow, /etc/sudoers, /etc/cron*, etc. are
|
||
# never visible inside the sandbox.
|
||
_DEFAULT_ETC_FILES = (
|
||
"/etc/resolv.conf",
|
||
"/etc/hosts",
|
||
"/etc/nsswitch.conf",
|
||
"/etc/passwd",
|
||
"/etc/group",
|
||
"/etc/localtime",
|
||
"/etc/ld.so.cache",
|
||
"/etc/ld.so.conf",
|
||
)
|
||
|
||
# Read-only directory binds for the multi-file /etc subtrees — linker
|
||
# search paths and TLS trust stores. Bound as directories so adding /
|
||
# updating cert bundles on the host stays transparent inside the
|
||
# sandbox.
|
||
_DEFAULT_ETC_DIRS = (
|
||
"/etc/ld.so.conf.d",
|
||
"/etc/ssl",
|
||
"/etc/ca-certificates",
|
||
"/etc/pki",
|
||
)
|
||
|
||
# Linux ``CLONE_NEW*`` flag bits. Any of these set in arg 0 of
|
||
# ``clone()`` would create a fresh namespace — the canonical user-
|
||
# namespace privilege-escalation pattern out of bwrap-style sandboxes.
|
||
# Defined here (instead of relying on a header import) because Python
|
||
# doesn't expose them in ``os`` and we don't want a libc lookup at
|
||
# activation time.
|
||
_CLONE_NEWNS = 0x00020000
|
||
_CLONE_NEWCGROUP = 0x02000000
|
||
_CLONE_NEWUTS = 0x04000000
|
||
_CLONE_NEWIPC = 0x08000000
|
||
_CLONE_NEWUSER = 0x10000000
|
||
_CLONE_NEWPID = 0x20000000
|
||
_CLONE_NEWNET = 0x40000000
|
||
|
||
_CLONE_NEW_FLAG_BITS = (
|
||
_CLONE_NEWNS,
|
||
_CLONE_NEWCGROUP,
|
||
_CLONE_NEWUTS,
|
||
_CLONE_NEWIPC,
|
||
_CLONE_NEWUSER,
|
||
_CLONE_NEWPID,
|
||
_CLONE_NEWNET,
|
||
)
|
||
|
||
# Socket families the helper is allowed to create. Everything else is
|
||
# denied with EPERM. Numeric values are inlined so the activation path
|
||
# never imports ``socket`` (which on import opens netlink sockets on
|
||
# some platforms — pointless work inside a hardened helper).
|
||
#
|
||
# References: ``include/linux/socket.h`` in the Linux source tree.
|
||
_ALLOWED_SOCKET_FAMILIES = (
|
||
1, # AF_UNIX — local IPC
|
||
2, # AF_INET — IPv4
|
||
10, # AF_INET6 — IPv6
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Backend
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class BwrapSandboxBackend(SandboxBackend):
|
||
"""
|
||
Bubblewrap-based sandbox backend.
|
||
|
||
Resolves a :class:`SandboxPolicy` from an :class:`OSEnvSpec`,
|
||
builds the ``bwrap`` argv at spawn time
|
||
(:meth:`wrap_launcher_argv`), and applies the hardened seccomp
|
||
profile inside the helper subprocess once it has exec'd
|
||
(:meth:`activate`).
|
||
|
||
Stateless: a single shared instance is registered with the
|
||
sandbox registry at module import time.
|
||
"""
|
||
|
||
type_name = "linux_bwrap"
|
||
|
||
def resolve(self, spec: OSEnvSpec, cwd: Path) -> SandboxPolicy:
|
||
"""
|
||
Build a :class:`SandboxPolicy` for the bwrap backend.
|
||
|
||
Three resolver behaviors specific to this backend:
|
||
|
||
- ``write_paths`` defaults to **empty** — cwd is read-only
|
||
unless the spec sets ``write_paths: ["."]`` explicitly.
|
||
- ``cwd_allow_hidden`` falls back to
|
||
:data:`_DEFAULT_CWD_ALLOW_HIDDEN` when the spec doesn't
|
||
declare one.
|
||
- The bwrap binary must be on ``PATH`` or the resolver fails
|
||
loud with an install hint — no silent fallback.
|
||
|
||
:param spec: The agent's :class:`OSEnvSpec`. ``spec.sandbox``
|
||
is read for backend tunables; the rest of the spec is
|
||
unused by this backend.
|
||
:param cwd: Effective working directory of the helper, e.g.
|
||
the project root. Used to resolve relative entries in
|
||
``read_paths`` / ``write_paths`` / ``write_files``.
|
||
:returns: A populated :class:`SandboxPolicy` ready to be
|
||
consumed by :meth:`wrap_launcher_argv` and
|
||
:meth:`activate`.
|
||
:raises OSError: If the host is not Linux or the ``bwrap``
|
||
binary cannot be located.
|
||
"""
|
||
sandbox_spec = spec.sandbox or OSEnvSandboxSpec(type=self.type_name)
|
||
|
||
if os.name != "posix" or not sys.platform.startswith("linux"):
|
||
raise OSError(
|
||
"linux_bwrap sandbox is only available on Linux. "
|
||
"Configure os_env.sandbox.type='none' on other OSes."
|
||
)
|
||
if shutil.which("bwrap") is None:
|
||
raise OSError(
|
||
"linux_bwrap sandbox requires the 'bwrap' binary on PATH. "
|
||
"Install bubblewrap (e.g. `apt install bubblewrap` or "
|
||
"`dnf install bubblewrap`), or set os_env.sandbox.type to "
|
||
"'none' to disable sandboxing."
|
||
)
|
||
|
||
read_roots: list[Path] | None = None
|
||
if sandbox_spec.read_paths is not None:
|
||
read_roots = [_resolve_root(cwd, root) for root in sandbox_spec.read_paths]
|
||
|
||
# Bwrap-specific default: cwd is RO unless the spec opts in.
|
||
# Empty default honors the "no surprise writes" contract from
|
||
# the design plan — agents that need an editable project tree
|
||
# opt in via ``write_paths: ["."]``.
|
||
write_paths_config = (
|
||
sandbox_spec.write_paths if sandbox_spec.write_paths is not None else []
|
||
)
|
||
write_roots = [_resolve_root(cwd, root) for root in write_paths_config]
|
||
|
||
write_files: list[Path] = []
|
||
if sandbox_spec.write_files is not None:
|
||
write_files.extend(_resolve_root(cwd, path) for path in sandbox_spec.write_files)
|
||
|
||
cwd_allow_hidden = (
|
||
list(sandbox_spec.cwd_allow_hidden)
|
||
if sandbox_spec.cwd_allow_hidden is not None
|
||
else list(_DEFAULT_CWD_ALLOW_HIDDEN)
|
||
)
|
||
|
||
return SandboxPolicy(
|
||
backend_type=self.type_name,
|
||
active=True,
|
||
read_roots=read_roots,
|
||
write_roots=write_roots,
|
||
write_files=write_files,
|
||
allow_network=sandbox_spec.allow_network,
|
||
cwd_allow_hidden=cwd_allow_hidden,
|
||
cwd_hidden_scan_max_entries=sandbox_spec.cwd_hidden_scan_max_entries,
|
||
cwd_hidden_scan_overflow=sandbox_spec.cwd_hidden_scan_overflow,
|
||
env_passthrough=(
|
||
list(sandbox_spec.env_passthrough)
|
||
if sandbox_spec.env_passthrough is not None
|
||
else None
|
||
),
|
||
credential_proxy=sandbox_spec.credential_proxy,
|
||
)
|
||
|
||
def wrap_launcher_argv(
|
||
self,
|
||
argv: list[str],
|
||
policy: SandboxPolicy,
|
||
cwd: Path,
|
||
chdir: Path | None = None,
|
||
target: str | None = None,
|
||
) -> list[str]:
|
||
"""
|
||
Build the ``bwrap`` argv that wraps *argv* with the hermetic
|
||
sandbox view described in this module's docstring.
|
||
|
||
:param argv: The unwrapped helper command, typically
|
||
``[sys.executable, "-m", "omnigent.inner.os_env",
|
||
"helper", "<encoded>"]``.
|
||
:param policy: The :class:`SandboxPolicy` produced by
|
||
:meth:`resolve`, possibly augmented by
|
||
:func:`omnigent.inner.sandbox.with_additional_write_roots`
|
||
(the parent adds the per-helper scratch tmpdir there
|
||
before calling this).
|
||
:param cwd: Workspace directory exposed to the helper. Bound
|
||
read-only (or read-write if a ``write_root`` matches it)
|
||
at its real absolute path, and walked for the dotfile /
|
||
escaping-symlink mask. Always exposed regardless of
|
||
*chdir* so the agent can reach project files via absolute
|
||
paths even when the helper starts elsewhere.
|
||
:param chdir: Optional separate target for ``--chdir``. When
|
||
``None`` (the default), the helper starts in *cwd*. When
|
||
set — typically to the per-helper scratch tmpdir for
|
||
``OSEnvSpec.start_in_scratch`` — the helper starts there
|
||
instead while *cwd* stays bound for reads.
|
||
:param target: Absolute path to the binary that the launcher
|
||
will exec as its final target after the re-exec (e.g. the
|
||
``claude`` CLI installed under ``node_modules/.bin/``).
|
||
When set and the path lives outside the default mounts,
|
||
:func:`_ensure_executable_visible` is called for it so
|
||
its directory chain is bind-mounted into the namespace —
|
||
the same treatment ``argv[0]`` (the Python interpreter)
|
||
already receives. ``None`` when the target is already
|
||
reachable via the standard mounts.
|
||
:returns: A complete ``bwrap`` argv ready for
|
||
``subprocess.Popen`` — never an empty list.
|
||
|
||
Environment containment is deliberately NOT done here with
|
||
``--clearenv`` / ``--setenv``: values embedded in the bwrap
|
||
argv are world-readable via ``/proc/<pid>/cmdline`` for the
|
||
sandbox's lifetime, which would leak spec-granted secrets to
|
||
every host process. The child instead inherits the spawner's
|
||
already-filtered environment, enforced by the
|
||
``SandboxPolicy.spawn_env_allowlist`` prune in
|
||
:func:`omnigent.inner.sandbox.run_launcher`.
|
||
"""
|
||
cwd_resolved = cwd.resolve(strict=False)
|
||
chdir_target = chdir.resolve(strict=False) if chdir is not None else cwd_resolved
|
||
bwrap_args: list[str] = ["bwrap"]
|
||
|
||
for path in _DEFAULT_RO_DIRS:
|
||
bwrap_args += ["--ro-bind-try", path, path]
|
||
for path in _DEFAULT_ETC_FILES:
|
||
bwrap_args += ["--ro-bind-try", path, path]
|
||
for path in _DEFAULT_ETC_DIRS:
|
||
bwrap_args += ["--ro-bind-try", path, path]
|
||
|
||
# /proc, /dev (filtered by bwrap to a safe minimal device set),
|
||
# and a private /tmp so the agent's writes there don't pollute
|
||
# the host.
|
||
bwrap_args += [
|
||
"--proc",
|
||
"/proc",
|
||
"--dev",
|
||
"/dev",
|
||
"--tmpfs",
|
||
"/tmp",
|
||
]
|
||
|
||
# Make sure argv[0] (the helper interpreter) is reachable inside
|
||
# the sandbox even when it lives outside the default mounts —
|
||
# e.g. a pyenv install under ``$HOME/.pyenv``.
|
||
bwrap_args += _ensure_executable_visible(argv, cwd_resolved)
|
||
|
||
# Make sure the final target binary (e.g. the claude CLI at
|
||
# node_modules/.bin/claude) is also reachable. The launcher
|
||
# re-execs itself into the bwrap namespace and then runs the
|
||
# target via subprocess.run — without this bind the target's
|
||
# directory is invisible inside the namespace and the exec
|
||
# fails with FileNotFoundError.
|
||
if target is not None:
|
||
bwrap_args += _ensure_executable_visible([target], cwd_resolved)
|
||
|
||
# cwd bind: writable iff a write_root resolves to cwd.
|
||
cwd_writable = any(_is_same_path(root, cwd_resolved) for root in policy.write_roots)
|
||
bwrap_args += [
|
||
"--bind" if cwd_writable else "--ro-bind",
|
||
str(cwd_resolved),
|
||
str(cwd_resolved),
|
||
]
|
||
|
||
# Additional write roots — typically the per-helper scratch
|
||
# tmpdir. We skip cwd here because it was bound above.
|
||
for root in policy.write_roots:
|
||
if _is_same_path(root, cwd_resolved):
|
||
continue
|
||
bwrap_args += ["--bind-try", str(root), str(root)]
|
||
|
||
# Per-file write grants. ``--bind-try`` with a file source
|
||
# creates a file-to-file bind mount; the parent dir must
|
||
# already exist inside the sandbox view (typically because the
|
||
# caller also added the parent to read_paths or write_paths).
|
||
for fpath in policy.write_files:
|
||
bwrap_args += ["--bind-try", str(fpath), str(fpath)]
|
||
|
||
# Extra read roots beyond the default toolchain bind set.
|
||
if policy.read_roots is not None:
|
||
for root in policy.read_roots:
|
||
bwrap_args += ["--ro-bind-try", str(root), str(root)]
|
||
|
||
# Mask dotfiles anywhere under cwd OR under any ``read_paths``
|
||
# root that aren't on the allowlist, plus any symlink (at any
|
||
# depth) whose target escapes the sandbox mount set
|
||
# (host-relative dereference defense). The walker prunes at
|
||
# masked dot-directories so ``.git/objects`` etc. don't count
|
||
# toward the cap.
|
||
#
|
||
# Mask emission MUST come AFTER the cwd / write_paths /
|
||
# read_paths binds above: bwrap mount semantics layer later
|
||
# mounts ON TOP of earlier ones, so a ``--tmpfs <root>/.aws``
|
||
# emitted before ``--ro-bind-try <root>`` would be shadowed
|
||
# by the read_paths bind and the helper would silently see
|
||
# the host's ``.aws/`` content. Emitting the mask last keeps
|
||
# the deny-wins guarantee on Linux regardless of how many
|
||
# broader binds the policy stacks underneath.
|
||
bwrap_args.extend(
|
||
_dotfile_and_symlink_mask_args(
|
||
cwd_resolved,
|
||
policy.cwd_allow_hidden if policy.cwd_allow_hidden is not None else [],
|
||
policy,
|
||
argv=argv,
|
||
)
|
||
)
|
||
|
||
# AF_UNIX control-socket masks. A denied socket
|
||
# (e.g. the managed tmux control socket) lives inside a bound
|
||
# write root — the instance ``private_dir`` — so the helper can
|
||
# otherwise reach it and ``connect(2)`` to the unsandboxed tmux
|
||
# server, whose ``run-shell`` would execute outside the sandbox.
|
||
# Overlay ``/dev/null`` onto each socket path so the helper sees
|
||
# a character device, not a socket, and ``connect`` fails. The
|
||
# host (which manages the server) keeps the real socket — the
|
||
# mask only exists inside the helper's mount namespace. Emitted
|
||
# AFTER the write-root binds so it is the last mount and wins
|
||
# (same deny-wins ordering as the dotfile mask above).
|
||
for sock in policy.deny_unix_socket_paths or []:
|
||
bwrap_args += ["--bind-try", "/dev/null", str(sock)]
|
||
|
||
# Set $TMPDIR (and friends) to the first write_root that lives
|
||
# under the system temp dir — the parent's
|
||
# ``set_temp_env`` already pointed the host env at the same
|
||
# path, but bwrap doesn't carry parent env vars across
|
||
# ``--unshare-*`` boundaries reliably, so re-emit them.
|
||
scratch = _scratch_tmpdir(policy.write_roots)
|
||
if scratch is not None:
|
||
for env_var in ("TMPDIR", "TMP", "TEMP", "TEMPDIR"):
|
||
bwrap_args += ["--setenv", env_var, str(scratch)]
|
||
|
||
# Isolate the network namespace when networking is disabled OR
|
||
# when egress rules are active (hard enforcement — the helper
|
||
# has no direct internet; only the relay->proxy path works).
|
||
if not policy.allow_network or policy.egress_relay_port is not None:
|
||
bwrap_args.append("--unshare-net")
|
||
|
||
# Namespacing + lifecycle hardening.
|
||
# - ``--unshare-pid``: the helper sees only its own process tree.
|
||
# - ``--unshare-uts``: hostname/domain isolation.
|
||
# - ``--unshare-ipc``: isolate SysV IPC and POSIX message queues.
|
||
# - ``--die-with-parent``: kernel kills the helper if the parent
|
||
# crashes (no orphan helpers left running on the host).
|
||
# - ``--new-session``: run the helper in its own session so a
|
||
# misbehaving signal handler can't reach the parent's TTY.
|
||
bwrap_args += [
|
||
"--unshare-pid",
|
||
"--unshare-uts",
|
||
"--unshare-ipc",
|
||
"--die-with-parent",
|
||
"--new-session",
|
||
"--chdir",
|
||
str(chdir_target),
|
||
"--",
|
||
]
|
||
bwrap_args.extend(argv)
|
||
return bwrap_args
|
||
|
||
def activate(self, policy: SandboxPolicy) -> None:
|
||
"""
|
||
Apply the in-helper hardening: ``PR_SET_NO_NEW_PRIVS`` plus
|
||
the two seccomp filters described at the top of this module.
|
||
|
||
Bwrap already sets ``PR_SET_NO_NEW_PRIVS`` for the helper
|
||
(it's how non-setuid bwrap installs run unprivileged user
|
||
namespaces), but we re-set it here for defense in depth — the
|
||
prctl is idempotent and cheap, and an explicit call documents
|
||
the seccomp prerequisite.
|
||
|
||
Two seccomp filters are loaded back-to-back (the kernel ANDs
|
||
them):
|
||
|
||
1. The shared k8s-derived baseline via
|
||
:func:`omnigent.inner._seccomp.apply_baseline_denylist`.
|
||
2. The bwrap-specific argument-filtered hardening built by
|
||
:func:`_bwrap_extra_seccomp_rules` (clone-with-CLONE_NEW*
|
||
mask, clone3 deny, socket family allowlist).
|
||
|
||
After seccomp, if the policy carries egress relay config, a
|
||
TCP-to-Unix relay daemon is started in a background thread so
|
||
that HTTP clients routed via ``HTTP_PROXY`` can reach the
|
||
parent's egress proxy through the bind-mounted Unix socket.
|
||
|
||
:param policy: The :class:`SandboxPolicy` for this helper.
|
||
Consulted for :attr:`egress_relay_port` and
|
||
:attr:`egress_socket_path` to conditionally start the
|
||
in-namespace relay.
|
||
"""
|
||
_set_no_new_privs()
|
||
apply_baseline_denylist()
|
||
apply_seccomp_filter(_bwrap_extra_seccomp_rules())
|
||
|
||
if policy.egress_relay_port is not None and policy.egress_socket_path is not None:
|
||
from omnigent.inner.egress.relay import start_relay
|
||
|
||
# Under bwrap the helper runs in its own network namespace
|
||
# so port collisions with host processes are impossible —
|
||
# the bind always succeeds and the relay listens on a
|
||
# private loopback. We still rely on
|
||
# :func:`start_relay`'s fail-loud bind contract so that
|
||
# any unexpected condition (e.g. seccomp denying socket)
|
||
# aborts the helper rather than running with no egress
|
||
# enforcement.
|
||
start_relay(
|
||
policy.egress_relay_port,
|
||
policy.egress_socket_path,
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Internals
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _bwrap_extra_seccomp_rules() -> list[SeccompRule]:
|
||
"""
|
||
Build the bwrap-specific argument-filtered seccomp rules layered
|
||
on top of :func:`omnigent.inner._seccomp.apply_baseline_denylist`.
|
||
|
||
Three rule families, all returning ``EPERM`` against the default
|
||
``SCMP_ACT_ALLOW``:
|
||
|
||
1. ``clone(flags, ...)`` denied when any ``CLONE_NEW*`` bit is set
|
||
(one rule per bit, matched with ``SCMP_CMP_MASKED_EQ``).
|
||
``fork()`` and ``pthread_create()`` keep working — they don't
|
||
set ``CLONE_NEW*``.
|
||
2. ``clone3`` denied with ``ENOSYS``. ``struct clone_args`` lives
|
||
in user memory that seccomp can't dereference, so we can't
|
||
arg-filter it. We return ``ENOSYS`` (not ``EPERM``) because
|
||
glibc's ``clone_internal`` only falls back to legacy ``clone``
|
||
on ``ENOSYS`` — returning ``EPERM`` propagates up and breaks
|
||
``pthread_create`` on glibc 2.34+ (Ubuntu 22.04+), which uses
|
||
``clone3`` for thread creation. The fallback to legacy
|
||
``clone`` is still caught by rule 1 if the spawn carries a
|
||
``CLONE_NEW*`` bit, so namespace-creation is still blocked.
|
||
3. ``socket(domain, ...)`` denied for all families NOT in
|
||
:data:`_ALLOWED_SOCKET_FAMILIES`. Uses range-based deny rules:
|
||
deny ``AF_UNSPEC`` (0), deny families 3–9 (gap between
|
||
``AF_INET`` and ``AF_INET6``), and deny ``>= 11`` via
|
||
``SCMP_CMP_GE`` — catching all current and future families
|
||
above ``AF_INET6`` without code changes.
|
||
|
||
These rules are bwrap-specific because (a) bwrap already set up a
|
||
fresh PID/UTS/IPC namespace so blocking nested namespace creation
|
||
is a true defense-in-depth layer rather than a workload break,
|
||
and (b) the bwrap helper has no legitimate use for raw netlink
|
||
or packet sockets. The shared baseline does not include them so
|
||
other backends can adopt the baseline without inheriting this
|
||
extra restriction.
|
||
|
||
:returns: The flattened list of :class:`SeccompRule` ready for
|
||
:func:`apply_seccomp_filter`.
|
||
"""
|
||
deny = scmp_act_errno(errno.EPERM)
|
||
# ``clone3`` is denied with ``ENOSYS`` (not ``EPERM``) so glibc's
|
||
# ``clone_internal`` falls back to the legacy ``clone`` syscall,
|
||
# which IS arg-filtered above. Returning ``EPERM`` here would
|
||
# break ``pthread_create`` outright on glibc 2.34+ (Ubuntu
|
||
# 22.04+ uses ``clone3`` for thread creation, and the glibc
|
||
# fallback path triggers only on ``ENOSYS``). The in-helper
|
||
# egress relay thread depends on ``pthread_create`` succeeding.
|
||
deny_clone3 = scmp_act_errno(errno.ENOSYS)
|
||
rules: list[SeccompRule] = []
|
||
|
||
for bit in _CLONE_NEW_FLAG_BITS:
|
||
rules.append(
|
||
SeccompRule(
|
||
syscall="clone",
|
||
action=deny,
|
||
arg_filters=(
|
||
SeccompArgFilter(
|
||
arg=0,
|
||
op=SCMP_CMP_MASKED_EQ,
|
||
datum_a=bit,
|
||
datum_b=bit,
|
||
),
|
||
),
|
||
)
|
||
)
|
||
|
||
rules.append(SeccompRule(syscall="clone3", action=deny_clone3))
|
||
|
||
# Socket allowlist: deny everything NOT in _ALLOWED_SOCKET_FAMILIES.
|
||
# We can't use a blanket deny + ALLOW holes (libseccomp rejects
|
||
# ALLOW rules when default_action is ALLOW). Instead we deny the
|
||
# complement using ranges:
|
||
# - AF_UNSPEC (0): denied individually
|
||
# - Families 3..9: denied individually (gap between AF_INET and AF_INET6)
|
||
# - Families >= 11: denied via SCMP_CMP_GE (catches all future families)
|
||
_AF_UNSPEC = 0
|
||
_FIRST_GAP_START = 3 # first family between AF_INET(2) and AF_INET6(10)
|
||
_FIRST_GAP_END = 9
|
||
_ABOVE_INET6 = 11 # first family after AF_INET6(10)
|
||
|
||
rules.append(
|
||
SeccompRule(
|
||
syscall="socket",
|
||
action=deny,
|
||
arg_filters=(SeccompArgFilter(arg=0, op=SCMP_CMP_EQ, datum_a=_AF_UNSPEC),),
|
||
)
|
||
)
|
||
for family in range(_FIRST_GAP_START, _FIRST_GAP_END + 1):
|
||
rules.append(
|
||
SeccompRule(
|
||
syscall="socket",
|
||
action=deny,
|
||
arg_filters=(SeccompArgFilter(arg=0, op=SCMP_CMP_EQ, datum_a=family),),
|
||
)
|
||
)
|
||
rules.append(
|
||
SeccompRule(
|
||
syscall="socket",
|
||
action=deny,
|
||
arg_filters=(SeccompArgFilter(arg=0, op=SCMP_CMP_GE, datum_a=_ABOVE_INET6),),
|
||
)
|
||
)
|
||
|
||
return rules
|
||
|
||
|
||
def _set_no_new_privs() -> None:
|
||
"""
|
||
Set ``PR_SET_NO_NEW_PRIVS`` on the current process via libc
|
||
``prctl``.
|
||
|
||
Idempotent — bwrap already does this for the helper, but applying
|
||
it explicitly here documents the seccomp prerequisite (the kernel
|
||
refuses ``seccomp_load`` without ``CAP_SYS_ADMIN`` unless
|
||
``NO_NEW_PRIVS`` is set).
|
||
|
||
:raises OSError: If ``prctl`` returns non-zero.
|
||
"""
|
||
libc = ctypes.CDLL(None, use_errno=True)
|
||
rc = libc.prctl(_PR_SET_NO_NEW_PRIVS, 1, 0, 0, 0)
|
||
if rc != 0:
|
||
err = ctypes.get_errno()
|
||
raise OSError(err, f"prctl(PR_SET_NO_NEW_PRIVS) failed: {os.strerror(err)}")
|
||
|
||
|
||
def _resolve_root(cwd: Path, root: str) -> Path:
|
||
"""
|
||
Resolve a spec-supplied path string against *cwd*, expanding
|
||
only ``~`` (NOT ``$VAR``) and normalising relative entries.
|
||
|
||
H4 (security): ``os.path.expandvars`` is intentionally NOT
|
||
applied to the spec string. Expanding environment variables
|
||
against the parent's env at resolve-time turned the env into a
|
||
sandbox-widening lever — an attacker who can influence
|
||
``$HOME`` / ``$XDG_DATA_HOME`` / ``$LOG_DIR`` (via a parent
|
||
shell, MCP server, supervisor agent, or unaudited spec
|
||
templating) could rewrite ``read_paths: ['$LOG_DIR/audit']``
|
||
into ``read_paths: ['/']`` and silently widen the sandbox to
|
||
the whole filesystem. Spec authors should write literal paths
|
||
or ``~`` and let the operator audit them; ``$`` in a spec is
|
||
now warned about so over-broad expansions stand out in logs.
|
||
See :data:`omnigent.inner.seatbelt_sandbox._resolve_root`
|
||
for the parallel hardening on the macOS backend.
|
||
|
||
:param cwd: The agent's effective working directory; relative
|
||
paths in the spec are resolved against it.
|
||
:param root: The raw path string from the YAML spec, e.g.
|
||
``"."``, ``"src"``, ``"~/projects"``, or ``"/etc"``.
|
||
:returns: An absolute, normalized :class:`Path` (without strict
|
||
existence check — the caller may bind a path that doesn't
|
||
exist yet).
|
||
"""
|
||
if "$" in root:
|
||
_LOGGER.warning(
|
||
"linux_bwrap: spec-supplied path %r contains '$' which is "
|
||
"no longer expanded against the parent environment (security "
|
||
"hardening — env-var expansion was a sandbox-widening lever "
|
||
"when an attacker could shape the parent's env). Use literal "
|
||
"paths or ~ instead.",
|
||
root,
|
||
)
|
||
expanded = os.path.expanduser(root)
|
||
path = Path(expanded)
|
||
if not path.is_absolute():
|
||
path = cwd / path
|
||
return path.resolve(strict=False)
|
||
|
||
|
||
def _is_same_path(a: Path, b: Path) -> bool:
|
||
"""
|
||
Return whether two paths reference the same filesystem location
|
||
after symlink-free resolution.
|
||
|
||
:param a: First path to compare.
|
||
:param b: Second path to compare.
|
||
:returns: ``True`` when both paths resolve to the same string.
|
||
"""
|
||
return a.resolve(strict=False) == b.resolve(strict=False)
|
||
|
||
|
||
def _ensure_executable_visible(argv: list[str], cwd: Path) -> list[str]:
|
||
"""
|
||
Return any extra ``--ro-bind-try`` args needed so ``argv[0]`` (the
|
||
helper interpreter) is reachable inside the sandbox.
|
||
|
||
Bwrap exec's ``argv[0]`` literally — it does NOT follow symlinks on
|
||
the host before exec — so every name in the path the kernel will
|
||
traverse must be visible inside the sandbox. This walks the symlink
|
||
chain at ``argv[0]`` hop-by-hop. At each hop, the literal path's
|
||
parent and grandparent are bound at their literal names, with
|
||
sources pointing at the fully-resolved real directories. That
|
||
handles intermediate directory-symlinks: e.g. uv installs a
|
||
version-floating symlink ``cpython-3.12 -> cpython-3.12.13``; the
|
||
kernel traverses the literal ``cpython-3.12`` name, so we mount the
|
||
real ``cpython-3.12.13`` contents there.
|
||
|
||
Common cases:
|
||
|
||
- ``/usr/bin/python3`` (literal == resolved): covered by
|
||
``--ro-bind-try /usr``; nothing extra emitted.
|
||
- ``./.venv/bin/python`` inside cwd, symlinking into ``/usr``:
|
||
covered by the cwd bind (literal) plus ``/usr`` (resolved);
|
||
nothing extra emitted.
|
||
- ``/home/me/omnigent/.venv/bin/python`` while the helper
|
||
runs in ``/tmp/scratch`` (uv-managed Python with intermediate
|
||
``cpython-3.12`` symlink): binds the venv bin/lib roots at the
|
||
literal paths and the uv-python install dir at its literal
|
||
name, with the real ``cpython-3.12.13`` mounted there.
|
||
- Pyenv-style ``$HOME/.pyenv/versions/3.12/bin/python``: same
|
||
walk produces the binds needed whether or not the
|
||
``versions/3.12`` component is itself a symlink.
|
||
|
||
:param argv: Helper argv. The function inspects ``argv[0]``.
|
||
:param cwd: Effective working directory; used to skip binds for
|
||
paths already covered by the cwd mount.
|
||
:returns: Extra bwrap args (possibly empty) to insert before the
|
||
cwd bind. Never includes a destination already covered by
|
||
:data:`_DEFAULT_RO_DIRS` or by *cwd*.
|
||
"""
|
||
if not argv:
|
||
return []
|
||
|
||
covered_prefixes = [Path(p) for p in _DEFAULT_RO_DIRS]
|
||
covered_prefixes.append(cwd)
|
||
|
||
extra: list[str] = []
|
||
seen_dest: set[Path] = set()
|
||
|
||
def _emit(src: Path, dst: Path) -> None:
|
||
if dst in seen_dest:
|
||
return
|
||
# The destination is the literal path bwrap/the kernel will
|
||
# traverse. Skip when that literal lives under a default mount.
|
||
if any(_is_within(dst, root, resolve=False) for root in covered_prefixes):
|
||
return
|
||
seen_dest.add(dst)
|
||
extra.extend(["--ro-bind-try", str(src), str(dst)])
|
||
|
||
def _emit_parent_pair(literal: Path) -> None:
|
||
"""Bind ``literal``'s parent and grandparent at their literal
|
||
paths, sourcing from each path's realpath so intermediate
|
||
directory-symlinks resolve correctly inside the sandbox."""
|
||
parent_literal = literal.parent
|
||
parent_real = Path(os.path.realpath(str(parent_literal)))
|
||
_emit(parent_real, parent_literal)
|
||
gp_literal = parent_literal.parent
|
||
if gp_literal != parent_literal:
|
||
gp_real = Path(os.path.realpath(str(gp_literal)))
|
||
_emit(gp_real, gp_literal)
|
||
|
||
# Walk the symlink chain hop-by-hop until either a regular file is
|
||
# reached or the visited-set catches a cycle. The 40-hop cap matches
|
||
# Linux's MAXSYMLINKS / ELOOP threshold.
|
||
visited: set[Path] = set()
|
||
current = Path(os.path.abspath(str(Path(argv[0]))))
|
||
for _ in range(40):
|
||
if current in visited:
|
||
break
|
||
visited.add(current)
|
||
_emit_parent_pair(current)
|
||
try:
|
||
if not current.is_symlink():
|
||
break
|
||
link = os.readlink(str(current))
|
||
except OSError:
|
||
break
|
||
if link.startswith("/"):
|
||
current = Path(link)
|
||
else:
|
||
current = Path(os.path.normpath(str(current.parent / link)))
|
||
|
||
return extra
|
||
|
||
|
||
def _is_within(path: Path, root: Path, *, resolve: bool = True) -> bool:
|
||
"""
|
||
Return whether *path* is *root* or a descendant of *root*.
|
||
|
||
By default both paths are passed through ``resolve(strict=False)``
|
||
before comparison, so symlinks under cwd that point into safe
|
||
roots (e.g. ``./.venv/bin/python`` → ``/usr/bin/python3.12``)
|
||
count as "within" the safe root for safety checks.
|
||
|
||
Set *resolve* to ``False`` for path-string-prefix checks where
|
||
symlink dereferencing would mask the literal path's location —
|
||
e.g. when verifying that ``argv[0]`` (the literal exec path that
|
||
bwrap will pass to ``execvp``) is reachable inside the sandbox.
|
||
|
||
:param path: Candidate path, e.g. ``/usr/bin/python3``.
|
||
:param root: Prefix path, e.g. ``/usr``.
|
||
:param resolve: Whether to resolve symlinks before comparison.
|
||
:returns: ``True`` when *path* equals or lives under *root* under
|
||
the chosen resolution mode.
|
||
"""
|
||
if resolve:
|
||
compare_path = path.resolve(strict=False)
|
||
compare_root = root.resolve(strict=False)
|
||
else:
|
||
# Use absolute() to normalise '..' / cwd-relative components
|
||
# without following symlinks. Drop trailing slashes by
|
||
# normalising both sides through pathlib.
|
||
compare_path = Path(os.path.abspath(str(path)))
|
||
compare_root = Path(os.path.abspath(str(root)))
|
||
try:
|
||
compare_path.relative_to(compare_root)
|
||
return True
|
||
except ValueError:
|
||
return False
|
||
|
||
|
||
def _bwrap_safe_roots(
|
||
cwd: Path, policy: SandboxPolicy, argv: Sequence[str] | None = None
|
||
) -> list[Path]:
|
||
"""
|
||
Build the "already exposed" path set used by the cwd walker to
|
||
decide which symlinks are escaping.
|
||
|
||
Includes the bwrap-specific default mounts (``/usr``, ``/lib*``,
|
||
``/bin``, ``/sbin``, ``/proc``, ``/dev``, ``/tmp``), the cwd, the
|
||
policy's read/write roots, AND — when *argv* is supplied — the
|
||
literal-and-resolved parent (and grandparent) of ``argv[0]`` so
|
||
the dotfile walker doesn't mask the helper interpreter symlink.
|
||
|
||
Without the interpreter widening, a ``read_paths`` grant covering
|
||
a workspace whose venv resolves into a non-default install root
|
||
(e.g. ``/opt/hostedtoolcache/Python/.../bin/python3.12`` on
|
||
GitHub-hosted Linux runners, where ``/opt`` is NOT in
|
||
:data:`_DEFAULT_RO_DIRS`) would have the S5 read_paths scan flag
|
||
``.venv/bin/python`` as an escaping symlink and emit a
|
||
``--bind /dev/null`` mask over it — making the helper unspawnable
|
||
with the bwrap error
|
||
``Can't create file at .venv/bin/python: No such file or directory``.
|
||
Including the resolved interpreter parents keeps the scanner
|
||
consistent with what :func:`_ensure_executable_visible` will
|
||
already bind in for the helper.
|
||
|
||
Backends that expose a different default mount set (e.g.
|
||
``darwin_seatbelt``) build their own list with the same shape.
|
||
|
||
:param cwd: Resolved cwd path.
|
||
:param policy: Sandbox policy whose read/write roots also count
|
||
as safe.
|
||
:param argv: Optional helper argv; when present, ``argv[0]``'s
|
||
literal-and-resolved parent / grandparent are added so the
|
||
walker doesn't mask the interpreter symlink.
|
||
:returns: List of safe root paths.
|
||
"""
|
||
safe_roots: list[Path] = [cwd]
|
||
safe_roots.extend(Path(p) for p in _DEFAULT_RO_DIRS)
|
||
safe_roots.append(Path("/proc"))
|
||
safe_roots.append(Path("/dev"))
|
||
safe_roots.append(Path("/tmp"))
|
||
if policy.read_roots is not None:
|
||
safe_roots.extend(policy.read_roots)
|
||
safe_roots.extend(policy.write_roots)
|
||
if argv:
|
||
safe_roots.extend(_interpreter_safe_roots(argv))
|
||
return safe_roots
|
||
|
||
|
||
def _interpreter_safe_roots(argv: Sequence[str]) -> list[Path]:
|
||
"""
|
||
Compute the safe-root widening that mirrors
|
||
:func:`_ensure_executable_visible`.
|
||
|
||
Returns the literal-and-resolved parent and grandparent of
|
||
``argv[0]`` so the dotfile walker treats the helper interpreter
|
||
(and the typical venv ``bin/`` + ``lib/`` siblings) as
|
||
already-exposed, regardless of whether the resolved Python sits
|
||
under a non-default mount root like ``/opt/hostedtoolcache`` on
|
||
GitHub Actions Linux runners.
|
||
|
||
:param argv: Helper argv. Only ``argv[0]`` is inspected.
|
||
:returns: List of safe root paths derived from the interpreter
|
||
location. Empty when *argv* is empty.
|
||
"""
|
||
if not argv:
|
||
return []
|
||
exe_literal = Path(argv[0])
|
||
exe_resolved = exe_literal.resolve(strict=False)
|
||
candidates: set[Path] = set()
|
||
for exe in (exe_literal, exe_resolved):
|
||
parent = exe.parent
|
||
if str(parent) not in {"", "/"}:
|
||
candidates.add(parent)
|
||
grandparent = parent.parent
|
||
if str(grandparent) not in {"", "/"}:
|
||
candidates.add(grandparent)
|
||
return sorted(candidates, key=str)
|
||
|
||
|
||
def _dotfile_and_symlink_mask_args(
|
||
cwd: Path,
|
||
allow_hidden: list[str],
|
||
policy: SandboxPolicy,
|
||
argv: Sequence[str] | None = None,
|
||
) -> list[str]:
|
||
"""
|
||
Build the bwrap mount args needed to mask dotfile / escaping
|
||
entries the helper must not see, across BOTH cwd and every
|
||
spec-supplied ``read_paths`` root.
|
||
|
||
Thin emitter over :func:`omnigent.inner._cwd_scan.scan_cwd_mask_entries`:
|
||
the shared walker decides *which* paths to mask (dotfiles by
|
||
basename + escaping symlinks); this function maps each
|
||
:class:`MaskedEntry` to the bwrap mount triple that hides it.
|
||
|
||
Mask shape depends on entry kind because bwrap mount targets are
|
||
type-checked at the kernel level:
|
||
|
||
- **Directories** are masked with ``--tmpfs <path>`` — the agent
|
||
sees an empty in-memory directory.
|
||
- **Files / symlinks / sockets** are masked with
|
||
``--bind-try /dev/null <path>`` — the agent sees what looks like
|
||
an empty file (reads return EOF, writes are discarded).
|
||
``/dev/null`` is always present inside the sandbox via the
|
||
``--dev /dev`` mount. ``--bind-try`` (rather than ``--bind``)
|
||
tolerates TOCTOU races: transient files (e.g. ``.coverage.*``)
|
||
may vanish between the scan and the ``bwrap`` exec; the
|
||
``-try`` variant silently skips the mount instead of aborting.
|
||
|
||
S5 (security): the walk covers each ``read_paths`` root in
|
||
addition to ``cwd`` so a broad grant like ``read_paths: ["~/"]``
|
||
does NOT leave ``~/.aws``, ``~/.ssh``, ``~/.config/gcloud`` etc.
|
||
readable just because the dotfile masker used to be cwd-only.
|
||
Roots that live under ``cwd`` are skipped — the cwd pass already
|
||
covered them. Per-path dedup runs across both passes.
|
||
|
||
See :mod:`omnigent.inner._cwd_scan` for the masking-decision
|
||
semantics, walk-cap behaviour, and the
|
||
``cwd_hidden_scan_max_entries`` / ``cwd_hidden_scan_overflow``
|
||
tuning knobs. Each root is walked under the same cap; a
|
||
``read_paths: ["~/"]`` grant will almost always trip it and the
|
||
resulting :class:`OSError` points at the same tunables (the
|
||
right answer in almost every case is to narrow the grant).
|
||
|
||
:param cwd: Effective working directory of the helper, already
|
||
resolved (no symlinks).
|
||
:param allow_hidden: Dotfile / dotdir basenames that pass through
|
||
unmasked at any depth, e.g. ``[".venv"]``.
|
||
:param policy: Sandbox policy. Provides the read/write roots that
|
||
contribute to the symlink-escape safe set and the
|
||
``cwd_hidden_scan_max_entries`` / ``cwd_hidden_scan_overflow``
|
||
knobs.
|
||
:returns: Flat list of bwrap argv tokens (alternating
|
||
``--tmpfs <path>`` and ``--bind-try /dev/null <path>`` triples).
|
||
Empty when nothing worth masking was found.
|
||
:raises OSError: When the entry cap is reached and the policy's
|
||
overflow mode is ``"error"``.
|
||
"""
|
||
safe_roots = _bwrap_safe_roots(cwd, policy, argv=argv)
|
||
seen_mask_paths: set[str] = set()
|
||
entries = scan_cwd_mask_entries(
|
||
cwd,
|
||
allow_hidden=allow_hidden,
|
||
safe_roots=safe_roots,
|
||
max_entries=policy.cwd_hidden_scan_max_entries,
|
||
overflow=policy.cwd_hidden_scan_overflow,
|
||
logger_name=__name__,
|
||
)
|
||
for entry in entries:
|
||
seen_mask_paths.add(str(entry.path))
|
||
# Extend the mask to every read_paths root that the cwd scan
|
||
# didn't already cover (skip roots that live under cwd — those
|
||
# were walked in the cwd pass).
|
||
for root in policy.read_roots or []:
|
||
if _is_within(root, cwd):
|
||
continue
|
||
try:
|
||
extra = scan_cwd_mask_entries(
|
||
root,
|
||
allow_hidden=allow_hidden,
|
||
safe_roots=safe_roots,
|
||
max_entries=policy.cwd_hidden_scan_max_entries,
|
||
overflow=policy.cwd_hidden_scan_overflow,
|
||
logger_name=__name__,
|
||
scope_label="read_paths",
|
||
)
|
||
except OSError as err:
|
||
# Re-raise with read_paths-specific advice, forwarding the
|
||
# walker's own message verbatim — it already names the
|
||
# overflowed root (passed as the walk's scope) and the
|
||
# unfinished directories, so we don't want to drop that
|
||
# detail by rewriting the text from scratch.
|
||
raise OSError(
|
||
f"dotfile mask scan overflowed while walking read_paths root "
|
||
f"{root}. Narrow the grant or tune the scan limits. {err}"
|
||
) from err
|
||
for entry in extra:
|
||
key = str(entry.path)
|
||
if key in seen_mask_paths:
|
||
continue
|
||
seen_mask_paths.add(key)
|
||
entries.append(entry)
|
||
args: list[str] = []
|
||
for entry in entries:
|
||
# Re-stat just before emitting: a mask overlays onto an EXISTING
|
||
# path, so a transient dotfile (e.g. coverage.py's `.coverage.*`)
|
||
# that vanished since the scan would force bwrap to create the
|
||
# mountpoint inside the ro-bound cwd and abort. Skipping a gone
|
||
# target is safe — persistent host dotfiles still exist and are
|
||
# still masked.
|
||
if not _path_exists_lstat(entry.path):
|
||
continue
|
||
if entry.kind == "dir":
|
||
args.extend(["--tmpfs", str(entry.path)])
|
||
else:
|
||
args.extend(["--bind-try", "/dev/null", str(entry.path)])
|
||
return args
|
||
|
||
|
||
def _path_exists_lstat(path: Path) -> bool:
|
||
"""
|
||
Return whether *path* exists without following a final symlink.
|
||
|
||
``lstat`` (not ``stat``) so a dangling / escaping symlink still
|
||
counts as present and gets masked.
|
||
|
||
:param path: Candidate mask target.
|
||
:returns: ``True`` when the path still exists (including broken
|
||
symlinks), ``False`` when it has been removed.
|
||
"""
|
||
try:
|
||
os.lstat(path)
|
||
except OSError:
|
||
return False
|
||
return True
|
||
|
||
|
||
def _scratch_tmpdir(write_roots: list[Path]) -> Path | None:
|
||
"""
|
||
Return the first ``write_root`` that lives under the system tempdir.
|
||
|
||
:func:`omnigent.inner.sandbox.create_private_tmpdir` returns a
|
||
path under ``tempfile.gettempdir()`` (typically ``/tmp/...``).
|
||
The helper-spawn site adds it to ``write_roots`` via
|
||
:func:`omnigent.inner.sandbox.with_additional_write_roots`, so
|
||
we identify it here by checking which ``write_root`` is under
|
||
the system tempdir prefix. Used to set ``$TMPDIR`` etc. inside
|
||
the sandbox.
|
||
|
||
:param write_roots: The policy's resolved write roots, including
|
||
any paths added by the spawn-site augmentation.
|
||
:returns: The scratch tmpdir path, or ``None`` when the helper
|
||
wasn't given one (e.g. tests that build a policy by hand).
|
||
"""
|
||
import tempfile
|
||
|
||
sys_tmp = Path(tempfile.gettempdir()).resolve(strict=False)
|
||
for root in write_roots:
|
||
try:
|
||
if _is_within(root, sys_tmp):
|
||
return root
|
||
except OSError:
|
||
continue
|
||
return None
|
||
|
||
|
||
register_backend(BwrapSandboxBackend())
|