Intercept execve/execveat under COW for upper layer binaries

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

Author: congwang-mk

src/sandlock/_context.py

@@ -117,6 +117,7 @@ def _notif_syscall_names(notif: "NotifPolicy") -> list[str]:
             "newfstatat", "statx", "faccessat",
             "symlinkat", "linkat", "fchmodat", "fchownat",
             "readlinkat", "truncate", "utimensat", "getdents64",
+            "execve", "execveat",
             # Non-at variants (x86_64 has both, aarch64 only has *at)
             "unlink", "rmdir", "mkdir", "rename",
             "stat", "lstat", "access",

src/sandlock/_notif.py

@@ -938,6 +938,76 @@ def _dispatch(self, notif: SeccompNotif) -> None:
                 self._respond_continue(notif.id)
                 return
 
+        # --- COW: execve / execveat path redirection ---
+        if self._cow_handler is not None:
+            nr_execve = _SYSCALL_NR.get("execve")
+            nr_execveat = _SYSCALL_NR.get("execveat")
+
+            if nr in (nr_execve, nr_execveat):
+                try:
+                    if nr == nr_execve:
+                        # execve(pathname, argv, envp)
+                        pathname_addr = notif.data.args[0]
+                        path = resolve_openat_path(pid, -100, pathname_addr)
+                    else:
+                        # execveat(dirfd, pathname, argv, envp, flags)
+                        dirfd = ctypes.c_int32(notif.data.args[0] & 0xFFFFFFFF).value
+                        exec_flags = notif.data.args[4]
+                        if exec_flags & 0x1000:  # AT_EMPTY_PATH — fd-based, no path
+                            self._respond_continue(notif.id)
+                            return
+                        pathname_addr = notif.data.args[1]
+                        path = resolve_openat_path(pid, dirfd, pathname_addr)
+                except OSError:
+                    self._respond_continue(notif.id)
+                    return
+
+                if not self._id_valid(notif.id):
+                    return
+
+                if not self._cow_handler.matches(path):
+                    self._respond_continue(notif.id)
+                    return
+
+                # Resolve through COW layer
+                real_path = self._cow_handler.handle_stat(path)
+                if real_path is None:
+                    # File deleted in COW
+                    self._respond_errno(notif.id, errno.ENOENT)
+                    return
+
+                # If unchanged (real_path == path), let kernel handle it
+                if real_path == path:
+                    self._respond_continue(notif.id)
+                    return
+
+                # File is in upper layer — inject fd then rewrite path
+                try:
+                    src_fd = os.open(real_path, os.O_RDONLY | os.O_CLOEXEC)
+                except OSError:
+                    self._respond_continue(notif.id)
+                    return
+
+                try:
+                    child_fd = self._inject_fd(notif.id, src_fd, cloexec=False)
+                finally:
+                    os.close(src_fd)
+
+                if child_fd < 0:
+                    self._respond_continue(notif.id)
+                    return
+
+                # Overwrite the pathname in child memory with /proc/self/fd/N
+                proc_path = f"/proc/self/fd/{child_fd}\0".encode()
+                try:
+                    write_bytes(pid, pathname_addr, proc_path)
+                except OSError:
+                    self._respond_continue(notif.id)
+                    return
+
+                self._respond_continue(notif.id)
+                return
+
         # --- Filesystem: open / openat virtualization + COW ---
         nr_openat = _SYSCALL_NR.get("openat")
         nr_open = _SYSCALL_NR.get("open")
@@ -1050,6 +1120,26 @@ def _respond_addfd(self, notif_id: int, src_fd: int) -> None:
             ctypes.byref(resp),
         )
 
+    def _inject_fd(self, notif_id: int, src_fd: int,
+                   cloexec: bool = False) -> int:
+        """Inject an fd into the child without completing the notification.
+
+        Returns the fd number in the child's table, or -1 on failure.
+        """
+        addfd = SeccompNotifAddfd()
+        addfd.id = notif_id
+        addfd.flags = 0  # Don't auto-send response
+        addfd.srcfd = src_fd
+        addfd.newfd = 0
+        addfd.newfd_flags = os.O_CLOEXEC if cloexec else 0
+
+        ret = _libc.ioctl(
+            ctypes.c_int(self._notify_fd),
+            ctypes.c_ulong(SECCOMP_IOCTL_NOTIF_ADDFD),
+            ctypes.byref(addfd),
+        )
+        return ret
+
     # Network, memory, and fork handlers moved to _network.py and _resource.py
 
     def _handle_port_remap(self, notif: SeccompNotif, nr: int) -> None:

tests/test_integration.py

@@ -1065,6 +1065,51 @@ def test_disk_quota_only_counts_delta(self, isolation):
             assert result.success
             assert b"orig=51200" in result.stdout
 
+    def test_cow_execve_runs_upper_binary(self, isolation):
+        """execve on a binary created in COW upper layer should work."""
+        import subprocess as sp
+
+        def create_and_exec():
+            with open('hello.sh', 'w') as f:
+                f.write('#!/bin/sh\necho HELLO_FROM_COW\n')
+            os.chmod('hello.sh', 0o755)
+            return sp.check_output(['./hello.sh']).decode().strip()
+
+        with tempfile.TemporaryDirectory() as td:
+            os.makedirs(f"{td}/project")
+            policy = _make_cow_policy(
+                f"{td}/project", td, isolation,
+                on_exit=BranchAction.ABORT,
+            )
+            result = Sandbox(policy).call(create_and_exec)
+            assert result.success, f"Failed: {result.error}"
+            assert result.value == "HELLO_FROM_COW"
+            assert not os.path.exists(f"{td}/project/hello.sh")
+
+    def test_cow_execve_modified_binary(self, isolation):
+        """execve on a binary modified in COW upper layer runs the new version."""
+        import subprocess as sp
+
+        with tempfile.TemporaryDirectory() as td:
+            os.makedirs(f"{td}/project")
+            with open(f"{td}/project/run.sh", "w") as f:
+                f.write("#!/bin/sh\necho ORIGINAL\n")
+            os.chmod(f"{td}/project/run.sh", 0o755)
+
+            def modify_and_exec():
+                with open('run.sh', 'w') as f:
+                    f.write('#!/bin/sh\necho MODIFIED\n')
+                return sp.check_output(['./run.sh']).decode().strip()
+
+            policy = _make_cow_policy(
+                f"{td}/project", td, isolation,
+                on_exit=BranchAction.ABORT,
+            )
+            result = Sandbox(policy).call(modify_and_exec)
+            assert result.success, f"Failed: {result.error}"
+            assert result.value == "MODIFIED"
+            assert open(f"{td}/project/run.sh").read() == "#!/bin/sh\necho ORIGINAL\n"
+
     def test_cow_chown_goes_to_upper(self, isolation):
         """chown on a COW file operates on the upper copy, not the original."""
         with tempfile.TemporaryDirectory() as td: