diff --git a/.github/workflows/release-please.yml b/.github/workflows/release-please.yml index 279ca1216..9351f930f 100644 --- a/.github/workflows/release-please.yml +++ b/.github/workflows/release-please.yml @@ -661,10 +661,10 @@ jobs: cargo test --manifest-path Cargo.toml - name: Lint standalone package with Ruff - run: uv run --with ruff ruff check src tests + run: uv run --with 'ruff==0.15.10' ruff check src tests - name: Check standalone package formatting with Ruff - run: uv run --with ruff ruff format --check src tests + run: uv run --with 'ruff==0.15.10' ruff format --check src tests - name: Type check standalone package with mypy run: uv run --with mypy mypy src tests diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 290103ea0..9b9ed7772 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -1015,15 +1015,15 @@ jobs: - name: Lint standalone package with Ruff run: | - uv run --with ruff ruff check src tests + uv run --with 'ruff==0.15.10' ruff check src tests - name: Check standalone package import organization with Ruff run: | - uv run --with ruff ruff check --select I src tests + uv run --with 'ruff==0.15.10' ruff check --select I src tests - name: Check standalone package formatting with Ruff run: | - uv run --with ruff ruff format --check src tests + uv run --with 'ruff==0.15.10' ruff format --check src tests - name: Type check standalone package with mypy run: | diff --git a/CHANGELOG.md b/CHANGELOG.md index 59c62e251..3069296c9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Bug Fixes + +- Prevent Windows cache identity probes from creating locked temporary files inside scanned directories. + ## [0.2.52](https://github.com/promptfoo/modelaudit/compare/v0.2.51...v0.2.52) (2026-07-22) ### Bug Fixes diff --git a/docs/agents/picklescan-package-split.md b/docs/agents/picklescan-package-split.md index dc6b62869..2adae870b 100644 --- a/docs/agents/picklescan-package-split.md +++ b/docs/agents/picklescan-package-split.md @@ -128,8 +128,8 @@ Standalone package checks run from `packages/modelaudit-picklescan`: ```bash uv lock --check -uv run --with ruff ruff check src tests -uv run --with ruff ruff format --check src tests +uv run --with 'ruff==0.15.10' ruff check src tests +uv run --with 'ruff==0.15.10' ruff format --check src tests uv run --with mypy mypy src tests uv run --with pytest --with pytest-xdist pytest -n auto tests --tb=short uv run --with pytest pytest tests -q diff --git a/modelaudit/cache/batch_operations.py b/modelaudit/cache/batch_operations.py index e8fe7082d..74bd5ca7a 100644 --- a/modelaudit/cache/batch_operations.py +++ b/modelaudit/cache/batch_operations.py @@ -6,6 +6,7 @@ from pathlib import Path from typing import Any +from ..utils.repository_context import REPOSITORY_SCAN_ROOT_CONFIG_KEY from .adaptive_cache_keys import AdaptiveCacheKeyGenerator from .cache_manager import CacheManager from .cache_policy import should_cache_scan_result @@ -73,9 +74,21 @@ def batch_lookup( cache_dir_groups = self._group_by_cache_directory(cache_lookups) # Process each cache directory concurrently + scan_config = version_context.get("scan_config") if version_context is not None else None + has_repository_scan_root = isinstance(scan_config, dict) and isinstance( + scan_config.get(REPOSITORY_SCAN_ROOT_CONFIG_KEY), str + ) with ThreadPoolExecutor(max_workers=max_workers) as executor: future_to_group = { - executor.submit(self._process_cache_directory_group, group_files): group_files + ( + executor.submit( + self._process_cache_directory_group, + group_files, + version_context=version_context, + ) + if has_repository_scan_root + else executor.submit(self._process_cache_directory_group, group_files) + ): group_files for group, group_files in cache_dir_groups.items() } @@ -119,7 +132,10 @@ def _group_by_cache_directory( return groups def _process_cache_directory_group( - self, group_files: list[tuple[str, str, os.stat_result]] + self, + group_files: list[tuple[str, str, os.stat_result]], + *, + version_context: dict[str, Any] | None = None, ) -> dict[str, dict[str, Any] | None]: """Process all cache files in a single directory efficiently.""" results: dict[str, dict[str, Any] | None] = {} @@ -128,11 +144,19 @@ def _process_cache_directory_group( try: # Use optimized cache lookup if self.cache_manager.cache is not None: - cached_result = self.cache_manager.cache.get_cached_result_by_key( - cache_key, - file_path=file_path, - file_stat=stat_result, - ) + if version_context is None: + cached_result = self.cache_manager.cache.get_cached_result_by_key( + cache_key, + file_path=file_path, + file_stat=stat_result, + ) + else: + cached_result = self.cache_manager.cache.get_cached_result_by_key( + cache_key, + file_path=file_path, + file_stat=stat_result, + version_context=version_context, + ) else: cached_result = None results[file_path] = cached_result diff --git a/modelaudit/cache/scan_results_cache.py b/modelaudit/cache/scan_results_cache.py index 63c38b6ec..9d36ad9b5 100644 --- a/modelaudit/cache/scan_results_cache.py +++ b/modelaudit/cache/scan_results_cache.py @@ -12,6 +12,7 @@ import time from collections.abc import Callable, Iterable from contextlib import suppress +from contextvars import ContextVar from dataclasses import asdict, dataclass from importlib.machinery import ( BYTECODE_SUFFIXES, @@ -26,11 +27,17 @@ import modelaudit_picklescan.call_graph as _picklescan_call_graph from ..utils.helpers.secure_hasher import SecureFileHasher +from ..utils.repository_context import REPOSITORY_SCAN_ROOT_CONFIG_KEY from .adaptive_cache_keys import AdaptiveCacheKeyGenerator from .optimized_config import build_cache_version_context logger = logging.getLogger(__name__) +_CAPTURE_REPOSITORY_SCAN_ROOT: ContextVar[Path | None] = ContextVar( + "modelaudit_capture_repository_scan_root", + default=None, +) + _CALL_GRAPH_SOURCE_FINGERPRINTS_KEY = "call_graph_source_fingerprints" _CALL_GRAPH_SOURCE_FINGERPRINT_MAX_BYTES = 1024 * 1024 _CALL_GRAPH_REGULAR_FILE_FINGERPRINT = "regular-file" @@ -410,6 +417,9 @@ def __init__(self, cache_dir: str | None = None): self.key_generator = AdaptiveCacheKeyGenerator() self._change_clock_probes: dict[int, tuple[BinaryIO, Path]] = {} self._change_clock_probe_lock = threading.Lock() + self._change_clock_probe_condition = threading.Condition(self._change_clock_probe_lock) + self._active_identity_captures = 0 + self._cache_clear_in_progress = False self._ensure_metadata_exists() @@ -493,6 +503,29 @@ def get_cached_result_with_identity( include_private_metadata=include_private_metadata, ) + @staticmethod + def _configured_repository_scan_root(version_context: dict[str, Any] | None) -> Path | None: + if version_context is None: + return None + scan_config = version_context.get("scan_config") + if not isinstance(scan_config, dict): + return None + configured_root = scan_config.get(REPOSITORY_SCAN_ROOT_CONFIG_KEY) + if not isinstance(configured_root, str): + return None + return Path(configured_root).resolve(strict=False) + + def _capture_identity_with_repository_scan_root( + self, + file_path: str, + scan_root: Path, + ) -> ScannedFileIdentity: + root_token = _CAPTURE_REPOSITORY_SCAN_ROOT.set(scan_root) + try: + return self.capture_file_identity(file_path) + finally: + _CAPTURE_REPOSITORY_SCAN_ROOT.reset(root_token) + def _get_cached_result_with_identity( self, file_path: str, @@ -507,10 +540,23 @@ def _get_cached_result_with_identity( logger.debug("Bypassing scan-result cache lookup for symlinked path %s", file_path) return None, None - if ( - file_stat is not None - and getattr(self.capture_file_identity, "__func__", None) is ScanResultsCache.capture_file_identity - ): + scan_root = self._configured_repository_scan_root(version_context) + + uses_default_capture = ( + getattr(self.capture_file_identity, "__func__", None) is ScanResultsCache.capture_file_identity + ) + if uses_default_capture and scan_root is not None: + file_identity = self._capture_file_identity( + file_path, + file_stat=file_stat, + scan_root=scan_root, + ) + elif scan_root is not None: + file_identity = self._capture_identity_with_repository_scan_root( + file_path, + scan_root, + ) + elif uses_default_capture and file_stat is not None: file_identity = self._capture_file_identity(file_path, file_stat=file_stat) else: # Preserve legacy subclasses that override the established signature. @@ -571,6 +617,7 @@ def get_cached_result_by_key( *, file_path: str | None = None, file_stat: os.stat_result | None = None, + version_context: dict[str, Any] | None = None, include_private_metadata: bool = False, ) -> dict[str, Any] | None: """ @@ -591,7 +638,23 @@ def get_cached_result_by_key( if not self._get_cache_file_path(cache_key).exists(): self._record_cache_miss("not_found") return None - file_identity = self.capture_file_identity(file_path) + scan_root = self._configured_repository_scan_root(version_context) + uses_default_capture = ( + getattr(self.capture_file_identity, "__func__", None) is ScanResultsCache.capture_file_identity + ) + if scan_root is not None and uses_default_capture: + file_identity = self._capture_file_identity( + file_path, + file_stat=file_stat, + scan_root=scan_root, + ) + elif scan_root is not None: + file_identity = self._capture_identity_with_repository_scan_root( + file_path, + scan_root, + ) + else: + file_identity = self.capture_file_identity(file_path) file_stat = file_identity[0] return self._get_cached_result_by_key( cache_key, @@ -890,13 +953,47 @@ def store_result( def capture_file_identity(self, file_path: str) -> ScannedFileIdentity: """Capture a stable stat, content hash, and platform change token before scanning.""" - return self._capture_file_identity(file_path, file_stat=None) + scan_root = _CAPTURE_REPOSITORY_SCAN_ROOT.get() + if scan_root is None: + return self._capture_file_identity(file_path, file_stat=None) + return self._capture_file_identity( + file_path, + file_stat=None, + scan_root=scan_root, + ) def _capture_file_identity( self, file_path: str, *, file_stat: os.stat_result | None, + scan_root: Path | None = None, + ) -> ScannedFileIdentity: + with self._change_clock_probe_condition: + while self._cache_clear_in_progress: + self._change_clock_probe_condition.wait() + self._active_identity_captures += 1 + + try: + if scan_root is None: + return self._capture_file_identity_leased(file_path, file_stat=file_stat) + return self._capture_file_identity_leased( + file_path, + file_stat=file_stat, + scan_root=scan_root, + ) + finally: + with self._change_clock_probe_condition: + self._active_identity_captures -= 1 + if self._active_identity_captures == 0: + self._change_clock_probe_condition.notify_all() + + def _capture_file_identity_leased( + self, + file_path: str, + *, + file_stat: os.stat_result | None, + scan_root: Path | None = None, ) -> ScannedFileIdentity: if self._path_has_symlink_component(file_path): raise ValueError(f"Symlinked paths are not cacheable: {file_path}") @@ -906,7 +1003,14 @@ def _capture_file_identity( for _capture_attempt in range(_MAX_IDENTITY_CAPTURE_ATTEMPTS): preliminary_stat = stat_hint if stat_hint is not None else os.stat(file_path) stat_hint = None - probe = self._get_change_clock_probe(file_path, preliminary_stat.st_dev) + if scan_root is None: + probe = self._get_change_clock_probe(file_path, preliminary_stat.st_dev) + else: + probe = self._get_change_clock_probe( + file_path, + preliminary_stat.st_dev, + scan_root=scan_root, + ) preliminary_change_token = self._get_file_change_token(file_path, preliminary_stat) preliminary_ancestor_identity = self._capture_ancestor_identity(file_path) @@ -978,31 +1082,85 @@ def _capture_file_identity( raise ValueError(f"File kept changing while capturing cache identity: {file_path}") from last_change_error - def _get_change_clock_probe(self, file_path: str, file_device: int) -> BinaryIO: + def _get_change_clock_probe( + self, + file_path: str, + file_device: int, + *, + scan_root: Path | None = None, + ) -> BinaryIO: """Return a reusable probe whose inode lives on the scanned file's filesystem.""" with self._change_clock_probe_lock: + try: + scanned_parent = Path(file_path).resolve(strict=False).parent + normalized_scan_root = scan_root.resolve(strict=False) if scan_root is not None else None + except (OSError, RuntimeError) as exc: + raise ValueError(f"No writable cache identity probe directory for: {file_path}") from exc + protected_root = normalized_scan_root or scanned_parent + existing = self._change_clock_probes.get(file_device) if existing is not None: - return existing[0] + try: + existing_directory = existing[1].resolve(strict=False) + except (OSError, RuntimeError) as exc: + raise ValueError(f"No writable cache identity probe directory for: {file_path}") from exc + if os.name == "nt" and ( + existing_directory == protected_root or protected_root in existing_directory.parents + ): + if normalized_scan_root is None or self._active_identity_captures > 1: + raise ValueError(f"No writable cache identity probe directory for: {file_path}") + + existing_probe = existing[0] + close_failed = False + try: + existing_probe.close() + except OSError: + close_failed = True + + if not existing_probe.closed and not close_failed: + underlying_probe = getattr(existing_probe, "file", None) + wrapper_closer = getattr(existing_probe, "_closer", None) + if ( + underlying_probe is not None + and underlying_probe is not existing_probe + and getattr(wrapper_closer, "file", None) is underlying_probe + and getattr(wrapper_closer, "close_called", False) + ): + with suppress(OSError): + underlying_probe.close() + + if not existing_probe.closed: + raise ValueError(f"No writable cache identity probe directory for: {file_path}") + self._change_clock_probes.pop(file_device, None) + else: + return existing[0] if os.name == "nt": - candidates = [Path(tempfile.gettempdir())] + # Windows keeps TemporaryFile names visible and locked until close. + candidates = [Path(tempfile.gettempdir()), self.cache_dir] + candidates.extend(protected_root.parents) else: candidates = [self.cache_dir, Path(tempfile.gettempdir())] - ancestor = Path(os.path.abspath(file_path)).parent - while True: - candidates.append(ancestor) - if ancestor.parent == ancestor: - break - ancestor = ancestor.parent - if os.name == "nt": - candidates.append(self.cache_dir) + ancestor = scanned_parent + while True: + candidates.append(ancestor) + if ancestor.parent == ancestor: + break + ancestor = ancestor.parent checked: set[Path] = set() for candidate in candidates: - if candidate in checked: + try: + normalized_candidate = candidate.resolve(strict=False) + except (OSError, RuntimeError): continue - checked.add(candidate) + if os.name == "nt" and ( + normalized_candidate == protected_root or protected_root in normalized_candidate.parents + ): + continue + if normalized_candidate in checked: + continue + checked.add(normalized_candidate) if not self._directory_is_on_device(candidate, file_device): continue @@ -2079,16 +2237,64 @@ def clear_cache(self) -> None: logger.debug("Clearing entire scan results cache") - # Remove all cache files except metadata - for item in self.cache_dir.iterdir(): - if item.name != "cache_metadata.json": - if item.is_dir(): - shutil.rmtree(item) - else: - item.unlink() + retained_probe_paths: set[Path] = set() + with self._change_clock_probe_condition: + while self._cache_clear_in_progress: + self._change_clock_probe_condition.wait() + self._cache_clear_in_progress = True + try: + while self._active_identity_captures: + self._change_clock_probe_condition.wait() + + for file_device, (probe, probe_directory) in tuple(self._change_clock_probes.items()): + close_failed = False + try: + probe.close() + except OSError: + close_failed = True + + if not probe.closed and not close_failed: + underlying_probe = getattr(probe, "file", None) + wrapper_closer = getattr(probe, "_closer", None) + if ( + underlying_probe is not None + and underlying_probe is not probe + and getattr(wrapper_closer, "file", None) is underlying_probe + and getattr(wrapper_closer, "close_called", False) + ): + with suppress(OSError): + underlying_probe.close() + + if not probe.closed: + probe_name = getattr(probe, "name", None) + if isinstance(probe_name, str): + probe_path = Path(os.path.abspath(probe_name)) + cache_path = Path(os.path.abspath(self.cache_dir)) + if ( + probe_directory == self.cache_dir + and probe_path.parent == cache_path + and probe_path.name.startswith(".modelaudit-cache-clock-") + ): + retained_probe_paths.add(probe_path) + continue + + self._change_clock_probes.pop(file_device, None) + + # Remove all cache files except metadata while excluding new probes. + for item in self.cache_dir.iterdir(): + if Path(os.path.abspath(item)) in retained_probe_paths: + continue + if item.name != "cache_metadata.json": + if item.is_dir(): + shutil.rmtree(item) + else: + item.unlink() + + self._create_initial_metadata() + finally: + self._cache_clear_in_progress = False + self._change_clock_probe_condition.notify_all() - # Reset metadata - self._create_initial_metadata() logger.debug("Cache cleared successfully") def _ensure_metadata_exists(self): diff --git a/packages/modelaudit-picklescan/AGENTS.md b/packages/modelaudit-picklescan/AGENTS.md index e68c5aa4d..df7509f3d 100644 --- a/packages/modelaudit-picklescan/AGENTS.md +++ b/packages/modelaudit-picklescan/AGENTS.md @@ -40,8 +40,8 @@ Run from `packages/modelaudit-picklescan/`: ```bash uv lock --check -uv run --with ruff ruff check src tests -uv run --with ruff ruff format --check src tests +uv run --with 'ruff==0.15.10' ruff check src tests +uv run --with 'ruff==0.15.10' ruff format --check src tests uv run --with mypy mypy src tests uv run --with pytest --with pytest-xdist pytest -n auto tests --tb=short diff --git a/packages/modelaudit-picklescan/tests/test_api.py b/packages/modelaudit-picklescan/tests/test_api.py index 229667049..4d24cfb22 100644 --- a/packages/modelaudit-picklescan/tests/test_api.py +++ b/packages/modelaudit-picklescan/tests/test_api.py @@ -9062,26 +9062,39 @@ def test_scan_bytes_warns_when_allowlisted_module_is_unresolved(monkeypatch: pyt @pytest.mark.parametrize( - ("module", "name"), + ("module", "name", "source_changes"), [ - ("joblib.numpy_pickle", "NumpyArrayWrapper"), - ("numpy._core.multiarray", "_reconstruct"), - ("torch._utils", "_rebuild_tensor_v2"), + ("joblib.numpy_pickle", "NumpyArrayWrapper", False), + ("joblib.numpy_pickle", "NumpyArrayWrapper", True), + ("numpy._core.multiarray", "_reconstruct", False), + ("torch._utils", "_rebuild_tensor_v2", False), ], ) def test_scan_bytes_warns_on_unresolved_framework_reconstruction_global( module: str, name: str, + source_changes: bool, monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.setattr( "modelaudit_picklescan.call_graph._trusted_module_origin_kind", lambda _module_name: "unresolved", ) + if source_changes: + + def raise_source_stability_error(_report_generation: int | None) -> None: + raise _CallGraphAnalysisLimitError("source changed during shared call-graph analysis") + + monkeypatch.setattr(package_api, "_ensure_shared_source_snapshot_stable", raise_source_stability_error) report = scan_bytes(f"c{module}\n{name}\n.".encode(), source="unresolved-framework-global.pkl") - assert report.status == ScanStatus.COMPLETE + if report.status == ScanStatus.INCONCLUSIVE: + _assert_call_graph_source_stability_error(report) + else: + assert report.status == ScanStatus.COMPLETE + if source_changes: + assert report.status == ScanStatus.INCONCLUSIVE assert report.verdict == SafetyVerdict.SUSPICIOUS assert any( finding.rule_code == "NON_ALLOWLISTED_GLOBAL" and finding.details.get("import_reference") == f"{module}.{name}" diff --git a/tests/cache/test_cache_correctness.py b/tests/cache/test_cache_correctness.py index 6da2cdd8a..4c00a7c8d 100644 --- a/tests/cache/test_cache_correctness.py +++ b/tests/cache/test_cache_correctness.py @@ -5,6 +5,7 @@ import os import sys import tempfile +import threading import time import zipfile from collections.abc import Iterator @@ -22,7 +23,7 @@ ) from pathlib import Path from types import FunctionType, ModuleType -from typing import Any +from typing import Any, BinaryIO, cast from zipimport import zipimporter import pytest @@ -37,6 +38,7 @@ from modelaudit_picklescan.call_graph import _source_resolution_context as _picklescan_source_resolution_context from modelaudit.cache import get_cache_manager, reset_cache_manager +from modelaudit.cache import scan_results_cache as scan_results_cache_module from modelaudit.cache.batch_operations import BatchCacheOperations from modelaudit.cache.optimized_config import ( ConfigurationExtractor, @@ -57,6 +59,7 @@ from modelaudit.config.rule_config import ModelAuditConfig, get_config, reset_config, set_config from modelaudit.scanner_results import INCONCLUSIVE_SCAN_OUTCOME, ScanResult from modelaudit.utils.helpers.cache_decorator import cached_scan +from modelaudit.utils.repository_context import REPOSITORY_SCAN_ROOT_CONFIG_KEY @pytest.fixture(autouse=True) @@ -176,7 +179,8 @@ def test_cache_config_hash_preserves_128_bits(tmp_path: Path) -> None: assert len(config_hash) == 32 -def test_capture_file_identity_uses_target_filesystem_probe( +@pytest.mark.skipif(os.name == "nt", reason="Windows probes must remain outside scanned content") +def test_posix_capture_file_identity_uses_target_filesystem_probe( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: @@ -200,6 +204,888 @@ def simulate_cache_and_system_temp_on_other_devices(directory: Path, device: int cache._change_clock_probes.clear() +def test_windows_change_clock_probe_avoids_scanned_ancestors( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + file_path = _make_cacheable_file(tmp_path) + cache = ScanResultsCache(str(tmp_path / "cache")) + attempted_probe_dirs: list[str] = [] + + def record_probe_attempt(*_args: Any, dir: str | Path, **_kwargs: Any) -> BinaryIO: + attempted_probe_dirs.append(str(dir)) + raise OSError("simulated probe creation failure") + + with monkeypatch.context() as patch: + patch.setattr(os, "name", "nt") + patch.setattr(scan_results_cache_module, "Path", type(file_path)) + patch.setattr( + cache, + "_directory_is_on_device", + lambda directory, _device: directory == file_path.parent, + ) + patch.setattr(tempfile, "TemporaryFile", record_probe_attempt) + + with pytest.raises(ValueError, match="No writable cache identity probe directory"): + cache._get_change_clock_probe(str(file_path), file_path.stat().st_dev) + + assert attempted_probe_dirs == [] + + +@pytest.mark.parametrize( + "unsafe_location", + ( + "cache_parent", + "cache_descendant", + "system_temp_parent", + "system_temp_descendant", + ), +) +def test_windows_change_clock_probe_rejects_scanned_directory_candidates( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + unsafe_location: str, +) -> None: + scanned_directory = tmp_path / "scanned" + scanned_directory.mkdir() + file_path = _make_cacheable_file(scanned_directory) + file_device = file_path.stat().st_dev + + if unsafe_location.startswith("cache_"): + if unsafe_location == "cache_parent": + cache_directory = scanned_directory + elif unsafe_location == "cache_ancestor": + cache_directory = tmp_path + else: + cache_directory = scanned_directory / ".scan-cache" + system_temp = tmp_path / "other-device-temp" + unsafe_candidate = cache_directory + else: + cache_directory = tmp_path / "other-device-cache" + if unsafe_location == "system_temp_parent": + system_temp = scanned_directory + elif unsafe_location == "system_temp_ancestor": + system_temp = tmp_path + else: + system_temp = scanned_directory / ".modelaudit-temp" + unsafe_candidate = system_temp + + cache = ScanResultsCache(str(cache_directory)) + attempted_probe_dirs: list[Path] = [] + + def record_probe_attempt(*_args: Any, dir: str | Path, **_kwargs: Any) -> BinaryIO: + attempted_probe_dirs.append(type(file_path)(dir)) + raise OSError("an overlapping Windows probe must never be created") + + with monkeypatch.context() as patch: + patch.setattr(os, "name", "nt") + patch.setattr(scan_results_cache_module, "Path", type(file_path)) + patch.setattr(tempfile, "gettempdir", lambda: str(system_temp)) + patch.setattr( + cache, + "_directory_is_on_device", + lambda directory, _device: directory == unsafe_candidate, + ) + patch.setattr(tempfile, "TemporaryFile", record_probe_attempt) + + with pytest.raises(ValueError, match="No writable cache identity probe directory"): + cache._get_change_clock_probe(str(file_path), file_device) + + assert attempted_probe_dirs == [] + assert cache._change_clock_probes == {} + + +@pytest.mark.parametrize( + "unsafe_location", + ("scanned_parent", "scanned_descendant"), +) +def test_windows_change_clock_probe_rejects_overlapping_existing_handle( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + unsafe_location: str, +) -> None: + scanned_directory = tmp_path / "scanned" + scanned_directory.mkdir() + file_path = _make_cacheable_file(scanned_directory) + file_device = file_path.stat().st_dev + cache = ScanResultsCache(str(tmp_path / "isolated-cache")) + if unsafe_location == "scanned_parent": + unsafe_directory = scanned_directory + elif unsafe_location == "scanned_ancestor": + unsafe_directory = tmp_path + else: + unsafe_directory = scanned_directory / ".existing-probes" + unsafe_directory.mkdir() + + attempted_probe_dirs: list[Path] = [] + + def record_probe_attempt(*_args: Any, dir: str | Path, **_kwargs: Any) -> BinaryIO: + attempted_probe_dirs.append(type(file_path)(dir)) + raise OSError("an overlapping Windows probe must never be replaced") + + with tempfile.TemporaryFile(mode="w+b", dir=unsafe_directory) as probe: + tracked_probe = cast(BinaryIO, probe) + cache._change_clock_probes[file_device] = (tracked_probe, unsafe_directory) + + with monkeypatch.context() as patch: + patch.setattr(os, "name", "nt") + patch.setattr(scan_results_cache_module, "Path", type(file_path)) + patch.setattr(tempfile, "TemporaryFile", record_probe_attempt) + + with pytest.raises(ValueError, match="No writable cache identity probe directory"): + cache._get_change_clock_probe(str(file_path), file_device) + + assert attempted_probe_dirs == [] + assert probe.closed is False + assert cache._change_clock_probes[file_device] == (tracked_probe, unsafe_directory) + + +@pytest.mark.parametrize("unsafe_location", ("cache_sibling", "system_temp_sibling")) +def test_windows_change_clock_probe_rejects_siblings_inside_actual_scan_root( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + unsafe_location: str, +) -> None: + scan_root = tmp_path / "scan-root" + model_directory = scan_root / "models" + model_directory.mkdir(parents=True) + file_path = _make_cacheable_file(model_directory) + file_device = file_path.stat().st_dev + if unsafe_location == "cache_sibling": + cache_directory = scan_root / "cache" + system_temp = tmp_path / "outside-temp" + unsafe_candidate = cache_directory + else: + cache_directory = tmp_path / "outside-cache" + system_temp = scan_root / "temp" + unsafe_candidate = system_temp + + unsafe_candidate.mkdir(parents=True, exist_ok=True) + cache = ScanResultsCache(str(cache_directory)) + attempted_probe_dirs: list[Path] = [] + + def record_probe_attempt(*_args: Any, dir: str | Path, **_kwargs: Any) -> BinaryIO: + attempted_probe_dirs.append(type(file_path)(dir)) + raise OSError("a probe inside the actual directory scan root must never be created") + + with monkeypatch.context() as patch: + patch.setattr(os, "name", "nt") + patch.setattr(scan_results_cache_module, "Path", type(file_path)) + patch.setattr(tempfile, "gettempdir", lambda: str(system_temp)) + patch.setattr( + cache, + "_directory_is_on_device", + lambda directory, _device: directory == unsafe_candidate, + ) + patch.setattr(tempfile, "TemporaryFile", record_probe_attempt) + + with pytest.raises(ValueError, match="No writable cache identity probe directory"): + cache._get_change_clock_probe(str(file_path), file_device, scan_root=scan_root) + + assert attempted_probe_dirs == [] + assert cache._change_clock_probes == {} + + +def test_windows_change_clock_probe_relocates_existing_handle_outside_actual_scan_root( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + scan_root = tmp_path / "scan-root" + model_directory = scan_root / "models" + model_directory.mkdir(parents=True) + file_path = _make_cacheable_file(model_directory) + unsafe_directory = scan_root / "cache" + unsafe_directory.mkdir() + file_device = file_path.stat().st_dev + cache = ScanResultsCache(str(tmp_path / "isolated-cache")) + attempted_probe_dirs: list[Path] = [] + + with ( + tempfile.TemporaryFile(mode="w+b", dir=unsafe_directory) as unsafe_probe, + tempfile.TemporaryFile(mode="w+b", dir=cache.cache_dir) as safe_probe, + ): + + def create_relocated_probe(*_args: Any, dir: str | Path, **_kwargs: Any) -> BinaryIO: + attempted_probe_dirs.append(type(file_path)(dir)) + return cast(BinaryIO, safe_probe) + + cache._change_clock_probes[file_device] = (cast(BinaryIO, unsafe_probe), unsafe_directory) + with monkeypatch.context() as patch: + patch.setattr(os, "name", "nt") + patch.setattr(scan_results_cache_module, "Path", type(file_path)) + patch.setattr(tempfile, "gettempdir", lambda: str(scan_root)) + patch.setattr( + cache, + "_directory_is_on_device", + lambda directory, _device: directory == cache.cache_dir, + ) + patch.setattr(tempfile, "TemporaryFile", create_relocated_probe) + selected_probe = cache._get_change_clock_probe( + str(file_path), + file_device, + scan_root=scan_root, + ) + + assert unsafe_probe.closed is True + assert selected_probe is safe_probe + assert attempted_probe_dirs == [cache.cache_dir] + assert cache._change_clock_probes[file_device] == (cast(BinaryIO, safe_probe), cache.cache_dir) + cache._change_clock_probes.clear() + + +def test_windows_cache_lookup_propagates_repository_scan_root( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + scan_root = tmp_path / "scan-root" + model_directory = scan_root / "models" + model_directory.mkdir(parents=True) + file_path = _make_cacheable_file(model_directory) + unsafe_cache = scan_root / "cache" + cache = ScanResultsCache(str(unsafe_cache)) + system_temp = tmp_path / "outside-temp" + attempted_probe_dirs: list[Path] = [] + + def record_probe_attempt(*_args: Any, dir: str | Path, **_kwargs: Any) -> BinaryIO: + attempted_probe_dirs.append(type(file_path)(dir)) + raise OSError("a repository-root cache probe must never be created") + + version_context = build_cache_version_context({REPOSITORY_SCAN_ROOT_CONFIG_KEY: str(scan_root)}) + with monkeypatch.context() as patch: + patch.setattr(os, "name", "nt") + patch.setattr(scan_results_cache_module, "Path", type(file_path)) + patch.setattr(tempfile, "gettempdir", lambda: str(system_temp)) + patch.setattr( + cache, + "_directory_is_on_device", + lambda directory, _device: directory == unsafe_cache, + ) + patch.setattr(tempfile, "TemporaryFile", record_probe_attempt) + + result = cache.get_cached_result_with_identity( + str(file_path), + version_context=version_context, + ) + + assert result == (None, None) + assert attempted_probe_dirs == [] + assert cache._change_clock_probes == {} + + +def test_windows_change_clock_probe_preserves_temporary_directory_ancestor( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + scanned_directory = tmp_path / "models" + scanned_directory.mkdir() + file_path = _make_cacheable_file(scanned_directory) + file_device = file_path.stat().st_dev + cache = ScanResultsCache(str(scanned_directory / "cache")) + attempted_probe_dirs: list[Path] = [] + + with tempfile.TemporaryFile(mode="w+b", dir=tmp_path) as probe: + + def use_temporary_ancestor(*_args: Any, dir: str | Path, **_kwargs: Any) -> BinaryIO: + attempted_probe_dirs.append(type(file_path)(dir)) + return cast(BinaryIO, probe) + + with monkeypatch.context() as patch: + patch.setattr(os, "name", "nt") + patch.setattr(scan_results_cache_module, "Path", type(file_path)) + patch.setattr(tempfile, "gettempdir", lambda: str(tmp_path)) + patch.setattr( + cache, + "_directory_is_on_device", + lambda directory, _device: directory == tmp_path, + ) + patch.setattr(tempfile, "TemporaryFile", use_temporary_ancestor) + + selected_probe = cache._get_change_clock_probe(str(file_path), file_device) + + assert selected_probe is probe + assert attempted_probe_dirs == [tmp_path] + assert cache._change_clock_probes[file_device][1] == tmp_path + cache._change_clock_probes.clear() + + +def test_windows_change_clock_probe_uses_safe_ancestor_without_scan_root( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + volume_root = tmp_path / "model-volume" + scanned_parent = volume_root / "models" + scanned_parent.mkdir(parents=True) + file_path = _make_cacheable_file(scanned_parent) + file_device = file_path.stat().st_dev + off_device_cache = tmp_path / "off-device-cache" + off_device_temp = tmp_path / "off-device-temp" + cache = ScanResultsCache(str(off_device_cache)) + safe_ancestor = scanned_parent.parent.resolve() + attempted_probe_dirs: list[Path] = [] + + with tempfile.TemporaryFile(mode="w+b", dir=safe_ancestor) as probe: + + def create_safe_ancestor_probe(*_args: Any, dir: str | Path, **_kwargs: Any) -> BinaryIO: + attempted_probe_dirs.append(type(file_path)(dir)) + return cast(BinaryIO, probe) + + with monkeypatch.context() as patch: + patch.setattr(os, "name", "nt") + patch.setattr(scan_results_cache_module, "Path", type(file_path)) + patch.setattr(tempfile, "gettempdir", lambda: str(off_device_temp)) + patch.setattr( + cache, + "_directory_is_on_device", + lambda directory, _device: directory == safe_ancestor, + ) + patch.setattr(tempfile, "TemporaryFile", create_safe_ancestor_probe) + + selected_probe = cache._get_change_clock_probe(str(file_path), file_device) + + assert selected_probe is probe + assert attempted_probe_dirs == [safe_ancestor] + assert safe_ancestor != scanned_parent + assert scanned_parent not in safe_ancestor.parents + assert cache._change_clock_probes[file_device] == (cast(BinaryIO, probe), safe_ancestor) + cache._change_clock_probes.clear() + + +def test_windows_change_clock_probe_uses_safe_ancestor_outside_scan_root( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + repository_root = tmp_path / "repository" + scan_root = repository_root / "snapshots" / "pinned-revision" + model_directory = scan_root / "models" + model_directory.mkdir(parents=True) + file_path = _make_cacheable_file(model_directory) + file_device = file_path.stat().st_dev + off_device_cache = tmp_path / "off-device-cache" + off_device_temp = tmp_path / "off-device-temp" + cache = ScanResultsCache(str(off_device_cache)) + safe_ancestor = scan_root.parent.resolve() + attempted_probe_dirs: list[Path] = [] + + with tempfile.TemporaryFile(mode="w+b", dir=safe_ancestor) as probe: + + def return_safe_ancestor_probe(*_args: Any, dir: str | Path, **_kwargs: Any) -> BinaryIO: + attempted_probe_dirs.append(type(file_path)(dir)) + return cast(BinaryIO, probe) + + with monkeypatch.context() as patch: + patch.setattr(os, "name", "nt") + patch.setattr(scan_results_cache_module, "Path", type(file_path)) + patch.setattr(tempfile, "gettempdir", lambda: str(off_device_temp)) + patch.setattr( + cache, + "_directory_is_on_device", + lambda directory, _device: directory == safe_ancestor, + ) + patch.setattr(tempfile, "TemporaryFile", return_safe_ancestor_probe) + + selected_probe = cache._get_change_clock_probe( + str(file_path), + file_device, + scan_root=scan_root, + ) + + assert selected_probe is probe + assert attempted_probe_dirs == [safe_ancestor] + assert safe_ancestor != scan_root + assert scan_root not in safe_ancestor.parents + assert cache._change_clock_probes[file_device][1] == safe_ancestor + cache._change_clock_probes.clear() + + +def test_windows_probe_protects_snapshot_root_for_external_hugging_face_blob( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + repository_root = tmp_path / "hub" / "model" + scan_root = repository_root / "snapshots" / "pinned-revision" + blobs_directory = repository_root / "blobs" + scan_root.mkdir(parents=True) + blobs_directory.mkdir() + blob_path = _make_cacheable_file(blobs_directory, name="secure-blob") + snapshot_path = scan_root / "model.pkl" + try: + snapshot_path.symlink_to(blob_path) + except OSError as exc: + pytest.skip(f"snapshot symlinks unavailable: {exc}") + assert snapshot_path.resolve() == blob_path + + unsafe_cache = scan_root / "cache" + cache = ScanResultsCache(str(unsafe_cache)) + off_device_temp = tmp_path / "off-device-temp" + attempted_probe_dirs: list[Path] = [] + + def record_probe_attempt(*_args: Any, dir: str | Path, **_kwargs: Any) -> BinaryIO: + attempted_probe_dirs.append(type(blob_path)(dir)) + raise OSError("a Hugging Face snapshot-root probe must never be created") + + version_context = build_cache_version_context({REPOSITORY_SCAN_ROOT_CONFIG_KEY: str(scan_root)}) + with monkeypatch.context() as patch: + patch.setattr(os, "name", "nt") + patch.setattr(scan_results_cache_module, "Path", type(blob_path)) + patch.setattr(tempfile, "gettempdir", lambda: str(off_device_temp)) + patch.setattr( + cache, + "_directory_is_on_device", + lambda directory, _device: directory == unsafe_cache, + ) + patch.setattr(tempfile, "TemporaryFile", record_probe_attempt) + + result = cache.get_cached_result_with_identity( + str(blob_path), + version_context=version_context, + ) + + assert result == (None, None) + assert attempted_probe_dirs == [] + assert cache._change_clock_probes == {} + + +@pytest.mark.parametrize("unsafe_location", ("cache_alias", "system_temp_alias")) +def test_windows_change_clock_probe_rejects_aliases_into_actual_scan_root( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + unsafe_location: str, +) -> None: + scan_root = tmp_path / "scan-root" + models = scan_root / "models" + models.mkdir(parents=True) + file_path = _make_cacheable_file(models) + actual_probe_directory = scan_root / "internal-probes" + actual_probe_directory.mkdir() + probe_alias = tmp_path / "outside-alias" + try: + probe_alias.symlink_to(actual_probe_directory, target_is_directory=True) + except OSError as exc: + pytest.skip(f"directory aliases unavailable: {exc}") + + if unsafe_location == "cache_alias": + cache_directory = probe_alias + system_temp = tmp_path / "off-device-temp" + else: + cache_directory = tmp_path / "off-device-cache" + system_temp = probe_alias + cache = ScanResultsCache(str(cache_directory)) + attempted_probe_dirs: list[Path] = [] + + def record_probe_attempt(*_args: Any, dir: str | Path, **_kwargs: Any) -> BinaryIO: + attempted_probe_dirs.append(type(file_path)(dir)) + raise OSError("an aliased scan-root probe must never be created") + + with monkeypatch.context() as patch: + patch.setattr(os, "name", "nt") + patch.setattr(scan_results_cache_module, "Path", type(file_path)) + patch.setattr(tempfile, "gettempdir", lambda: str(system_temp)) + patch.setattr( + cache, + "_directory_is_on_device", + lambda directory, _device: directory == probe_alias, + ) + patch.setattr(tempfile, "TemporaryFile", record_probe_attempt) + + with pytest.raises(ValueError, match="No writable cache identity probe directory"): + cache._get_change_clock_probe( + str(file_path), + file_path.stat().st_dev, + scan_root=scan_root, + ) + + assert attempted_probe_dirs == [] + assert cache._change_clock_probes == {} + + +def test_windows_change_clock_probe_relocates_reused_alias_outside_scan_root( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + scan_root = tmp_path / "scan-root" + models = scan_root / "models" + models.mkdir(parents=True) + file_path = _make_cacheable_file(models) + actual_probe_directory = scan_root / "internal-probes" + actual_probe_directory.mkdir() + probe_alias = tmp_path / "outside-alias" + try: + probe_alias.symlink_to(actual_probe_directory, target_is_directory=True) + except OSError as exc: + pytest.skip(f"directory aliases unavailable: {exc}") + + cache = ScanResultsCache(str(tmp_path / "isolated-cache")) + file_device = file_path.stat().st_dev + attempted_probe_dirs: list[Path] = [] + with ( + tempfile.TemporaryFile(mode="w+b", dir=actual_probe_directory) as unsafe_probe, + tempfile.TemporaryFile(mode="w+b", dir=cache.cache_dir) as safe_probe, + ): + + def create_relocated_probe(*_args: Any, dir: str | Path, **_kwargs: Any) -> BinaryIO: + attempted_probe_dirs.append(type(file_path)(dir)) + return cast(BinaryIO, safe_probe) + + cache._change_clock_probes[file_device] = (cast(BinaryIO, unsafe_probe), probe_alias) + with monkeypatch.context() as patch: + patch.setattr(os, "name", "nt") + patch.setattr(scan_results_cache_module, "Path", type(file_path)) + patch.setattr(tempfile, "gettempdir", lambda: str(actual_probe_directory)) + patch.setattr( + cache, + "_directory_is_on_device", + lambda directory, _device: directory == cache.cache_dir, + ) + patch.setattr(tempfile, "TemporaryFile", create_relocated_probe) + selected_probe = cache._get_change_clock_probe( + str(file_path), + file_device, + scan_root=scan_root, + ) + + assert unsafe_probe.closed is True + assert selected_probe is safe_probe + assert attempted_probe_dirs == [cache.cache_dir] + assert cache._change_clock_probes[file_device] == (cast(BinaryIO, safe_probe), cache.cache_dir) + cache._change_clock_probes.clear() + + +def test_windows_change_clock_probe_retains_shared_unsafe_handle( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + scan_root = tmp_path / "scan-root" + models = scan_root / "models" + models.mkdir(parents=True) + file_path = _make_cacheable_file(models) + unsafe_directory = scan_root / "cache" + unsafe_directory.mkdir() + cache = ScanResultsCache(str(tmp_path / "isolated-cache")) + file_device = file_path.stat().st_dev + attempted_probe_dirs: list[Path] = [] + + def forbid_replacement(*_args: Any, dir: str | Path, **_kwargs: Any) -> BinaryIO: + attempted_probe_dirs.append(type(file_path)(dir)) + raise OSError("an active shared probe must never be closed or replaced") + + with tempfile.TemporaryFile(mode="w+b", dir=unsafe_directory) as probe: + tracked_probe = cast(BinaryIO, probe) + cache._change_clock_probes[file_device] = (tracked_probe, unsafe_directory) + cache._active_identity_captures = 2 + with monkeypatch.context() as patch: + patch.setattr(os, "name", "nt") + patch.setattr(scan_results_cache_module, "Path", type(file_path)) + patch.setattr(tempfile, "TemporaryFile", forbid_replacement) + with pytest.raises(ValueError, match="No writable cache identity probe directory"): + cache._get_change_clock_probe( + str(file_path), + file_device, + scan_root=scan_root, + ) + + assert probe.closed is False + assert attempted_probe_dirs == [] + assert cache._change_clock_probes[file_device] == (tracked_probe, unsafe_directory) + cache._active_identity_captures = 0 + + +def test_windows_change_clock_probe_retains_handle_after_failed_relocation_close( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + scan_root = tmp_path / "scan-root" + models = scan_root / "models" + models.mkdir(parents=True) + file_path = _make_cacheable_file(models) + unsafe_directory = scan_root / "cache" + unsafe_directory.mkdir() + cache = ScanResultsCache(str(tmp_path / "isolated-cache")) + file_device = file_path.stat().st_dev + close_attempts: list[str] = [] + attempted_probe_dirs: list[Path] = [] + + class UnclosableProbe: + closed = False + + def close(self) -> None: + close_attempts.append("close") + raise OSError("the underlying Windows probe is still locked") + + unsafe_probe = UnclosableProbe() + cache._change_clock_probes[file_device] = (cast(BinaryIO, unsafe_probe), unsafe_directory) + + def forbid_replacement(*_args: Any, dir: str | Path, **_kwargs: Any) -> BinaryIO: + attempted_probe_dirs.append(type(file_path)(dir)) + raise OSError("an unclosed probe must never be replaced") + + with monkeypatch.context() as patch: + patch.setattr(os, "name", "nt") + patch.setattr(scan_results_cache_module, "Path", type(file_path)) + patch.setattr(tempfile, "TemporaryFile", forbid_replacement) + with pytest.raises(ValueError, match="No writable cache identity probe directory"): + cache._get_change_clock_probe( + str(file_path), + file_device, + scan_root=scan_root, + ) + + assert close_attempts == ["close"] + assert unsafe_probe.closed is False + assert attempted_probe_dirs == [] + assert cache._change_clock_probes[file_device] == (cast(BinaryIO, unsafe_probe), unsafe_directory) + + +def test_windows_batch_lookup_preserves_actual_repository_scan_root( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + scan_root = tmp_path / "scan-root" + models = scan_root / "models" + models.mkdir(parents=True) + file_path = _make_cacheable_file(models) + unsafe_cache = scan_root / "cache" + cache_manager = get_cache_manager(str(unsafe_cache), enabled=True) + assert cache_manager.cache is not None + cache = cache_manager.cache + batch_operations = BatchCacheOperations(cache_manager) + version_context = build_cache_version_context({REPOSITORY_SCAN_ROOT_CONFIG_KEY: str(scan_root)}) + expected = { + "checks": [], + "issues": [], + "metadata": {}, + "scanner": "test", + "success": True, + } + + assert cache_manager.store_result( + str(file_path), + expected, + version_context=version_context, + **_identity_kwargs(cache, str(file_path)), + ) + cache_key = cache.generate_cache_key(str(file_path), version_context=version_context) + assert cache_key is not None + assert cache._get_cache_file_path(cache_key).exists() + for existing_probe, _probe_directory in tuple(cache._change_clock_probes.values()): + existing_probe.close() + cache._change_clock_probes.clear() + + off_device_temp = tmp_path / "off-device-temp" + attempted_probe_dirs: list[Path] = [] + + def record_probe_attempt(*_args: Any, dir: str | Path, **_kwargs: Any) -> BinaryIO: + attempted_probe_dirs.append(type(file_path)(dir)) + raise OSError("a batch repository-root probe must never be created") + + with monkeypatch.context() as patch: + patch.setattr(os, "name", "nt") + patch.setattr(scan_results_cache_module, "Path", type(file_path)) + patch.setattr(tempfile, "gettempdir", lambda: str(off_device_temp)) + patch.setattr( + cache, + "_directory_is_on_device", + lambda directory, _device: directory == unsafe_cache, + ) + patch.setattr(tempfile, "TemporaryFile", record_probe_attempt) + + results = batch_operations.batch_lookup( + [str(file_path)], + version_context=version_context, + ) + + assert results[str(file_path)] is None + assert attempted_probe_dirs == [] + assert cache._change_clock_probes == {} + + +def test_windows_cache_lookup_propagates_repository_root_through_legacy_override( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + scan_root = tmp_path / "scan-root" + models = scan_root / "models" + models.mkdir(parents=True) + file_path = _make_cacheable_file(models) + unsafe_cache = scan_root / "cache" + off_device_temp = tmp_path / "off-device-temp" + attempted_probe_dirs: list[Path] = [] + + class LegacyCaptureCache(ScanResultsCache): + def __init__(self, cache_dir: str) -> None: + super().__init__(cache_dir) + self.capture_called = False + + def capture_file_identity(self, path: str) -> Any: + self.capture_called = True + return super().capture_file_identity(path) + + cache = LegacyCaptureCache(str(unsafe_cache)) + version_context = build_cache_version_context({REPOSITORY_SCAN_ROOT_CONFIG_KEY: str(scan_root)}) + + def forbid_internal_probe(*_args: Any, dir: str | Path, **_kwargs: Any) -> BinaryIO: + attempted_probe_dirs.append(type(file_path)(dir)) + raise OSError("a legacy override must not create a probe in the repository") + + with monkeypatch.context() as patch: + patch.setattr(os, "name", "nt") + patch.setattr(scan_results_cache_module, "Path", type(file_path)) + patch.setattr(tempfile, "gettempdir", lambda: str(off_device_temp)) + patch.setattr( + cache, + "_directory_is_on_device", + lambda directory, _device: directory == unsafe_cache, + ) + patch.setattr(tempfile, "TemporaryFile", forbid_internal_probe) + + result = cache.get_cached_result_with_identity( + str(file_path), + version_context=version_context, + ) + + assert result == (None, None) + assert cache.capture_called is True + assert attempted_probe_dirs == [] + assert cache._change_clock_probes == {} + assert scan_results_cache_module._CAPTURE_REPOSITORY_SCAN_ROOT.get() is None + + +def test_windows_batch_lookup_propagates_repository_root_through_legacy_override( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + scan_root = tmp_path / "scan-root" + models = scan_root / "models" + models.mkdir(parents=True) + file_path = _make_cacheable_file(models) + unsafe_cache = scan_root / "cache" + off_device_temp = tmp_path / "off-device-temp" + + class LegacyCaptureCache(ScanResultsCache): + def __init__(self, cache_dir: str) -> None: + super().__init__(cache_dir) + self.capture_called = False + + def capture_file_identity(self, path: str) -> Any: + self.capture_called = True + return super().capture_file_identity(path) + + cache_manager = get_cache_manager(str(unsafe_cache), enabled=True) + cache = LegacyCaptureCache(str(unsafe_cache)) + cache_manager.cache = cache + batch_operations = BatchCacheOperations(cache_manager) + version_context = build_cache_version_context({REPOSITORY_SCAN_ROOT_CONFIG_KEY: str(scan_root)}) + expected = { + "checks": [], + "issues": [], + "metadata": {}, + "scanner": "test", + "success": True, + } + + assert cache_manager.store_result( + str(file_path), + expected, + version_context=version_context, + **_identity_kwargs(cache, str(file_path)), + ) + cache_key = cache.generate_cache_key(str(file_path), version_context=version_context) + assert cache_key is not None + assert cache._get_cache_file_path(cache_key).exists() + for existing_probe, _probe_directory in tuple(cache._change_clock_probes.values()): + existing_probe.close() + cache._change_clock_probes.clear() + cache.capture_called = False + attempted_probe_dirs: list[Path] = [] + + def forbid_internal_probe(*_args: Any, dir: str | Path, **_kwargs: Any) -> BinaryIO: + attempted_probe_dirs.append(type(file_path)(dir)) + raise OSError("a legacy batch override must not create a repository-root probe") + + with monkeypatch.context() as patch: + patch.setattr(os, "name", "nt") + patch.setattr(scan_results_cache_module, "Path", type(file_path)) + patch.setattr(tempfile, "gettempdir", lambda: str(off_device_temp)) + patch.setattr( + cache, + "_directory_is_on_device", + lambda directory, _device: directory == unsafe_cache, + ) + patch.setattr(tempfile, "TemporaryFile", forbid_internal_probe) + + results = batch_operations.batch_lookup( + [str(file_path)], + version_context=version_context, + ) + + assert results[str(file_path)] is None + assert cache.capture_called is True + assert attempted_probe_dirs == [] + assert cache._change_clock_probes == {} + assert scan_results_cache_module._CAPTURE_REPOSITORY_SCAN_ROOT.get() is None + + +def test_repository_scan_root_context_resets_after_legacy_override_failure( + tmp_path: Path, +) -> None: + scan_root = tmp_path / "scan-root" + scan_root.mkdir() + file_path = _make_cacheable_file(scan_root) + + class FailingLegacyCaptureCache(ScanResultsCache): + def capture_file_identity(self, _path: str) -> Any: + raise ValueError("legacy identity capture deliberately failed") + + cache = FailingLegacyCaptureCache(str(tmp_path / "isolated-cache")) + version_context = build_cache_version_context({REPOSITORY_SCAN_ROOT_CONFIG_KEY: str(scan_root)}) + + assert cache.get_cached_result_with_identity( + str(file_path), + version_context=version_context, + ) == (None, None) + assert scan_results_cache_module._CAPTURE_REPOSITORY_SCAN_ROOT.get() is None + + +@pytest.mark.parametrize("safe_location", ("cache", "system_temp")) +def test_windows_change_clock_probe_preserves_isolated_same_device_fallback( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + safe_location: str, +) -> None: + scanned_directory = tmp_path / "scanned" + scanned_directory.mkdir() + file_path = _make_cacheable_file(scanned_directory) + file_device = file_path.stat().st_dev + cache = ScanResultsCache(str(tmp_path / "isolated-cache")) + system_temp = tmp_path / "isolated-temp" + system_temp.mkdir() + safe_candidate = cache.cache_dir if safe_location == "cache" else system_temp + attempted_probe_dirs: list[Path] = [] + + with tempfile.TemporaryFile(mode="w+b", dir=safe_candidate) as probe: + + def return_isolated_probe(*_args: Any, dir: str | Path, **_kwargs: Any) -> BinaryIO: + attempted_probe_dirs.append(type(file_path)(dir)) + return cast(BinaryIO, probe) + + with monkeypatch.context() as patch: + patch.setattr(os, "name", "nt") + patch.setattr(scan_results_cache_module, "Path", type(file_path)) + patch.setattr(tempfile, "gettempdir", lambda: str(system_temp)) + patch.setattr( + cache, + "_directory_is_on_device", + lambda directory, _device: directory == safe_candidate, + ) + patch.setattr(tempfile, "TemporaryFile", return_isolated_probe) + + selected_probe = cache._get_change_clock_probe(str(file_path), file_device) + + assert selected_probe is probe + assert attempted_probe_dirs == [safe_candidate] + assert cache._change_clock_probes[file_device][1] == safe_candidate + cache._change_clock_probes.clear() + + def test_change_clock_probe_prefers_isolated_directory(tmp_path: Path) -> None: file_path = _make_cacheable_file(tmp_path) cache = ScanResultsCache(str(tmp_path / "cache")) @@ -207,8 +1093,256 @@ def test_change_clock_probe_prefers_isolated_directory(tmp_path: Path) -> None: file_stat, _file_hash, _change_token, ancestor_identity = cache.capture_file_identity(str(file_path)) assert ancestor_identity - expected_probe_dir = Path(tempfile.gettempdir()) if os.name == "nt" else cache.cache_dir - assert cache._change_clock_probes[file_stat.st_dev][1] == expected_probe_dir + expected_probe_dirs = {Path(tempfile.gettempdir()), cache.cache_dir} if os.name == "nt" else {cache.cache_dir} + assert cache._change_clock_probes[file_stat.st_dev][1] in expected_probe_dirs + + +def test_clear_cache_closes_reusable_change_clock_probes( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + cache = ScanResultsCache(str(tmp_path / "cache")) + original_unlink = Path.unlink + + with tempfile.NamedTemporaryFile(mode="w+b", dir=cache.cache_dir, delete=False) as probe: + probe_path = Path(probe.name) + cache._change_clock_probes[probe_path.stat().st_dev] = (cast(BinaryIO, probe), cache.cache_dir) + + def reject_locked_probe(path: Path, *, missing_ok: bool = False) -> None: + if path == probe_path and not probe.closed: + raise PermissionError("Windows cannot unlink an open cache clock probe") + original_unlink(path, missing_ok=missing_ok) + + monkeypatch.setattr(Path, "unlink", reject_locked_probe) + + cache.clear_cache() + + assert probe.closed is True + + assert cache._change_clock_probes == {} + assert not probe_path.exists() + assert cache.metadata_file.exists() + + +def test_clear_cache_closes_remaining_probes_after_close_failure( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + cache = ScanResultsCache(str(tmp_path / "cache")) + stale_clean_result = cache.cache_dir / "stale-clean-result.json" + stale_clean_result.write_text('{"verdict":"CLEAN","findings":[]}', encoding="utf-8") + failed_probe_path = cache.cache_dir / ".modelaudit-cache-clock-retry" + failed_probe_path.write_bytes(b"locked probe") + close_attempts: list[str] = [] + removed_paths: list[Path] = [] + original_unlink = Path.unlink + + class CachedProbe: + def __init__( + self, + label: str, + *, + probe_path: Path | None = None, + fail_close: bool = False, + ) -> None: + self.label = label + self.name = str(probe_path) if probe_path is not None else label + self.fail_close = fail_close + self.closed = False + + def close(self) -> None: + close_attempts.append(self.label) + if self.fail_close: + raise OSError("simulated probe close failure") + self.closed = True + + first_probe = CachedProbe("first", probe_path=failed_probe_path, fail_close=True) + second_probe = CachedProbe("second") + + def record_unlink(path: Path, *, missing_ok: bool = False) -> None: + removed_paths.append(path) + if path == failed_probe_path and not first_probe.closed: + raise PermissionError("Windows cannot unlink an open cache clock probe") + original_unlink(path, missing_ok=missing_ok) + + cache._change_clock_probes[1] = (cast(BinaryIO, first_probe), cache.cache_dir) + cache._change_clock_probes[2] = (cast(BinaryIO, second_probe), cache.cache_dir) + monkeypatch.setattr(Path, "unlink", record_unlink) + + cache.clear_cache() + + assert close_attempts == ["first", "second"] + assert (first_probe.closed, second_probe.closed) == (False, True) + assert set(cache._change_clock_probes) == {1} + assert cache._change_clock_probes[1][0] is cast(BinaryIO, first_probe) + assert stale_clean_result in removed_paths + assert not stale_clean_result.exists() + assert failed_probe_path not in removed_paths + assert failed_probe_path.exists() + assert cache.metadata_file not in removed_paths + assert cache.metadata_file.exists() + + first_probe.fail_close = False + cache.clear_cache() + + assert close_attempts == ["first", "second", "first"] + assert first_probe.closed is True + assert cache._change_clock_probes == {} + assert failed_probe_path in removed_paths + assert not failed_probe_path.exists() + assert cache.metadata_file.exists() + metadata = json.loads(cache.metadata_file.read_text(encoding="utf-8")) + assert metadata["statistics"]["total_entries"] == 0 + + +def test_clear_cache_retries_real_temporary_file_wrapper( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + cache = ScanResultsCache(str(tmp_path / "cache")) + stale_result = cache.cache_dir / "stale-result.json" + stale_result.write_text('{"verdict":"CLEAN"}', encoding="utf-8") + close_attempts: list[str] = [] + removed_paths: list[Path] = [] + original_unlink = Path.unlink + + with tempfile.NamedTemporaryFile( + mode="w+b", + dir=cache.cache_dir, + prefix=".modelaudit-cache-clock-", + delete=False, + ) as probe: + probe_path = Path(probe.name) + underlying_probe = probe.file + original_close = underlying_probe.close + + def close_underlying_probe() -> None: + close_attempts.append("underlying") + if len(close_attempts) == 1: + raise OSError("transient underlying Windows probe close failure") + original_close() + + def reject_locked_probe(path: Path, *, missing_ok: bool = False) -> None: + if path == probe_path and not underlying_probe.closed: + raise PermissionError("Windows cannot remove an open cache clock probe") + removed_paths.append(path) + original_unlink(path, missing_ok=missing_ok) + + monkeypatch.setattr(underlying_probe, "close", close_underlying_probe) + monkeypatch.setattr(Path, "unlink", reject_locked_probe) + file_device = probe_path.stat().st_dev + cache._change_clock_probes[file_device] = (cast(BinaryIO, probe), cache.cache_dir) + + cache.clear_cache() + + assert (close_attempts, probe.closed) == (["underlying"], False) + assert cache._change_clock_probes[file_device][0] is cast(BinaryIO, probe) + assert probe_path not in removed_paths + assert probe_path.exists() + assert stale_result in removed_paths + assert cache.metadata_file.exists() + + cache.clear_cache() + + assert (close_attempts, probe.closed) == (["underlying", "underlying"], True) + assert cache._change_clock_probes == {} + assert probe_path in removed_paths + assert not probe_path.exists() + assert cache.metadata_file.exists() + + +def test_clear_cache_holds_probe_lock_through_deletion_and_metadata_reset( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + cache = ScanResultsCache(str(tmp_path / "cache")) + stale_result = cache.cache_dir / "stale-result.json" + stale_result.write_text('{"verdict":"CLEAN"}', encoding="utf-8") + observed_phases: list[str] = [] + original_iterdir = Path.iterdir + original_create_metadata = cache._create_initial_metadata + + def observe_locked_directory(path: Path) -> Iterator[Path]: + if path == cache.cache_dir: + assert cache._change_clock_probe_lock.locked() + assert cache._change_clock_probe_lock.acquire(blocking=False) is False + observed_phases.append("delete") + return original_iterdir(path) + + def observe_locked_metadata() -> None: + assert cache._change_clock_probe_lock.locked() + assert cache._change_clock_probe_lock.acquire(blocking=False) is False + observed_phases.append("metadata") + original_create_metadata() + + monkeypatch.setattr(Path, "iterdir", observe_locked_directory) + monkeypatch.setattr(cache, "_create_initial_metadata", observe_locked_metadata) + + cache.clear_cache() + + assert observed_phases == ["delete", "metadata"] + assert not stale_result.exists() + assert cache.metadata_file.exists() + assert cache._change_clock_probe_lock.locked() is False + + +def test_clear_cache_waits_for_active_identity_capture( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + file_path = _make_cacheable_file(tmp_path) + cache = ScanResultsCache(str(tmp_path / "cache")) + capture_hash_entered = threading.Event() + allow_capture_hash = threading.Event() + clear_started = threading.Event() + clear_completed = threading.Event() + worker_errors: list[BaseException] = [] + original_hash = cache.hasher.hash_file_with_stat + + def block_identity_hash(path: str, file_stat: os.stat_result) -> str: + capture_hash_entered.set() + if not allow_capture_hash.wait(timeout=10): + raise TimeoutError("identity capture was not released") + return original_hash(path, file_stat) + + def capture_identity() -> None: + try: + identity = cache.capture_file_identity(str(file_path)) + cache.release_ancestor_identity(identity[-1]) + except BaseException as exc: + worker_errors.append(exc) + + def clear_identity_cache() -> None: + clear_started.set() + try: + cache.clear_cache() + except BaseException as exc: + worker_errors.append(exc) + finally: + clear_completed.set() + + monkeypatch.setattr(cache.hasher, "hash_file_with_stat", block_identity_hash) + capture_thread = threading.Thread(target=capture_identity, daemon=True) + clear_thread = threading.Thread(target=clear_identity_cache, daemon=True) + capture_thread.start() + + try: + assert capture_hash_entered.wait(timeout=10) + clear_thread.start() + assert clear_started.wait(timeout=10) + assert clear_completed.wait(timeout=0.05) is False + finally: + allow_capture_hash.set() + capture_thread.join(timeout=10) + if clear_thread.is_alive(): + clear_thread.join(timeout=10) + + assert not capture_thread.is_alive() + assert not clear_thread.is_alive() + assert worker_errors == [] + assert clear_completed.is_set() + assert cache._active_identity_captures == 0 + assert cache.metadata_file.exists() def test_windows_change_clock_probe_uses_existing_handle( diff --git a/tests/test_release_workflow.py b/tests/test_release_workflow.py index 15bebdc46..507501ef0 100644 --- a/tests/test_release_workflow.py +++ b/tests/test_release_workflow.py @@ -98,6 +98,40 @@ def test_root_release_accepts_current_picklescan_version() -> None: assert Requirement(picklescan_requirements[0]).specifier.contains(picklescan_project["version"]) +def test_standalone_package_lint_uses_locked_root_ruff_version() -> None: + root_dir = Path(__file__).resolve().parents[1] + root_lock = tomllib.loads((root_dir / "uv.lock").read_text(encoding="utf-8")) + ruff_package = next(package for package in root_lock["package"] if package["name"] == "ruff") + expected_requirement = f"ruff=={ruff_package['version']}" + python_workflow = yaml.safe_load((root_dir / ".github" / "workflows" / "test.yml").read_text(encoding="utf-8")) + assert isinstance(python_workflow, dict) + + for workflow, job_name in ( + (_load_release_workflow(), "build-picklescan-package"), + (python_workflow, "picklescan-package"), + ): + ruff_runs = [ + run + for step in _job_steps(workflow, job_name) + if isinstance(run := step.get("run"), str) and ("ruff check" in run or "ruff format" in run) + ] + assert ruff_runs, f"{job_name} must run Ruff" + for run in ruff_runs: + assert f"uv run --with '{expected_requirement}' ruff" in run + + expected_commands = ( + f"uv run --with '{expected_requirement}' ruff check src tests", + f"uv run --with '{expected_requirement}' ruff format --check src tests", + ) + for guide in ( + root_dir / "packages/modelaudit-picklescan/AGENTS.md", + root_dir / "docs/agents/picklescan-package-split.md", + ): + guide_lines = guide.read_text(encoding="utf-8").splitlines() + for expected_command in expected_commands: + assert expected_command in guide_lines + + def test_release_workflow_manual_dispatch_inputs_and_guardrails() -> None: workflow = _load_release_workflow()