Add COW interception for chown/fchownat and utimensat/futimesat

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

Author: congwang-mk

src/sandlock/_context.py

@@ -115,12 +115,13 @@ def _notif_syscall_names(notif: "NotifPolicy") -> list[str]:
         cow_syscalls = [
             "unlinkat", "mkdirat", "renameat2",
             "newfstatat", "statx", "faccessat",
-            "symlinkat", "linkat", "fchmodat",
-            "readlinkat", "truncate", "getdents64",
+            "symlinkat", "linkat", "fchmodat", "fchownat",
+            "readlinkat", "truncate", "utimensat", "getdents64",
             # Non-at variants (x86_64 has both, aarch64 only has *at)
             "unlink", "rmdir", "mkdir", "rename",
             "stat", "lstat", "access",
-            "symlink", "link", "chmod", "readlink",
+            "symlink", "link", "chmod", "chown", "lchown",
+            "readlink", "futimesat",
         ]
         for name in cow_syscalls:
             if name in _SYSCALL_NR:

src/sandlock/_notif.py

@@ -660,18 +660,26 @@ def _dispatch(self, notif: SeccompNotif) -> None:
             nr_link = _SYSCALL_NR.get("link")
             nr_fchmodat = _SYSCALL_NR.get("fchmodat")
             nr_chmod = _SYSCALL_NR.get("chmod")
+            nr_fchownat = _SYSCALL_NR.get("fchownat")
+            nr_chown = _SYSCALL_NR.get("chown")
+            nr_lchown = _SYSCALL_NR.get("lchown")
             nr_readlinkat = _SYSCALL_NR.get("readlinkat")
             nr_readlink = _SYSCALL_NR.get("readlink")
             nr_truncate = _SYSCALL_NR.get("truncate")
+            nr_utimensat = _SYSCALL_NR.get("utimensat")
+            nr_futimesat = _SYSCALL_NR.get("futimesat")
 
             # *at variants: dirfd=arg0, pathname=arg1
             cow_at_nrs = {nr_unlinkat, nr_mkdirat, nr_renameat2,
                           nr_newfstatat, nr_statx, nr_faccessat,
-                          nr_fchmodat, nr_readlinkat} - {None}
+                          nr_fchmodat, nr_fchownat, nr_readlinkat,
+                          nr_utimensat} - {None}
             # non-at variants: pathname=arg0
             cow_plain_nrs = {nr_unlink, nr_rmdir, nr_mkdir, nr_rename,
                              nr_stat, nr_lstat, nr_access,
-                             nr_chmod, nr_readlink, nr_truncate} - {None}
+                             nr_chmod, nr_chown, nr_lchown,
+                             nr_readlink, nr_truncate,
+                             nr_futimesat} - {None}
             # Special arg layouts
             cow_special_nrs = {nr_symlinkat, nr_symlink,
                                nr_linkat, nr_link} - {None}
@@ -838,6 +846,74 @@ def _dispatch(self, notif: SeccompNotif) -> None:
                             self._respond_continue(notif.id)
                         return
 
+                    # chown / fchownat / lchown
+                    if nr in (nr_fchownat, nr_chown, nr_lchown):
+                        if nr == nr_fchownat:
+                            uid = ctypes.c_int32(notif.data.args[2] & 0xFFFFFFFF).value
+                            gid = ctypes.c_int32(notif.data.args[3] & 0xFFFFFFFF).value
+                            follow = not bool(notif.data.args[4] & 0x100)  # AT_SYMLINK_NOFOLLOW
+                        elif nr == nr_chown:
+                            uid = ctypes.c_int32(notif.data.args[1] & 0xFFFFFFFF).value
+                            gid = ctypes.c_int32(notif.data.args[2] & 0xFFFFFFFF).value
+                            follow = True
+                        else:  # lchown
+                            uid = ctypes.c_int32(notif.data.args[1] & 0xFFFFFFFF).value
+                            gid = ctypes.c_int32(notif.data.args[2] & 0xFFFFFFFF).value
+                            follow = False
+                        if self._cow_handler.handle_chown(path, uid, gid,
+                                                          follow_symlinks=follow):
+                            self._respond_val(notif.id, 0)
+                        else:
+                            self._respond_continue(notif.id)
+                        return
+
+                    # utimensat / futimesat
+                    if nr in (nr_utimensat, nr_futimesat):
+                        UTIME_NOW = (1 << 30) - 1
+                        UTIME_OMIT = (1 << 30) - 2
+                        times = None  # default: current time
+                        if nr == nr_utimensat:
+                            # utimensat(dirfd, path, times[2], flags)
+                            times_addr = notif.data.args[2]
+                            follow = not bool(notif.data.args[3] & 0x100)
+                            if times_addr != 0:
+                                from ._procfs import read_bytes as _read
+                                raw = _read(pid, times_addr, 32)
+                                a_s, a_ns = struct.unpack_from("<qQ", raw, 0)
+                                m_s, m_ns = struct.unpack_from("<qQ", raw, 16)
+                                atime = None if a_ns == UTIME_OMIT else (
+                                    None if a_ns == UTIME_NOW else a_s + a_ns / 1e9)
+                                mtime = None if m_ns == UTIME_OMIT else (
+                                    None if m_ns == UTIME_NOW else m_s + m_ns / 1e9)
+                                if atime is not None or mtime is not None:
+                                    # Need current stat for OMIT fields
+                                    real = self._cow_handler.handle_stat(path)
+                                    if real:
+                                        st = os.stat(real)
+                                        if atime is None:
+                                            atime = st.st_atime
+                                        if mtime is None:
+                                            mtime = st.st_mtime
+                                    times = (atime or 0, mtime or 0)
+                                # both UTIME_NOW → times=None (current time)
+                        else:
+                            # futimesat(dirfd, path, times[2])
+                            times_addr = notif.data.args[2]
+                            follow = True
+                            if times_addr != 0:
+                                from ._procfs import read_bytes as _read
+                                raw = _read(pid, times_addr, 32)
+                                a_s, a_us = struct.unpack_from("<qQ", raw, 0)
+                                m_s, m_us = struct.unpack_from("<qQ", raw, 16)
+                                times = (a_s + a_us / 1e6,
+                                         m_s + m_us / 1e6)
+                        if self._cow_handler.handle_utimens(path, times,
+                                                            follow_symlinks=follow):
+                            self._respond_val(notif.id, 0)
+                        else:
+                            self._respond_continue(notif.id)
+                        return
+
                     # truncate (path-based)
                     if nr == nr_truncate:
                         length = notif.data.args[1]

