[security] fix(payloads): validate store collection names

rampart/payloads/_store.py

@@ -34,6 +34,14 @@
 from rampart.core.types import Payload, PayloadFormat
 
 logger = logging.getLogger(__name__)
+_WINDOWS_RESERVED_BASENAMES = {
+    "AUX",
+    "CON",
+    "NUL",
+    "PRN",
+    *(f"COM{number}" for number in range(1, 10)),
+    *(f"LPT{number}" for number in range(1, 10)),
+}
 
 
 class PayloadStore:
@@ -165,6 +173,7 @@ def exists(self, name: str) -> bool:
             bool: True if the collection directory and its
                 ``payloads.jsonl`` file both exist.
         """
+        self._validate_collection_name(name)
         return self._collection_path(name).exists()
 
     def list_collections(self) -> list[str]:
@@ -185,6 +194,7 @@ def list_collections(self) -> list[str]:
 
     def delete(self, name: str) -> None:
         """Remove a collection from disk."""
+        self._validate_collection_name(name)
         collection_dir = self._root / name
         if collection_dir.exists():
             shutil.rmtree(collection_dir)
@@ -202,6 +212,7 @@ def manifest(self, name: str) -> dict[str, Any]:
         Raises:
             FileNotFoundError: If the collection does not exist.
         """
+        self._validate_collection_name(name)
         path = self._root / name / "manifest.json"
         if not path.exists():
             msg = f"No manifest for collection '{name}'"
@@ -211,14 +222,24 @@ def manifest(self, name: str) -> dict[str, Any]:
 
     @staticmethod
     def _validate_collection_name(name: str) -> None:
-        """Reject names that would escape the store root.
+        """Reject names that would escape or alias another collection.
 
         Raises:
-            ValueError: If the name is empty, contains path separators,
-                or is a relative-path component (``.`` / ``..``).
+            ValueError: If the name is not a portable directory name.
         """
-        if not name or "/" in name or "\\" in name or name in {".", ".."}:
-            msg = f"Invalid collection name: {name!r}. Must be a simple directory name."
+        basename = name.partition(".")[0].upper()
+        is_unsafe_path = (
+            not name
+            or name in {".", ".."}
+            or any(separator in name for separator in ("/", "\\"))
+        )
+        is_windows_alias = (
+            name.endswith((".", " ")) or basename in _WINDOWS_RESERVED_BASENAMES
+        )
+        if is_unsafe_path or is_windows_alias:
+            msg = (
+                f"Invalid collection name: {name!r}. Must be a portable directory name."
+            )
             raise ValueError(msg)
 
     def _collection_path(self, name: str) -> Path:

tests/unit/payloads/test_store.py

@@ -136,6 +136,74 @@ def test_manifest_missing_raises(self, store: PayloadStore) -> None:
         with pytest.raises(FileNotFoundError, match="No manifest"):
             store.manifest("ghost")
 
+    def test_exists_rejects_collection_path_traversal(
+        self,
+        store: PayloadStore,
+    ) -> None:
+        with pytest.raises(ValueError, match="Invalid collection name"):
+            store.exists("../outside")
+
+    def test_delete_rejects_collection_path_traversal(self, tmp_path: Path) -> None:
+        root = tmp_path / "store"
+        root.mkdir()
+        sentinel = tmp_path / "sentinel.txt"
+        sentinel.write_text("do not delete")
+        store = PayloadStore(root=root)
+
+        with pytest.raises(ValueError, match="Invalid collection name"):
+            store.delete("..")
+
+        assert sentinel.exists()
+        assert root.exists()
+
+    def test_manifest_rejects_collection_path_traversal(self, tmp_path: Path) -> None:
+        root = tmp_path / "store"
+        root.mkdir()
+        outside_manifest = tmp_path / "manifest.json"
+        outside_manifest.write_text('{"collection": "outside"}')
+        store = PayloadStore(root=root)
+
+        with pytest.raises(ValueError, match="Invalid collection name"):
+            store.manifest("..")
+
+    @pytest.mark.parametrize("alias", ["victim.", "victim...", "victim "])
+    def test_delete_rejects_windows_trailing_alias(
+        self,
+        store: PayloadStore,
+        alias: str,
+    ) -> None:
+        store.save("victim", payloads=[Payload(content="safe", id="p1")])
+
+        with pytest.raises(ValueError, match="Invalid collection name"):
+            store.delete(alias)
+
+        assert store.exists("victim")
+
+    @pytest.mark.parametrize("name", ["CON", "NUL", "COM1", "LPT9", "CON.txt"])
+    def test_save_rejects_windows_reserved_basename(
+        self,
+        store: PayloadStore,
+        name: str,
+    ) -> None:
+        with pytest.raises(ValueError, match="Invalid collection name"):
+            store.save(name, payloads=[Payload(content="x", id="p1")])
+
+    @pytest.mark.parametrize("name", ["team payloads", "チーム", "x" * 129])
+    def test_existing_collection_name_compatibility(
+        self,
+        store: PayloadStore,
+        name: str,
+    ) -> None:
+        store.save(name, payloads=[Payload(content="x", id="p1")])
+
+        assert name in store.list_collections()
+        assert store.exists(name)
+        assert store.load(name)[0].content == "x"
+        assert store.manifest(name)["collection"] == name
+
+        store.delete(name)
+        assert not store.exists(name)
+
 
 class TestPayloadStorePathPayload:
     def test_path_based_payload_roundtrip(