diff --git a/.gitignore b/.gitignore index 11b4a67d..8accea35 100644 --- a/.gitignore +++ b/.gitignore @@ -301,6 +301,13 @@ __marimo__/ /e agora*.txt /*.txt +# OPSEC: operator-local paths — never `git add -A` (explicit `git add ` only). +# docs/ops/today-mode/ hub files already on origin stay tracked; this blocks new untracked drafts. +docs/ops/today-mode/ +scripts/beastie-ecosystem-sync.ps1 +scripts/beastie-ecosystem-sync.sh +api/static/mascot/bata_boar_mascot_pot.png + # Generated POC test corpus (on-demand, not tracked) tests/synthetic_corpus/ .cursor/rules/dossier-update-on-evidence.mdc diff --git a/connectors/filesystem_connector.py b/connectors/filesystem_connector.py index ed30cda2..37886767 100644 --- a/connectors/filesystem_connector.py +++ b/connectors/filesystem_connector.py @@ -10,13 +10,15 @@ import stat import tempfile import zipfile -from collections.abc import Iterator +from collections.abc import Callable, Iterator from pathlib import Path from typing import Any from core.connector_registry import register from core.archives import ( ArchiveUnsupportedError, + SCAN_FAILURE_REASON_ENCRYPTED_NO_PASSWORD, + SCAN_FAILURE_REASON_WRONG_PASSWORD, default_compressed_extensions, detect_archive_type, iter_archive_members, @@ -287,6 +289,7 @@ def _read_text_sample( ocr_max_dimension: int = 2000, ocr_lang: str = "eng", scan_for_stego: bool = False, + on_read_barrier: Callable[[str, str], None] | None = None, ) -> str: """Extract text from file for sensitivity scan; return empty on error. No content stored after return. @@ -346,8 +349,18 @@ def finalize(text: str) -> str: if reader.is_encrypted: password = pw.get(ext) or pw.get("default") if password: - reader.decrypt(password) + decrypt_status = reader.decrypt(password) + if decrypt_status == 0: + if on_read_barrier is not None: + on_read_barrier( + SCAN_FAILURE_REASON_WRONG_PASSWORD, str(path) + ) + return finalize("") else: + if on_read_barrier is not None: + on_read_barrier( + SCAN_FAILURE_REASON_ENCRYPTED_NO_PASSWORD, str(path) + ) return finalize("") return finalize( " ".join(p.extract_text() or "" for p in reader.pages[:5])[:max_chars] @@ -778,6 +791,10 @@ def run(self) -> None: _size = _stat.st_size except OSError: # noqa: BLE001 pass # stat() best-effort for file-identity metadata + + def _record_read_barrier(reason: str, details: str) -> None: + self.db_manager.save_failure(target_name, reason, details) + content = _read_text_sample( file_path, effective_ext, @@ -788,6 +805,7 @@ def run(self) -> None: ocr_max_dimension=self.ocr_max_dimension, ocr_lang=self.ocr_lang, scan_for_stego=self.scan_for_stego, + on_read_barrier=_record_read_barrier, ) res = self.scanner.scan_file_content(content, file_path) if res is None: @@ -891,12 +909,17 @@ def scan_archive_at_path( except (TypeError, ValueError): max_size = DEFAULT_MAX_INNER_SIZE try: + + def _on_member_read_failure(reason: str, member: str, details: str) -> None: + db_manager.save_failure(target_name, reason, details) + for member_name, data in iter_archive_members( archive_path, archive_type, max_size, extensions, file_passwords, + on_member_read_failure=_on_member_read_failure, ): ext = Path(member_name).suffix.lower() tmp_path: Path | None = None @@ -910,6 +933,16 @@ def scan_archive_at_path( ext_use = choose_effective_rich_media_extension( ext_use, use_content_type, tmp_path ) + + def _on_read_barrier( + reason: str, details: str, *, _member: str = member_name + ) -> None: + db_manager.save_failure( + target_name, + reason, + f"{archive_display_name}|{_member}: {details}", + ) + content = _read_text_sample( tmp_path, ext_use, @@ -920,6 +953,7 @@ def scan_archive_at_path( ocr_max_dimension=ocr_max_dimension, ocr_lang=ocr_lang, scan_for_stego=scan_for_stego, + on_read_barrier=_on_read_barrier, ) res = scanner.scan_file_content(content, member_name) if res is None: diff --git a/core/archives.py b/core/archives.py index 216c8d00..6971eae2 100644 --- a/core/archives.py +++ b/core/archives.py @@ -9,11 +9,19 @@ from __future__ import annotations import zipfile +from collections.abc import Callable from pathlib import Path from typing import Iterable, Iterator import tarfile +# scan_failures reasons for unreadable archive members (#828; shared taxonomy with SQL sampling_error). +SCAN_FAILURE_REASON_ENCRYPTED_NO_PASSWORD = "encrypted_no_password" +SCAN_FAILURE_REASON_WRONG_PASSWORD = "wrong_password" +SCAN_FAILURE_REASON_ARCHIVE_READ_ERROR = "archive_read_error" + +MemberReadFailureCallback = Callable[[str, str, str], None] + # Tier 1 – stdlib-backed formats (see PLAN_COMPRESSED_FILES.md) _TIER1_EXTENSIONS: set[str] = { @@ -201,12 +209,55 @@ class ArchiveUnsupportedError(Exception): pass +def classify_zip_member_read_failure( + exc: BaseException, + *, + encrypted: bool, + password_provided: bool, +) -> str: + """Map zipfile member read errors to scan_failures reason codes (#828).""" + if encrypted and not password_provided: + return SCAN_FAILURE_REASON_ENCRYPTED_NO_PASSWORD + if encrypted and password_provided: + msg = str(exc).lower() + if ( + "bad password" in msg + or "incorrect password" in msg + or "bad crc-32" in msg + or "bad crc" in msg + ): + return SCAN_FAILURE_REASON_WRONG_PASSWORD + return SCAN_FAILURE_REASON_ARCHIVE_READ_ERROR + + +def classify_7z_member_read_failure( + exc: BaseException, + *, + password_provided: bool, +) -> str: + """Map py7zr member read errors to scan_failures reason codes (#828).""" + try: + from py7zr import exceptions as py7zr_exc + except ImportError: + py7zr_exc = None # type: ignore[misc, assignment] + if py7zr_exc is not None and isinstance(exc, py7zr_exc.PasswordRequired): + return SCAN_FAILURE_REASON_ENCRYPTED_NO_PASSWORD + if password_provided: + if py7zr_exc is not None and isinstance(exc, py7zr_exc.CrcError): + return SCAN_FAILURE_REASON_WRONG_PASSWORD + msg = str(exc).lower() + if "password" in msg or "wrong" in msg or "crc" in msg: + return SCAN_FAILURE_REASON_WRONG_PASSWORD + return SCAN_FAILURE_REASON_ARCHIVE_READ_ERROR + + def iter_archive_members( path: Path, archive_type: str, max_inner_size: int, allowed_extensions: Iterable[str] | set[str], file_passwords: dict[str, str] | None = None, + on_member_read_failure: MemberReadFailureCallback | None = None, ) -> Iterator[tuple[str, bytes]]: """ Yield (member_name, content_bytes) for each file inside the archive whose @@ -238,7 +289,19 @@ def iter_archive_members( continue try: data = z.read(name) - except (RuntimeError, zipfile.BadZipFile): + except (RuntimeError, zipfile.BadZipFile) as e: + if on_member_read_failure is not None: + encrypted = bool(info.flag_bits & 0x1) + reason = classify_zip_member_read_failure( + e, + encrypted=encrypted, + password_provided=bool(pwd), + ) + on_member_read_failure( + reason, + name, + f"{path.name}|{name}: {type(e).__name__}: {e}", + ) continue yield (name, data) except (zipfile.BadZipFile, OSError): @@ -278,6 +341,8 @@ def iter_archive_members( "on a host without liblzma see issue #926" ) pwd = pw.get(".7z") or pw.get("default") or None + from py7zr.exceptions import PasswordRequired + try: with py7zr.SevenZipFile(path, "r", password=pwd) as archive: # py7zr >= 0.22: use list(); older docs used files_list (removed in 1.x). @@ -296,8 +361,25 @@ def iter_archive_members( if bio is not None: data = bio.read() yield (member.filename, data) - except (Exception, OSError): + except (Exception, OSError) as e: + if on_member_read_failure is not None: + reason = classify_7z_member_read_failure( + e, password_provided=bool(pwd) + ) + on_member_read_failure( + reason, + member.filename, + f"{path.name}|{member.filename}: {type(e).__name__}: {e}", + ) continue + except PasswordRequired as e: + if on_member_read_failure is not None: + on_member_read_failure( + SCAN_FAILURE_REASON_ENCRYPTED_NO_PASSWORD, + path.name, + f"{path.name}: {type(e).__name__}: {e}", + ) + return except (py7zr.Bad7zFile, OSError): return diff --git a/core/database.py b/core/database.py index 21b1463e..225e1833 100644 --- a/core/database.py +++ b/core/database.py @@ -169,6 +169,17 @@ def failure_hint(reason: str) -> str: "evaluated — this is not a clean result. Check dialect-specific sampling notes, credentials, " "and the detailed message; fix the target or sampling plan before re-running." ) + if r == "encrypted_no_password": + return ( + "File or archive member is encrypted and no password was configured. This is not a clean " + "result — add the password under file_scan.file_passwords (or the matching extension key) " + "and re-run, or exclude the path if decryption is out of scope." + ) + if r == "wrong_password": + return ( + "Password was provided but decryption failed. Verify file_scan.file_passwords for this " + "extension; a wrong password is not a clean scan result." + ) return ( "Unexpected error. Review the detailed message and audit log, verify the target configuration " "(host, port, path, credentials) and test connectivity manually before re-running." diff --git a/tests/test_archives_scan_failures.py b/tests/test_archives_scan_failures.py new file mode 100644 index 00000000..d72d2103 --- /dev/null +++ b/tests/test_archives_scan_failures.py @@ -0,0 +1,220 @@ +"""#828 — archive member read failures must surface in scan_failures, not silent skip.""" + +import zipfile +from unittest.mock import MagicMock + +import pytest + +from connectors.filesystem_connector import FilesystemConnector, scan_archive_at_path +from core.archives import ( + SCAN_FAILURE_REASON_ENCRYPTED_NO_PASSWORD, + SCAN_FAILURE_REASON_WRONG_PASSWORD, + classify_7z_member_read_failure, + classify_zip_member_read_failure, +) +from core.scanner import DataScanner + + +class _DummyDB: + def __init__(self) -> None: + self.failures: list[tuple[str, str, str]] = [] + self.findings: list[dict] = [] + + def save_failure(self, target_name: str, reason: str, details: str) -> None: + self.failures.append((target_name, reason, details)) + + def save_finding(self, *args, **kwargs) -> None: + self.findings.append(kwargs) + + +def test_classify_zip_member_read_failure_encrypted_no_password(): + err = RuntimeError("File inner.txt is encrypted, password required for extraction") + assert ( + classify_zip_member_read_failure(err, encrypted=True, password_provided=False) + == SCAN_FAILURE_REASON_ENCRYPTED_NO_PASSWORD + ) + + +def test_classify_zip_member_read_failure_wrong_password(): + err = RuntimeError("Bad password for file 'inner.txt'") + assert ( + classify_zip_member_read_failure(err, encrypted=True, password_provided=True) + == SCAN_FAILURE_REASON_WRONG_PASSWORD + ) + + +def test_classify_zip_member_read_failure_bad_crc32_wrong_password(): + err = zipfile.BadZipFile("Bad CRC-32 for file 'inner.txt'") + assert ( + classify_zip_member_read_failure(err, encrypted=True, password_provided=True) + == SCAN_FAILURE_REASON_WRONG_PASSWORD + ) + + +def test_classify_7z_member_read_failure_password_required(): + pytest.importorskip("py7zr") + from py7zr.exceptions import PasswordRequired + + assert ( + classify_7z_member_read_failure(PasswordRequired(), password_provided=False) + == SCAN_FAILURE_REASON_ENCRYPTED_NO_PASSWORD + ) + + +def test_classify_7z_member_read_failure_crc_wrong_password(): + pytest.importorskip("py7zr") + from py7zr.exceptions import CrcError + + assert ( + classify_7z_member_read_failure( + CrcError(0, 1, "member.txt"), password_provided=True + ) + == SCAN_FAILURE_REASON_WRONG_PASSWORD + ) + assert ( + classify_7z_member_read_failure( + OSError("crc check failed"), password_provided=True + ) + == SCAN_FAILURE_REASON_WRONG_PASSWORD + ) + + +def test_scan_archive_at_path_inner_sample_uses_on_read_barrier( + tmp_path, monkeypatch: pytest.MonkeyPatch +): + """Encrypted PDF inside a zip must record scan_failures via on_read_barrier (#828).""" + import connectors.filesystem_connector as fs_mod + + zip_path = tmp_path / "bundle.zip" + with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as z: + z.writestr("inner.pdf", b"%PDF-1.4") + + db = _DummyDB() + scanner = MagicMock() + scanner.scan_file_content.return_value = None + + def _capture_barrier_read(*_args, **kwargs): # type: ignore[no-untyped-def] + barrier = kwargs.get("on_read_barrier") + if barrier is not None: + barrier("encrypted_no_password", "encrypted inner pdf") + return "" + + monkeypatch.setattr(fs_mod, "_read_text_sample", _capture_barrier_read) + + scan_archive_at_path( + archive_path=zip_path, + archive_display_name=zip_path.name, + target_name="T", + path_display=str(tmp_path), + scanner=scanner, + db_manager=db, + extensions={".pdf"}, + max_inner_size=1_000_000, + file_passwords={}, + file_sample_max_chars=5000, + ) + + encrypted = [ + f for f in db.failures if f[1] == SCAN_FAILURE_REASON_ENCRYPTED_NO_PASSWORD + ] + assert len(encrypted) == 1 + assert "bundle.zip|inner.pdf" in encrypted[0][2] + + +def test_scan_compressed_encrypted_zip_member_records_failure( + tmp_path, monkeypatch: pytest.MonkeyPatch +): + """Encrypted inner member without password must not masquerade as a clean archive scan.""" + root = tmp_path / "root" + root.mkdir() + zip_path = root / "locked.zip" + with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as z: + z.writestr("inner.txt", "CPF 123.456.789-00") + + original_getinfo = zipfile.ZipFile.getinfo + + def _encrypted_getinfo(self, name): # type: ignore[no-untyped-def] + info = original_getinfo(self, name) + info.flag_bits = info.flag_bits | 0x1 + return info + + def _encrypted_read(self, name, pwd=None): # type: ignore[no-untyped-def] + raise RuntimeError( + f"File {name!r} is encrypted, password required for extraction" + ) + + monkeypatch.setattr(zipfile.ZipFile, "getinfo", _encrypted_getinfo) + monkeypatch.setattr(zipfile.ZipFile, "read", _encrypted_read) + + db = _DummyDB() + scanner = DataScanner() + target = { + "name": "FS", + "type": "filesystem", + "path": str(root), + "recursive": True, + "file_scan": { + "scan_compressed": True, + "compressed_extensions": [".zip"], + }, + } + connector = FilesystemConnector( + target, + scanner, + db, + extensions=[".zip", ".txt"], + scan_sqlite_as_db=False, + sample_limit=5000, + file_passwords={}, + ) + connector.run() + + encrypted_failures = [ + f for f in db.failures if f[1] == SCAN_FAILURE_REASON_ENCRYPTED_NO_PASSWORD + ] + assert len(encrypted_failures) >= 1 + assert any("locked.zip" in f[2] for f in encrypted_failures) + inner_findings = [f for f in db.findings if "|" in f.get("file_name", "")] + assert len(inner_findings) == 0 + + +def test_scan_archive_at_path_wrong_password_records_failure( + tmp_path, monkeypatch: pytest.MonkeyPatch +): + zip_path = tmp_path / "bad_pwd.zip" + with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as z: + z.writestr("inner.txt", "hello") + + db = _DummyDB() + scanner = MagicMock() + scanner.scan_file_content.return_value = None + original_getinfo = zipfile.ZipFile.getinfo + + def _encrypted_getinfo(self, name): # type: ignore[no-untyped-def] + info = original_getinfo(self, name) + info.flag_bits = info.flag_bits | 0x1 + return info + + def _bad_password_read(self, name, pwd=None): # type: ignore[no-untyped-def] + if getattr(self, "pwd", None): + raise RuntimeError("Bad password for file 'inner.txt'") + raise RuntimeError("File 'inner.txt' is encrypted, password required") + + monkeypatch.setattr(zipfile.ZipFile, "getinfo", _encrypted_getinfo) + monkeypatch.setattr(zipfile.ZipFile, "read", _bad_password_read) + + scan_archive_at_path( + archive_path=zip_path, + archive_display_name=zip_path.name, + target_name="T", + path_display=str(tmp_path), + scanner=scanner, + db_manager=db, + extensions={".txt"}, + max_inner_size=1_000_000, + file_passwords={".zip": "wrong-secret"}, + file_sample_max_chars=5000, + ) + + wrong = [f for f in db.failures if f[1] == SCAN_FAILURE_REASON_WRONG_PASSWORD] + assert len(wrong) >= 1 diff --git a/tests/test_database.py b/tests/test_database.py index dbd362fb..e41d33b3 100644 --- a/tests/test_database.py +++ b/tests/test_database.py @@ -220,6 +220,8 @@ def test_failure_hint_other_reasons(): assert "auth" in failure_hint("auth_failed").lower() assert "permission" in failure_hint("permission_denied").lower() assert "sampling" in failure_hint("sampling_error").lower() + assert "encrypted" in failure_hint("encrypted_no_password").lower() + assert "password" in failure_hint("wrong_password").lower() assert ( "unexpected" in failure_hint("unknown").lower() or "error" in failure_hint("unknown").lower()