diff --git a/CHANGELOG.md b/CHANGELOG.md index a35827d21..9883ec2cc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Bug Fixes - Avoid network false positives for bounded README examples that download sample images over HTTPS from Hugging Face. +- Preserve Windows source-fingerprint cache hits while continuing to reject swapped or modified files. - Prevent Windows cache identity probes from creating locked temporary files inside scanned directories. - Preserve locked Windows cache probes reached through directory aliases while clearing stale scan results. - Place cross-volume Windows cache identity probes near the volume root instead of the nearest ancestor of the scanned path, so a probe can no longer appear inside a directory tree that a concurrent scan is walking. diff --git a/modelaudit/cache/scan_results_cache.py b/modelaudit/cache/scan_results_cache.py index ef4ba3323..c6b601ea6 100644 --- a/modelaudit/cache/scan_results_cache.py +++ b/modelaudit/cache/scan_results_cache.py @@ -1715,6 +1715,20 @@ def _regular_file_identity_fingerprint(file_stat: os.stat_result) -> str: digest = hashlib.sha256(identity.encode()).hexdigest() return f"{_CALL_GRAPH_REGULAR_FILE_FINGERPRINT}:{digest}" + @staticmethod + def _cross_view_file_stat_identity(file_stat: os.stat_result) -> tuple[int, ...]: + """Compare path and descriptor views without Windows-only stat differences.""" + identity = ( + file_stat.st_dev, + file_stat.st_ino, + stat.S_IFMT(file_stat.st_mode), + file_stat.st_size, + file_stat.st_mtime_ns, + ) + if os.name == "nt": + return identity + return (*identity, file_stat.st_ctime_ns) + @staticmethod def _bounded_source_fingerprint(path: Path) -> str | None: flags = os.O_RDONLY | getattr(os, "O_BINARY", 0) | getattr(os, "O_NONBLOCK", 0) @@ -1766,15 +1780,9 @@ def _bounded_source_fingerprint(path: Path) -> str | None: path_stat = path.stat() except OSError as error: raise ValueError("source fingerprint candidate path changed while being read") from error - path_identity = ( - path_stat.st_dev, - path_stat.st_ino, - path_stat.st_mode, - path_stat.st_size, - path_stat.st_mtime_ns, - path_stat.st_ctime_ns, - ) - if before_identity != after_identity or after_identity != path_identity: + if before_identity != after_identity or ScanResultsCache._cross_view_file_stat_identity( + after + ) != ScanResultsCache._cross_view_file_stat_identity(path_stat): raise ValueError("source fingerprint candidate changed while being read") if is_extension: return ScanResultsCache._regular_file_identity_fingerprint(after) @@ -1873,7 +1881,20 @@ def _bounded_read_fingerprint(path: Path, read_limit: int, require_complete: boo path_stat.st_mtime_ns, path_stat.st_ctime_ns, ) - if before_identity != after_identity or after_identity != path_identity: + initial_path_identity = ( + path_before.st_dev, + path_before.st_ino, + path_before.st_mode, + path_before.st_size, + path_before.st_mtime_ns, + path_before.st_ctime_ns, + ) + if ( + before_identity != after_identity + or initial_path_identity != path_identity + or ScanResultsCache._cross_view_file_stat_identity(after) + != ScanResultsCache._cross_view_file_stat_identity(path_stat) + ): raise ValueError("read fingerprint candidate changed while being read") if require_complete and len(source) > read_limit: raise ValueError("read fingerprint budget exceeded") diff --git a/tests/cache/test_cache_correctness.py b/tests/cache/test_cache_correctness.py index 5978d0d1e..2b9b1ab10 100644 --- a/tests/cache/test_cache_correctness.py +++ b/tests/cache/test_cache_correctness.py @@ -3,6 +3,7 @@ import hashlib import json import os +import stat import sys import tempfile import threading @@ -22,7 +23,7 @@ SourcelessFileLoader, ) from pathlib import Path -from types import FunctionType, ModuleType +from types import FunctionType, ModuleType, SimpleNamespace from typing import Any, BinaryIO, cast from zipimport import zipimporter @@ -3749,6 +3750,112 @@ def test_scan_cache_invalidates_replaced_large_extension_candidate(tmp_path: Pat assert cache.get_cached_result(str(file_path), version_context=version_context) is None +@pytest.mark.parametrize("fingerprint_kind", ["source", "read"]) +def test_source_fingerprints_accept_windows_cross_view_stat_differences( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + fingerprint_kind: str, +) -> None: + source_path = tmp_path / "helper.py" + source = b"def entrypoint():\n return 1\n" + source_path.write_bytes(source) + original_fstat = os.fstat + + def windows_descriptor_stat(file_descriptor: int) -> os.stat_result: + file_stat = original_fstat(file_descriptor) + return cast( + os.stat_result, + SimpleNamespace( + st_dev=file_stat.st_dev, + st_ino=file_stat.st_ino, + st_mode=file_stat.st_mode ^ 0o111, + st_size=file_stat.st_size, + st_mtime_ns=file_stat.st_mtime_ns, + st_ctime_ns=file_stat.st_ctime_ns + 1, + ), + ) + + with monkeypatch.context() as windows: + windows.setattr(os, "fstat", windows_descriptor_stat) + windows.setattr(os, "name", "nt") + if fingerprint_kind == "source": + fingerprint = ScanResultsCache._bounded_source_fingerprint(source_path) + expected = hashlib.sha256(source).hexdigest() + else: + fingerprint = ScanResultsCache._bounded_read_fingerprint(source_path, 64 * 1024, True) + expected = hashlib.sha256(b"file\0" + source).hexdigest() + + assert fingerprint == expected + + +@pytest.mark.parametrize("fingerprint_kind", ["source", "read"]) +def test_source_fingerprints_reject_windows_cross_view_file_replacement( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + fingerprint_kind: str, +) -> None: + source_path = tmp_path / "helper.py" + source_path.write_bytes(b"def entrypoint():\n return 1\n") + original_fstat = os.fstat + + def replaced_descriptor_stat(file_descriptor: int) -> os.stat_result: + file_stat = original_fstat(file_descriptor) + return cast( + os.stat_result, + SimpleNamespace( + st_dev=file_stat.st_dev, + st_ino=file_stat.st_ino + 1, + st_mode=stat.S_IFMT(file_stat.st_mode) | 0o666, + st_size=file_stat.st_size, + st_mtime_ns=file_stat.st_mtime_ns, + st_ctime_ns=file_stat.st_ctime_ns + 1, + ), + ) + + with monkeypatch.context() as windows: + windows.setattr(os, "fstat", replaced_descriptor_stat) + windows.setattr(os, "name", "nt") + with pytest.raises(ValueError, match="changed while being read"): + if fingerprint_kind == "source": + ScanResultsCache._bounded_source_fingerprint(source_path) + else: + ScanResultsCache._bounded_read_fingerprint(source_path, 64 * 1024, True) + + +@pytest.mark.skipif(os.name == "nt", reason="Windows ctime differs across stat views") +@pytest.mark.parametrize("fingerprint_kind", ["source", "read"]) +def test_source_fingerprints_reject_posix_cross_view_ctime_change( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + fingerprint_kind: str, +) -> None: + source_path = tmp_path / "helper.py" + source_path.write_bytes(b"def entrypoint():\n return 1\n") + original_fstat = os.fstat + + def changed_descriptor_stat(file_descriptor: int) -> os.stat_result: + file_stat = original_fstat(file_descriptor) + return cast( + os.stat_result, + SimpleNamespace( + st_dev=file_stat.st_dev, + st_ino=file_stat.st_ino, + st_mode=file_stat.st_mode, + st_size=file_stat.st_size, + st_mtime_ns=file_stat.st_mtime_ns, + st_ctime_ns=file_stat.st_ctime_ns + 1, + ), + ) + + monkeypatch.setattr(os, "fstat", changed_descriptor_stat) + + with pytest.raises(ValueError, match="changed while being read"): + if fingerprint_kind == "source": + ScanResultsCache._bounded_source_fingerprint(source_path) + else: + ScanResultsCache._bounded_read_fingerprint(source_path, 64 * 1024, True) + + @pytest.mark.skipif(sys.platform == "win32", reason="Windows prevents replacing an open source file") def test_source_fingerprint_rejects_path_replacement_during_read( tmp_path: Path,