Filter /proc readdir to hide foreign PIDs via getdents64 interception

Signed-off-by: Cong Wang <cwang@multikernel.io>

Author: congwang-mk

src/sandlock/_context.py

@@ -80,7 +80,7 @@ def _pidfd_poll(pidfd: int, timeout_s: float) -> bool:
 
 # --- Syscalls to intercept for notification ---
 
-def _notif_syscall_names(policy: "Policy") -> list[str]:
+def _notif_syscall_names(notif: "NotifPolicy") -> list[str]:
     """Return the list of syscalls to intercept via user notification.
 
     openat is always intercepted.  open is added on x86_64 (not
@@ -91,13 +91,16 @@ def _notif_syscall_names(policy: "Policy") -> list[str]:
     names = ["openat"]
     if "open" in _SYSCALL_NR:
         names.append("open")
-    notif = policy.notif_policy
     if notif is not None and notif.allowed_ips:
         names.extend(["connect", "sendto"])
     if notif is not None and notif.max_memory_bytes > 0:
         names.extend(["mmap", "munmap", "brk", "mremap"])
     if notif is not None and notif.max_processes > 0:
         names.extend(["clone", "fork", "vfork"])
+    if notif is not None and notif.isolate_pids:
+        names.append("getdents64")
+        if "getdents" in _SYSCALL_NR:
+            names.append("getdents")
     # Deduplicate (clone/open may already be in the list)
     return list(dict.fromkeys(names))
 
@@ -298,8 +301,18 @@ def _close_pidfd(self) -> None:
             self._control_fd = -1
 
     def __enter__(self) -> "SandboxContext":
-        notif_policy = self._policy.notif_policy
-        use_notif = notif_policy is not None
+        # Auto-enable /proc PID isolation when /proc is readable
+        self._notif_policy = self._policy.notif_policy
+        if self._notif_policy is None and any(
+            p == "/proc" or p.rstrip("/") == "/proc"
+            for p in self._policy.fs_readable
+        ):
+            from ._notif_policy import NotifPolicy, default_proc_rules
+            self._notif_policy = NotifPolicy(
+                rules=default_proc_rules(),
+                isolate_pids=True,
+            )
+        use_notif = self._notif_policy is not None
 
         # Pre-import modules used in the child BEFORE fork — the child's
         # Landlock policy won't include the sandlock source directory, so
@@ -442,7 +455,7 @@ def __enter__(self) -> "SandboxContext":
                         from ._landlock import _set_no_new_privs
                         _set_no_new_privs()
                         notify_fd = install_notif_filter(
-                            _notif_syscall_names(self._policy),
+                            _notif_syscall_names(self._notif_policy),
                             deny_syscalls=deny,
                             allow_syscalls=allow,
                         )
@@ -547,7 +560,7 @@ def __enter__(self) -> "SandboxContext":
                     parent_sock.close()
                     pids_fn = lambda pgid=pid: _pids_by_pgid(pgid)  # noqa: E731
                     self._supervisor = NotifSupervisor(
-                        notify_fd, pid, notif_policy,
+                        notify_fd, pid, self._notif_policy,
                         pids_fn=pids_fn,
                     )
                     self._supervisor.start()

src/sandlock/_notif.py

@@ -24,7 +24,7 @@
 
 from .exceptions import NotifError
 from ._notif_policy import NotifAction, NotifPolicy
-from ._procfs import read_bytes, resolve_openat_path
+from ._procfs import read_bytes, write_bytes, resolve_openat_path
 from ._seccomp import (
     AUDIT_ARCH,
     BPF_ABS,
@@ -355,6 +355,72 @@ def _parse_msghdr_dest_ip(pid: int, msghdr_addr: int) -> str | None:
     return _parse_dest_ip(pid, name_addr, name_len)
 
 
+# --- getdents64 helpers ---
+
+def _build_dirent64(d_ino: int, d_off: int, d_type: int, name: str) -> bytes:
+    """Build a single linux_dirent64 entry.
+
+    struct linux_dirent64 {
+        u64  d_ino;      // 0
+        s64  d_off;      // 8
+        u16  d_reclen;   // 16
+        u8   d_type;     // 18
+        char d_name[];   // 19+
+    };
+    d_reclen is 8-byte aligned.
+    """
+    name_bytes = name.encode("utf-8") + b"\0"
+    # 19 bytes header + name + padding to 8-byte alignment
+    reclen = 19 + len(name_bytes)
+    reclen = (reclen + 7) & ~7  # align to 8
+    buf = bytearray(reclen)
+    struct.pack_into("QqHB", buf, 0, d_ino, d_off, reclen, d_type)
+    buf[19:19 + len(name_bytes)] = name_bytes
+    return bytes(buf)
+
+
+def _build_filtered_dirents(sandbox_pids: set[int]) -> list[bytes]:
+    """Build a list of dirent64 entries for /proc, filtering out foreign PIDs.
+
+    Reads the real /proc directory in the supervisor process and builds
+    synthetic dirent64 entries, excluding PID directories not in sandbox_pids.
+    """
+    DT_DIR = 4
+    DT_REG = 8
+    DT_LNK = 10
+    entries = []
+    d_off = 0
+    try:
+        with os.scandir("/proc") as it:
+            for entry in it:
+                name = entry.name
+                # Filter out foreign PID directories
+                if name.isdigit():
+                    if int(name) not in sandbox_pids:
+                        continue
+
+                d_off += 1
+                try:
+                    if entry.is_dir(follow_symlinks=False):
+                        d_type = DT_DIR
+                    elif entry.is_symlink():
+                        d_type = DT_LNK
+                    else:
+                        d_type = DT_REG
+                except OSError:
+                    d_type = DT_REG
+
+                try:
+                    d_ino = entry.inode()
+                except OSError:
+                    d_ino = 0
+
+                entries.append(_build_dirent64(d_ino, d_off, d_type, name))
+    except OSError:
+        pass
+    return entries
+
+
 # --- Notification supervisor ---
 
 class NotifSupervisor:
@@ -384,6 +450,8 @@ def __init__(
         self._brk_base: dict[int, int] = {}  # pid → last known brk
         self._proc_count: int = 1     # Start at 1 (the initial child)
         self._proc_pids: set[int] = {child_pid}  # All known sandbox PIDs
+        # getdents /proc filtering: fd → list of remaining dirent entries
+        self._proc_dir_cache: dict[int, list[bytes]] = {}
 
     def start(self) -> None:
         """Start the supervisor thread."""
@@ -499,6 +567,14 @@ def _dispatch(self, notif: SeccompNotif) -> None:
             self._handle_net(notif, nr)
             return
 
+        # --- /proc readdir PID filtering ---
+        nr_getdents64 = _SYSCALL_NR.get("getdents64")
+        nr_getdents = _SYSCALL_NR.get("getdents")
+
+        if nr in (nr_getdents64, nr_getdents) and self._policy.isolate_pids:
+            self._handle_getdents(notif)
+            return
+
         # --- Filesystem: open / openat virtualization ---
         nr_openat = _SYSCALL_NR.get("openat")
         nr_open = _SYSCALL_NR.get("open")
@@ -658,8 +734,77 @@ def _handle_fork(self, notif: SeccompNotif, nr: int) -> None:
         # The new child's PID is unknown until it makes its first
         # intercepted syscall — tracked lazily via _record_pid.
         self._proc_pids.add(notif.pid)
+        # Invalidate /proc readdir cache so new PIDs appear
+        self._proc_dir_cache.clear()
         self._respond_continue(notif.id)
 
+    def _handle_getdents(self, notif: SeccompNotif) -> None:
+        """Handle getdents64/getdents — filter /proc readdir to hide foreign PIDs.
+
+        On first call for a given fd, reads all /proc entries from the
+        supervisor, filters out foreign PIDs, builds dirent64 entries,
+        and caches them.  Each call returns as many cached entries as fit
+        in the child's buffer, then returns 0 when exhausted.
+        """
+        pid = notif.pid
+        child_fd_num = notif.data.args[0] & 0xFFFFFFFF
+        buf_addr = notif.data.args[1]
+        buf_size = notif.data.args[2] & 0xFFFFFFFF
+
+        # Check if the fd points to /proc
+        try:
+            target = os.readlink(f"/proc/{pid}/fd/{child_fd_num}")
+        except OSError:
+            self._respond_continue(notif.id)
+            return
+
+        if target != "/proc":
+            self._respond_continue(notif.id)
+            return
+
+        # Build cache on first call for this fd
+        cache_key = (pid, child_fd_num)
+        if cache_key not in self._proc_dir_cache:
+            sandbox_pids = None
+            if self._pids_fn is not None:
+                sandbox_pids = set(self._pids_fn())
+            if sandbox_pids is None:
+                self._respond_continue(notif.id)
+                return
+
+            entries = _build_filtered_dirents(sandbox_pids)
+            self._proc_dir_cache[cache_key] = entries
+
+        entries = self._proc_dir_cache[cache_key]
+
+        if not self._id_valid(notif.id):
+            return
+
+        # Pack as many entries as fit into buf_size
+        result = bytearray()
+        consumed = 0
+        for entry in entries:
+            if len(result) + len(entry) > buf_size:
+                break
+            result.extend(entry)
+            consumed += 1
+
+        # Remove consumed entries from cache
+        if consumed > 0:
+            self._proc_dir_cache[cache_key] = entries[consumed:]
+        elif not entries:
+            # All entries consumed — clean up cache
+            del self._proc_dir_cache[cache_key]
+
+        # Write to child memory and return byte count
+        try:
+            if result:
+                write_bytes(pid, buf_addr, bytes(result))
+            self._respond_val(notif.id, len(result))
+        except OSError:
+            self._proc_dir_cache.pop(cache_key, None)
+            self._respond_continue(notif.id)
+
     def _id_valid(self, notif_id: int) -> bool:
         """Check if a notification ID is still valid (TOCTTOU check)."""
         id_val = ctypes.c_uint64(notif_id)
@@ -683,6 +828,19 @@ def _respond_continue(self, notif_id: int) -> None:
             ctypes.byref(resp),
         )
 
+    def _respond_val(self, notif_id: int, val: int) -> None:
+        """Return a specific value as the syscall result."""
+        resp = SeccompNotifResp()
+        resp.id = notif_id
+        resp.val = val
+        resp.error = 0
+        resp.flags = 0
+        _libc.ioctl(
+            ctypes.c_int(self._notify_fd),
+            ctypes.c_ulong(SECCOMP_IOCTL_NOTIF_SEND),
+            ctypes.byref(resp),
+        )
+
     def _respond_errno(self, notif_id: int, errno_code: int) -> None:
         """Deny the syscall with the given errno."""
         resp = SeccompNotifResp()

src/sandlock/_procfs.py

@@ -56,6 +56,24 @@ def read_bytes(pid: int, addr: int, length: int) -> bytes:
         os.close(fd)
 
 
+def write_bytes(pid: int, addr: int, data: bytes) -> None:
+    """Write raw bytes to a child process's memory.
+
+    Args:
+        pid: Target process ID.
+        addr: Virtual address in the target's address space.
+        data: Bytes to write.
+
+    Raises:
+        OSError: If /proc/<pid>/mem cannot be written.
+    """
+    fd = os.open(f"/proc/{pid}/mem", os.O_WRONLY)
+    try:
+        os.pwrite(fd, data, addr)
+    finally:
+        os.close(fd)
+
+
 def resolve_openat_path(pid: int, dirfd: int, pathname_addr: int) -> str:
     """Resolve the full path for an openat(dirfd, pathname, ...) call.
 