src/sandlock/cowfs/_handler.py

@@ -209,3 +209,25 @@ def handle_truncate(self, path: str, length: int) -> bool:
         upper_file = self._branch.ensure_cow_copy(rel_path)
         os.truncate(str(upper_file), length)
         return True
+
+    def handle_chown(self, path: str, uid: int, gid: int,
+                     follow_symlinks: bool = True) -> bool:
+        """Handle chown/fchownat: chown in upper (COW copy if needed)."""
+        rel_path = os.path.relpath(path, self._workdir_str)
+        upper_file = self._branch.ensure_cow_copy(rel_path)
+        os.chown(str(upper_file), uid, gid,
+                 follow_symlinks=follow_symlinks)
+        return True
+
+    def handle_utimens(self, path: str,
+                       times: tuple[float, float] | None,
+                       follow_symlinks: bool = True) -> bool:
+        """Handle utimensat: set timestamps in upper (COW copy if needed).
+
+        times is (atime, mtime) as floats, or None for current time.
+        """
+        rel_path = os.path.relpath(path, self._workdir_str)
+        upper_file = self._branch.ensure_cow_copy(rel_path)
+        os.utime(str(upper_file), times=times,
+                 follow_symlinks=follow_symlinks)
+        return True

tests/test_integration.py

@@ -1065,6 +1065,78 @@ def test_disk_quota_only_counts_delta(self, isolation):
             assert result.success
             assert b"orig=51200" in result.stdout
 
+    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:
+            os.makedirs(f"{td}/project")
+            with open(f"{td}/project/file.txt", "w") as f:
+                f.write("data")
+            orig_stat = os.stat(f"{td}/project/file.txt")
+
+            policy = _make_cow_policy(
+                f"{td}/project", td, isolation,
+                on_exit=BranchAction.ABORT,
+            )
+            result = Sandbox(policy).run(
+                ["python3", "-c", """
+import os
+# chown to same uid/gid (always allowed, but triggers COW copy)
+st = os.stat('file.txt')
+os.chown('file.txt', st.st_uid, st.st_gid)
+# Verify the file is still readable after chown
+print('OK', open('file.txt').read())
+"""]
+            )
+            assert result.success
+            assert b"OK data" in result.stdout
+            # Original unchanged after abort
+            assert open(f"{td}/project/file.txt").read() == "data"
+
+    def test_cow_utimensat_goes_to_upper(self, isolation):
+        """utime on a COW file operates on the upper copy, not the original."""
+        with tempfile.TemporaryDirectory() as td:
+            os.makedirs(f"{td}/project")
+            with open(f"{td}/project/file.txt", "w") as f:
+                f.write("data")
+            orig_mtime = os.stat(f"{td}/project/file.txt").st_mtime
+
+            policy = _make_cow_policy(
+                f"{td}/project", td, isolation,
+                on_exit=BranchAction.ABORT,
+            )
+            result = Sandbox(policy).run(
+                ["python3", "-c", """
+import os
+os.utime('file.txt', (1000000, 1000000))
+st = os.stat('file.txt')
+print(f'MTIME {int(st.st_mtime)}')
+"""]
+            )
+            assert result.success
+            assert b"MTIME 1000000" in result.stdout
+            # Original mtime unchanged after abort
+            assert os.stat(f"{td}/project/file.txt").st_mtime == orig_mtime
+
+    def test_cow_utimensat_committed(self, isolation):
+        """utime changes are committed on success."""
+        with tempfile.TemporaryDirectory() as td:
+            os.makedirs(f"{td}/project")
+            with open(f"{td}/project/file.txt", "w") as f:
+                f.write("data")
+
+            policy = _make_cow_policy(f"{td}/project", td, isolation)
+            result = Sandbox(policy).run(
+                ["python3", "-c", """
+import os
+os.utime('file.txt', (2000000, 2000000))
+print('OK')
+"""]
+            )
+            assert result.success
+            assert b"OK" in result.stdout
+            # After commit, mtime should be updated
+            assert int(os.stat(f"{td}/project/file.txt").st_mtime) == 2000000
+
 
 # --- Deterministic randomness ---