From da32e601d87da883a0cc38edd787fb49509cfd1f Mon Sep 17 00:00:00 2001 From: hinotoi-agent Date: Thu, 21 May 2026 10:38:09 +0800 Subject: [PATCH 1/2] fix: validate payload store collection names --- rampart/payloads/_store.py | 12 ++++++++++-- tests/unit/payloads/test_store.py | 30 ++++++++++++++++++++++++++++++ 2 files changed, 40 insertions(+), 2 deletions(-) diff --git a/rampart/payloads/_store.py b/rampart/payloads/_store.py index 91cef07..63ef037 100644 --- a/rampart/payloads/_store.py +++ b/rampart/payloads/_store.py @@ -26,6 +26,7 @@ import json import logging +import re import shutil import tempfile from pathlib import Path @@ -34,6 +35,7 @@ from rampart.core.types import Payload, PayloadFormat logger = logging.getLogger(__name__) +_COLLECTION_NAME_PATTERN = re.compile(r"^[A-Za-z0-9._-]{1,128}$") class PayloadStore: @@ -162,6 +164,7 @@ def load( def exists(self, name: str) -> bool: """Check whether a collection exists on disk.""" + self._validate_collection_name(name) return self._collection_path(name).exists() def list_collections(self) -> list[str]: @@ -176,6 +179,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) @@ -193,6 +197,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}'" @@ -205,8 +210,11 @@ def manifest(self, name: str) -> dict[str, Any]: @staticmethod def _validate_collection_name(name: str) -> None: """Reject names that would escape the store root.""" - 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." + if not _COLLECTION_NAME_PATTERN.fullmatch(name) or name in (".", ".."): + msg = ( + f"Invalid collection name: {name!r}. Must be a filename-safe " + "identifier using letters, numbers, dots, underscores, or hyphens." + ) raise ValueError( msg, ) diff --git a/tests/unit/payloads/test_store.py b/tests/unit/payloads/test_store.py index 30e3cdd..b9a7715 100644 --- a/tests/unit/payloads/test_store.py +++ b/tests/unit/payloads/test_store.py @@ -136,6 +136,36 @@ 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("..") + class TestPayloadStorePathPayload: def test_path_based_payload_roundtrip( From 62eaf0a88c88df86c8dc179d282f4e4614b00343 Mon Sep 17 00:00:00 2001 From: hinotoi-agent Date: Thu, 23 Jul 2026 09:45:57 +0800 Subject: [PATCH 2/2] fix: preserve compatible payload collection names Signed-off-by: hinotoi-agent --- rampart/payloads/_store.py | 28 +++++++++++++++++------ tests/unit/payloads/test_store.py | 38 +++++++++++++++++++++++++++++++ 2 files changed, 59 insertions(+), 7 deletions(-) diff --git a/rampart/payloads/_store.py b/rampart/payloads/_store.py index 0f8e80b..40716a0 100644 --- a/rampart/payloads/_store.py +++ b/rampart/payloads/_store.py @@ -26,7 +26,6 @@ import json import logging -import re import shutil import tempfile from pathlib import Path @@ -35,7 +34,14 @@ from rampart.core.types import Payload, PayloadFormat logger = logging.getLogger(__name__) -_COLLECTION_NAME_PATTERN = re.compile(r"^[A-Za-z0-9._-]{1,128}$") +_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: @@ -216,15 +222,23 @@ 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 not a filename-safe identifier. + ValueError: If the name is not a portable directory name. """ - if not _COLLECTION_NAME_PATTERN.fullmatch(name) or name in {".", ".."}: + 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 filename-safe " - "identifier using letters, numbers, dots, underscores, or hyphens." + f"Invalid collection name: {name!r}. Must be a portable directory name." ) raise ValueError(msg) diff --git a/tests/unit/payloads/test_store.py b/tests/unit/payloads/test_store.py index b9a7715..74f4483 100644 --- a/tests/unit/payloads/test_store.py +++ b/tests/unit/payloads/test_store.py @@ -166,6 +166,44 @@ def test_manifest_rejects_collection_path_traversal(self, tmp_path: Path) -> Non 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(