Fix path traversal and symlink escape in COW filesystem handler

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

Author: congwang-mk

src/sandlock/cowfs/_branch.py

@@ -106,8 +106,13 @@ def ensure_cow_copy(self, rel_path: str) -> Path:
 
         upper_file.parent.mkdir(parents=True, exist_ok=True)
 
-        if lower_file.exists():
-            shutil.copy2(str(lower_file), str(upper_file))
+        if lower_file.exists() and not lower_file.is_symlink():
+            shutil.copy2(str(lower_file), str(upper_file),
+                         follow_symlinks=False)
+        elif lower_file.is_symlink():
+            # Copy symlink itself, not its target
+            link_target = os.readlink(str(lower_file))
+            os.symlink(link_target, str(upper_file))
 
         return upper_file
 

src/sandlock/cowfs/_handler.py

@@ -44,6 +44,16 @@ def matches(self, path: str) -> bool:
         """Check if a path is under the COW workdir."""
         return path.startswith(self._workdir_str + "/") or path == self._workdir_str
 
+    def _safe_rel(self, path: str) -> str | None:
+        """Compute relative path and reject traversal escapes.
+
+        Returns the relative path, or None if it escapes the workdir.
+        """
+        rel = os.path.relpath(path, self._workdir_str)
+        if rel == ".." or rel.startswith("../"):
+            return None
+        return rel
+
     def handle_open(self, path: str, flags: int) -> str | None:
         """Determine the real path to open for a COW-intercepted openat.
 
@@ -52,7 +62,9 @@ def handle_open(self, path: str, flags: int) -> str | None:
         if flags & O_DIRECTORY:
             return None
 
-        rel_path = os.path.relpath(path, self._workdir_str)
+        rel_path = self._safe_rel(path)
+        if rel_path is None:
+            return None
 
         # Deleted file — new create or ENOENT
         if self._branch.is_deleted(rel_path):
@@ -78,7 +90,9 @@ def handle_unlink(self, path: str, is_dir: bool = False) -> bool:
 
         Returns True if handled, False to let kernel handle.
         """
-        rel_path = os.path.relpath(path, self._workdir_str)
+        rel_path = self._safe_rel(path)
+        if rel_path is None:
+            return False
         upper_file = self._branch.upper_dir / rel_path
         lower_file = Path(self._workdir_str) / rel_path
 
@@ -102,7 +116,9 @@ def handle_unlink(self, path: str, is_dir: bool = False) -> bool:
 
     def handle_mkdir(self, path: str, mode: int) -> bool:
         """Handle mkdirat: create directory in upper."""
-        rel_path = os.path.relpath(path, self._workdir_str)
+        rel_path = self._safe_rel(path)
+        if rel_path is None:
+            return False
         self._branch._deleted.discard(rel_path)
         upper_dir = self._branch.upper_dir / rel_path
         upper_dir.mkdir(parents=True, exist_ok=True)
@@ -113,7 +129,9 @@ def handle_stat(self, path: str) -> str | None:
 
         Returns the real path to stat, or None if deleted/nonexistent.
         """
-        rel_path = os.path.relpath(path, self._workdir_str)
+        rel_path = self._safe_rel(path)
+        if rel_path is None:
+            return None
 
         if self._branch.is_deleted(rel_path):
             return None
@@ -125,8 +143,10 @@ def handle_stat(self, path: str) -> str | None:
 
     def handle_rename(self, old_path: str, new_path: str) -> bool:
         """Handle rename: rename in upper dir."""
-        old_rel = os.path.relpath(old_path, self._workdir_str)
-        new_rel = os.path.relpath(new_path, self._workdir_str)
+        old_rel = self._safe_rel(old_path)
+        new_rel = self._safe_rel(new_path)
+        if old_rel is None or new_rel is None:
+            return False
 
         old_upper = self._branch.ensure_cow_copy(old_rel)
         new_upper = self._branch.upper_dir / new_rel
@@ -162,8 +182,17 @@ def list_merged_dir(self, rel_path: str) -> list[str]:
         return sorted(entries)
 
     def handle_symlink(self, target: str, linkpath: str) -> bool:
-        """Handle symlink: create symlink in upper."""
-        rel_path = os.path.relpath(linkpath, self._workdir_str)
+        """Handle symlink: create symlink in upper.
+
+        Rejects absolute or traversal symlink targets to prevent
+        escaping the COW layer on commit.
+        """
+        rel_path = self._safe_rel(linkpath)
+        if rel_path is None:
+            return False
+        # Reject symlink targets that escape the workdir
+        if os.path.isabs(target) or ".." in target.split("/"):
+            return False
         self._branch._deleted.discard(rel_path)
         upper_link = self._branch.upper_dir / rel_path
         upper_link.parent.mkdir(parents=True, exist_ok=True)
@@ -172,8 +201,10 @@ def handle_symlink(self, target: str, linkpath: str) -> bool:
 
     def handle_link(self, oldpath: str, newpath: str) -> bool:
         """Handle link: create hard link in upper."""
-        old_rel = os.path.relpath(oldpath, self._workdir_str)
-        new_rel = os.path.relpath(newpath, self._workdir_str)
+        old_rel = self._safe_rel(oldpath)
+        new_rel = self._safe_rel(newpath)
+        if old_rel is None or new_rel is None:
+            return False
         old_upper = self._branch.ensure_cow_copy(old_rel)
         new_upper = self._branch.upper_dir / new_rel
         new_upper.parent.mkdir(parents=True, exist_ok=True)
@@ -182,14 +213,18 @@ def handle_link(self, oldpath: str, newpath: str) -> bool:
 
     def handle_chmod(self, path: str, mode: int) -> bool:
         """Handle chmod: chmod in upper (COW copy if needed)."""
-        rel_path = os.path.relpath(path, self._workdir_str)
+        rel_path = self._safe_rel(path)
+        if rel_path is None:
+            return False
         upper_file = self._branch.ensure_cow_copy(rel_path)
         os.chmod(str(upper_file), mode)
         return True
 
     def handle_readlink(self, path: str) -> str | None:
         """Handle readlink: resolve symlink from upper or lower."""
-        rel_path = os.path.relpath(path, self._workdir_str)
+        rel_path = self._safe_rel(path)
+        if rel_path is None:
+            return None
 
         if self._branch.is_deleted(rel_path):
             return None
@@ -205,15 +240,19 @@ def handle_readlink(self, path: str) -> str | None:
 
     def handle_truncate(self, path: str, length: int) -> bool:
         """Handle truncate: truncate in upper (COW copy if needed)."""
-        rel_path = os.path.relpath(path, self._workdir_str)
+        rel_path = self._safe_rel(path)
+        if rel_path is None:
+            return False
         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)
+        rel_path = self._safe_rel(path)
+        if rel_path is None:
+            return False
         upper_file = self._branch.ensure_cow_copy(rel_path)
         os.chown(str(upper_file), uid, gid,
                  follow_symlinks=follow_symlinks)
@@ -226,7 +265,9 @@ def handle_utimens(self, path: str,
 
         times is (atime, mtime) as floats, or None for current time.
         """
