feat: add bounded artifact reader

src/research_signal_context_pipelines/__init__.py

@@ -1,5 +1,6 @@
 """Research signal context artifact helpers."""
 
+from .artifact_io import DEFAULT_MAX_ARTIFACT_BYTES, read_bounded_artifact_bytes
 from .overlay_backtest import OverlayPolicy, backtest_overlay
 from .context_bundle import DEFAULT_UNIVERSE, build_context_bundle, build_context_from_source
 from .price_history import PriceExtractionSummary, write_filtered_price_history
@@ -16,13 +17,15 @@
     "PriceExtractionSummary",
     "SignalValidationError",
     "DEFAULT_UNIVERSE",
+    "DEFAULT_MAX_ARTIFACT_BYTES",
     "backtest_overlay",
     "build_context_bundle",
     "build_context_from_source",
     "build_theme_context",
     "build_theme_momentum_snapshot",
     "load_symbol_theme_exposure",
     "load_theme_taxonomy",
+    "read_bounded_artifact_bytes",
     "validate_signal",
     "validate_theme_momentum_snapshot",
     "write_filtered_price_history",

src/research_signal_context_pipelines/artifact_io.py

@@ -0,0 +1,106 @@
+from __future__ import annotations
+
+import os
+import stat
+from pathlib import Path
+
+from .schema import SignalValidationError
+
+
+DEFAULT_MAX_ARTIFACT_BYTES = 4 * 1024 * 1024
+_OPEN_SUPPORTS_DIR_FD = os.open in os.supports_dir_fd
+
+
+def read_bounded_artifact_bytes(
+    source_path: str | Path,
+    *,
+    base_dir: str | Path,
+    max_bytes: int = DEFAULT_MAX_ARTIFACT_BYTES,
+) -> bytes:
+    """Read one regular artifact beneath a directory without following symlinks."""
+    if not isinstance(max_bytes, int) or isinstance(max_bytes, bool) or max_bytes <= 0:
+        raise SignalValidationError("artifact size limit must be a positive integer")
+    _require_safe_open_support()
+
+    raw_base = Path(os.path.abspath(os.fspath(base_dir)))
+    try:
+        canonical_base = raw_base.resolve(strict=True)
+    except OSError as exc:
+        raise SignalValidationError("artifact base directory is unavailable") from exc
+    if not canonical_base.is_dir():
+        raise SignalValidationError("artifact base must be a directory")
+
+    parts = _relative_source_parts(source_path, raw_base=raw_base, canonical_base=canonical_base)
+    opened_fds: list[int] = []
+    try:
+        base_fd = os.open(
+            canonical_base,
+            os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC,
+        )
+        opened_fds.append(base_fd)
+        parent_fd = base_fd
+        for part in parts[:-1]:
+            parent_fd = os.open(
+                part,
+                os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC,
+                dir_fd=parent_fd,
+            )
+            opened_fds.append(parent_fd)
+        artifact_fd = os.open(
+            parts[-1],
+            os.O_RDONLY | os.O_NONBLOCK | os.O_NOFOLLOW | os.O_CLOEXEC,
+            dir_fd=parent_fd,
+        )
+        opened_fds.append(artifact_fd)
+        metadata = os.fstat(artifact_fd)
+        if not stat.S_ISREG(metadata.st_mode):
+            raise SignalValidationError("artifact source must be a regular file")
+        if metadata.st_size > max_bytes:
+            raise SignalValidationError("artifact exceeds the configured size limit")
+        with os.fdopen(artifact_fd, "rb", closefd=False) as stream:
+            payload = stream.read(max_bytes + 1)
+        if len(payload) > max_bytes:
+            raise SignalValidationError("artifact exceeds the configured size limit")
+        return payload
+    except SignalValidationError:
+        raise
+    except (OSError, ValueError) as exc:
+        raise SignalValidationError("artifact source must be a readable regular file beneath base") from exc
+    finally:
+        for fd in reversed(opened_fds):
+            try:
+                os.close(fd)
+            except OSError:
+                pass
+
+
+def _require_safe_open_support() -> None:
+    required_flags = ("O_CLOEXEC", "O_DIRECTORY", "O_NOFOLLOW", "O_NONBLOCK")
+    if any(not hasattr(os, flag) for flag in required_flags) or not _OPEN_SUPPORTS_DIR_FD:
+        raise SignalValidationError("safe descriptor-relative artifact reads are unsupported")
+
+
+def _relative_source_parts(
+    source_path: str | Path,
+    *,
+    raw_base: Path,
+    canonical_base: Path,
+) -> tuple[str, ...]:
+    source = Path(source_path)
+    if source.is_absolute():
+        relative = _relative_to_either_base(source, raw_base=raw_base, canonical_base=canonical_base)
+    else:
+        relative = source
+    parts = relative.parts
+    if not parts or any(part in {"", ".", ".."} for part in parts):
+        raise SignalValidationError("artifact source must remain beneath base")
+    return parts
+
+
+def _relative_to_either_base(source: Path, *, raw_base: Path, canonical_base: Path) -> Path:
+    for base in (raw_base, canonical_base):
+        try:
+            return source.relative_to(base)
+        except ValueError:
+            continue
+    raise SignalValidationError("artifact source must remain beneath base")

tests/test_artifact_io.py

