File size: 7,723 Bytes
9e551a1 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 | # Tensor Roll — Recursive CUDA-Q Model Quantizer
# Copyright (C) 2026 SnapKitty Collective
# SPDX-License-Identifier: AGPL-3.0-or-later
"""OS-level sandbox supervisor (spec section 12).
Architecture (enforced below):
USER -> TERMINAL -> TENSOR ROLL -> MODEL -> TOOL REQUEST
-> SANDBOX SUPERVISOR -> POLICY -> DENY | EXECUTE -> ISOLATED PROCESS
Enforced: process isolation (separate session/process group), filesystem
confinement (all path arguments must resolve inside the jail), resource
limits (RLIMIT_CPU/AS/FSIZE/NPROC/NOFILE via preexec_fn), environment
filtering (whitelist), child-process restrictions (own process group,
killed on timeout), signal handling, and a JSON-lines audit log.
Network namespaces are attempted best-effort (unshare(CLONE_NEWNET));
when the kernel refuses, the supervisor records net_isolated=false and
falls back to command-level policy (network fetch tools are denied).
What is enforced vs. best-effort is reported honestly, never assumed.
"""
from __future__ import annotations
import os
import shlex
import shutil
import signal
import subprocess
import tempfile
import time
from dataclasses import dataclass, field
from .util import AuditLog, ensure_dir, utc_now_iso
try:
import resource as _resource
except ImportError: # pragma: no cover
_resource = None
DENIED_BINS = {
# destructive
"rm", "dd", "mkfs", "mkfs.ext4", "shutdown", "reboot", "halt",
"poweroff", "init", "kill", "killall", "pkill",
# network fetch / exfiltration
"curl", "wget", "aria2c", "scp", "sftp", "ftp", "telnet", "nc",
"ncat", "socat", "ssh",
# privilege / package management
"sudo", "su", "doas", "apt", "apt-get", "dpkg", "pip", "pip3",
"snap", "flatpak",
# shells that escape argument scanning
"sh", "bash", "zsh", "fish", "dash",
}
ENV_WHITELIST = {"PATH", "HOME", "LANG", "LC_ALL", "LC_CTYPE", "TMPDIR",
"PYTHONPATH", "PYTHONHOME", "TERM", "TZ"}
@dataclass
class Policy:
max_cpu_s: int = 30
max_mem_mb: int = 1024
max_fsize_mb: int = 256
max_nproc: int = 32
max_nofile: int = 128
timeout_s: int = 60
@dataclass
class Verdict:
allowed: bool
reason: str
resolved_bin: str = ""
class SandboxSupervisor:
def __init__(self, workdir: str, policy: Policy | None = None):
self.workdir = ensure_dir(workdir)
self.policy = policy or Policy()
self.jail = ensure_dir(os.path.join(self.workdir, "jail"))
self.audit = AuditLog(os.path.join(self.workdir, "audit.log"))
self.netns_available = self._probe_netns()
@staticmethod
def _probe_netns() -> bool:
"""Honest capability probe: can we actually unshare a net namespace?"""
try:
r = subprocess.run(["unshare", "--net", "true"],
capture_output=True, timeout=10)
return r.returncode == 0
except Exception:
return False
# -- policy ---------------------------------------------------------
def check(self, argv: list[str]) -> Verdict:
if not argv:
return Verdict(False, "empty command")
name = os.path.basename(argv[0])
if name in DENIED_BINS:
return Verdict(False, f"denied binary: {name}")
resolved = shutil.which(argv[0], path="/usr/bin:/bin")
if resolved is None:
# allow binaries shipped inside the jail itself
cand = os.path.join(self.jail, os.path.basename(argv[0]))
if os.path.isfile(cand) and os.access(cand, os.X_OK):
resolved = cand
else:
return Verdict(False, f"binary not found in restricted PATH: {argv[0]}")
for arg in argv[1:]:
if ".." in arg.split(os.sep):
return Verdict(False, f"path traversal rejected: {arg}")
if os.path.isabs(arg):
real = os.path.realpath(arg)
if not (real == self.jail or real.startswith(self.jail + os.sep)):
return Verdict(False,
f"absolute path outside jail rejected: {arg}")
return Verdict(True, "policy pass", resolved)
# -- execution ------------------------------------------------------
def _preexec(self):
# NOTE: start_new_session=True already puts the child in its own
# session + process group (the child-process restriction anchor).
if _resource is not None:
p = self.policy
_resource.setrlimit(_resource.RLIMIT_CPU, (p.max_cpu_s, p.max_cpu_s))
mem = p.max_mem_mb * 1024 * 1024
_resource.setrlimit(_resource.RLIMIT_AS, (mem, mem))
_resource.setrlimit(_resource.RLIMIT_FSIZE,
(p.max_fsize_mb * 1024 * 1024,) * 2)
_resource.setrlimit(_resource.RLIMIT_NPROC, (p.max_nproc, p.max_nproc))
_resource.setrlimit(_resource.RLIMIT_NOFILE, (p.max_nofile, p.max_nofile))
# best-effort network namespace; failure is recorded, not hidden
try:
os.unshare(os.CLONE_NEWNET)
except OSError:
pass
def run(self, cmd: str, *, timeout: int | None = None) -> dict:
argv = shlex.split(cmd)
verdict = self.check(argv)
rec = {"ts": utc_now_iso(), "cmd": cmd, "allowed": verdict.allowed,
"reason": verdict.reason}
if not verdict.allowed:
self.audit.record(event="sandbox.deny", **rec)
rec["returncode"] = None
return rec
env = {k: v for k, v in os.environ.items() if k in ENV_WHITELIST}
env["PATH"] = "/usr/bin:/bin"
env["TMPDIR"] = self.jail
t0 = time.perf_counter()
proc = subprocess.Popen(
[verdict.resolved_bin] + argv[1:],
cwd=self.jail, env=env, preexec_fn=self._preexec,
stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
start_new_session=True) # own process group: child restriction anchor
try:
out, err = proc.communicate(timeout=timeout or self.policy.timeout_s)
rc = proc.returncode
except subprocess.TimeoutExpired:
rc, out, err = "timeout", "", ""
try:
os.killpg(proc.pid, signal.SIGKILL)
except Exception:
pass
try:
out, err = proc.communicate(timeout=5)
except Exception:
pass
# reap any stray children via the process group
try:
os.killpg(proc.pid, signal.SIGKILL)
except (ProcessLookupError, PermissionError):
pass
rec.update({"returncode": rc,
"stdout": (out or "")[-4000:],
"stderr": (err or "")[-4000:],
"elapsed_s": time.perf_counter() - t0,
"netns_isolated": self.netns_available})
self.audit.record(event="sandbox.execute", **rec)
return rec
def demo(self) -> list[dict]:
"""Demonstrably ALLOW a permitted action and DENY forbidden ones."""
results = []
results.append(("ALLOW echo", self.run("echo hello-tensor-roll")))
with tempfile.NamedTemporaryFile("w", suffix=".txt",
dir=self.jail, delete=False) as f:
f.write("jail-write-ok")
results.append(("ALLOW jail write+read",
self.run(f"cat {f.name}")))
results.append(("DENY shadow", self.run("cat /etc/shadow")))
results.append(("DENY curl", self.run("curl http://example.com")))
results.append(("DENY rm", self.run("rm -rf /")))
return results
|