Harden sandbox transport and capability tokens

Author: Parad0x-Labs

core/agent_runtime/builder/controller.py

@@ -356,7 +356,7 @@ def workspace_build_verification(
         return {"status": "skipped", "ok": False, "response_text": "Verification skipped for non-Python scaffold."}
     execution = execute_runtime_tool_fn(
         "sandbox.run_command",
-        {"command": f"python3 -m compileall -q {root_dir}/src"},
+        {"command": f"python3 -m compileall -q {root_dir}/src", "_trusted_local_only": True},
         source_context=source_context,
     )
     if execution is None:

core/agent_runtime/builder/support.py

@@ -179,7 +179,10 @@ def workspace_build_verification_payload(*, target: dict[str, str]) -> dict[str,
         return None
     return {
         "intent": "sandbox.run_command",
-        "arguments": {"command": f"python3 -m compileall -q {root_dir}/src"},
+        "arguments": {
+            "command": f"python3 -m compileall -q {root_dir}/src",
+            "_trusted_local_only": True,
+        },
     }
 
 

core/capability_tokens.py

@@ -218,7 +218,7 @@ def verify_assignment_capability(
     remembered = load_capability_token(token_id)
     if remembered is not None:
         status = str(remembered.get("status") or "active")
-        if status not in {"active", "used"}:
+        if status != "active":
             return CapabilityDecision(False, f"Capability token is {status}.")
         local_signature = str(remembered.get("signature") or "")
         if local_signature and local_signature != signature:

core/runtime_execution_tools.py

@@ -11,6 +11,7 @@
 from core.execution_gate import ExecutionGate
 from core.runtime_paths import resolve_workspace_root
 from core.runtime_tool_contracts import runtime_tool_contract_map, runtime_tool_contracts
+from sandbox.network_guard import parse_command
 from sandbox.sandbox_runner import SandboxRunner
 
 _EXECUTION_REQUEST_MARKERS = (
@@ -887,7 +888,11 @@ def _run_command(arguments: dict[str, Any], *, workspace_root: Path) -> RuntimeE
     cwd = _resolve_workspace_path(raw_cwd, workspace_root=workspace_root) if raw_cwd else workspace_root
     if cwd.is_file():
         cwd = cwd.parent
-    runner = SandboxRunner(ExecutionGate(), str(workspace_root))
+    runner = SandboxRunner(
+        ExecutionGate(),
+        str(workspace_root),
+        network_isolation_mode=_trusted_local_network_mode(command, arguments=arguments),
+    )
     result = runner.run_command(command, cwd=str(cwd))
     relative_cwd = _relative_path(cwd, workspace_root=workspace_root)
     status = str(result.get("status") or "")
@@ -1012,3 +1017,16 @@ def _run_command(arguments: dict[str, Any], *, workspace_root: Path) -> RuntimeE
             ),
         },
     )
+
+
+def _trusted_local_network_mode(command: str, *, arguments: dict[str, Any]) -> str | None:
+    if not bool(arguments.get("_trusted_local_only", False)):
+        return None
+    argv = parse_command(command)
+    if len(argv) < 4:
+        return None
+    if argv[0] not in {"python", "python3"}:
+        return None
+    if argv[1:4] != ["-m", "compileall", "-q"]:
+        return None
+    return "heuristic_only"

network/transport.py

@@ -2,6 +2,7 @@
 
 import base64
 import contextlib
+import errno
 import os
 import secrets
 import socket
@@ -182,42 +183,18 @@ def _parse_frag_header(packet: bytes) -> tuple[str, int, int] | None:
     return transfer_id, index, total
 
 
-def _kill_stale_udp_holder(port: int) -> bool:
-    """Find and kill a stale process holding a UDP port (Windows + POSIX)."""
-    import subprocess
-    my_pid = os.getpid()
-    try:
-        result = subprocess.run(
-            ["netstat", "-ano", "-p", "UDP"],
-            capture_output=True, text=True, timeout=5,
-        )
-        for line in (result.stdout or "").splitlines():
-            if "UDP" not in line or f":{port}" not in line:
-                continue
-            parts = line.split()
-            if len(parts) < 3:
-                continue
-            try:
-                holder_pid = int(parts[-1])
-            except ValueError:
-                continue
-            if holder_pid == my_pid or holder_pid == 0:
-                continue
-            audit_logger.log(
-                "killing_stale_port_holder",
-                target_id=f"pid={holder_pid}",
-                target_type="transport",
-                details={"port": port},
-            )
-            with contextlib.suppress(Exception):
-                subprocess.run(
-                    ["taskkill", "/F", "/PID", str(holder_pid)],
-                    capture_output=True, timeout=5,
-                )
-            return True
-    except Exception:
-        pass
-    return False
+def _is_address_in_use_error(exc: OSError) -> bool:
+    err_no = getattr(exc, "errno", None)
+    winerror = getattr(exc, "winerror", None)
+    return err_no == errno.EADDRINUSE or err_no == 10048 or winerror == 10048
+
+
+def _raise_udp_bind_conflict(host: str, port: int, exc: OSError) -> None:
+    raise OSError(
+        getattr(exc, "errno", None) or getattr(exc, "winerror", None) or errno.EADDRINUSE,
+        f"UDP transport cannot bind {host}:{port}: port already in use. "
+        "Stop the conflicting process or choose a different port.",
+    ) from exc
 
 
 def _configure_udp_socket_buffers(sock: socket.socket) -> None:
@@ -285,20 +262,10 @@ def start(self) -> TransportRuntime:
         try:
             sock.bind((self.host, self.port))
         except OSError as bind_err:
-            if getattr(bind_err, "winerror", None) == 10048 or bind_err.errno == 10048:
-                killed = _kill_stale_udp_holder(self.port)
-                if killed:
-                    time.sleep(0.5)
-                    try:
-                        sock.bind((self.host, self.port))
-                    except OSError:
-                        sock.close()
-                        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
-                        sock.bind((self.host, self.port))
-                else:
-                    raise
-            else:
-                raise
+            sock.close()
+            if _is_address_in_use_error(bind_err):
+                _raise_udp_bind_conflict(self.host, self.port, bind_err)
+            raise
 
         bound_port = int(sock.getsockname()[1])
 

sandbox/job_runner.py

@@ -65,9 +65,11 @@ def _with_network_isolation(self, argv: list[str]) -> list[str]:
         isolated = self._kernel_network_isolation_prefix(argv)
         if isolated is not None:
             return isolated
-        if mode == "os_enforced":
+        if mode in {"auto", "os_enforced"}:
             raise ValueError(
-                "OS-level network isolation is required but unavailable (expected one of: bwrap, unshare, firejail)."
+                "OS-level network isolation is required but unavailable "
+                "(expected one of: bwrap, unshare, firejail). "
+                "Set network_isolation_mode='heuristic_only' only for an explicit unsafe local override."
             )
         return argv
 

sandbox/sandbox_runner.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import os
 import subprocess
 from pathlib import Path
 
@@ -10,6 +11,16 @@
 from sandbox.resource_limits import ExecutionPolicy
 
 
+def _network_isolation_mode_from_env() -> str:
+    raw = str(os.environ.get("NULLA_SANDBOX_NETWORK_MODE") or "").strip().lower()
+    if raw in {"os_enforced", "heuristic_only"}:
+        return raw
+    unsafe = str(os.environ.get("NULLA_SANDBOX_UNSAFE_EXPLICIT") or "").strip().lower()
+    if unsafe in {"1", "true", "yes", "on"}:
+        return "heuristic_only"
+    return "auto"
+
+
 class SandboxRunner:
     """
     Acts as the single point of OS-level command execution.
@@ -29,7 +40,7 @@ class SandboxRunner:
 
     _PACKAGE_INSTALL_COMMANDS = {"pip", "pip3", "npm", "npx", "yarn", "pnpm", "cargo"}
 
-    def __init__(self, gate: ExecutionGate, workspace_path: str):
+    def __init__(self, gate: ExecutionGate, workspace_path: str, *, network_isolation_mode: str | None = None):
         self.gate = gate
         self.workspace = workspace_path
         self._workspace_path = Path(workspace_path).resolve()
@@ -40,6 +51,7 @@ def __init__(self, gate: ExecutionGate, workspace_path: str):
                 max_seconds=120,
                 max_output_kb=256,
                 allow_network_egress=False,
+                network_isolation_mode=network_isolation_mode or _network_isolation_mode_from_env(),
             )
         )
 

tests/test_capability_tokens.py

@@ -11,6 +11,7 @@
     issue_assignment_capability,
     load_capability_token,
     mark_capability_token_used,
+    remember_capability_token,
     verify_assignment_capability,
 )
 from core.task_capsule import build_task_capsule
@@ -131,6 +132,42 @@ def test_expire_stale_capability_tokens_marks_expired(self) -> None:
         remembered = load_capability_token(str(token["token_id"]))
         self.assertEqual(str((remembered or {}).get("status")), "expired")
 
+    def test_assignment_capability_rejects_previously_used_token(self) -> None:
+        task_id = f"task-{uuid.uuid4()}"
+        capsule = _capsule(task_id)
+        helper_peer_id = "helper-peer-used"
+
+        token = {
+            "token_id": str(uuid.uuid4()),
+            "capability_name": "EXECUTE_TASK_CAPSULE",
+            "task_id": task_id,
+            "granted_by": get_local_peer_id(),
+            "granted_to": helper_peer_id,
+            "scope": {
+                "task_id": task_id,
+                "capsule_hash": capsule.capsule_hash,
+                "allowed_operations": list(capsule.allowed_operations),
+                "max_response_bytes": int(capsule.max_response_bytes),
+                "assignment_mode": "single",
+            },
+            "created_at": datetime.now(timezone.utc).isoformat(),
+            "expires_at": (datetime.now(timezone.utc) + timedelta(minutes=10)).isoformat(),
+            "signature": "sig-test",
+        }
+        remember_capability_token(token, status="used")
+
+        with unittest.mock.patch("core.capability_tokens.signer.verify", return_value=True):
+            decision = verify_assignment_capability(
+                token,
+                task_id=task_id,
+                parent_peer_id=get_local_peer_id(),
+                helper_peer_id=helper_peer_id,
+                capsule=capsule,
+            )
+
+        self.assertFalse(decision.ok)
+        self.assertIn("used", decision.reason)
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_job_runner.py

@@ -39,6 +39,35 @@ def test_os_enforced_network_isolation_requires_backend(self) -> None:
                 with self.assertRaises(ValueError):
                     runner.run(["python3", "-c", "print('safe')"])
 
+    @unittest.skipIf(sys.platform == "win32", "PosixPath cannot be instantiated on Windows")
+    def test_auto_network_isolation_now_fails_closed_without_backend(self) -> None:
+        with tempfile.TemporaryDirectory() as tmpdir:
+            runner = JobRunner(
+                ExecutionPolicy(
+                    workspace_root=Path(tmpdir),
+                    network_isolation_mode="auto",
+                )
+            )
+            with patch("sandbox.job_runner.os.name", "posix"), patch("sandbox.job_runner.sys.platform", "darwin"), patch(
+                "sandbox.job_runner.shutil.which", return_value=None
+            ):
+                with self.assertRaises(ValueError):
+                    runner.run(["python3", "-c", "print('safe')"])
+
+    def test_heuristic_only_mode_remains_explicit_opt_in(self) -> None:
+        with tempfile.TemporaryDirectory() as tmpdir:
+            runner = JobRunner(
+                ExecutionPolicy(
+                    workspace_root=Path(tmpdir),
+                    network_isolation_mode="heuristic_only",
+                )
+            )
+            with patch("sandbox.job_runner.os.name", "posix"), patch("sandbox.job_runner.sys.platform", "darwin"), patch(
+                "sandbox.job_runner.shutil.which", return_value=None
+            ):
+                argv = runner._with_network_isolation(["python3", "-c", "print('x')"])
+                self.assertEqual(argv, ["python3", "-c", "print('x')"])
+
     def test_linux_unshare_prefix_applied_when_available(self) -> None:
         with tempfile.TemporaryDirectory() as tmpdir:
             runner = JobRunner(

tests/test_runtime_execution_tools.py

@@ -9,7 +9,11 @@
 
 import pytest
 
-from core.runtime_execution_tools import execute_runtime_tool, extract_observation_followup_hints
+from core.runtime_execution_tools import (
+    _trusted_local_network_mode,
+    execute_runtime_tool,
+    extract_observation_followup_hints,
+)
 
 _UNSHARE_AVAILABLE = os.system("unshare -r true >/dev/null 2>&1") == 0
 
@@ -232,6 +236,27 @@ def test_sandbox_run_command_blocks_network_commands(self) -> None:
             self.assertEqual(result.details["observation"]["tool_surface"], "sandbox")
             self.assertEqual(result.details["observation"]["status"], "blocked_by_policy")
 
+    def test_trusted_local_network_mode_only_applies_to_internal_compileall_verification(self) -> None:
+        self.assertEqual(
+            _trusted_local_network_mode(
+                "python3 -m compileall -q generated/telegram-bot/src",
+                arguments={"_trusted_local_only": True},
+            ),
+            "heuristic_only",
+        )
+        self.assertIsNone(
+            _trusted_local_network_mode(
+                "python3 app.py",
+                arguments={"_trusted_local_only": True},
+            )
+        )
+        self.assertIsNone(
+            _trusted_local_network_mode(
+                "python3 -m compileall -q generated/telegram-bot/src",
+                arguments={},
+            )
+        )
+
 
 if __name__ == "__main__":
     unittest.main()