[FIX]: contain deserialized payload artifacts

rampart/payloads/_store.py

@@ -331,6 +331,62 @@ 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 _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/."
+        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:
+            artifact_relative = artifact_path.relative_to("artifacts")
+        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=artifacts_dir,
+            description="artifact path",
+        )
+        return resolved
+
     @staticmethod
     def _deserialize(
         *,
@@ -349,12 +405,16 @@ def _deserialize(
 
         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(
+                collection_dir=collection_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,121 @@
+# 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")
+
+
+@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")