src/sandlock/cli.py

@@ -76,21 +76,6 @@ def cmd_run(args: argparse.Namespace) -> int:
     else:
         policy = Policy(**cli_kwargs)
 
-    # Auto-enable /proc pid isolation when /proc is readable
-    readable = list(policy.fs_readable)
-    if readable and any(
-        p == "/proc" or p.rstrip("/") == "/proc" for p in readable
-    ):
-        from ._notif_policy import NotifPolicy, default_proc_rules
-        if policy.notif_policy is None:
-            import dataclasses
-            policy = dataclasses.replace(
-                policy,
-                notif_policy=NotifPolicy(
-                    rules=default_proc_rules(),
-                    isolate_pids=True,
-                ),
-            )
     sb = Sandbox(policy)
 
     if args.interactive:

tests/test_integration.py

@@ -694,6 +694,63 @@ def test_proc_auto_isolation_blocks_foreign_pid(self):
         assert result.success
         assert b"HIDDEN" in result.stdout
 
+    def test_getdents_hides_foreign_pids(self):
+        """readdir(/proc) should only show sandbox PIDs when isolate_pids=True."""
+        from sandlock._notif_policy import NotifPolicy, default_proc_rules
+
+        policy = Policy(
+            fs_readable=_PYTHON_READABLE,
+            notif_policy=NotifPolicy(
+                rules=default_proc_rules(),
+                isolate_pids=True,
+            ),
+        )
+        result = Sandbox(policy).run(
+            ["python3", "-c", """
+import os
+# List /proc and collect numeric (PID) entries
+pids = [e for e in os.listdir('/proc') if e.isdigit()]
+my_pid = str(os.getpid())
+# Our own PID should be visible
+if my_pid in pids:
+    print(f'SELF_VISIBLE')
+else:
+    print(f'SELF_HIDDEN')
+# PID 1 (init) should NOT be visible
+if '1' in pids:
+    print('INIT_VISIBLE')
+else:
+    print('INIT_HIDDEN')
+print(f'PID_COUNT={len(pids)}')
+"""]
+        )
+        assert result.success
+        assert b"INIT_HIDDEN" in result.stdout
+        # Should have very few PIDs (just the sandbox's own)
+        for line in result.stdout.decode().splitlines():
+            if line.startswith("PID_COUNT="):
+                count = int(line.split("=")[1])
+                assert count < 10, f"Too many PIDs visible: {count}"
+
+    def test_always_isolates_when_proc_readable(self):
+        """PID isolation is always on when /proc is in fs_readable."""
+        policy = Policy(
+            fs_readable=_PYTHON_READABLE,
+        )
+        result = Sandbox(policy).run(
+            ["python3", "-c", """
+import os
+pids = [e for e in os.listdir('/proc') if e.isdigit()]
+if '1' in pids:
+    print('INIT_VISIBLE')
+else:
+    print('INIT_HIDDEN')
+print(f'PID_COUNT={len(pids)}')
+"""]
+        )
+        assert result.success
+        assert b"INIT_HIDDEN" in result.stdout
+
 
 # --- Resource limits (seccomp notif based) ---