[FIX]: contain deserialized payload artifacts

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)
@@ -331,30 +339,96 @@ 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.
+
+        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):
+            msg = f"Invalid {description}: {path!s} escapes {directory!s}"
+            raise ValueError(msg)
+
+    @staticmethod
+    def _validate_artifact_reference(artifact: object) -> Path:
+        """Validate and normalize a serialized artifact reference.
+
+        Returns:
+            Path: The artifact path relative to the artifacts directory.
+
+        Raises:
+            ValueError: If the reference is not a relative path under artifacts/.
+        """
+        msg = f"Invalid artifact path: {artifact!r}. Must be under artifacts/."
+        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)
+        try:
+            artifact_relative = artifact_path.relative_to("artifacts")
+        except ValueError as exc:
+            raise ValueError(msg) from exc
+
+        if artifact_relative == Path():
+            raise ValueError(msg)
+        return artifact_relative
+
+    @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,
+            description="artifact path",
+        )
+        return resolved
+
     @staticmethod
     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.
 
         Raises:
             FileNotFoundError: If a referenced artifact is missing.
+            ValueError: If a referenced artifact path is invalid.
         """
         fmt = PayloadFormat(data["format"])
 
         artifact: Path | None = None
         if "artifact" in data:
-            artifact_path = collection_dir / data["artifact"]
+            artifact_path = PayloadStore._resolve_artifact_path(
+                artifacts_dir=artifacts_dir,
+                artifact=data["artifact"],
+            )
             if not artifact_path.exists():
                 msg = f"Missing artifact: {artifact_path}"
                 raise FileNotFoundError(msg)

tests/unit/payloads/test_payload_store_security.py

@@ -0,0 +1,126 @@
+# 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: object) -> 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",
+        encoding="utf-8",
+    )
+
+
+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")