diff --git a/rampart/payloads/_store.py b/rampart/payloads/_store.py index 91cef07..8884ca5 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) @@ -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) 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..b471348 --- /dev/null +++ b/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")