From 8d576393463c46ce84181b289052a42fe8619898 Mon Sep 17 00:00:00 2001 From: hinotoi-agent Date: Wed, 8 Jul 2026 08:24:24 +0800 Subject: [PATCH 1/8] fix: contain loaded payload artifacts --- rampart/payloads/_store.py | 42 +++++++++++- .../payloads/test_payload_store_security.py | 67 +++++++++++++++++++ 2 files changed, 108 insertions(+), 1 deletion(-) create mode 100644 tests/unit/payloads/test_payload_store_security.py diff --git a/rampart/payloads/_store.py b/rampart/payloads/_store.py index 91cef07..267654f 100644 --- a/rampart/payloads/_store.py +++ b/rampart/payloads/_store.py @@ -34,6 +34,7 @@ from rampart.core.types import Payload, PayloadFormat logger = logging.getLogger(__name__) +_MIN_ARTIFACT_PATH_PARTS = 2 class PayloadStore: @@ -331,6 +332,42 @@ def _copy_file_artifact( shutil.copy2(source, artifacts_dir / filename) return f"artifacts/{filename}" + @staticmethod + def _ensure_within_directory( + *, + path: Path, + directory: Path, + description: str, + ) -> None: + """Raise if a resolved path escapes a required directory.""" + resolved_path = path.resolve(strict=False) + resolved_directory = directory.resolve(strict=False) + if not resolved_path.is_relative_to(resolved_directory): + msg = f"Invalid {description}: {path!s} escapes {directory!s}" + raise ValueError(msg) + + @staticmethod + def _resolve_artifact_path(*, collection_dir: Path, artifact: str) -> Path: + """Resolve a serialized artifact path inside collection artifacts/.""" + artifact_path = Path(artifact) + if artifact_path.is_absolute() or ".." in artifact_path.parts: + msg = f"Invalid artifact path: {artifact!r}. Must stay under artifacts/." + raise ValueError(msg) + if ( + len(artifact_path.parts) < _MIN_ARTIFACT_PATH_PARTS + or artifact_path.parts[0] != "artifacts" + ): + msg = f"Invalid artifact path: {artifact!r}. Must be under artifacts/." + raise ValueError(msg) + + resolved = collection_dir / artifact_path + PayloadStore._ensure_within_directory( + path=resolved, + directory=collection_dir / "artifacts", + description="artifact path", + ) + return resolved + @staticmethod def _deserialize( *, @@ -354,7 +391,10 @@ def _deserialize( artifact: Path | None = None if "artifact" in data: - artifact_path = collection_dir / data["artifact"] + artifact_path = PayloadStore._resolve_artifact_path( + collection_dir=collection_dir, + artifact=data["artifact"], + ) if not artifact_path.exists(): msg = f"Missing artifact: {artifact_path}" raise FileNotFoundError(msg) diff --git a/tests/unit/payloads/test_payload_store_security.py b/tests/unit/payloads/test_payload_store_security.py new file mode 100644 index 0000000..d639fe4 --- /dev/null +++ b/tests/unit/payloads/test_payload_store_security.py @@ -0,0 +1,67 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT license. + +"""Security tests for payload artifact path handling.""" + +import json +from pathlib import Path + +import pytest + +from rampart.payloads._store import PayloadStore + + +def _write_collection_record(collection_dir: Path, artifact: str) -> None: + collection_dir.mkdir(parents=True, exist_ok=True) + record: dict[str, object] = { + "id": "safe-id", + "content": "binary", + "format": "pdf", + "metadata": {}, + "artifact": artifact, + } + (collection_dir / "payloads.jsonl").write_text(json.dumps(record) + "\n") + + +@pytest.mark.parametrize( + "artifact", + [ + "../outside.pdf", + "artifacts/../outside.pdf", + "/tmp/outside.pdf", + "outside.pdf", + ], +) +def test_payload_store_rejects_deserialized_artifact_escape( + tmp_path: Path, + artifact: str, +) -> None: + """Serialized artifact paths must stay under the collection artifacts dir.""" + collection_dir = tmp_path / "store" / "collection" + _write_collection_record(collection_dir, artifact) + + store = PayloadStore(root=tmp_path / "store") + with pytest.raises(ValueError, match="Invalid artifact path"): + store.load("collection") + + +def test_payload_store_rejects_deserialized_artifact_symlink_escape( + tmp_path: Path, +) -> None: + """Serialized artifact paths cannot resolve through symlinks outside artifacts.""" + collection_dir = tmp_path / "store" / "collection" + artifacts_dir = collection_dir / "artifacts" + artifacts_dir.mkdir(parents=True) + outside = tmp_path / "outside.pdf" + outside.write_bytes(b"outside") + symlink = artifacts_dir / "linked.pdf" + try: + symlink.symlink_to(outside) + except OSError as exc: + pytest.skip(f"symlinks are not available on this platform: {exc}") + + _write_collection_record(collection_dir, "artifacts/linked.pdf") + + store = PayloadStore(root=tmp_path / "store") + with pytest.raises(ValueError, match="escapes"): + store.load("collection") From 261a9669f5594e1f6049a4b5744e9d069a0eab67 Mon Sep 17 00:00:00 2001 From: hinotoi-agent Date: Wed, 8 Jul 2026 10:41:10 +0800 Subject: [PATCH 2/8] refactor: use relative_to for artifact prefix check --- rampart/payloads/_store.py | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/rampart/payloads/_store.py b/rampart/payloads/_store.py index 267654f..fe90d45 100644 --- a/rampart/payloads/_store.py +++ b/rampart/payloads/_store.py @@ -34,7 +34,6 @@ from rampart.core.types import Payload, PayloadFormat logger = logging.getLogger(__name__) -_MIN_ARTIFACT_PATH_PARTS = 2 class PayloadStore: @@ -353,10 +352,13 @@ def _resolve_artifact_path(*, collection_dir: Path, artifact: str) -> Path: if artifact_path.is_absolute() or ".." in artifact_path.parts: msg = f"Invalid artifact path: {artifact!r}. Must stay under artifacts/." raise ValueError(msg) - if ( - len(artifact_path.parts) < _MIN_ARTIFACT_PATH_PARTS - or artifact_path.parts[0] != "artifacts" - ): + try: + artifact_relative = artifact_path.relative_to("artifacts") + except ValueError: + msg = f"Invalid artifact path: {artifact!r}. Must be under artifacts/." + raise ValueError(msg) from None + + if not artifact_relative.parts: msg = f"Invalid artifact path: {artifact!r}. Must be under artifacts/." raise ValueError(msg) From 2e19dadad751d12deb0156233421ad5d4b0d8639 Mon Sep 17 00:00:00 2001 From: hinotoi-agent Date: Wed, 22 Jul 2026 08:07:00 +0800 Subject: [PATCH 3/8] test: cover artifact path validation branches --- rampart/payloads/_store.py | 6 ++---- tests/unit/payloads/test_payload_store_security.py | 13 +++++++++++++ 2 files changed, 15 insertions(+), 4 deletions(-) diff --git a/rampart/payloads/_store.py b/rampart/payloads/_store.py index fe90d45..4458a42 100644 --- a/rampart/payloads/_store.py +++ b/rampart/payloads/_store.py @@ -348,18 +348,16 @@ def _ensure_within_directory( @staticmethod def _resolve_artifact_path(*, collection_dir: Path, artifact: str) -> Path: """Resolve a serialized artifact path inside collection artifacts/.""" + msg = f"Invalid artifact path: {artifact!r}. Must be under artifacts/." artifact_path = Path(artifact) if artifact_path.is_absolute() or ".." in artifact_path.parts: - msg = f"Invalid artifact path: {artifact!r}. Must stay under artifacts/." raise ValueError(msg) try: artifact_relative = artifact_path.relative_to("artifacts") except ValueError: - msg = f"Invalid artifact path: {artifact!r}. Must be under artifacts/." - raise ValueError(msg) from None + raise ValueError(msg) # noqa: B904 - preserve the exception chain if not artifact_relative.parts: - msg = f"Invalid artifact path: {artifact!r}. Must be under artifacts/." raise ValueError(msg) resolved = collection_dir / artifact_path diff --git a/tests/unit/payloads/test_payload_store_security.py b/tests/unit/payloads/test_payload_store_security.py index d639fe4..59c34f7 100644 --- a/tests/unit/payloads/test_payload_store_security.py +++ b/tests/unit/payloads/test_payload_store_security.py @@ -30,6 +30,7 @@ def _write_collection_record(collection_dir: Path, artifact: str) -> None: "artifacts/../outside.pdf", "/tmp/outside.pdf", "outside.pdf", + "artifacts", ], ) def test_payload_store_rejects_deserialized_artifact_escape( @@ -65,3 +66,15 @@ def test_payload_store_rejects_deserialized_artifact_symlink_escape( store = PayloadStore(root=tmp_path / "store") with pytest.raises(ValueError, match="escapes"): store.load("collection") + + +def test_payload_store_rejects_missing_deserialized_artifact( + tmp_path: Path, +) -> None: + """A valid serialized artifact path must refer to an existing file.""" + collection_dir = tmp_path / "store" / "collection" + _write_collection_record(collection_dir, "artifacts/gone.pdf") + + store = PayloadStore(root=tmp_path / "store") + with pytest.raises(FileNotFoundError, match="Missing artifact"): + store.load("collection") From b6cef53af593660185325034aac44ad8d59456af Mon Sep 17 00:00:00 2001 From: hinotoi-agent Date: Wed, 22 Jul 2026 08:43:36 +0800 Subject: [PATCH 4/8] fix: reject symlinked artifact directories --- rampart/payloads/_store.py | 12 +++++++--- .../payloads/test_payload_store_security.py | 24 +++++++++++++++++++ 2 files changed, 33 insertions(+), 3 deletions(-) diff --git a/rampart/payloads/_store.py b/rampart/payloads/_store.py index 4458a42..fe1170c 100644 --- a/rampart/payloads/_store.py +++ b/rampart/payloads/_store.py @@ -354,16 +354,22 @@ def _resolve_artifact_path(*, collection_dir: Path, artifact: str) -> Path: raise ValueError(msg) try: artifact_relative = artifact_path.relative_to("artifacts") - except ValueError: - raise ValueError(msg) # noqa: B904 - preserve the exception chain + except ValueError as exc: + raise ValueError(msg) from exc if not artifact_relative.parts: raise ValueError(msg) + artifacts_dir = collection_dir / "artifacts" + PayloadStore._ensure_within_directory( + path=artifacts_dir, + directory=collection_dir, + description="artifacts directory", + ) resolved = collection_dir / artifact_path PayloadStore._ensure_within_directory( path=resolved, - directory=collection_dir / "artifacts", + directory=artifacts_dir, description="artifact path", ) return resolved diff --git a/tests/unit/payloads/test_payload_store_security.py b/tests/unit/payloads/test_payload_store_security.py index 59c34f7..d204fd8 100644 --- a/tests/unit/payloads/test_payload_store_security.py +++ b/tests/unit/payloads/test_payload_store_security.py @@ -68,6 +68,30 @@ def test_payload_store_rejects_deserialized_artifact_symlink_escape( store.load("collection") +def test_payload_store_rejects_deserialized_artifacts_directory_symlink_escape( + tmp_path: Path, +) -> None: + """The collection artifacts directory cannot resolve outside the collection.""" + collection_dir = tmp_path / "store" / "collection" + collection_dir.mkdir(parents=True) + outside_dir = tmp_path / "outside" + outside_dir.mkdir() + (outside_dir / "linked.pdf").write_bytes(b"outside") + try: + (collection_dir / "artifacts").symlink_to( + outside_dir, + target_is_directory=True, + ) + except OSError as exc: + pytest.skip(f"symlinks are not available on this platform: {exc}") + + _write_collection_record(collection_dir, "artifacts/linked.pdf") + + store = PayloadStore(root=tmp_path / "store") + with pytest.raises(ValueError, match=r"artifacts directory.*escapes"): + store.load("collection") + + def test_payload_store_rejects_missing_deserialized_artifact( tmp_path: Path, ) -> None: From 107e8ce1e79ac7e8d7cb0b7fff4d6a05b91ec6dc Mon Sep 17 00:00:00 2001 From: hinotoi-agent Date: Thu, 23 Jul 2026 09:52:08 +0800 Subject: [PATCH 5/8] fix: normalize invalid artifact reference types Signed-off-by: hinotoi-agent --- rampart/payloads/_store.py | 24 +++++++++++++++---- .../payloads/test_payload_store_security.py | 19 ++++++++++++++- 2 files changed, 37 insertions(+), 6 deletions(-) diff --git a/rampart/payloads/_store.py b/rampart/payloads/_store.py index fe1170c..f2427de 100644 --- a/rampart/payloads/_store.py +++ b/rampart/payloads/_store.py @@ -29,7 +29,7 @@ import shutil import tempfile from pathlib import Path -from typing import Any +from typing import Any, cast from rampart.core.types import Payload, PayloadFormat @@ -338,7 +338,11 @@ def _ensure_within_directory( directory: Path, description: str, ) -> None: - """Raise if a resolved path escapes a required directory.""" + """Raise if a resolved path escapes a required directory. + + Raises: + ValueError: If the resolved path is outside the directory. + """ resolved_path = path.resolve(strict=False) resolved_directory = directory.resolve(strict=False) if not resolved_path.is_relative_to(resolved_directory): @@ -346,10 +350,20 @@ def _ensure_within_directory( raise ValueError(msg) @staticmethod - def _resolve_artifact_path(*, collection_dir: Path, artifact: str) -> Path: - """Resolve a serialized artifact path inside collection artifacts/.""" + def _resolve_artifact_path(*, collection_dir: Path, artifact: object) -> Path: + """Resolve a serialized artifact path inside collection artifacts/. + + Returns: + Path: The validated artifact path. + + Raises: + ValueError: If the artifact reference is not a contained string path. + """ msg = f"Invalid artifact path: {artifact!r}. Must be under artifacts/." - artifact_path = Path(artifact) + try: + artifact_path = Path(cast("str", artifact)) + except TypeError as exc: + raise ValueError(msg) from exc if artifact_path.is_absolute() or ".." in artifact_path.parts: raise ValueError(msg) try: diff --git a/tests/unit/payloads/test_payload_store_security.py b/tests/unit/payloads/test_payload_store_security.py index d204fd8..c505002 100644 --- a/tests/unit/payloads/test_payload_store_security.py +++ b/tests/unit/payloads/test_payload_store_security.py @@ -11,7 +11,7 @@ from rampart.payloads._store import PayloadStore -def _write_collection_record(collection_dir: Path, artifact: str) -> None: +def _write_collection_record(collection_dir: Path, artifact: object) -> None: collection_dir.mkdir(parents=True, exist_ok=True) record: dict[str, object] = { "id": "safe-id", @@ -46,6 +46,23 @@ def test_payload_store_rejects_deserialized_artifact_escape( store.load("collection") +@pytest.mark.parametrize( + "artifact", + [None, 7, ["artifacts/file.pdf"], {"path": "artifacts/file.pdf"}], +) +def test_payload_store_rejects_non_string_deserialized_artifact( + tmp_path: Path, + artifact: object, +) -> None: + """Serialized artifact references must be strings.""" + collection_dir = tmp_path / "store" / "collection" + _write_collection_record(collection_dir, artifact) + + store = PayloadStore(root=tmp_path / "store") + with pytest.raises(ValueError, match="Invalid artifact path"): + store.load("collection") + + def test_payload_store_rejects_deserialized_artifact_symlink_escape( tmp_path: Path, ) -> None: From c48d3ce31e79ccbfd5e9b529d84fa8b20d023993 Mon Sep 17 00:00:00 2001 From: hinotoi-agent Date: Tue, 28 Jul 2026 12:27:13 +0800 Subject: [PATCH 6/8] fix: clarify artifact reference validation Signed-off-by: hinotoi-agent --- rampart/payloads/_store.py | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/rampart/payloads/_store.py b/rampart/payloads/_store.py index f2427de..e1342ba 100644 --- a/rampart/payloads/_store.py +++ b/rampart/payloads/_store.py @@ -29,7 +29,7 @@ import shutil import tempfile from pathlib import Path -from typing import Any, cast +from typing import Any from rampart.core.types import Payload, PayloadFormat @@ -360,10 +360,9 @@ def _resolve_artifact_path(*, collection_dir: Path, artifact: object) -> Path: ValueError: If the artifact reference is not a contained string path. """ msg = f"Invalid artifact path: {artifact!r}. Must be under artifacts/." - try: - artifact_path = Path(cast("str", artifact)) - except TypeError as exc: - raise ValueError(msg) from exc + if not isinstance(artifact, str): + raise ValueError(msg) # noqa: TRY004 - stable deserialization error + artifact_path = Path(artifact) if artifact_path.is_absolute() or ".." in artifact_path.parts: raise ValueError(msg) try: @@ -406,6 +405,7 @@ def _deserialize( Raises: FileNotFoundError: If a referenced artifact is missing. + ValueError: If a referenced artifact path is invalid. """ fmt = PayloadFormat(data["format"]) From 7de1c10ee1bd9edbe309b1a8818e6187f6412110 Mon Sep 17 00:00:00 2001 From: hinotoi-agent Date: Thu, 6 Aug 2026 09:33:47 +0800 Subject: [PATCH 7/8] refactor: address artifact containment review Signed-off-by: hinotoi-agent --- rampart/payloads/_store.py | 45 ++-- .../payloads/test_payload_store_security.py | 194 +++++++++--------- 2 files changed, 127 insertions(+), 112 deletions(-) diff --git a/rampart/payloads/_store.py b/rampart/payloads/_store.py index e1342ba..8aa6b34 100644 --- a/rampart/payloads/_store.py +++ b/rampart/payloads/_store.py @@ -145,6 +145,14 @@ def load( msg, ) + collection_dir = payloads_path.parent + artifacts_dir = collection_dir / "artifacts" + self._ensure_within_directory( + path=artifacts_dir, + directory=collection_dir, + description="artifacts directory", + ) + payloads: list[Payload] = [] with payloads_path.open("r", encoding="utf-8") as f: for raw_line in f: @@ -153,7 +161,7 @@ def load( continue payload = self._deserialize( data=json.loads(line), - collection_dir=payloads_path.parent, + artifacts_dir=artifacts_dir, ) if format_filter is None or payload.format == format_filter: payloads.append(payload) @@ -350,14 +358,14 @@ def _ensure_within_directory( raise ValueError(msg) @staticmethod - def _resolve_artifact_path(*, collection_dir: Path, artifact: object) -> Path: - """Resolve a serialized artifact path inside collection artifacts/. + def _validate_artifact_reference(artifact: object) -> Path: + """Validate and normalize a serialized artifact reference. Returns: - Path: The validated artifact path. + Path: The artifact path relative to the artifacts directory. Raises: - ValueError: If the artifact reference is not a contained string path. + ValueError: If the reference is not a relative path under artifacts/. """ msg = f"Invalid artifact path: {artifact!r}. Must be under artifacts/." if not isinstance(artifact, str): @@ -372,14 +380,20 @@ def _resolve_artifact_path(*, collection_dir: Path, artifact: object) -> Path: if not artifact_relative.parts: raise ValueError(msg) + return artifact_relative - artifacts_dir = collection_dir / "artifacts" - PayloadStore._ensure_within_directory( - path=artifacts_dir, - directory=collection_dir, - description="artifacts directory", - ) - resolved = collection_dir / artifact_path + @staticmethod + def _resolve_artifact_path(*, artifacts_dir: Path, artifact: object) -> Path: + """Resolve a serialized artifact path inside the artifacts directory. + + Returns: + Path: The validated artifact path. + + Raises: + ValueError: If the artifact reference escapes the artifacts directory. + """ + artifact_relative = PayloadStore._validate_artifact_reference(artifact) + resolved = artifacts_dir / artifact_relative PayloadStore._ensure_within_directory( path=resolved, directory=artifacts_dir, @@ -391,14 +405,13 @@ def _resolve_artifact_path(*, collection_dir: Path, artifact: object) -> Path: def _deserialize( *, data: dict[str, Any], - collection_dir: Path, + artifacts_dir: Path, ) -> Payload: """Deserialize a JSON record back to a Payload. Args: data (dict[str, Any]): JSON record from JSONL. - collection_dir (Path): Collection directory for resolving - artifact paths. + artifacts_dir (Path): Directory for resolving artifact paths. Returns: Payload: Reconstituted Payload. @@ -412,7 +425,7 @@ def _deserialize( artifact: Path | None = None if "artifact" in data: artifact_path = PayloadStore._resolve_artifact_path( - collection_dir=collection_dir, + artifacts_dir=artifacts_dir, artifact=data["artifact"], ) if not artifact_path.exists(): diff --git a/tests/unit/payloads/test_payload_store_security.py b/tests/unit/payloads/test_payload_store_security.py index c505002..5722ee7 100644 --- a/tests/unit/payloads/test_payload_store_security.py +++ b/tests/unit/payloads/test_payload_store_security.py @@ -23,99 +23,101 @@ def _write_collection_record(collection_dir: Path, artifact: object) -> None: (collection_dir / "payloads.jsonl").write_text(json.dumps(record) + "\n") -@pytest.mark.parametrize( - "artifact", - [ - "../outside.pdf", - "artifacts/../outside.pdf", - "/tmp/outside.pdf", - "outside.pdf", - "artifacts", - ], -) -def test_payload_store_rejects_deserialized_artifact_escape( - tmp_path: Path, - artifact: str, -) -> None: - """Serialized artifact paths must stay under the collection artifacts dir.""" - collection_dir = tmp_path / "store" / "collection" - _write_collection_record(collection_dir, artifact) - - store = PayloadStore(root=tmp_path / "store") - with pytest.raises(ValueError, match="Invalid artifact path"): - store.load("collection") - - -@pytest.mark.parametrize( - "artifact", - [None, 7, ["artifacts/file.pdf"], {"path": "artifacts/file.pdf"}], -) -def test_payload_store_rejects_non_string_deserialized_artifact( - tmp_path: Path, - artifact: object, -) -> None: - """Serialized artifact references must be strings.""" - collection_dir = tmp_path / "store" / "collection" - _write_collection_record(collection_dir, artifact) - - store = PayloadStore(root=tmp_path / "store") - with pytest.raises(ValueError, match="Invalid artifact path"): - store.load("collection") - - -def test_payload_store_rejects_deserialized_artifact_symlink_escape( - tmp_path: Path, -) -> None: - """Serialized artifact paths cannot resolve through symlinks outside artifacts.""" - collection_dir = tmp_path / "store" / "collection" - artifacts_dir = collection_dir / "artifacts" - artifacts_dir.mkdir(parents=True) - outside = tmp_path / "outside.pdf" - outside.write_bytes(b"outside") - symlink = artifacts_dir / "linked.pdf" - try: - symlink.symlink_to(outside) - except OSError as exc: - pytest.skip(f"symlinks are not available on this platform: {exc}") - - _write_collection_record(collection_dir, "artifacts/linked.pdf") - - store = PayloadStore(root=tmp_path / "store") - with pytest.raises(ValueError, match="escapes"): - store.load("collection") - - -def test_payload_store_rejects_deserialized_artifacts_directory_symlink_escape( - tmp_path: Path, -) -> None: - """The collection artifacts directory cannot resolve outside the collection.""" - collection_dir = tmp_path / "store" / "collection" - collection_dir.mkdir(parents=True) - outside_dir = tmp_path / "outside" - outside_dir.mkdir() - (outside_dir / "linked.pdf").write_bytes(b"outside") - try: - (collection_dir / "artifacts").symlink_to( - outside_dir, - target_is_directory=True, - ) - except OSError as exc: - pytest.skip(f"symlinks are not available on this platform: {exc}") - - _write_collection_record(collection_dir, "artifacts/linked.pdf") - - store = PayloadStore(root=tmp_path / "store") - with pytest.raises(ValueError, match=r"artifacts directory.*escapes"): - store.load("collection") - - -def test_payload_store_rejects_missing_deserialized_artifact( - tmp_path: Path, -) -> None: - """A valid serialized artifact path must refer to an existing file.""" - collection_dir = tmp_path / "store" / "collection" - _write_collection_record(collection_dir, "artifacts/gone.pdf") - - store = PayloadStore(root=tmp_path / "store") - with pytest.raises(FileNotFoundError, match="Missing artifact"): - store.load("collection") +class TestPayloadStoreArtifactContainment: + @pytest.mark.parametrize( + "artifact", + [ + "../outside.pdf", + "artifacts/../outside.pdf", + "/tmp/outside.pdf", + "outside.pdf", + "artifacts", + ], + ) + def test_payload_store_rejects_deserialized_artifact_escape( + self, + tmp_path: Path, + artifact: str, + ) -> None: + """Serialized artifact paths must stay under the collection artifacts dir.""" + collection_dir = tmp_path / "store" / "collection" + _write_collection_record(collection_dir, artifact) + + store = PayloadStore(root=tmp_path / "store") + with pytest.raises(ValueError, match="Invalid artifact path"): + store.load("collection") + + @pytest.mark.parametrize( + "artifact", + [None, 7, ["artifacts/file.pdf"], {"path": "artifacts/file.pdf"}], + ) + def test_payload_store_rejects_non_string_deserialized_artifact( + self, + tmp_path: Path, + artifact: object, + ) -> None: + """Serialized artifact references must be strings.""" + collection_dir = tmp_path / "store" / "collection" + _write_collection_record(collection_dir, artifact) + + store = PayloadStore(root=tmp_path / "store") + with pytest.raises(ValueError, match="Invalid artifact path"): + store.load("collection") + + def test_payload_store_rejects_deserialized_artifact_symlink_escape( + self, + tmp_path: Path, + ) -> None: + """Reject artifact paths resolving through symlinks outside artifacts.""" + collection_dir = tmp_path / "store" / "collection" + artifacts_dir = collection_dir / "artifacts" + artifacts_dir.mkdir(parents=True) + outside = tmp_path / "outside.pdf" + outside.write_bytes(b"outside") + symlink = artifacts_dir / "linked.pdf" + try: + symlink.symlink_to(outside) + except OSError as exc: + pytest.skip(f"symlinks are not available on this platform: {exc}") + + _write_collection_record(collection_dir, "artifacts/linked.pdf") + + store = PayloadStore(root=tmp_path / "store") + with pytest.raises(ValueError, match="escapes"): + store.load("collection") + + def test_payload_store_rejects_deserialized_artifacts_directory_symlink_escape( + self, + tmp_path: Path, + ) -> None: + """The collection artifacts directory cannot resolve outside the collection.""" + collection_dir = tmp_path / "store" / "collection" + collection_dir.mkdir(parents=True) + outside_dir = tmp_path / "outside" + outside_dir.mkdir() + (outside_dir / "linked.pdf").write_bytes(b"outside") + try: + (collection_dir / "artifacts").symlink_to( + outside_dir, + target_is_directory=True, + ) + except OSError as exc: + pytest.skip(f"symlinks are not available on this platform: {exc}") + + _write_collection_record(collection_dir, "artifacts/linked.pdf") + + store = PayloadStore(root=tmp_path / "store") + with pytest.raises(ValueError, match=r"artifacts directory.*escapes"): + store.load("collection") + + def test_payload_store_rejects_missing_deserialized_artifact( + self, + tmp_path: Path, + ) -> None: + """A valid serialized artifact path must refer to an existing file.""" + collection_dir = tmp_path / "store" / "collection" + _write_collection_record(collection_dir, "artifacts/gone.pdf") + + store = PayloadStore(root=tmp_path / "store") + with pytest.raises(FileNotFoundError, match="Missing artifact"): + store.load("collection") From 81028a97a1d4ee53dc70271407788d47b2d07e1d Mon Sep 17 00:00:00 2001 From: hinotoi-agent Date: Fri, 7 Aug 2026 23:27:37 +0800 Subject: [PATCH 8/8] fix: address artifact validation lint and review --- rampart/payloads/_store.py | 7 ++++--- tests/unit/payloads/test_payload_store_security.py | 5 ++++- 2 files changed, 8 insertions(+), 4 deletions(-) diff --git a/rampart/payloads/_store.py b/rampart/payloads/_store.py index 8aa6b34..8884ca5 100644 --- a/rampart/payloads/_store.py +++ b/rampart/payloads/_store.py @@ -368,8 +368,9 @@ def _validate_artifact_reference(artifact: object) -> Path: ValueError: If the reference is not a relative path under artifacts/. """ msg = f"Invalid artifact path: {artifact!r}. Must be under artifacts/." - if not isinstance(artifact, str): - raise ValueError(msg) # noqa: TRY004 - stable deserialization error + is_string_reference = isinstance(artifact, str) + if not is_string_reference: + raise ValueError(msg) artifact_path = Path(artifact) if artifact_path.is_absolute() or ".." in artifact_path.parts: raise ValueError(msg) @@ -378,7 +379,7 @@ def _validate_artifact_reference(artifact: object) -> Path: except ValueError as exc: raise ValueError(msg) from exc - if not artifact_relative.parts: + if artifact_relative == Path(): raise ValueError(msg) return artifact_relative diff --git a/tests/unit/payloads/test_payload_store_security.py b/tests/unit/payloads/test_payload_store_security.py index 5722ee7..b471348 100644 --- a/tests/unit/payloads/test_payload_store_security.py +++ b/tests/unit/payloads/test_payload_store_security.py @@ -20,7 +20,10 @@ def _write_collection_record(collection_dir: Path, artifact: object) -> None: "metadata": {}, "artifact": artifact, } - (collection_dir / "payloads.jsonl").write_text(json.dumps(record) + "\n") + (collection_dir / "payloads.jsonl").write_text( + json.dumps(record) + "\n", + encoding="utf-8", + ) class TestPayloadStoreArtifactContainment: