Add path sanitizing checks to bionemo.core.data.load() [BIO-409] (#1546)

### Description

Make sure that tar paths from files downloaded by bionemo.core.data.load
are safe and local.

### Type of changes

<!-- Mark the relevant option with an [x] -->

- [x] Bug fix (non-breaking change which fixes an issue)
- [ ] New feature (non-breaking change which adds functionality)
- [ ] Refactor
- [ ] Documentation update
- [ ] Other (please describe):

### Pre-submit Checklist

<!--- Ensure all items are completed before submitting -->

- [x] I have tested these changes locally
- [x] I have updated the documentation accordingly
- [x] I have added/updated tests as needed
- [x] All existing tests pass successfully


<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->

## Summary by CodeRabbit

* **Bug Fixes**
* Implemented path-traversal protection for ZIP and TAR archive
extraction, preventing malicious archives from writing files outside
their designated extraction directory during loading.

* **Tests**
* Added comprehensive test coverage for archive extraction security
validations, including path traversal attack scenarios and safe
extraction verification.

<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Signed-off-by: John St John <jstjohn@nvidia.com>

Author: jstjohn

sub-packages/bionemo-core/src/bionemo/core/data/load.py

@@ -18,7 +18,9 @@
 import os
 import shutil
 import sys
+import tarfile
 import tempfile
+import zipfile
 from dataclasses import dataclass
 from pathlib import Path
 from typing import TYPE_CHECKING, Literal, Optional, Sequence, TextIO
@@ -225,6 +227,47 @@ def load(
     return path
 
 
+def _validate_archive_members(member_names: list[str], extract_dir: str) -> None:
+    """Validate that no archive members would be extracted outside the target directory.
+
+    This prevents Zip Slip / path traversal attacks where malicious archives contain entries
+    with relative paths (e.g., ``../../etc/passwd``) that escape the extraction directory.
+
+    Args:
+        member_names: List of member names from the archive.
+        extract_dir: The directory where files will be extracted.
+
+    Raises:
+        ValueError: If any member would be extracted outside the target directory.
+    """
+    safe_dir = os.path.normpath(extract_dir)
+    for name in member_names:
+        member_path = os.path.normpath(os.path.join(safe_dir, name))
+        if not (member_path == safe_dir or member_path.startswith(safe_dir + os.sep)):
+            raise ValueError(
+                f"Archive member '{name}' would be extracted outside "
+                f"the target directory '{safe_dir}'. This is a potential Zip Slip security risk."
+            )
+
+
+class _SafeUntar(pooch.Untar):
+    """Untar processor with path traversal validation."""
+
+    def _extract_file(self, fname, extract_dir):
+        with tarfile.open(fname, "r") as tar_file:
+            _validate_archive_members([m.name for m in tar_file.getmembers()], extract_dir)
+        super()._extract_file(fname, extract_dir)
+
+
+class _SafeUnzip(pooch.Unzip):
+    """Unzip processor with path traversal validation."""
+
+    def _extract_file(self, fname, extract_dir):
+        with zipfile.ZipFile(fname, "r") as zip_file:
+            _validate_archive_members(zip_file.namelist(), extract_dir)
+        super()._extract_file(fname, extract_dir)
+
+
 def _get_processor(extension: str, unpack: bool | None, decompress: bool | None):
     """Get the processor for a given file extension.
 
@@ -242,10 +285,10 @@ def _get_processor(extension: str, unpack: bool | None, decompress: bool | None)
         return pooch.Decompress()
 
     elif extension in {".tar", ".tar.gz"} and unpack is None:
-        return pooch.Untar()
+        return _SafeUntar()
 
     elif extension == ".zip" and unpack is None:
-        return pooch.Unzip()
+        return _SafeUnzip()
 
     else:
         return None

sub-packages/bionemo-core/tests/bionemo/core/data/test_load.py

@@ -18,13 +18,21 @@
 import io
 import subprocess
 import tarfile
+import zipfile
 from pathlib import Path
 from unittest.mock import Mock, patch
 
 import ngcbase.environ
 import pytest
 
-from bionemo.core.data.load import default_ngc_client, default_pbss_client, load
+from bionemo.core.data.load import (
+    _SafeUntar,
+    _SafeUnzip,
+    _validate_archive_members,
+    default_ngc_client,
+    default_pbss_client,
+    load,
+)
 from bionemo.core.data.resource import get_all_resources
 
 
@@ -344,3 +352,99 @@ def mocked_ngc_download(url, destination, file_patterns):
     assert file_path.read_text() == "test"
 
     mocked_ngc_client.registry.resource.download_version.assert_called_once()
+
+
+# --- Zip Slip / path traversal tests ---
+
+
+def test_validate_archive_members_rejects_path_traversal():
+    with pytest.raises(ValueError, match="Zip Slip"):
+        _validate_archive_members(["../../etc/passwd"], "/tmp/extract")
+
+
+def test_validate_archive_members_rejects_absolute_path():
+    with pytest.raises(ValueError, match="Zip Slip"):
+        _validate_archive_members(["/etc/passwd"], "/tmp/extract")
+
+
+def test_validate_archive_members_accepts_safe_paths():
+    _validate_archive_members(["subdir/file.txt", "file.txt", "a/b/c/d.txt"], "/tmp/extract")
+
+
+def test_safe_untar_rejects_path_traversal(tmp_path):
+    """Verify _SafeUntar rejects tar archives with path traversal entries."""
+    malicious_tar = tmp_path / "malicious.tar"
+    with tarfile.open(malicious_tar, "w") as tar:
+        data = b"malicious content"
+        info = tarfile.TarInfo(name="../../escaped.txt")
+        info.size = len(data)
+        tar.addfile(info, io.BytesIO(data))
+
+    extract_dir = str(tmp_path / "extracted")
+    processor = _SafeUntar()
+    with pytest.raises(ValueError, match="Zip Slip"):
+        processor._extract_file(str(malicious_tar), extract_dir)
+
+
+def test_safe_untar_allows_safe_archive(tmp_path):
+    """Verify _SafeUntar allows normal tar archives."""
+    safe_tar = tmp_path / "safe.tar"
+    with tarfile.open(safe_tar, "w") as tar:
+        data = b"safe content"
+        info = tarfile.TarInfo(name="subdir/file.txt")
+        info.size = len(data)
+        tar.addfile(info, io.BytesIO(data))
+
+    extract_dir = str(tmp_path / "extracted")
+    processor = _SafeUntar()
+    processor._extract_file(str(safe_tar), extract_dir)
+    assert (tmp_path / "extracted" / "subdir" / "file.txt").read_text() == "safe content"
+
+
+def test_safe_unzip_rejects_path_traversal(tmp_path):
+    """Verify _SafeUnzip rejects zip archives with path traversal entries."""
+    malicious_zip = tmp_path / "malicious.zip"
+    with zipfile.ZipFile(malicious_zip, "w") as zf:
+        zf.writestr("../../escaped.txt", "malicious content")
+
+    extract_dir = str(tmp_path / "extracted")
+    processor = _SafeUnzip()
+    with pytest.raises(ValueError, match="Zip Slip"):
+        processor._extract_file(str(malicious_zip), extract_dir)
+
+
+def test_safe_unzip_allows_safe_archive(tmp_path):
+    """Verify _SafeUnzip allows normal zip archives."""
+    safe_zip = tmp_path / "safe.zip"
+    with zipfile.ZipFile(safe_zip, "w") as zf:
+        zf.writestr("subdir/file.txt", "safe content")
+
+    extract_dir = str(tmp_path / "extracted")
+    processor = _SafeUnzip()
+    processor._extract_file(str(safe_zip), extract_dir)
+    assert (tmp_path / "extracted" / "subdir" / "file.txt").read_text() == "safe content"
+
+
+@patch("bionemo.core.data.load._s3_download")
+def test_load_rejects_malicious_tar(mocked_s3_download, tmp_path):
+    """End-to-end test: loading a resource with a malicious tar archive raises ValueError."""
+    (tmp_path / "foo.yaml").write_text(
+        """
+        - tag: "evil"
+          pbss: "s3://test/evil.tar"
+          owner: Test <test@test.com>
+          sha256: null
+        """
+    )
+
+    def write_malicious_tar(_1, output_file: str, _2):
+        with tarfile.open(output_file, "w") as tar:
+            data = b"malicious"
+            info = tarfile.TarInfo(name="../../escaped.txt")
+            info.size = len(data)
+            tar.addfile(info, io.BytesIO(data))
+
+    mocked_s3_download.side_effect = write_malicious_tar
+
+    with pytest.raises(ValueError, match="Zip Slip"):
+        load("foo/evil", resources=get_all_resources(tmp_path), cache_dir=tmp_path, source="pbss")