From 49f8b0e8eac857a6fadf0c665594bd47d4da0dbb Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 20 Jul 2026 22:28:44 +0000 Subject: [PATCH] Enforce filesystem policy on guest processes with Landlock Translates the sandbox's filesystem policy into a kernel Landlock ruleset applied to the guest process, so a guest that defeats the Python open guard and reaches the real open/openat can still only touch permitted paths. Applied after PR_SET_NO_NEW_PRIVS (which landlock_restrict_self requires) and before the seccomp lockdown. - runtime/landlock.py: ABI-versioned ruleset construction. Queries the kernel Landlock ABI and masks the handled access-rights set to it (requesting an unsupported right fails ruleset creation); grants read+execute on the interpreter's runtime paths (stdlib, shared libs) so the guest can keep importing, read on policy read paths, and read+write on policy write paths; everything else the ruleset handles is denied. Packed path_beneath struct, O_PATH parent fds. - confine.apply_confinement now sets no_new_privs first, then rlimits, then Landlock (only when the policy names paths -- a default-deny ruleset with no allow-list would just break the interpreter), then seccomp. - process_backend extracts read vs write paths from the policy and passes them in the bootstrap; the confinement report gains landlock fields. - Fixes a latent bug in _extract_fs_tcp: the legacy Policy exposes allow_fs/allow_tcp as *methods*, so it must be discriminated from RuntimePolicy (which exposes rule tuples) by type, not getattr. Landlock is unavailable on many kernels; it is applied best-effort and recorded in the report (hardened mode can require it). Enforcement tests are gated on landlock_supported() plus PYISOLATE_LIVE_LANDLOCK_TESTS, like the live BPF test; the ABI-masking, struct-packing, and policy-extraction logic is unit-tested unconditionally. CI runs the live Landlock test on the cross-kernel matrix. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01DDSofWX5HwawsrwbN2nSXR --- .github/workflows/ci.yml | 6 + pyisolate/runtime/child.py | 5 + pyisolate/runtime/confine.py | 55 +++++- pyisolate/runtime/landlock.py | 247 +++++++++++++++++++++++++++ pyisolate/runtime/process_backend.py | 68 +++++++- pyisolate/supervisor.py | 1 + tests/test_landlock.py | 145 ++++++++++++++++ 7 files changed, 516 insertions(+), 11 deletions(-) create mode 100644 pyisolate/runtime/landlock.py create mode 100644 tests/test_landlock.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 7c54ff8..4eb09e8 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -129,6 +129,12 @@ jobs: run: python -m pip install -e .[dev] - name: Run kernel-facing tests run: pytest -q tests/test_bpf_manager.py tests/test_watchdog.py tests/test_cgroup.py + - name: Run process-backend confinement tests + run: pytest -q tests/test_process_backend.py tests/test_process_confinement.py tests/test_landlock.py + - name: Run live Landlock enforcement (kernels that support it) + env: + PYISOLATE_LIVE_LANDLOCK_TESTS: "1" + run: pytest -q tests/test_landlock.py coverage: runs-on: ubuntu-24.04 diff --git a/pyisolate/runtime/child.py b/pyisolate/runtime/child.py index c9885d5..98b892f 100644 --- a/pyisolate/runtime/child.py +++ b/pyisolate/runtime/child.py @@ -194,7 +194,10 @@ def _serve(sock: socket.socket) -> None: report = apply_confinement( mem_bytes=bootstrap.get("mem_bytes"), cpu_seconds=bootstrap.get("cpu_seconds"), + fs_read=bootstrap.get("fs_read"), + fs_write=bootstrap.get("fs_write"), require_seccomp=bool(bootstrap.get("require_seccomp", False)), + require_landlock=bool(bootstrap.get("require_landlock", False)), ) _send_frame( sock, @@ -203,6 +206,8 @@ def _serve(sock: socket.socket) -> None: "seccomp": report.seccomp, "seccomp_denied": report.seccomp_denied, "rlimits": report.rlimits, + "landlock": report.landlock, + "landlock_rules": report.landlock_rules, "skipped": report.skipped, }, ) diff --git a/pyisolate/runtime/confine.py b/pyisolate/runtime/confine.py index 12b45c7..d2e4150 100644 --- a/pyisolate/runtime/confine.py +++ b/pyisolate/runtime/confine.py @@ -31,6 +31,8 @@ import sys from dataclasses import dataclass, field +from . import landlock as _landlock + # prctl options _PR_SET_NO_NEW_PRIVS = 38 _PR_SET_SECCOMP = 22 @@ -114,9 +116,21 @@ class ConfinementReport: seccomp: bool = False seccomp_denied: int = 0 rlimits: list[str] = field(default_factory=list) + landlock: bool = False + landlock_rules: int = 0 skipped: list[str] = field(default_factory=list) +def _set_no_new_privs() -> bool: + """Set ``PR_SET_NO_NEW_PRIVS`` on the current process. Returns success. + + Required both to install a seccomp filter and to call + ``landlock_restrict_self`` while unprivileged. + """ + libc = ctypes.CDLL(None, use_errno=True) + return libc.prctl(_PR_SET_NO_NEW_PRIVS, 1, 0, 0, 0) == 0 + + def seccomp_supported() -> bool: """Return whether the seccomp filter in this module can be installed here.""" return sys.platform.startswith("linux") and platform.machine() == "x86_64" @@ -201,22 +215,55 @@ def _apply_rlimits( report.skipped.append("rlimit_cpu") +def _apply_landlock( + report: ConfinementReport, + *, + fs_read: list[str] | None, + fs_write: list[str] | None, + require_landlock: bool, +) -> None: + # Only restrict the filesystem when the policy actually names paths; without + # an allow-list a default-deny ruleset would just break the interpreter. + if not fs_read and not fs_write: + return + landlock_report = _landlock.apply_landlock( + fs_read, fs_write, require=require_landlock + ) + if landlock_report.applied: + report.landlock = True + report.landlock_rules = landlock_report.rules + else: + report.skipped.append(f"landlock:{landlock_report.skipped}") + + def apply_confinement( *, mem_bytes: int | None = None, cpu_seconds: int | None = None, + fs_read: list[str] | None = None, + fs_write: list[str] | None = None, seccomp: bool = True, require_seccomp: bool = False, + require_landlock: bool = False, ) -> ConfinementReport: """Confine the *current* process before it runs guest code. - Resource limits are always attempted (best-effort, recorded in the report). - The seccomp filter is installed when ``seccomp`` is true and the platform - supports it; ``require_seccomp`` turns an unsupported platform or a failed - install into a raised error instead of a skip. + Order matters: ``no_new_privs`` is set first (required by both Landlock and + seccomp), then resource limits, then the Landlock filesystem ruleset, and + finally the seccomp filter as the last lockdown step. ``require_*`` turns an + unsupported platform or a failed install into a raised error instead of a + best-effort skip recorded in the report. """ report = ConfinementReport() + report.no_new_privs = _set_no_new_privs() + _apply_rlimits(report, mem_bytes=mem_bytes, cpu_seconds=cpu_seconds) + _apply_landlock( + report, + fs_read=fs_read, + fs_write=fs_write, + require_landlock=require_landlock, + ) if not seccomp: report.skipped.append("seccomp_disabled") diff --git a/pyisolate/runtime/landlock.py b/pyisolate/runtime/landlock.py new file mode 100644 index 0000000..abe7e7e --- /dev/null +++ b/pyisolate/runtime/landlock.py @@ -0,0 +1,247 @@ +"""Landlock filesystem confinement for ``backend="process"`` guest processes. + +Landlock lets an unprivileged process irrevocably restrict its own filesystem +access to an allow-list of path hierarchies, enforced by the kernel. Applied to +a guest process (which already has ``PR_SET_NO_NEW_PRIVS`` set by +:mod:`pyisolate.runtime.confine`), it turns the sandbox's filesystem policy into +a real kernel boundary: even guest code that defeats the Python ``open`` guard +and reaches the libc ``open``/``openat`` can only touch permitted paths. + +Landlock is default-deny for every access right the ruleset *handles*: once a +ruleset handling ``READ_FILE``/``WRITE_FILE``/... is applied, those actions are +denied everywhere except beneath the paths added as rules. So the guest must be +granted read+execute on the Python installation and system libraries it needs to +keep running, in addition to the policy's own read/write paths. + +The set of access rights and struct layout depend on the kernel's Landlock ABI +version, which this module queries and masks against; requesting an unsupported +right makes ``landlock_create_ruleset`` fail. Landlock is Linux-specific and +unavailable on many kernels (``landlock_create_ruleset`` returns ``ENOSYS``); +callers must treat :func:`landlock_supported` as authoritative. +""" + +from __future__ import annotations + +import ctypes +import os +import sys +from dataclasses import dataclass, field + +# Landlock syscall numbers (x86-64). +_NR_LANDLOCK_CREATE_RULESET = 444 +_NR_LANDLOCK_ADD_RULE = 445 +_NR_LANDLOCK_RESTRICT_SELF = 446 + +_LANDLOCK_CREATE_RULESET_VERSION = 1 << 0 +_LANDLOCK_RULE_PATH_BENEATH = 1 + +# Filesystem access rights (bit positions are a stable kernel ABI). +ACCESS_FS = { + "EXECUTE": 1 << 0, + "WRITE_FILE": 1 << 1, + "READ_FILE": 1 << 2, + "READ_DIR": 1 << 3, + "REMOVE_DIR": 1 << 4, + "REMOVE_FILE": 1 << 5, + "MAKE_CHAR": 1 << 6, + "MAKE_DIR": 1 << 7, + "MAKE_REG": 1 << 8, + "MAKE_SOCK": 1 << 9, + "MAKE_FIFO": 1 << 10, + "MAKE_BLOCK": 1 << 11, + "MAKE_SYM": 1 << 12, + "REFER": 1 << 13, # ABI >= 2 + "TRUNCATE": 1 << 14, # ABI >= 3 + "IOCTL_DEV": 1 << 15, # ABI >= 5 +} + +_READ = ACCESS_FS["READ_FILE"] | ACCESS_FS["READ_DIR"] +_READ_EXEC = _READ | ACCESS_FS["EXECUTE"] +_WRITE = ( + _READ + | ACCESS_FS["WRITE_FILE"] + | ACCESS_FS["MAKE_REG"] + | ACCESS_FS["MAKE_DIR"] + | ACCESS_FS["MAKE_SYM"] + | ACCESS_FS["REMOVE_FILE"] + | ACCESS_FS["REMOVE_DIR"] + | ACCESS_FS["TRUNCATE"] +) + + +class _RulesetAttr(ctypes.Structure): + _fields_ = [("handled_access_fs", ctypes.c_uint64)] + + +class _PathBeneathAttr(ctypes.Structure): + # ``struct landlock_path_beneath_attr`` is packed: no padding between the + # 8-byte access mask and the 4-byte fd. + _pack_ = 1 + _fields_ = [ + ("allowed_access", ctypes.c_uint64), + ("parent_fd", ctypes.c_int32), + ] + + +@dataclass +class LandlockReport: + """Outcome of applying Landlock to the current process.""" + + applied: bool = False + abi: int = 0 + rules: int = 0 + skipped: str | None = None + denied_paths: list[str] = field(default_factory=list) + + +def _libc() -> ctypes.CDLL: + libc = ctypes.CDLL(None, use_errno=True) + libc.syscall.restype = ctypes.c_long + return libc + + +def abi_version() -> int: + """Return the kernel's Landlock ABI version, or 0 if unavailable.""" + if not sys.platform.startswith("linux"): + return 0 + libc = _libc() + result = libc.syscall( + _NR_LANDLOCK_CREATE_RULESET, None, 0, _LANDLOCK_CREATE_RULESET_VERSION + ) + return int(result) if result > 0 else 0 + + +def landlock_supported() -> bool: + """Return whether Landlock filesystem rules can be applied on this host.""" + return abi_version() >= 1 + + +def handled_access_fs(abi: int) -> int: + """Return the set of FS access rights to handle at Landlock ABI *abi*. + + Requesting a right the running kernel does not support makes + ``landlock_create_ruleset`` fail, so the handled set is masked to the ABI. + """ + handled = 0 + for name, bit in ACCESS_FS.items(): + if name == "REFER" and abi < 2: + continue + if name == "TRUNCATE" and abi < 3: + continue + if name == "IOCTL_DEV" and abi < 5: + continue + handled |= bit + return handled + + +def _runtime_read_paths() -> list[str]: + """Paths the interpreter must keep reading/executing to run guest code. + + Without these, a Landlock ruleset that handles ``READ_FILE``/``EXECUTE`` + would stop the guest from importing the standard library or loading shared + objects, breaking the interpreter rather than the guest's intent. + """ + candidates = [ + sys.base_prefix, + sys.base_exec_prefix, + sys.prefix, + sys.exec_prefix, + "/usr/lib", + "/usr/lib64", + "/usr/local/lib", + "/lib", + "/lib64", + "/etc", # ld.so.cache, ssl certs, locale data + ] + candidates.extend(p for p in sys.path if p) + seen: dict[str, None] = {} + for path in candidates: + if path and os.path.exists(path): + seen.setdefault(os.path.realpath(path), None) + return list(seen) + + +def _add_path_rule( + libc: ctypes.CDLL, ruleset_fd: int, path: str, access: int, handled: int +) -> bool: + allowed = access & handled + if not allowed: + return False + try: + parent_fd = os.open(path, os.O_PATH | os.O_CLOEXEC) + except OSError: + return False + try: + attr = _PathBeneathAttr(allowed_access=allowed, parent_fd=parent_fd) + result = libc.syscall( + _NR_LANDLOCK_ADD_RULE, + ruleset_fd, + _LANDLOCK_RULE_PATH_BENEATH, + ctypes.byref(attr), + 0, + ) + finally: + os.close(parent_fd) + return result == 0 + + +def apply_landlock( + read_paths: list[str] | None, + write_paths: list[str] | None, + *, + require: bool = False, +) -> LandlockReport: + """Restrict the current process's filesystem access to the policy paths. + + ``read_paths`` are granted read access, ``write_paths`` read+write, and the + interpreter's runtime paths are granted read+execute so it can keep running. + Everything else that the ruleset handles is denied. On a kernel without + Landlock this is a no-op unless ``require`` is set, in which case it raises. + """ + report = LandlockReport() + abi = abi_version() + report.abi = abi + if abi < 1: + if require: + raise RuntimeError("Landlock confinement required but unsupported") + report.skipped = "unsupported" + return report + + handled = handled_access_fs(abi) + libc = _libc() + attr = _RulesetAttr(handled_access_fs=handled) + ruleset_fd = libc.syscall( + _NR_LANDLOCK_CREATE_RULESET, ctypes.byref(attr), ctypes.sizeof(attr), 0 + ) + if ruleset_fd < 0: + err = ctypes.get_errno() + if require: + raise OSError(err, "landlock_create_ruleset failed") + report.skipped = f"create_ruleset_failed:{err}" + return report + + try: + for path in _runtime_read_paths(): + if _add_path_rule(libc, ruleset_fd, path, _READ_EXEC, handled): + report.rules += 1 + for path in read_paths or []: + if _add_path_rule(libc, ruleset_fd, path, _READ, handled): + report.rules += 1 + for path in write_paths or []: + if _add_path_rule(libc, ruleset_fd, path, _WRITE, handled): + report.rules += 1 + + # PR_SET_NO_NEW_PRIVS is set by apply_confinement before this runs, which + # landlock_restrict_self requires when unprivileged. + result = libc.syscall(_NR_LANDLOCK_RESTRICT_SELF, ruleset_fd, 0) + if result != 0: + err = ctypes.get_errno() + if require: + raise OSError(err, "landlock_restrict_self failed") + report.skipped = f"restrict_self_failed:{err}" + return report + finally: + os.close(ruleset_fd) + + report.applied = True + return report diff --git a/pyisolate/runtime/process_backend.py b/pyisolate/runtime/process_backend.py index 8381b70..52f9720 100644 --- a/pyisolate/runtime/process_backend.py +++ b/pyisolate/runtime/process_backend.py @@ -27,6 +27,7 @@ from typing import Any, Optional from .. import errors +from ..policy.model import RuntimePolicy from .thread import Stats _LEN = struct.Struct("!I") @@ -52,15 +53,15 @@ def _extract_fs_tcp(policy: Any) -> tuple[Optional[list[str]], Optional[list[str fs: Optional[list[str]] = None tcp: Optional[list[str]] = None - allow_fs = getattr(policy, "allow_fs", None) - allow_tcp = getattr(policy, "allow_tcp", None) - if allow_fs is not None or allow_tcp is not None: - if allow_fs: - fs = [rule.path for rule in allow_fs] - if allow_tcp: - tcp = [rule.destination for rule in allow_tcp] + if isinstance(policy, RuntimePolicy): + if policy.allow_fs: + fs = [rule.path for rule in policy.allow_fs] + if policy.allow_tcp: + tcp = [rule.destination for rule in policy.allow_tcp] return fs, tcp + # Legacy Policy: ``allow_fs``/``allow_tcp`` are *methods*, not rule + # collections, so read the plain ``.fs``/``.tcp`` string attributes. p_fs = getattr(policy, "fs", None) p_tcp = getattr(policy, "tcp", None) if p_fs: @@ -70,6 +71,54 @@ def _extract_fs_tcp(policy: Any) -> tuple[Optional[list[str]], Optional[list[str return fs, tcp +def _extract_fs_read_write( + policy: Any, +) -> tuple[Optional[list[str]], Optional[list[str]]]: + """Split policy filesystem paths into read-only and writable sets. + + Used to build a Landlock ruleset that grants read on read paths and + read+write on write paths. Handles the compiled ``RuntimePolicy`` + (``allow_fs`` rules with an ``access`` mode) and the legacy ``Policy`` + (``ReadPath``/``WritePath`` capabilities plus ``.fs`` read+write paths). + """ + if policy is None: + return None, None + + read: list[str] = [] + write: list[str] = [] + + if isinstance(policy, RuntimePolicy): + for rule in policy.allow_fs: + if getattr(rule, "access", "readwrite") == "read": + read.append(rule.path) + else: + write.append(rule.path) + return _dedupe_read_write(read, write) + + for cap in getattr(policy, "capabilities", None) or []: + path = getattr(cap, "path", None) + if path is None: + continue + if getattr(cap, "kind", None) == "write_path": + write.append(str(path)) + elif getattr(cap, "kind", None) == "read_path": + read.append(str(path)) + for item in getattr(policy, "fs", None) or []: + write.append(str(item)) + return _dedupe_read_write(read, write) + + +def _dedupe_read_write( + read: list[str], write: list[str] +) -> tuple[Optional[list[str]], Optional[list[str]]]: + # A write grant already implies read, so drop any read path that is also + # writable and de-duplicate each set (order-preserving). + write_unique = list(dict.fromkeys(write)) + write_set = set(write_unique) + read_unique = [p for p in dict.fromkeys(read) if p not in write_set] + return (read_unique or None, write_unique or None) + + class ProcessSandbox: """Runs guest code in a confined child process behind a JSON channel.""" @@ -84,6 +133,7 @@ def __init__( cpu_seconds: Optional[int] = None, confine: bool = True, require_seccomp: bool = False, + require_landlock: bool = False, ) -> None: self.name = name self._backend = backend @@ -109,6 +159,7 @@ def __init__( self._confined = threading.Event() fs, tcp = _extract_fs_tcp(policy) + fs_read, fs_write = _extract_fs_read_write(policy) parent_sock, child_sock = socket.socketpair(socket.AF_UNIX, socket.SOCK_STREAM) try: @@ -133,10 +184,13 @@ def __init__( "allowed_imports": allowed_imports, "fs": fs, "tcp": tcp, + "fs_read": fs_read, + "fs_write": fs_write, "confine": confine, "mem_bytes": mem_bytes, "cpu_seconds": cpu_seconds, "require_seccomp": require_seccomp, + "require_landlock": require_landlock, } ) diff --git a/pyisolate/supervisor.py b/pyisolate/supervisor.py index 578c952..a6cdd19 100644 --- a/pyisolate/supervisor.py +++ b/pyisolate/supervisor.py @@ -489,6 +489,7 @@ def _spawn_process( backend="process", mem_bytes=mem_bytes, require_seccomp=self._rollout_mode == "hardened", + require_landlock=self._rollout_mode == "hardened", ) except Exception: if usage_reserved and tenant: diff --git a/tests/test_landlock.py b/tests/test_landlock.py new file mode 100644 index 0000000..850dd7b --- /dev/null +++ b/tests/test_landlock.py @@ -0,0 +1,145 @@ +"""Tests for Landlock filesystem confinement of the process backend. + +Landlock enforcement is unavailable on many kernels (including where this +suite typically runs), so the end-to-end enforcement test is gated on +``landlock_supported()`` plus an opt-in env flag, mirroring the live BPF +kernel-enforcement test. The mechanism-level logic that does not require the +kernel is tested unconditionally. +""" + +import os +import sys +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT)) + +import pytest + +import pyisolate as iso +from pyisolate.runtime import landlock +from pyisolate.runtime.process_backend import ( + _extract_fs_read_write, + _extract_fs_tcp, +) + +requires_landlock = pytest.mark.skipif( + not landlock.landlock_supported(), + reason="kernel does not support Landlock", +) + +live_landlock = pytest.mark.skipif( + not ( + landlock.landlock_supported() + and os.environ.get("PYISOLATE_LIVE_LANDLOCK_TESTS") == "1" + ), + reason="live Landlock enforcement test requires Landlock and " + "PYISOLATE_LIVE_LANDLOCK_TESTS=1", +) + + +def test_handled_access_fs_is_masked_by_abi(): + # Newer access rights must only be requested on ABIs that support them, + # otherwise landlock_create_ruleset rejects the ruleset. + abi1 = landlock.handled_access_fs(1) + abi2 = landlock.handled_access_fs(2) + abi3 = landlock.handled_access_fs(3) + abi5 = landlock.handled_access_fs(5) + assert not abi1 & landlock.ACCESS_FS["REFER"] + assert abi2 & landlock.ACCESS_FS["REFER"] + assert not abi2 & landlock.ACCESS_FS["TRUNCATE"] + assert abi3 & landlock.ACCESS_FS["TRUNCATE"] + assert not abi3 & landlock.ACCESS_FS["IOCTL_DEV"] + assert abi5 & landlock.ACCESS_FS["IOCTL_DEV"] + # Read/write/execute are available from ABI 1. + for name in ("READ_FILE", "WRITE_FILE", "EXECUTE", "READ_DIR"): + assert abi1 & landlock.ACCESS_FS[name] + + +def test_path_beneath_attr_is_packed_to_twelve_bytes(): + # struct landlock_path_beneath_attr is packed (8-byte access + 4-byte fd); + # any padding would misalign the fd and make the kernel reject the rule. + import ctypes + + assert ctypes.sizeof(landlock._PathBeneathAttr) == 12 + + +def test_abi_and_support_are_consistent(): + assert landlock.landlock_supported() == (landlock.abi_version() >= 1) + + +def test_extract_fs_read_write_from_legacy_policy_dedupes(): + policy = iso.policy.Policy().allow_fs("/data").allow_read("/etc/hostname") + read, write = _extract_fs_read_write(policy) + # allow_fs grants read+write, so /data is writable only; the explicit + # read path stays read-only and nothing is duplicated across the sets. + assert write == ["/data"] + assert read == ["/etc/hostname"] + + +def test_extract_fs_tcp_handles_legacy_policy_methods(): + # Regression: Policy.allow_fs/allow_tcp are methods, not rule collections; + # the extractor must read the .fs/.tcp attributes instead of iterating them. + policy = iso.policy.Policy().allow_fs("/data").allow_tcp("127.0.0.1:80") + fs, tcp = _extract_fs_tcp(policy) + assert fs == ["/data"] + assert tcp == ["127.0.0.1:80"] + + +def test_apply_landlock_is_a_noop_when_unsupported(): + if landlock.landlock_supported(): + pytest.skip("kernel supports Landlock; skip the unsupported-path check") + report = landlock.apply_landlock(["/tmp"], None) + assert report.applied is False + assert report.skipped == "unsupported" + + +def test_apply_landlock_requires_support_when_required(): + if landlock.landlock_supported(): + pytest.skip("kernel supports Landlock; skip the required-failure check") + with pytest.raises(RuntimeError): + landlock.apply_landlock(["/tmp"], None, require=True) + + +@requires_landlock +def test_process_sandbox_reports_landlock_applied(tmp_path): + allowed = tmp_path / "allowed" + allowed.mkdir() + policy = iso.policy.Policy().allow_fs(str(allowed)) + with iso.spawn("ll-applied", policy=policy, backend="process") as sb: + report = sb._thread.wait_confined(timeout=5) + assert report is not None + assert report["landlock"] is True + assert report["landlock_rules"] >= 1 + + +# Recover the real ``open`` from a stdlib module's globals, bypassing the +# process backend's blocked-open guard, so the test exercises Landlock (the +# kernel layer) rather than the Python guard. +_REAL_READ = """ +def _real_open(path): + for cls in ().__class__.__base__.__subclasses__(): + if cls.__name__ == "catch_warnings": + return cls()._module.__builtins__["open"](path) + raise RuntimeError("no real open") +post(_real_open({path!r}).read()) +""" + + +@live_landlock +def test_landlock_blocks_disallowed_reads_but_allows_permitted(tmp_path): + allowed = tmp_path / "allowed" + allowed.mkdir() + (allowed / "ok.txt").write_text("ok", encoding="utf-8") + secret = tmp_path / "secret.txt" + secret.write_text("secret", encoding="utf-8") + + policy = iso.policy.Policy().allow_fs(str(allowed)) + with iso.spawn("ll-enforce", policy=policy, backend="process") as sb: + # Reading an allowed file via the real open is permitted by Landlock. + sb.exec(_REAL_READ.format(path=str(allowed / "ok.txt"))) + assert sb.recv(timeout=5) == "ok" + # Reading outside the allow-list is denied at the kernel level. + sb.exec(_REAL_READ.format(path=str(secret))) + with pytest.raises(iso.SandboxError): + sb.recv(timeout=5)