diff --git a/README.md b/README.md index de12fc5..03b1c7a 100644 --- a/README.md +++ b/README.md @@ -234,7 +234,9 @@ See [docs/execution-model.md](docs/execution-model.md). We keep this model small * `PR_SET_NO_NEW_PRIVS` + a seccomp deny-list that kills the process on dangerous syscalls (`execve`, `ptrace`, mount/namespace ops, `bpf`, module load, `process_vm_*`, ...) - x86-64 Linux; - * **Landlock** filesystem rules from policy, where the kernel supports it; + * **Landlock** filesystem rules from policy, plus **Landlock TCP-egress** + rules (ABI ≥ 4) that deny `connect()` to any port outside the policy's + allow-list, where the kernel supports it; * a coarse per-cgroup **eBPF/LSM** `deny_mask`, where BPF-LSM is available; * `rlimit` and cgroup resource caps. Each kernel layer is best-effort and recorded in the sandbox's confinement diff --git a/SECURITY.md b/SECURITY.md index 73422ee..8700c02 100644 --- a/SECURITY.md +++ b/SECURITY.md @@ -48,16 +48,21 @@ silently. deny-list, **not** a proof that only a fixed syscall allow-list is reachable. 3. **Filesystem policy** — Landlock confines the guest to the policy's read/write paths, on kernels that support Landlock. -4. **Coarse capability gating** — A per-cgroup eBPF/LSM `deny_mask` denies whole +4. **Network-egress policy** — On Landlock ABI >= 4 (Linux 6.7+) the policy's TCP + allow-list is mapped to allowed `connect()` ports and the kernel denies egress + to every other port. Landlock keys network rules on port, not address, so this + is a coarse kernel backstop beneath the userspace host:port guard; it is + applied only when every allow-listed destination carries a parseable port. +5. **Coarse capability gating** — A per-cgroup eBPF/LSM `deny_mask` denies whole capability classes (process creation, ptrace/mount/bpf, and filesystem or network when the policy grants nothing in that class), on kernels with BPF-LSM. This is coarse: it cannot express per-path allow-lists — that is Landlock's job (§2.3) and the broker's. -5. **Broker integrity & replay protection** — Control-plane frames are AEAD-sealed +6. **Broker integrity & replay protection** — Control-plane frames are AEAD-sealed (X25519 → HKDF-SHA-256 → ChaCha20-Poly1305, optional Kyber-768 hybrid) with a strictly increasing per-direction counter; forged, reordered, or replayed frames are rejected. -6. **Crash isolation** — A crash in a guest process cannot bring down the +7. **Crash isolation** — A crash in a guest process cannot bring down the supervisor. Resource quotas (CPU/RAM/I/O) are enforced by `rlimit` and cgroup v2 controls @@ -79,7 +84,8 @@ code. Confinement is applied before any guest code runs, in order: * **seccomp** deny-list — kills the process on high-risk syscalls; inherited across `fork`/`clone` and irremovable once `no_new_privs` is set. -* **Landlock** — filesystem access restricted to the policy's paths (kernels +* **Landlock** — filesystem access restricted to the policy's paths, and (ABI + >= 4) TCP `connect()` restricted to the policy's allow-listed ports (kernels with Landlock support). * **eBPF/LSM** — a per-cgroup `deny_mask` on `file_open`, `socket_connect`, `task_alloc`, `bprm_check_security`, `ptrace`, `sb_mount`, and `bpf` hooks, diff --git a/docs/threat-model.md b/docs/threat-model.md index b8790a4..3a9fb53 100644 --- a/docs/threat-model.md +++ b/docs/threat-model.md @@ -29,7 +29,8 @@ answer below as conditional on the backend you select. dangerous syscalls (`execve`, `ptrace`, mount/namespace ops, `bpf`, kernel module load, `process_vm_*`, …) — x86-64 Linux; - **Landlock** filesystem rules derived from policy, where the kernel supports - it; + it, plus **Landlock TCP-egress** rules that allow-list the connect ports in + the policy (Landlock ABI >= 4 / Linux 6.7+; keyed on port, not address); - a **coarse per-cgroup eBPF/LSM deny-mask** (deny whole capability classes), where BPF-LSM is available; - `rlimit` and cgroup resource caps. @@ -60,6 +61,9 @@ high-assurance multitenancy, run one sandbox per process inside a VM or microVM. - process isolation from the supervisor's address space, - a seccomp deny-list that kills the process on dangerous syscalls, - Landlock filesystem rules from policy (where supported), + - Landlock TCP-egress rules that deny connect() to any port the policy did + not allow-list (ABI >= 4; port-granular, a coarse backstop beneath the + userspace host:port guard), - a coarse per-cgroup eBPF/LSM deny-mask (where BPF-LSM is available), - broker-mediated privileged operations and per-sandbox resource limits. @@ -196,6 +200,11 @@ Any semantic change to defended/not-defended status requires: ### History +- **2026-07-21** — Added a Landlock TCP-egress layer to `backend="process"`: + where the kernel's Landlock ABI is >= 4, the policy's TCP allow-list is mapped + to allowed connect ports and the kernel denies egress to every other port, + backstopping the userspace host:port guard. Recorded in the confinement report + (`landlock_net` / `landlock_net_ports`). - **2026-07-20** — Made the hostile-Python answer backend-conditional. Landed a real `backend="process"` boundary: separate-process isolation, a seccomp deny-list with `no_new_privs`, Landlock filesystem enforcement from policy, diff --git a/pyisolate/runtime/child.py b/pyisolate/runtime/child.py index 98b892f..38fa7a8 100644 --- a/pyisolate/runtime/child.py +++ b/pyisolate/runtime/child.py @@ -42,6 +42,7 @@ from typing import Any from .. import errors +from . import landlock as _landlock from .confine import apply_confinement from .thread import _SAFE_BUILTINS, _blocked_open, _make_importer, _thread_local @@ -110,6 +111,24 @@ def request(self, capability: str, action: str, payload: Any = None) -> None: ) +def _net_connect_ports(tcp: list[str] | None) -> list[int] | None: + """Derive the Landlock TCP-egress allow-list from the policy destinations. + + Returns the list of allowed connect ports, or ``None`` to leave network + egress unrestricted by Landlock. Egress is restricted only when the policy + names a TCP allow-list *and* every destination carries a parseable port: + Landlock keys on port and is default-deny for ports it does not name, so an + allow-list we cannot represent exactly is left to the userspace host:port + guard rather than silently blocking a permitted destination. + """ + if not tcp: + return None + ports, exact = _landlock.connect_ports_from_destinations(tcp) + if exact and ports: + return ports + return None + + def _install_guest_context( *, allowed_imports: list[str] | None, @@ -196,6 +215,7 @@ def _serve(sock: socket.socket) -> None: cpu_seconds=bootstrap.get("cpu_seconds"), fs_read=bootstrap.get("fs_read"), fs_write=bootstrap.get("fs_write"), + net_connect_ports=_net_connect_ports(bootstrap.get("tcp")), require_seccomp=bool(bootstrap.get("require_seccomp", False)), require_landlock=bool(bootstrap.get("require_landlock", False)), ) @@ -208,6 +228,8 @@ def _serve(sock: socket.socket) -> None: "rlimits": report.rlimits, "landlock": report.landlock, "landlock_rules": report.landlock_rules, + "landlock_net": report.landlock_net, + "landlock_net_ports": report.landlock_net_ports, "skipped": report.skipped, }, ) diff --git a/pyisolate/runtime/confine.py b/pyisolate/runtime/confine.py index d2e4150..47a0263 100644 --- a/pyisolate/runtime/confine.py +++ b/pyisolate/runtime/confine.py @@ -118,6 +118,8 @@ class ConfinementReport: rlimits: list[str] = field(default_factory=list) landlock: bool = False landlock_rules: int = 0 + landlock_net: bool = False + landlock_net_ports: int = 0 skipped: list[str] = field(default_factory=list) @@ -220,20 +222,31 @@ def _apply_landlock( *, fs_read: list[str] | None, fs_write: list[str] | None, + net_connect_ports: list[int] | 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: + # Only handle an access class when the policy actually names an allow-list; + # without one a default-deny ruleset would break the interpreter (FS) or + # sever egress the policy meant to permit (network). + if not fs_read and not fs_write and not net_connect_ports: return landlock_report = _landlock.apply_landlock( - fs_read, fs_write, require=require_landlock + fs_read, + fs_write, + connect_ports=net_connect_ports, + require=require_landlock, ) if landlock_report.applied: report.landlock = True report.landlock_rules = landlock_report.rules - else: + elif fs_read or fs_write: report.skipped.append(f"landlock:{landlock_report.skipped}") + if landlock_report.net_applied: + report.landlock_net = True + report.landlock_net_ports = landlock_report.net_rules + elif net_connect_ports: + reason = landlock_report.net_skipped or landlock_report.skipped + report.skipped.append(f"landlock_net:{reason}") def apply_confinement( @@ -242,6 +255,7 @@ def apply_confinement( cpu_seconds: int | None = None, fs_read: list[str] | None = None, fs_write: list[str] | None = None, + net_connect_ports: list[int] | None = None, seccomp: bool = True, require_seccomp: bool = False, require_landlock: bool = False, @@ -249,8 +263,8 @@ def apply_confinement( """Confine the *current* process before it runs guest code. 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 + seccomp), then resource limits, then the Landlock filesystem/TCP-egress + 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. """ @@ -262,6 +276,7 @@ def apply_confinement( report, fs_read=fs_read, fs_write=fs_write, + net_connect_ports=net_connect_ports, require_landlock=require_landlock, ) diff --git a/pyisolate/runtime/landlock.py b/pyisolate/runtime/landlock.py index abe7e7e..cc4a390 100644 --- a/pyisolate/runtime/landlock.py +++ b/pyisolate/runtime/landlock.py @@ -1,4 +1,4 @@ -"""Landlock filesystem confinement for ``backend="process"`` guest processes. +"""Landlock filesystem and TCP-egress confinement for ``backend="process"``. Landlock lets an unprivileged process irrevocably restrict its own filesystem access to an allow-list of path hierarchies, enforced by the kernel. Applied to @@ -7,6 +7,14 @@ a real kernel boundary: even guest code that defeats the Python ``open`` guard and reaches the libc ``open``/``openat`` can only touch permitted paths. +From Landlock ABI 4 (Linux 6.7) the same mechanism can allow-list the TCP ports +a process may ``connect()`` to. PyIsolate maps the policy's TCP allow-list onto +that layer so network egress is denied by the kernel to any port the policy did +not permit, even if guest code walks ``object.__subclasses__()`` to a raw +socket. Landlock's network rules are keyed on port, not address, so this is a +coarse kernel backstop beneath the userspace host:port guard, not a replacement +for it. + 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 @@ -34,6 +42,12 @@ _LANDLOCK_CREATE_RULESET_VERSION = 1 << 0 _LANDLOCK_RULE_PATH_BENEATH = 1 +_LANDLOCK_RULE_NET_PORT = 2 + +# Landlock ABI that introduced TCP network rules (Linux 6.7). Below it the +# kernel has no notion of ``handled_access_net`` and network egress cannot be +# confined by Landlock. +_NET_ABI = 4 # Filesystem access rights (bit positions are a stable kernel ABI). ACCESS_FS = { @@ -55,6 +69,12 @@ "IOCTL_DEV": 1 << 15, # ABI >= 5 } +# TCP network access rights (ABI >= 4). Bit positions are a stable kernel ABI. +ACCESS_NET = { + "BIND_TCP": 1 << 0, + "CONNECT_TCP": 1 << 1, +} + _READ = ACCESS_FS["READ_FILE"] | ACCESS_FS["READ_DIR"] _READ_EXEC = _READ | ACCESS_FS["EXECUTE"] _WRITE = ( @@ -73,6 +93,25 @@ class _RulesetAttr(ctypes.Structure): _fields_ = [("handled_access_fs", ctypes.c_uint64)] +class _RulesetAttrNet(ctypes.Structure): + # ``struct landlock_ruleset_attr`` grew ``handled_access_net`` at ABI 4. + # Passing a struct sized to include it is only valid on a kernel that knows + # the field; on older kernels the fs-only :class:`_RulesetAttr` is used. + _fields_ = [ + ("handled_access_fs", ctypes.c_uint64), + ("handled_access_net", ctypes.c_uint64), + ] + + +class _NetPortAttr(ctypes.Structure): + # ``struct landlock_net_port_attr { __u64 allowed_access; __u64 port; }`` — + # both members are naturally aligned u64s, so no packing is required. + _fields_ = [ + ("allowed_access", ctypes.c_uint64), + ("port", 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. @@ -92,6 +131,10 @@ class LandlockReport: rules: int = 0 skipped: str | None = None denied_paths: list[str] = field(default_factory=list) + net_applied: bool = False + net_rules: int = 0 + allowed_ports: list[int] = field(default_factory=list) + net_skipped: str | None = None def _libc() -> ctypes.CDLL: @@ -134,6 +177,68 @@ def handled_access_fs(abi: int) -> int: return handled +def net_supported() -> bool: + """Return whether Landlock TCP network rules can be applied on this host.""" + return abi_version() >= _NET_ABI + + +def handled_access_net(abi: int) -> int: + """Return the set of TCP network rights to handle at Landlock ABI *abi*. + + Only ``CONNECT_TCP`` is handled: this layer confines outbound egress. TCP + ``bind`` (listening) is out of scope for the network-egress boundary and is + left to seccomp/policy. Returns 0 below the ABI that introduced net rules. + """ + if abi < _NET_ABI: + return 0 + return ACCESS_NET["CONNECT_TCP"] + + +def _parse_port(destination: str) -> int | None: + """Extract the TCP port from a ``"host:port"`` allow-list destination. + + Returns ``None`` when the destination has no parseable ``:port`` suffix in + the valid 1-65535 range (for example a bare hostname, or an unbracketed + IPv6 literal). Landlock can only allow-list ports, so such a destination + cannot be represented and the caller must decide how to degrade. + """ + _, sep, tail = destination.rpartition(":") + if not sep: + return None + try: + port = int(tail) + except ValueError: + return None + if 1 <= port <= 65535: + return port + return None + + +def connect_ports_from_destinations( + destinations: list[str], +) -> tuple[list[int], bool]: + """Turn a TCP allow-list into a de-duplicated set of connect ports. + + Returns ``(ports, exact)``. ``exact`` is ``False`` when any destination + lacked a parseable port; a network Landlock ruleset is default-deny for + every port it does not name, so an inexact allow-list would silently block a + destination the policy actually permits. Callers should skip network + Landlock in that case and rely on the userspace guard instead. + """ + ports: list[int] = [] + seen: set[int] = set() + exact = True + for dest in destinations: + port = _parse_port(dest) + if port is None: + exact = False + continue + if port not in seen: + seen.add(port) + ports.append(port) + return ports, exact + + def _runtime_read_paths() -> list[str]: """Paths the interpreter must keep reading/executing to run guest code. @@ -185,18 +290,74 @@ def _add_path_rule( return result == 0 +def _add_net_port_rule( + libc: ctypes.CDLL, ruleset_fd: int, port: int, access: int +) -> bool: + attr = _NetPortAttr(allowed_access=access, port=port) + result = libc.syscall( + _NR_LANDLOCK_ADD_RULE, + ruleset_fd, + _LANDLOCK_RULE_NET_PORT, + ctypes.byref(attr), + 0, + ) + return result == 0 + + +def _populate_fs_rules( + libc: ctypes.CDLL, + ruleset_fd: int, + read_paths: list[str] | None, + write_paths: list[str] | None, + handled_fs: int, + report: LandlockReport, +) -> None: + """Add the interpreter, read, and write path rules to *ruleset_fd*.""" + for path in _runtime_read_paths(): + if _add_path_rule(libc, ruleset_fd, path, _READ_EXEC, handled_fs): + report.rules += 1 + for path in read_paths or []: + if _add_path_rule(libc, ruleset_fd, path, _READ, handled_fs): + report.rules += 1 + for path in write_paths or []: + if _add_path_rule(libc, ruleset_fd, path, _WRITE, handled_fs): + report.rules += 1 + + +def _populate_net_rules( + libc: ctypes.CDLL, + ruleset_fd: int, + connect_ports: list[int], + report: LandlockReport, +) -> None: + """Add a CONNECT_TCP allow rule for each policy port to *ruleset_fd*.""" + for port in connect_ports: + if _add_net_port_rule(libc, ruleset_fd, port, ACCESS_NET["CONNECT_TCP"]): + report.net_rules += 1 + report.allowed_ports.append(port) + + def apply_landlock( read_paths: list[str] | None, write_paths: list[str] | None, *, + connect_ports: list[int] | None = None, require: bool = False, ) -> LandlockReport: - """Restrict the current process's filesystem access to the policy paths. + """Restrict the current process's filesystem and TCP-egress access to policy. ``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. + ``connect_ports`` (Landlock ABI >= 4, Linux 6.7+) allow-lists the TCP ports + the guest may ``connect()`` to; every other port is denied by the kernel. + Both layers share a single ruleset. A layer whose allow-list is empty/None + is not handled at all, so a default-deny ruleset never breaks the + interpreter or blocks egress the policy did not mean to restrict. + + On a kernel without Landlock this is a no-op unless ``require`` is set, in + which case it raises. When network confinement is requested but the kernel's + Landlock ABI predates net rules, the filesystem layer is still applied and + the network layer is recorded as skipped (or raises under ``require``). """ report = LandlockReport() abi = abi_version() @@ -207,9 +368,33 @@ def apply_landlock( report.skipped = "unsupported" return report - handled = handled_access_fs(abi) + handle_fs = bool(read_paths or write_paths) + want_net = connect_ports is not None + handle_net = want_net and abi >= _NET_ABI + if want_net and not handle_net: + if require: + raise RuntimeError( + "Landlock network confinement required but unsupported " + f"(ABI {abi} < {_NET_ABI})" + ) + report.net_skipped = f"net_unsupported_abi:{abi}" + + if not handle_fs and not handle_net: + # Nothing to restrict. A ruleset that handled an access class with no + # allow-list rules would be default-deny and break the guest. + report.skipped = "no_rules" + return report + + handled_fs = handled_access_fs(abi) if handle_fs else 0 + handled_net = handled_access_net(abi) if handle_net else 0 + libc = _libc() - attr = _RulesetAttr(handled_access_fs=handled) + if handled_net: + attr: ctypes.Structure = _RulesetAttrNet( + handled_access_fs=handled_fs, handled_access_net=handled_net + ) + else: + attr = _RulesetAttr(handled_access_fs=handled_fs) ruleset_fd = libc.syscall( _NR_LANDLOCK_CREATE_RULESET, ctypes.byref(attr), ctypes.sizeof(attr), 0 ) @@ -221,15 +406,12 @@ def apply_landlock( 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 + if handle_fs: + _populate_fs_rules( + libc, ruleset_fd, read_paths, write_paths, handled_fs, report + ) + if handle_net: + _populate_net_rules(libc, ruleset_fd, connect_ports or [], report) # PR_SET_NO_NEW_PRIVS is set by apply_confinement before this runs, which # landlock_restrict_self requires when unprivileged. @@ -243,5 +425,6 @@ def apply_landlock( finally: os.close(ruleset_fd) - report.applied = True + report.applied = handle_fs + report.net_applied = handle_net return report diff --git a/tests/test_landlock.py b/tests/test_landlock.py index 850dd7b..c4d99eb 100644 --- a/tests/test_landlock.py +++ b/tests/test_landlock.py @@ -18,6 +18,7 @@ import pyisolate as iso from pyisolate.runtime import landlock +from pyisolate.runtime.child import _net_connect_ports from pyisolate.runtime.process_backend import ( _extract_fs_read_write, _extract_fs_tcp, @@ -28,6 +29,20 @@ reason="kernel does not support Landlock", ) +requires_landlock_net = pytest.mark.skipif( + not landlock.net_supported(), + reason="kernel does not support Landlock network rules (ABI < 4)", +) + +live_landlock_net = pytest.mark.skipif( + not ( + landlock.net_supported() + and os.environ.get("PYISOLATE_LIVE_LANDLOCK_TESTS") == "1" + ), + reason="live Landlock network test requires ABI >= 4 and " + "PYISOLATE_LIVE_LANDLOCK_TESTS=1", +) + live_landlock = pytest.mark.skipif( not ( landlock.landlock_supported() @@ -126,6 +141,134 @@ def _real_open(path): """ +def test_handled_access_net_is_masked_by_abi(): + # TCP network rules only exist from ABI 4; requesting them earlier makes + # landlock_create_ruleset reject the ruleset. + assert landlock.handled_access_net(3) == 0 + assert landlock.handled_access_net(4) == landlock.ACCESS_NET["CONNECT_TCP"] + assert landlock.handled_access_net(6) & landlock.ACCESS_NET["CONNECT_TCP"] + # Only egress (connect) is confined by this layer; bind is out of scope. + assert not landlock.handled_access_net(4) & landlock.ACCESS_NET["BIND_TCP"] + + +def test_net_port_attr_and_ruleset_attr_net_layout(): + import ctypes + + # struct landlock_net_port_attr is two u64s (allowed_access, port). + assert ctypes.sizeof(landlock._NetPortAttr) == 16 + # struct landlock_ruleset_attr with handled_access_net is two u64s; passing + # this larger struct is only valid on an ABI >= 4 kernel. + assert ctypes.sizeof(landlock._RulesetAttrNet) == 16 + + +def test_net_supported_matches_abi(): + assert landlock.net_supported() == (landlock.abi_version() >= 4) + + +def test_connect_ports_from_destinations_parses_and_dedupes(): + ports, exact = landlock.connect_ports_from_destinations( + ["1.2.3.4:443", "example.com:80", "10.0.0.1:443"] + ) + # 443 appears twice but is de-duplicated; order is preserved. + assert ports == [443, 80] + assert exact is True + + +def test_connect_ports_from_destinations_flags_unparseable(): + # A bare hostname has no port to allow-list, so the result is inexact and + # the caller must not build a default-deny network ruleset from it. + ports, exact = landlock.connect_ports_from_destinations( + ["example.com", "1.2.3.4:443"] + ) + assert ports == [443] + assert exact is False + + +def test_connect_ports_from_destinations_rejects_out_of_range(): + ports, exact = landlock.connect_ports_from_destinations(["h:0", "h:70000"]) + assert ports == [] + assert exact is False + + +def test_net_connect_ports_helper_gates_on_exactness(): + # No allow-list -> no network Landlock at all. + assert _net_connect_ports(None) is None + assert _net_connect_ports([]) is None + # A fully-parseable allow-list yields the port set. + assert _net_connect_ports(["1.2.3.4:443", "h:80"]) == [443, 80] + # Any unrepresentable entry degrades to the userspace guard (None), rather + # than a kernel rule that would block the permitted-but-portless entry. + assert _net_connect_ports(["example.com", "1.2.3.4:443"]) is None + + +def test_apply_landlock_net_required_but_unsupported_raises(): + if landlock.net_supported(): + pytest.skip("kernel supports Landlock net; skip the required-failure check") + with pytest.raises(RuntimeError): + landlock.apply_landlock(None, None, connect_ports=[443], require=True) + + +@requires_landlock_net +def test_process_sandbox_reports_landlock_net_applied(): + policy = iso.policy.Policy().allow_tcp("127.0.0.1:9") + with iso.spawn("ll-net-applied", policy=policy, backend="process") as sb: + report = sb._thread.wait_confined(timeout=5) + assert report is not None + assert report["landlock_net"] is True + assert report["landlock_net_ports"] >= 1 + + +# Recover a real socket and attempt a TCP connect, bypassing the userspace +# network guard, so the test exercises Landlock (the kernel layer) rather than +# the Python guard. +_REAL_CONNECT = """ +import socket +def _connect(host, port): + s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + s.settimeout(2) + try: + s.connect((host, port)) + finally: + s.close() +_connect({host!r}, {port!r}) +post("connected") +""" + + +@live_landlock_net +def test_landlock_blocks_disallowed_ports_but_allows_permitted(): + import socket as _socket + import threading + + server = _socket.socket(_socket.AF_INET, _socket.SOCK_STREAM) + server.bind(("127.0.0.1", 0)) + server.listen(1) + allowed_port = server.getsockname()[1] + + def _accept_loop(): + while True: + try: + conn, _ = server.accept() + except OSError: + return + conn.close() + + threading.Thread(target=_accept_loop, daemon=True).start() + + policy = iso.policy.Policy().allow_tcp(f"127.0.0.1:{allowed_port}") + with iso.spawn( + "ll-net-enforce", policy=policy, backend="process", allowed_imports=["socket"] + ) as sb: + # Connecting to the allow-listed port is permitted by Landlock. + sb.exec(_REAL_CONNECT.format(host="127.0.0.1", port=allowed_port)) + assert sb.recv(timeout=5) == "connected" + # A port the policy never named is denied at the kernel level. + sb.exec(_REAL_CONNECT.format(host="127.0.0.1", port=allowed_port + 1)) + with pytest.raises(iso.SandboxError): + sb.recv(timeout=5) + server.close() + + @live_landlock def test_landlock_blocks_disallowed_reads_but_allows_permitted(tmp_path): allowed = tmp_path / "allowed"