diff --git a/rampart/payloads/_store.py b/rampart/payloads/_store.py index 0e2ec1d..40716a0 100644 --- a/rampart/payloads/_store.py +++ b/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: diff --git a/tests/unit/payloads/test_store.py b/tests/unit/payloads/test_store.py index 30e3cdd..74f4483 100644 --- a/tests/unit/payloads/test_store.py +++ b/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(