Wire the compiled policy into the eBPF map the LSM program enforces

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

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(
             [

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,

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()