@@ -0,0 +1,129 @@
+from __future__ import annotations
+
+import os
+from pathlib import Path
+
+import pytest
+
+from research_signal_context_pipelines import SignalValidationError, read_bounded_artifact_bytes
+
+
+def test_reads_relative_source_once_from_base(tmp_path: Path) -> None:
+    source = tmp_path / "artifact.json"
+    source.write_bytes(b'{"ok": true}')
+
+    assert read_bounded_artifact_bytes("artifact.json", base_dir=tmp_path) == b'{"ok": true}'
+
+
+def test_reads_absolute_source_through_symlinked_base(tmp_path: Path) -> None:
+    release = tmp_path / "releases" / "current"
+    release.mkdir(parents=True)
+    source = release / "artifact.json"
+    source.write_bytes(b"payload")
+    deployed = tmp_path / "deploy-current"
+    deployed.symlink_to(release, target_is_directory=True)
+
+    assert read_bounded_artifact_bytes(deployed / source.name, base_dir=deployed) == b"payload"
+
+
+def test_reads_absolute_source_through_lexically_normalized_symlinked_base(tmp_path: Path) -> None:
+    deploys = tmp_path / "deploys"
+    release = tmp_path / "releases" / "current"
+    deploys.mkdir()
+    release.mkdir(parents=True)
+    source = release / "artifact.json"
+    source.write_bytes(b"payload")
+    deployed = deploys / "current"
+    deployed.symlink_to(release, target_is_directory=True)
+    configured_base = deploys / ".." / "deploys" / "current"
+
+    assert read_bounded_artifact_bytes(deployed / source.name, base_dir=configured_base) == b"payload"
+
+
+def test_reads_absolute_source_under_canonical_base(tmp_path: Path) -> None:
+    release = tmp_path / "release"
+    release.mkdir()
+    source = release / "artifact.json"
+    source.write_bytes(b"payload")
+    deployed = tmp_path / "deploy-current"
+    deployed.symlink_to(release, target_is_directory=True)
+
+    assert read_bounded_artifact_bytes(source, base_dir=deployed) == b"payload"
+
+
+@pytest.mark.parametrize("source", ["../outside.json", "nested/../../outside.json"])
+def test_rejects_relative_escape(tmp_path: Path, source: str) -> None:
+    (tmp_path.parent / "outside.json").write_bytes(b"outside")
+
+    with pytest.raises(SignalValidationError, match="beneath base"):
+        read_bounded_artifact_bytes(source, base_dir=tmp_path)
+
+
+def test_rejects_absolute_source_outside_base(tmp_path: Path) -> None:
+    base = tmp_path / "base"
+    base.mkdir()
+    outside = tmp_path / "outside.json"
+    outside.write_bytes(b"outside")
+
+    with pytest.raises(SignalValidationError, match="beneath base"):
+        read_bounded_artifact_bytes(outside, base_dir=base)
+
+
+def test_rejects_leaf_symlink(tmp_path: Path) -> None:
+    target = tmp_path / "target.json"
+    target.write_bytes(b"payload")
+    (tmp_path / "artifact.json").symlink_to(target)
+
+    with pytest.raises(SignalValidationError, match="readable regular file"):
+        read_bounded_artifact_bytes("artifact.json", base_dir=tmp_path)
+
+
+def test_rejects_intermediate_symlink(tmp_path: Path) -> None:
+    real_dir = tmp_path / "real"
+    real_dir.mkdir()
+    (real_dir / "artifact.json").write_bytes(b"payload")
+    (tmp_path / "linked").symlink_to(real_dir, target_is_directory=True)
+
+    with pytest.raises(SignalValidationError, match="readable regular file"):
+        read_bounded_artifact_bytes("linked/artifact.json", base_dir=tmp_path)
+
+
+def test_rejects_fifo_without_blocking(tmp_path: Path) -> None:
+    fifo = tmp_path / "artifact.pipe"
+    os.mkfifo(fifo)
+
+    with pytest.raises(SignalValidationError, match="regular file"):
+        read_bounded_artifact_bytes(fifo.name, base_dir=tmp_path)
+
+
+def test_rejects_oversized_artifact(tmp_path: Path) -> None:
+    source = tmp_path / "artifact.json"
+    source.write_bytes(b"12345")
+
+    with pytest.raises(SignalValidationError, match="size limit"):
+        read_bounded_artifact_bytes(source.name, base_dir=tmp_path, max_bytes=4)
+
+
+def test_closes_all_opened_descriptors(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    source = tmp_path / "nested" / "artifact.json"
+    source.parent.mkdir()
+    source.write_bytes(b"payload")
+    real_open = os.open
+    real_close = os.close
+    opened: list[int] = []
+    closed: list[int] = []
+
+    def tracking_open(*args, **kwargs) -> int:
+        fd = real_open(*args, **kwargs)
+        opened.append(fd)
+        return fd
+
+    def tracking_close(fd: int) -> None:
+        closed.append(fd)
+        real_close(fd)
+
+    monkeypatch.setattr(os, "open", tracking_open)
+    monkeypatch.setattr(os, "close", tracking_close)
+
+    assert read_bounded_artifact_bytes("nested/artifact.json", base_dir=tmp_path) == b"payload"
+    assert set(opened) <= set(closed)