Validate Directory node names

src/buildstream/_cas/cascache.py

@@ -185,7 +185,11 @@ def checkout(self, dest, tree, *, can_link=False, _fetch=True):
         with open(self.objpath(tree), "rb") as f:
             directory.ParseFromString(f.read())
 
+        names = set[str]()
+
         for filenode in directory.files:
+            _validate_cas_node_name(filenode.name, names)
+
             # regular file, create hardlink
             fullpath = os.path.join(dest, filenode.name)
 
@@ -214,11 +218,13 @@ def checkout(self, dest, tree, *, can_link=False, _fetch=True):
                 os.chmod(fullpath, mode)
 
         for dirnode in directory.directories:
+            _validate_cas_node_name(dirnode.name, names)
             fullpath = os.path.join(dest, dirnode.name)
             self.checkout(fullpath, dirnode.digest, can_link=can_link, _fetch=False)
 
         for symlinknode in directory.symlinks:
             # symlink
+            _validate_cas_node_name(symlinknode.name, names)
             fullpath = os.path.join(dest, symlinknode.name)
             os.symlink(symlinknode.target, fullpath)
 
@@ -823,3 +829,22 @@ def _grouper(iterable, n):
         except StopIteration:
             return
         yield itertools.chain([current], itertools.islice(iterable, n - 1))
+
+
+# _validate_cas_node_name():
+#
+# Validates that a CAS Directory entry name is a single path segment.
+#
+# Args:
+#    name (str): The name of a CAS Directory entry
+#    names (set): The set of names in the directory
+#
+def _validate_cas_node_name(name: str, names: set[str]):
+    if os.path.basename(name) != name or name in ("", ".", ".."):
+        raise CASCacheError(
+            f"Invalid name in a CAS Directory node: '{name}' (every child in the directory must have a path of exactly one segment)"
+        )
+
+    if name in names:
+        raise CASCacheError(f"Duplicate name in a CAS Directory node: '{name}'")
+    names.add(name)

src/buildstream/storage/_casbaseddirectory.py

@@ -653,11 +653,15 @@ def __populate_index(self, digest) -> None:
             if prop.name == "SubtreeReadOnly":
                 self.__subtree_read_only = prop.value == "true"
 
+        names = set[str]()
+
         for dentry in pb2_directory.directories:
+            _validate_cas_node_name(dentry.name, names)
             self.__index[dentry.name] = _IndexEntry(
                 self.__cas_cache, dentry.name, FileType.DIRECTORY, digest=dentry.digest
             )
         for entry in pb2_directory.files:
+            _validate_cas_node_name(entry.name, names)
             mtime: Optional[timestamp_pb2.Timestamp]
             if entry.node_properties.HasField("mtime"):
                 mtime = entry.node_properties.mtime
@@ -673,6 +677,7 @@ def __populate_index(self, digest) -> None:
                 mtime=mtime,
             )
         for lentry in pb2_directory.symlinks:
+            _validate_cas_node_name(lentry.name, names)
             self.__index[lentry.name] = _IndexEntry(
                 self.__cas_cache, lentry.name, FileType.SYMLINK, target=lentry.target
             )
@@ -908,3 +913,22 @@ def __add_files_to_result(self, *, path_prefix: str, result: FileListResult) ->
                 subdir.__add_files_to_result(path_prefix=relative_pathname, result=result)
             else:
                 result.files_written.append(relative_pathname)
+
+
+# _validate_cas_node_name():
+#
+# Validates that a CAS Directory entry name is a single path segment.
+#
+# Args:
+#    name (str): The name of a CAS Directory entry
+#    names (set): The set of names in the directory
+#
+def _validate_cas_node_name(name: str, names: set[str]):
+    if os.path.basename(name) != name or name in ("", ".", ".."):
+        raise DirectoryError(
+            f"Invalid name in a CAS Directory node: '{name}' (every child in the directory must have a path of exactly one segment)"
+        )
+
+    if name in names:
+        raise DirectoryError(f"Duplicate name in a CAS Directory node: '{name}'")
+    names.add(name)