diff --git a/pyisolate/bpf/contract.py b/pyisolate/bpf/contract.py new file mode 100644 index 0000000..138e94d --- /dev/null +++ b/pyisolate/bpf/contract.py @@ -0,0 +1,114 @@ +"""Authoritative contract between the policy compiler and the eBPF LSM program. + +The kernel program ``syscall_filter.bpf.c`` enforces policy by looking up +``sandbox_policy[bpf_get_current_cgroup_id()]`` and denying operations whose +capability-class bit is set in ``deny_mask``. For that lookup to hit, userspace +must write the map with: + +* **key** = the sandbox's cgroup id, which is the cgroup directory's inode + number (what ``bpf_get_current_cgroup_id`` returns), as a little-endian u64; +* **value** = ``struct pyisolate_policy { __u32 deny_mask; __u32 audit_only; }``, + little-endian. + +This module is the single source of truth for those constants and their byte +encoding, shared by the manager (which writes the map) and the tests (which +assert the encoding matches the kernel struct). + +Granularity note: ``deny_mask`` is a coarse per-cgroup, per-capability-class +switch. It can express "deny all filesystem" or "deny all network", but not +"allow only ``/srv/data``". Fine-grained filesystem access is enforced by +Landlock and fine-grained network by the broker; the eBPF mask handles the +always-denied classes (process creation, ptrace/mount/bpf) and whole-class +denial when the policy grants nothing in a class. +""" + +from __future__ import annotations + +import os + +from ..policy.model import RuntimePolicy + +# Capability-class deny bits. These MUST match the ``PYI_DENY_*`` defines in +# syscall_filter.bpf.c. +DENY_FS = 1 << 0 +DENY_NET = 1 << 1 +DENY_PROCESS = 1 << 2 +DENY_RISKY = 1 << 3 + +#: Name of the pinned map the LSM program consults (defined in the .bpf.c). +SANDBOX_POLICY_MAP = "sandbox_policy" + + +def _policy_allows_any_fs(policy: object) -> bool: + if policy is None: + return False + if isinstance(policy, RuntimePolicy): + return bool(policy.allow_fs) + if getattr(policy, "fs", None): + return True + for cap in getattr(policy, "capabilities", None) or []: + if getattr(cap, "kind", None) in ("read_path", "write_path"): + return True + return False + + +def _policy_allows_any_net(policy: object) -> bool: + if policy is None: + return False + if isinstance(policy, RuntimePolicy): + return bool(policy.allow_tcp) + if getattr(policy, "tcp", None): + return True + for cap in getattr(policy, "capabilities", None) or []: + if getattr(cap, "kind", None) == "connect_tcp": + return True + return False + + +def compile_deny_mask(policy: object) -> int: + """Derive the coarse per-cgroup ``deny_mask`` enforced by the LSM program. + + Process creation and risky operations (ptrace/mount/bpf) are always denied + inside a sandbox cgroup -- the sandbox never execs and the broker performs + privileged work from a different cgroup. Filesystem and network are denied + as whole classes only when the policy grants nothing in that class; anything + finer is delegated to Landlock and the broker. + """ + mask = DENY_PROCESS | DENY_RISKY + if not _policy_allows_any_fs(policy): + mask |= DENY_FS + if not _policy_allows_any_net(policy): + mask |= DENY_NET + return mask + + +def _byte_tokens(data: bytes) -> list[str]: + """Return ``["0x..", ...]`` tokens ``bpftool map update`` consumes.""" + return [f"0x{byte:02x}" for byte in data] + + +def encode_sandbox_policy_key(cgroup_id: int) -> list[str]: + """Encode a cgroup id as the little-endian u64 map key.""" + return _byte_tokens(int(cgroup_id).to_bytes(8, "little")) + + +def encode_sandbox_policy_value(deny_mask: int, audit_only: bool = False) -> list[str]: + """Encode ``struct pyisolate_policy`` (two little-endian u32s).""" + return _byte_tokens( + int(deny_mask).to_bytes(4, "little") + + (1 if audit_only else 0).to_bytes(4, "little") + ) + + +def cgroup_id_for_path(path: "str | os.PathLike[str] | None") -> int | None: + """Return the cgroup id (directory inode) for *path*, or ``None``. + + ``bpf_get_current_cgroup_id()`` returns the cgroupfs directory inode, so the + inode is exactly the key the kernel program will look the policy up under. + """ + if path is None: + return None + try: + return os.stat(path).st_ino + except OSError: + return None diff --git a/pyisolate/bpf/manager.py b/pyisolate/bpf/manager.py index 5140acd..19a9cff 100644 --- a/pyisolate/bpf/manager.py +++ b/pyisolate/bpf/manager.py @@ -391,6 +391,43 @@ def hot_reload(self, policy_path: str) -> None: # paths have one representation. self.policy_maps = new_policy_maps + def set_sandbox_policy( + self, + cgroup_id: int, + deny_mask: int, + *, + audit_only: bool = False, + strict: bool = False, + ) -> bool: + """Write one sandbox's coarse deny-mask into the pinned ``sandbox_policy`` + map, keyed by its cgroup id, so the LSM program enforces it. + + Returns whether the update succeeded. When ``strict`` is set (hardened + rollout), a failure raises instead of being reported as ``False``. + """ + from .contract import ( + SANDBOX_POLICY_MAP, + encode_sandbox_policy_key, + encode_sandbox_policy_value, + ) + + pinned = self._bpffs_root / SANDBOX_POLICY_MAP + return self._run( + [ + "bpftool", + "map", + "update", + "pinned", + str(pinned), + "key", + *encode_sandbox_policy_key(cgroup_id), + "value", + *encode_sandbox_policy_value(deny_mask, audit_only), + "any", + ], + raise_on_error=strict, + ) + def _update_bpf_map(self, map_name: str, key: str, value: str) -> None: self._run( [ diff --git a/pyisolate/supervisor.py b/pyisolate/supervisor.py index 578c952..108b4cb 100644 --- a/pyisolate/supervisor.py +++ b/pyisolate/supervisor.py @@ -364,6 +364,7 @@ def spawn( mode=cast(cgroup.RolloutMode, self._rollout_mode), ) cg_path = cg_status.path + self._apply_kernel_policy(cg_path, policy) temp_dir = recovery.allocate_temp_dir(name) reset_config = { "policy": policy, @@ -449,6 +450,32 @@ def spawn( NAME_PATTERN = DEFAULT_NAME_PATTERN return Sandbox(thread, self) + def _apply_kernel_policy(self, cg_path: Any, policy: Any) -> None: + """Publish this sandbox's coarse deny-mask to the eBPF ``sandbox_policy`` + map, keyed by its cgroup id, so the LSM program enforces it. + + Best-effort in dev/compatibility rollout; in hardened mode a failure to + program the kernel map propagates so the spawn fails closed. + """ + if cg_path is None: + return + from .bpf.contract import cgroup_id_for_path, compile_deny_mask + + strict = self._rollout_mode == "hardened" + cgroup_id = cgroup_id_for_path(cg_path) + if cgroup_id is None: + if strict: + raise RuntimeError(f"cannot resolve cgroup id for {cg_path}") + return + try: + self._bpf.set_sandbox_policy( + cgroup_id, compile_deny_mask(policy), strict=strict + ) + except Exception: + if strict: + raise + logger.debug("sandbox_policy map update skipped for %s", cg_path) + def _spawn_process( self, name: str, diff --git a/tests/test_ebpf_contract.py b/tests/test_ebpf_contract.py new file mode 100644 index 0000000..977f9a5 --- /dev/null +++ b/tests/test_ebpf_contract.py @@ -0,0 +1,138 @@ +"""Tests for the userspace<->eBPF policy-map contract. + +The kernel LSM program enforces policy via ``sandbox_policy[cgroup_id]``. These +tests verify that userspace derives the right deny-mask, encodes the map key and +value exactly as the kernel struct expects, and -- crucially -- writes the map +the kernel program actually defines. The last point is a guard against the +class of bug where userspace programs a map name the kernel never reads. +""" + +import re +import sys +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT)) + +import pyisolate as iso +import pyisolate.bpf.manager as manager +from pyisolate.bpf import contract + +_BPF_SRC = ROOT / "pyisolate" / "bpf" / "syscall_filter.bpf.c" + + +def _kernel_map_names() -> set[str]: + text = _BPF_SRC.read_text(encoding="utf-8") + return set(re.findall(r"\}\s*([A-Za-z_]\w*)\s*SEC\(\"\.maps\"\)", text)) + + +def _kernel_deny_defines() -> dict[str, int]: + text = _BPF_SRC.read_text(encoding="utf-8") + found = {} + for name, shift in re.findall(r"#define\s+(PYI_DENY_\w+)\s+\(1U\s*<<\s*(\d+)\)", text): + found[name] = 1 << int(shift) + return found + + +def test_enforcement_map_is_defined_in_the_kernel_program(): + # The map userspace writes for enforcement must be one the LSM program reads. + assert contract.SANDBOX_POLICY_MAP in _kernel_map_names() + + +def test_deny_mask_bits_match_kernel_defines(): + # Guard against the Python constants drifting from the .bpf.c #defines. + kernel = _kernel_deny_defines() + assert kernel["PYI_DENY_FS"] == contract.DENY_FS + assert kernel["PYI_DENY_NET"] == contract.DENY_NET + assert kernel["PYI_DENY_PROCESS"] == contract.DENY_PROCESS + assert kernel["PYI_DENY_RISKY"] == contract.DENY_RISKY + + +def test_compile_deny_mask_denies_everything_without_a_policy(): + mask = contract.compile_deny_mask(None) + assert mask == ( + contract.DENY_FS | contract.DENY_NET | contract.DENY_PROCESS | contract.DENY_RISKY + ) + + +def test_compile_deny_mask_always_denies_process_and_risky(): + for policy in (None, iso.policy.Policy().allow_fs("/tmp").allow_tcp("127.0.0.1:80")): + mask = contract.compile_deny_mask(policy) + assert mask & contract.DENY_PROCESS + assert mask & contract.DENY_RISKY + + +def test_compile_deny_mask_clears_class_bits_when_granted(): + fs_only = contract.compile_deny_mask(iso.policy.Policy().allow_fs("/tmp")) + assert not fs_only & contract.DENY_FS + assert fs_only & contract.DENY_NET # nothing granted for net + + net_only = contract.compile_deny_mask(iso.policy.Policy().allow_tcp("10.0.0.1:443")) + assert not net_only & contract.DENY_NET + assert net_only & contract.DENY_FS + + +def test_sandbox_policy_key_is_little_endian_u64(): + tokens = contract.encode_sandbox_policy_key(0x1122334455667788) + raw = bytes(int(t, 16) for t in tokens) + assert len(raw) == 8 + assert int.from_bytes(raw, "little") == 0x1122334455667788 + + +def test_sandbox_policy_value_encodes_two_little_endian_u32(): + tokens = contract.encode_sandbox_policy_value(0xF, audit_only=True) + raw = bytes(int(t, 16) for t in tokens) + assert len(raw) == 8 + assert int.from_bytes(raw[:4], "little") == 0xF + assert int.from_bytes(raw[4:], "little") == 1 + + +def test_cgroup_id_for_path_returns_directory_inode(tmp_path): + assert contract.cgroup_id_for_path(tmp_path) == tmp_path.stat().st_ino + assert contract.cgroup_id_for_path(tmp_path / "missing") is None + assert contract.cgroup_id_for_path(None) is None + + +def test_set_sandbox_policy_targets_the_pinned_map(monkeypatch): + mgr = manager.BPFManager() + captured = {} + + def fake_run(cmd, *, raise_on_error=False): + captured["cmd"] = cmd + return True + + monkeypatch.setattr(mgr, "_run", fake_run) + assert mgr.set_sandbox_policy(11, contract.DENY_FS | contract.DENY_NET) is True + + cmd = captured["cmd"] + assert cmd[:4] == ["bpftool", "map", "update", "pinned"] + assert cmd[4].endswith("/pyisolate/sandbox_policy") + key_tokens = cmd[cmd.index("key") + 1 : cmd.index("value")] + assert bytes(int(t, 16) for t in key_tokens) == (11).to_bytes(8, "little") + + +def test_spawn_programs_the_sandbox_policy_map(monkeypatch): + calls = [] + + def fake_set(self, cgroup_id, deny_mask, **kwargs): + calls.append((cgroup_id, deny_mask)) + return True + + monkeypatch.setattr(manager.BPFManager, "set_sandbox_policy", fake_set) + + sup = iso.Supervisor() + try: + sb = sup.spawn("kpol", allowed_imports=["math"]) + assert len(calls) == 1 + cgroup_id, deny_mask = calls[0] + assert isinstance(cgroup_id, int) and cgroup_id > 0 + # No fs/net granted -> every class denied. + assert deny_mask == ( + contract.DENY_FS + | contract.DENY_NET + | contract.DENY_PROCESS + | contract.DENY_RISKY + ) + sb.close() + finally: + sup.shutdown()