-        rel_path = os.path.relpath(path, self._workdir_str)
+        rel_path = self._safe_rel(path)
+        if rel_path is None:
+            return False
         upper_file = self._branch.ensure_cow_copy(rel_path)
         os.utime(str(upper_file), times=times,
                  follow_symlinks=follow_symlinks)

tests/test_branchfs.py

@@ -296,6 +296,98 @@ def test_nested_sandbox_inherits_parent_branch(self, mock_create, mock_mount):
         )
 
 
+class TestCowHandlerPathTraversal(unittest.TestCase):
+    """Verify that COW handler rejects path traversal attacks."""
+
+    def setUp(self):
+        import tempfile
+        self._tmpdir = tempfile.mkdtemp(prefix="sandlock_cow_test_")
+        self._workdir = Path(self._tmpdir) / "workdir"
+        self._workdir.mkdir()
+        # Create a file in the workdir
+        (self._workdir / "hello.txt").write_text("hello")
+
+        from sandlock.cowfs._branch import CowBranch
+        from sandlock.cowfs._handler import CowHandler
+        self._branch = CowBranch(self._workdir)
+        self._branch.create()
+        self._handler = CowHandler(self._branch)
+
+    def tearDown(self):
+        import shutil
+        shutil.rmtree(self._tmpdir, ignore_errors=True)
+
+    def test_open_traversal_rejected(self):
+        result = self._handler.handle_open(
+            str(self._workdir) + "/../../../etc/passwd", 0)
+        self.assertIsNone(result)
+
+    def test_stat_traversal_rejected(self):
+        result = self._handler.handle_stat(
+            str(self._workdir) + "/../../../etc/passwd")
+        self.assertIsNone(result)
+
+    def test_unlink_traversal_rejected(self):
+        result = self._handler.handle_unlink(
+            str(self._workdir) + "/../../../etc/passwd")
+        self.assertFalse(result)
+
+    def test_mkdir_traversal_rejected(self):
+        result = self._handler.handle_mkdir(
+            str(self._workdir) + "/../../../tmp/evil", 0o755)
+        self.assertFalse(result)
+
+    def test_rename_traversal_rejected(self):
+        result = self._handler.handle_rename(
+            str(self._workdir) + "/hello.txt",
+            str(self._workdir) + "/../../../tmp/stolen")
+        self.assertFalse(result)
+
+    def test_symlink_absolute_target_rejected(self):
+        result = self._handler.handle_symlink(
+            "/etc/shadow",
+            str(self._workdir) + "/evil_link")
+        self.assertFalse(result)
+
+    def test_symlink_traversal_target_rejected(self):
+        result = self._handler.handle_symlink(
+            "../../../etc/shadow",
+            str(self._workdir) + "/evil_link")
+        self.assertFalse(result)
+
+    def test_symlink_safe_target_allowed(self):
+        result = self._handler.handle_symlink(
+            "hello.txt",
+            str(self._workdir) + "/good_link")
+        self.assertTrue(result)
+
+    def test_link_traversal_rejected(self):
+        result = self._handler.handle_link(
+            str(self._workdir) + "/../../../etc/passwd",
+            str(self._workdir) + "/stolen")
+        self.assertFalse(result)
+
+    def test_chmod_traversal_rejected(self):
+        result = self._handler.handle_chmod(
+            str(self._workdir) + "/../../../etc/passwd", 0o777)
+        self.assertFalse(result)
+
+    def test_readlink_traversal_rejected(self):
+        result = self._handler.handle_readlink(
+            str(self._workdir) + "/../../../etc/passwd")
+        self.assertIsNone(result)
+
+    def test_truncate_traversal_rejected(self):
+        result = self._handler.handle_truncate(
+            str(self._workdir) + "/../../../etc/passwd", 0)
+        self.assertFalse(result)
+
+    def test_legitimate_path_works(self):
+        result = self._handler.handle_open(
+            str(self._workdir) + "/hello.txt", 0)
+        self.assertIsNotNone(result)
+
+
 class TestExports(unittest.TestCase):
     def test_enums_importable(self):
         from sandlock import FsIsolation, BranchAction