From e0539edff9c40454d83b75ec5f6280e07652fa01 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 20 Jul 2026 20:33:55 +0000 Subject: [PATCH] Align tests, typing, and tooling with the documented spec - Fix test_concurrent_roundtrip flakiness: the broker protocol mandates strictly ordered receive counters (docs/protocol.md), so concurrent workers now retry out-of-order deliveries instead of racing frame() against unframe() and failing on scheduler interleavings. - Fix a portability bug in the conformance cgroup probe: read back threading.get_native_id() (what attach_current writes) instead of os.gettid(), which is missing on some libc builds and could also disagree with the written TID. - Resolve all 74 mypy errors across the tree: annotate SandboxThread runtime state, CryptoBroker counters, watchdog ring-buffer iterator, alert callbacks, and module-proxy helpers; align optional-dependency fallback stub signatures in pyisolate/__init__; rename shadowed loop variables in policy model/compiler; type the operator sandbox map. - Pre-commit: give mypy the PyYAML stubs, and run pytest via 'python -m pytest' so the hook uses the interpreter the package is installed into rather than whatever pytest is first on PATH. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01DDSofWX5HwawsrwbN2nSXR --- .pre-commit-config.yaml | 9 ++++++-- pyisolate/__init__.py | 8 ++++--- pyisolate/broker/crypto.py | 3 +++ pyisolate/broker/uring.py | 3 ++- pyisolate/capabilities.py | 2 +- pyisolate/conformance.py | 6 +++-- pyisolate/nogil.py | 2 +- pyisolate/observability/alerts.py | 7 ++++-- pyisolate/operator/__init__.py | 7 ++++-- pyisolate/policy/__init__.py | 13 ++++++----- pyisolate/policy/compiler.py | 6 ++--- pyisolate/policy/model.py | 16 ++++++------- pyisolate/runtime/thread.py | 37 ++++++++++++++++++------------- pyisolate/supervisor.py | 7 ++++-- pyisolate/watchdog.py | 6 +++-- tests/test_checkpoint.py | 2 +- tests/test_crypto.py | 14 +++++++++++- 17 files changed, 97 insertions(+), 51 deletions(-) diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 4b367ed..6681fd9 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -23,11 +23,16 @@ repos: rev: v1.10.0 hooks: - id: mypy - additional_dependencies: [] + # types-PyYAML: mypy reports missing PyYAML stubs even with + # ignore_missing_imports, because typeshed tracks them separately. + additional_dependencies: [types-PyYAML] - repo: local hooks: - id: pytest name: pytest - entry: pytest -q + # python -m pytest keeps the hook on the interpreter the package is + # installed into; a bare ``pytest`` can resolve to an unrelated + # standalone install on PATH. + entry: python -m pytest -q language: system pass_filenames: false diff --git a/pyisolate/__init__.py b/pyisolate/__init__.py index a172a66..d3f8290 100644 --- a/pyisolate/__init__.py +++ b/pyisolate/__init__.py @@ -42,10 +42,10 @@ if isinstance(exc, ImportError) and "cryptography" not in str(exc): raise - def checkpoint(*args, **kwargs): # type: ignore[no-redef] + def checkpoint(sandbox: "Sandbox", key: bytes) -> bytes: # type: ignore[no-redef] raise ModuleNotFoundError("cryptography is required for checkpoint support") - def restore(*args, **kwargs): # type: ignore[no-redef] + def restore(blob: bytes, key: bytes) -> "Sandbox": # type: ignore[no-redef] raise ModuleNotFoundError("cryptography is required for checkpoint support") @@ -91,7 +91,9 @@ def restore(*args, **kwargs): # type: ignore[no-redef] if isinstance(exc, ImportError) and "cryptography" not in str(exc): raise - def migrate(*args, **kwargs): # type: ignore[no-redef] + def migrate( # type: ignore[no-redef] + sandbox: "Sandbox", host: str, key: bytes + ) -> "MigrationResponse": raise ModuleNotFoundError("cryptography is required for migration support") diff --git a/pyisolate/broker/crypto.py b/pyisolate/broker/crypto.py index 18030e8..545e58f 100644 --- a/pyisolate/broker/crypto.py +++ b/pyisolate/broker/crypto.py @@ -59,6 +59,9 @@ class CryptoBroker: threads; counter and key updates are protected by an internal lock. """ + _tx_ctr: int + _rx_ctr: int + def __init__( self, private_key: bytes, diff --git a/pyisolate/broker/uring.py b/pyisolate/broker/uring.py index 58f1150..6f2ee3d 100644 --- a/pyisolate/broker/uring.py +++ b/pyisolate/broker/uring.py @@ -9,6 +9,7 @@ import asyncio import os +from typing import Any try: import uring # type: ignore @@ -20,7 +21,7 @@ class IOUring: """Minimal wrapper around ``io_uring`` for async file descriptor I/O.""" def __init__(self, entries: int = 8): - self._ring = None + self._ring: Any = None if uring is not None: ring = uring.io_uring() # handle API differences across wrapper versions diff --git a/pyisolate/capabilities.py b/pyisolate/capabilities.py index 465c1cb..e8be31c 100644 --- a/pyisolate/capabilities.py +++ b/pyisolate/capabilities.py @@ -40,7 +40,7 @@ class Authority(Capability[str]): kind: str - def to_policy_rule(self) -> dict[str, Any]: + def to_policy_rule(self) -> dict[str, Any] | str: """Return a YAML-serializable representation of this authority.""" raise NotImplementedError diff --git a/pyisolate/conformance.py b/pyisolate/conformance.py index 51cabd9..12bc65c 100644 --- a/pyisolate/conformance.py +++ b/pyisolate/conformance.py @@ -9,12 +9,12 @@ import argparse import json -import os import platform import shutil import subprocess import sys import sysconfig +import threading import time import uuid from dataclasses import asdict, dataclass @@ -348,8 +348,10 @@ def _probe_cgroup_behavior(self) -> ProbeResult: if created: cgroup.attach_current(cg_path) threads_file = Path(cg_path) / "cgroup.threads" + # attach_current writes threading.get_native_id(), so read back the + # same identifier; os.gettid() is unavailable on some libc builds. attached = threads_file.exists() and str( - os.gettid() + threading.get_native_id() ) in threads_file.read_text(encoding="utf-8") cgroup.delete(cg_path) deleted = not Path(cg_path).exists() diff --git a/pyisolate/nogil.py b/pyisolate/nogil.py index e5bf794..343dc5b 100644 --- a/pyisolate/nogil.py +++ b/pyisolate/nogil.py @@ -50,7 +50,7 @@ def _module_origin(module: ModuleType) -> str | None: def _is_native_origin(origin: str | None) -> bool: - return bool(origin) and origin.endswith(_NATIVE_SUFFIXES) + return origin is not None and origin.endswith(_NATIVE_SUFFIXES) def imported_native_extensions() -> list[dict[str, Any]]: diff --git a/pyisolate/observability/alerts.py b/pyisolate/observability/alerts.py index b18ea3e..7cb60e7 100644 --- a/pyisolate/observability/alerts.py +++ b/pyisolate/observability/alerts.py @@ -1,15 +1,18 @@ import logging +from collections.abc import Callable logger = logging.getLogger(__name__) +AlertCallback = Callable[[str, Exception], object] + class AlertManager: """Dispatch callbacks on policy violations.""" def __init__(self) -> None: - self._subs: list[callable] = [] + self._subs: list[AlertCallback] = [] - def register(self, callback) -> None: + def register(self, callback: AlertCallback) -> None: self._subs.append(callback) def notify(self, sandbox: str, error: Exception) -> list[Exception]: diff --git a/pyisolate/operator/__init__.py b/pyisolate/operator/__init__.py index 29fe7a7..5dcd775 100644 --- a/pyisolate/operator/__init__.py +++ b/pyisolate/operator/__init__.py @@ -2,10 +2,13 @@ from __future__ import annotations import logging -from typing import Dict +from typing import TYPE_CHECKING, Dict from ..supervisor import Supervisor +if TYPE_CHECKING: + from ..supervisor import Sandbox + __all__ = ["run_operator", "scale_sandboxes"] logger = logging.getLogger(__name__) @@ -19,7 +22,7 @@ def run_operator(namespace: str = "default") -> None: api = client.CustomObjectsApi() w = watch.Watch() sup = Supervisor() - sandboxes: Dict[str, object] = {} + sandboxes: Dict[str, "Sandbox"] = {} for event in w.stream( api.list_namespaced_custom_object, group="pyisolate.dev", diff --git a/pyisolate/policy/__init__.py b/pyisolate/policy/__init__.py index 4259137..803ebf4 100644 --- a/pyisolate/policy/__init__.py +++ b/pyisolate/policy/__init__.py @@ -5,6 +5,7 @@ import socket import tempfile import urllib.request +from collections.abc import Iterable, Mapping from dataclasses import dataclass, field from importlib import resources from pathlib import Path @@ -42,8 +43,8 @@ def _mini_load(text: str) -> dict: from typing import cast assert isinstance(result[current], list) - lst = cast(list[str], result[current]) - lst.append(_unquote(item)) + str_lst = cast(list[str], result[current]) + str_lst.append(_unquote(item)) continue if ":" not in line: @@ -66,7 +67,7 @@ def safe_load(stream): return _mini_load(stream.read()) return _mini_load(stream) - yaml = _MiniYaml() + yaml = _MiniYaml() # type: ignore[assignment] from ..capabilities import ConnectTCP, CpuBudget, Import, ReadPath, WritePath @@ -188,11 +189,13 @@ def to_yaml(self, name: str = "default") -> str: lines = ["version: 1.0", "sandboxes:", f" {name}:"] sandbox = data["sandboxes"][name] # type: ignore[index] - def append_rule_list(section: str, rules: object) -> None: + def append_rule_list( + section: str, rules: Iterable[Mapping[str, object]] | None + ) -> None: if not rules: return lines.append(f" {section}:") - for rule in rules: # type: ignore[union-attr] + for rule in rules: key, value = next(iter(rule.items())) lines.append(f" - {key}: {value!r}") diff --git a/pyisolate/policy/compiler.py b/pyisolate/policy/compiler.py index 388e2e4..9a6ed2d 100644 --- a/pyisolate/policy/compiler.py +++ b/pyisolate/policy/compiler.py @@ -299,10 +299,10 @@ def compile_policy(path: str | Path) -> CompiledPolicy: capabilities.append(ReadPath(rule.path)) elif rule.action == "write": capabilities.append(WritePath(rule.path)) - for rule in tcp_compiled: - if rule.action == "connect": + for tcp_rule in tcp_compiled: + if tcp_rule.action == "connect": try: - capabilities.append(ConnectTCP.from_address(rule.addr)) + capabilities.append(ConnectTCP.from_address(tcp_rule.addr)) except ValueError: pass capabilities.extend(Import(module) for module in imports) diff --git a/pyisolate/policy/model.py b/pyisolate/policy/model.py index 230f3f2..1068fab 100644 --- a/pyisolate/policy/model.py +++ b/pyisolate/policy/model.py @@ -280,11 +280,11 @@ def from_sandbox_policy(policy: Any) -> RuntimePolicy: if isinstance(item, str): allow_tcp.append(NetworkRule("connect", item)) continue - rule = _coerce_compiled_tcp(item) - if rule.action == "connect": - allow_tcp.append(rule) + tcp_rule = _coerce_compiled_tcp(item) + if tcp_rule.action == "connect": + allow_tcp.append(tcp_rule) else: - deny_tcp.append(rule) + deny_tcp.append(tcp_rule) imports = tuple(str(module) for module in (getattr(policy, "imports", None) or ())) return RuntimePolicy( @@ -382,13 +382,13 @@ def to_bpf_map_entries(policy_set: RuntimePolicySet) -> list[tuple[str, str, str entries.append(("policy_fs_allow", f"{sandbox_name}:{index}", rule.path)) for index, rule in enumerate(policy.deny_fs): entries.append(("policy_fs_deny", f"{sandbox_name}:{index}", rule.path)) - for index, rule in enumerate(policy.allow_tcp): + for index, net_rule in enumerate(policy.allow_tcp): entries.append( - ("policy_net_allow", f"{sandbox_name}:{index}", rule.destination) + ("policy_net_allow", f"{sandbox_name}:{index}", net_rule.destination) ) - for index, rule in enumerate(policy.deny_tcp): + for index, net_rule in enumerate(policy.deny_tcp): entries.append( - ("policy_net_deny", f"{sandbox_name}:{index}", rule.destination) + ("policy_net_deny", f"{sandbox_name}:{index}", net_rule.destination) ) for index, module in enumerate(policy.imports): entries.append(("policy_import_allow", f"{sandbox_name}:{index}", module)) diff --git a/pyisolate/runtime/thread.py b/pyisolate/runtime/thread.py index 36886ea..74ee409 100644 --- a/pyisolate/runtime/thread.py +++ b/pyisolate/runtime/thread.py @@ -48,7 +48,7 @@ from ..numa import bind_current_thread from ..observability.trace import Tracer from ..policy.model import RuntimePolicy, from_sandbox_policy -from ..telemetry import DenialEvent +from ..telemetry import Decision, DenialEvent from .protocol import ( AttachCgroupRequest, BrokerRequest, @@ -123,8 +123,8 @@ def _deny( policy_rule: str, message: str, *, - kernel_decision: str = "not_evaluated", - broker_decision: str = "deny", + kernel_decision: Decision = "not_evaluated", + broker_decision: Decision = "deny", ) -> errors.PolicyError: sandbox = _active_sandbox() cell = sandbox.name if sandbox is not None else "" @@ -670,12 +670,12 @@ def _check_network_destination(address: Iterable[str]) -> None: raise errors.NetworkExceeded() -def _guarded_connect(self_socket: socket.socket, address: Iterable[str]): +def _guarded_connect(self_socket: socket.socket, address: Any): _check_network_destination(address) return _ORIG_SOCKET_CONNECT(self_socket, address) -def _guarded_connect_ex(self_socket: socket.socket, address: Iterable[str]): +def _guarded_connect_ex(self_socket: socket.socket, address: Any): _check_network_destination(address) return _ORIG_SOCKET_CONNECT_EX(self_socket, address) @@ -771,7 +771,7 @@ def _run_with_accounting(*r_args, **r_kwargs): return SandboxedThread -def _module_proxy(name: str, module) -> types.ModuleType: +def _module_proxy(name: str, module) -> Any: proxy = types.ModuleType(name, getattr(module, "__doc__", None)) proxy.__dict__.update({k: getattr(module, k) for k in dir(module)}) proxy.__dict__["__name__"] = name @@ -840,7 +840,7 @@ def _wrap_module(name: str, module): return mod if base == "time": - def _require_clock() -> ClockCapability: + def _require_clock() -> ClockCapability | None: cap = getattr(_thread_local, "clock_capability", None) return cap @@ -885,7 +885,8 @@ def __init__(self, family=-1, type=-1, proto=-1, fileno=None): connect = _guarded_connect connect_ex = _guarded_connect_ex - sendto = _guarded_sendto + # The guard collapses socket.sendto's overloads into one signature. + sendto = _guarded_sendto # type: ignore[assignment] def _create_connection( address, timeout=socket._GLOBAL_DEFAULT_TIMEOUT, source_address=None @@ -936,7 +937,7 @@ def _create_connection( if base == "pathlib": mod = _module_proxy("pathlib", module) - class SandboxedPath(type(module.Path())): + class SandboxedPath(type(module.Path())): # type: ignore[misc] def open( self, mode="r", @@ -1048,6 +1049,12 @@ class Stats: class SandboxThread(threading.Thread): """Thread that runs guest code and communicates via a queue.""" + # Assigned by the supervisor after construction/reset rather than in + # ``__init__``; declared here so the attributes are part of the class + # contract. + _backend: str + _temp_dir: Path + @staticmethod def _merge_allowed_imports(policy, allowed_imports: Optional[Iterable[str]]): imports: set[str] = set() @@ -1066,7 +1073,7 @@ def _merge_allowed_imports(policy, allowed_imports: Optional[Iterable[str]]): imports.update(runtime_policy.imports) allowed_imports_provided = allowed_imports is not None - if allowed_imports_provided: + if allowed_imports is not None: imports.update(allowed_imports) if not imports and not allowed_imports_provided: @@ -1116,7 +1123,7 @@ def _init_config_wiring( self.child_work_max = child_work_max self.allowed_imports = self._merge_allowed_imports(policy, allowed_imports) self.numa_node = numa_node - self._bound_numa_node = None + self._bound_numa_node: Optional[int] = None self._cgroup_path = cgroup_path self.quota_enforcement = enforcement_status self._capabilities = deserialize_capabilities(capabilities) @@ -1145,15 +1152,15 @@ def _reset_runtime_state(self) -> None: self._cpu_time = 0.0 self._mem_peak = 0 self._mem_base = 0 - self._start_time = None + self._start_time: Optional[float] = None self._ops = 0 self._errors = 0 self._latency = {"0.5": 0, "1": 0, "5": 0, "10": 0, "inf": 0} self._latency_sum = 0.0 self._trace_enabled = False self._syscall_log: list[str] = [] - self._quarantine_reason = None - self.termination_reason = None + self._quarantine_reason: Optional[str] = None + self.termination_reason: Optional[str] = None self._open_files = 0 self._network_ops = 0 self._output_bytes = 0 @@ -1172,7 +1179,7 @@ def __init__( output_bytes_max: Optional[int] = None, child_work_max: Optional[int] = None, allowed_imports: Optional[list[str]] = None, - on_violation: Optional[Callable[[str, Exception], None]] = None, + on_violation: Optional[Callable[[str, Exception], object]] = None, tracer: Optional["Tracer"] = None, numa_node: Optional[int] = None, cgroup_path=None, diff --git a/pyisolate/supervisor.py b/pyisolate/supervisor.py index 9938427..d537cc2 100644 --- a/pyisolate/supervisor.py +++ b/pyisolate/supervisor.py @@ -13,7 +13,7 @@ import re import threading from pathlib import Path -from typing import Dict, Literal, Optional +from typing import Dict, Literal, Optional, cast from . import cgroup, recovery from .capabilities import ROOT, RootCapability @@ -324,7 +324,10 @@ def spawn( reused_warm = False try: cg_status = cgroup.create( - name, cpu_ms, mem_bytes, mode=self._rollout_mode + name, + cpu_ms, + mem_bytes, + mode=cast(cgroup.RolloutMode, self._rollout_mode), ) cg_path = cg_status.path temp_dir = recovery.allocate_temp_dir(name) diff --git a/pyisolate/watchdog.py b/pyisolate/watchdog.py index e95fe9b..90f1d0e 100644 --- a/pyisolate/watchdog.py +++ b/pyisolate/watchdog.py @@ -10,7 +10,7 @@ import logging import threading import time -from collections.abc import Mapping +from collections.abc import Iterator, Mapping from typing import TYPE_CHECKING from . import errors @@ -29,7 +29,7 @@ def __init__(self, supervisor: "Supervisor", interval: float = 0.05): self._supervisor = supervisor self._interval = interval self._stop_event = threading.Event() - self._rb_iter = None + self._rb_iter: Iterator[object] | None = None def stop(self, timeout: float = 0.2) -> None: self._stop_event.set() @@ -56,6 +56,8 @@ def run(self) -> None: continue name = event.get("name") + if not isinstance(name, str): + continue cpu_ms = event.get("cpu_ms", 0) rss = event.get("rss_bytes", 0) # The quota comparisons below run outside their try/except blocks, so diff --git a/tests/test_checkpoint.py b/tests/test_checkpoint.py index b63f176..44f73d6 100644 --- a/tests/test_checkpoint.py +++ b/tests/test_checkpoint.py @@ -44,7 +44,7 @@ def open_ring_buffer(self): return iter(()) -bpf_manager.BPFManager = DummyBPFManager +setattr(bpf_manager, "BPFManager", DummyBPFManager) # Install the stub while this module imports the rest of the package, but keep a # reference to the real module so it can be restored. The supervisor imports diff --git a/tests/test_crypto.py b/tests/test_crypto.py index 228bb99..76706cc 100644 --- a/tests/test_crypto.py +++ b/tests/test_crypto.py @@ -1,4 +1,5 @@ import threading +import time import pytest from cryptography.hazmat.primitives import serialization @@ -96,7 +97,18 @@ def test_concurrent_roundtrip(): def worker(i: int) -> None: msg = f"hi{i}".encode() frame = a.frame(msg) - results[i] = b.unframe(frame) + # The receive counter is strictly ordered (docs/protocol.md), so the + # receiver only accepts the next expected frame. Threads race to + # deliver, so retry until this thread's frame is the expected one. + deadline = time.monotonic() + 30 + while True: + try: + results[i] = b.unframe(frame) + return + except ValueError: + if time.monotonic() > deadline: + raise + time.sleep(0) threads = [threading.Thread(target=worker, args=(i,)) for i in range(100)] for t in threads: