diff --git a/tools/drive-archaeologist/src/drive_archaeologist/classifier.py b/tools/drive-archaeologist/src/drive_archaeologist/classifier.py index 69b49ec..4524ad4 100644 --- a/tools/drive-archaeologist/src/drive_archaeologist/classifier.py +++ b/tools/drive-archaeologist/src/drive_archaeologist/classifier.py @@ -2,10 +2,22 @@ File classification based on extension profiles. """ +import re from pathlib import Path from .profiles import CLASSIFICATION_PROFILES +# RINEX 2 short filename: ssssdddh.yyt (site, DOY, session, 2-digit year, type). +# The profile extension list only enumerates years .15–.22; real archives span +# far more (e.g. .05o, .23o) and include Hatanaka (.yyd) and met (.yym) files. +# o=obs n=nav g=GLONASS-nav d=Hatanaka m=met +_RINEX_SHORT_RE = re.compile(r"^[a-z0-9]{4}\d{3}[a-x0-9]\.\d{2}[ondgm]$", re.IGNORECASE) + +# Leica raw GNSS: .m00, .m01, ... — an extension family, so it cannot live in +# the static profile map. 3,665 of these surfaced unclassified on the first +# real GNSS-bearing drive (DOSTB20150918 $RECYCLE.BIN). +_LEICA_RAW_RE = re.compile(r"\.m\d{2}$", re.IGNORECASE) + class Classifier: """ @@ -33,7 +45,8 @@ def _build_extension_map(self) -> dict[str, str]: def classify(self, filepath: Path) -> str | None: """ - Classify a file based on its extension. + Classify a file based on its extension, with a RINEX short-name + regex fallback for year-extensions absent from the profile list. Args: filepath: The path to the file. @@ -42,7 +55,14 @@ def classify(self, filepath: Path) -> str | None: The classification category as a string, or None if no match is found. """ extension = filepath.suffix.lower() - return self._extension_map.get(extension) + category = self._extension_map.get(extension) + if category is not None: + return category + if _RINEX_SHORT_RE.match(filepath.name): + return "GNSS Data" + if _LEICA_RAW_RE.search(filepath.name): + return "GNSS Raw (Leica)" + return None def classify_by_ext(self, ext: str) -> str | None: return self._extension_map.get(ext.lower()) diff --git a/tools/drive-archaeologist/src/drive_archaeologist/cli.py b/tools/drive-archaeologist/src/drive_archaeologist/cli.py index adeb3fa..2f44519 100644 --- a/tools/drive-archaeologist/src/drive_archaeologist/cli.py +++ b/tools/drive-archaeologist/src/drive_archaeologist/cli.py @@ -1,12 +1,14 @@ """ CLI interface for drive-archaeologist using Click framework. -Provides the main 'scan' command for Phase 0. +Provides the 'scan' command (full JSONL catalog) and the 'survey' +command (fast wipe/keep triage, DA-003). """ from pathlib import Path import click from rich.console import Console +from rich.table import Table from .scanner import DeepScanner @@ -29,13 +31,43 @@ def main(): help="Output file path (default: scan__.jsonl)", ) @click.option("--resume", "-r", is_flag=True, help="Resume a previous interrupted scan") +@click.option("--force", is_flag=True, help="Overwrite an existing output file (otherwise refused)") +@click.option( + "--include-hidden", + is_flag=True, + help="Also scan hidden (dot-prefixed) and system entries ($RECYCLE.BIN, .Trash*, ...)", +) +@click.option( + "--exclude", + "-x", + "excludes", + multiple=True, + help="Glob (relative to scan root, or bare name) to skip; repeatable", +) +@click.option( + "--max-archive-depth", + type=int, + default=3, + show_default=True, + help="How many levels of nested archives to extract (0 = never extract)", +) @click.option( "--ingest", is_flag=True, help="Dispatch classified GNSS files to the ingestion pipeline" ) @click.option( "--dry-run", is_flag=True, help="Log what would be dispatched without sending to Celery" ) -def scan(path: Path, output: Path | None, resume: bool, ingest: bool, dry_run: bool): +def scan( + path: Path, + output: Path | None, + resume: bool, + force: bool, + include_hidden: bool, + excludes: tuple[str, ...], + max_archive_depth: int, + ingest: bool, + dry_run: bool, +): """Scan a drive or directory and produce a JSONL file with metadata""" on_classified = None if ingest or dry_run: @@ -44,7 +76,16 @@ def scan(path: Path, output: Path | None, resume: bool, ingest: bool, dry_run: b on_classified = make_dispatch_callback(dry_run=dry_run) try: - scanner = DeepScanner(path, output_file=output, resume=resume, on_classified=on_classified) + scanner = DeepScanner( + path, + output_file=output, + resume=resume, + on_classified=on_classified, + include_hidden=include_hidden, + excludes=list(excludes), + max_archive_depth=max_archive_depth, + force=force, + ) scanner.scan() except KeyboardInterrupt: console.print("\n[yellow]Warning: Scan interrupted by user[/yellow]") @@ -55,5 +96,83 @@ def scan(path: Path, output: Path | None, resume: bool, ingest: bool, dry_run: b raise click.Abort() from None +@main.command() +@click.argument("path", type=click.Path(exists=True, path_type=Path)) +@click.option( + "--include-hidden/--no-include-hidden", + default=True, + show_default=True, + help="Survey hidden/system entries too (a stick whose content sits in .Trash-1000 " + "would otherwise read as empty)", +) +@click.option( + "--exclude", + "-x", + "excludes", + multiple=True, + help="Glob (relative to survey root, or bare name) to skip; repeatable", +) +@click.option( + "--extract-archives", + is_flag=True, + help="Also look inside archives (slower; extracts to $TMPDIR)", +) +@click.option("--top", type=int, default=12, show_default=True, help="Rows in the category table") +def survey( + path: Path, + include_hidden: bool, + excludes: tuple[str, ...], + extract_archives: bool, + top: int, +): + """Fast triage: what's on this drive, and is it safe to wipe? + + Walks once with the same classifier as `scan` but writes NO catalog, + computes NO hashes, and (by default) opens NO archives. Prints a + category/extension breakdown and a wipe/keep verdict with explicit + disclosure of anything the walk did not cover. + """ + try: + scanner = DeepScanner( + path, + stats_only=True, + include_hidden=include_hidden, + excludes=list(excludes), + max_archive_depth=1 if extract_archives else 0, + ) + scanner.scan() + except KeyboardInterrupt: + console.print("\n[yellow]Survey interrupted[/yellow]") + raise click.Abort() from None + except Exception as e: + console.print(f"[red]Error: {e}[/red]") + raise click.Abort() from None + + stats = scanner.stats + table = Table(title=f"Survey: {path}") + table.add_column("Category", style="cyan") + table.add_column("Files", justify="right") + table.add_column("Top extensions", style="dim") + ext_by_cat: dict[str, list[str]] = {} + for category, count in stats.categories.most_common(top): + exts = [ + e + for e, _ in stats.extensions.most_common() + if scanner.classifier.classify_by_ext(e) == category + ][:4] + ext_by_cat[category] = exts + table.add_row(category, f"{count:,}", " ".join(exts)) + console.print(table) + console.print( + f"[bold]Total:[/bold] {scanner.file_count:,} files, {stats.total_bytes / (1024**3):.2f} GiB" + ) + + verdict, warnings = scanner.survey_verdict() + for w in warnings: + console.print(f"[yellow]⚠ {w}[/yellow]") + color = "red" if stats.gnss_files or stats.metadata_inconsistent else "green" + console.print(f"[bold {color}]Verdict: {verdict}[/bold {color}]") + + if __name__ == "__main__": main() diff --git a/tools/drive-archaeologist/src/drive_archaeologist/scanner.py b/tools/drive-archaeologist/src/drive_archaeologist/scanner.py index 6687e8e..1873976 100644 --- a/tools/drive-archaeologist/src/drive_archaeologist/scanner.py +++ b/tools/drive-archaeologist/src/drive_archaeologist/scanner.py @@ -2,11 +2,20 @@ Core scanner implementation with resume capability and progress tracking. Streams results to JSONL format for memory efficiency. Includes support for scanning inside archives. + +Hardened against corrupt filesystems (DA-002): capacity-sanity gate for +bogus directory-entry sizes, mojibake filename detection, symlink +non-traversal, itemized skip reporting, exclude globs, archive recursion +depth cap, and an output clobber guard. Also provides a stats-only mode +backing the `survey` CLI (DA-003). """ +import fnmatch import json +import os import shutil import time +from collections import Counter from collections.abc import Callable from datetime import datetime from pathlib import Path @@ -17,10 +26,40 @@ from .archive_handler import ArchiveHandler from .classifier import Classifier from .utils.checkpoint import CheckpointManager -from .utils.paths import sanitize_for_json, should_skip_path +from .utils.paths import is_suspect_name, sanitize_for_json, should_skip_path console = Console() +# Categories that make a drive ineligible for wiping (DA-003 verdict) +GNSS_CATEGORIES = {"GNSS Data", "GNSS Raw (Trimble)", "GNSS Raw (Leica)"} + +CORRUPT_CATEGORY = "Corrupt Direntry" +SYMLINK_CATEGORY = "Symlink" + +# Cap on how many skipped/excluded roots are itemized (memory bound) +_MAX_ITEMIZED = 200 + + +class ScanStats: + """Aggregate counters accumulated during a scan (drives the survey verdict).""" + + def __init__(self): + self.total_bytes = 0 # sane files only — corrupt claims excluded + self.claimed_bytes = 0 # every direntry's claimed size, lies included + self.categories: Counter[str] = Counter() + self.extensions: Counter[str] = Counter() + self.gnss_files = 0 + self.corrupt_entries = 0 + self.symlinks = 0 + self.hardlink_dups = 0 + self.archives_seen = 0 + self.archives_extracted = 0 + self.archives_depth_capped = 0 + self.excluded_count = 0 + self.skipped_roots: list[str] = [] + self.excluded_roots: list[str] = [] + self.metadata_inconsistent = False + class DeepScanner: """ @@ -33,21 +72,52 @@ def __init__( output_file: Path | None = None, resume: bool = False, on_classified: Callable[[dict], None] | None = None, + *, + include_hidden: bool = False, + excludes: list[str] | None = None, + max_archive_depth: int = 3, + force: bool = False, + stats_only: bool = False, + fs_capacity_bytes: int | None = None, ): self.root = Path(root_path).resolve() self.timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") self.scan_id = self.root.name.replace("/", "_").replace("\\", "_") self.on_classified = on_classified + self.include_hidden = include_hidden + self.excludes = list(excludes) if excludes else [] + self.max_archive_depth = max_archive_depth + self.force = force + self.stats_only = stats_only - if output_file: - self.output_file = Path(output_file) + if stats_only: + self.output_file = None + self.log_file = None + self.checkpoint = None else: - self.output_file = Path(f"scan_{self.scan_id}_{self.timestamp}.jsonl") + if output_file: + self.output_file = Path(output_file) + else: + self.output_file = Path(f"scan_{self.scan_id}_{self.timestamp}.jsonl") + self.log_file = self.output_file.with_suffix(".log") + self.checkpoint = ( + CheckpointManager(self.scan_id, checkpoint_dir=self.output_file.parent) + if resume + else None + ) - self.log_file = self.output_file.with_suffix(".log") - self.checkpoint = CheckpointManager(self.scan_id, checkpoint_dir=self.output_file.parent) if resume else None self.classifier = Classifier() self.archive_handler = ArchiveHandler() + self.stats = ScanStats() + self._seen_inodes: set[tuple[int, int]] = set() + + if fs_capacity_bytes is not None: + self._fs_capacity: int | None = fs_capacity_bytes + else: + try: + self._fs_capacity = shutil.disk_usage(str(self.root)).total + except OSError: + self._fs_capacity = None self.file_count = 0 self.error_count = 0 @@ -58,9 +128,15 @@ def __init__( def log(self, message: str, level: str = "INFO"): timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") - log_line = f"[{timestamp}] [{level}] {message}" + log_line = f"[{timestamp}] [{level}] {sanitize_for_json(message)}" + if self.log_file is None: + if level in ("WARNING", "ERROR"): + console.print(f"[yellow]{log_line}[/yellow]") + return try: - with open(self.log_file, "a", encoding="utf-8") as f: + # errors="replace": corrupt filesystems produce undecodable path + # bytes; a log write must never kill the scan + with open(self.log_file, "a", encoding="utf-8", errors="replace") as f: f.write(log_line + "\n") except Exception: # Errors during logging shouldn't crash the scan @@ -68,13 +144,30 @@ def log(self, message: str, level: str = "INFO"): def scan(self): """Main scanning loop with progress tracking""" + if ( + self.output_file is not None + and self.output_file.exists() + and not self.checkpoint + and not self.force + ): + raise FileExistsError( + f"Output file exists: {self.output_file} — re-running would overwrite the " + "previous scan. Use --resume to continue it or --force to overwrite." + ) + self.log(f"Starting deep scan of: {self.root}") - self.log(f"Output: {self.output_file}") console.print(f"[bold blue]Scanning:[/bold blue] {self.root}") - console.print(f"[bold green]Output:[/bold green] {self.output_file}") + if self.output_file is not None: + self.log(f"Output: {self.output_file}") + console.print(f"[bold green]Output:[/bold green] {self.output_file}") mode = "a" if self.checkpoint else "w" - with open(self.output_file, mode, encoding="utf-8") as outfile: + outfile = ( + open(self.output_file, mode, encoding="utf-8", errors="replace") + if self.output_file is not None + else None + ) + try: with Progress( SpinnerColumn(), TextColumn("[progress.description]{task.description}"), @@ -86,6 +179,9 @@ def scan(self): self.progress = progress self.task = self.progress.add_task("[cyan]Scanning...", total=None) self._scan_directory(self.root, outfile) + finally: + if outfile is not None: + outfile.close() if self.checkpoint: self.checkpoint.save_checkpoint() @@ -93,53 +189,160 @@ def scan(self): self._print_summary() - def _scan_directory(self, dir_path: Path, outfile, archive_path: Path | None = None): + def _is_excluded(self, path: Path) -> bool: + if not self.excludes: + return False + try: + rel = str(path.relative_to(self.root)) + except ValueError: + # Inside an archive temp dir — exclude globs are root-relative only + rel = path.name + return any( + fnmatch.fnmatch(rel, pat) or fnmatch.fnmatch(path.name, pat) for pat in self.excludes + ) + + def _record_itemized(self, bucket: list[str], path: Path): + if len(bucket) < _MAX_ITEMIZED: + bucket.append(sanitize_for_json(str(path))) + + def _scan_directory( + self, dir_path: Path, outfile, archive_path: Path | None = None, archive_depth: int = 0 + ): """Recursively scan a directory.""" - for filepath in dir_path.iterdir(): + try: + entries = list(dir_path.iterdir()) + except OSError as e: + self.error_count += 1 + self.log(f"Error listing {dir_path}: {e}") + return + + for filepath in entries: if self.checkpoint and self.checkpoint.is_scanned(filepath): continue - if should_skip_path(filepath): + if self._is_excluded(filepath): + self.stats.excluded_count += 1 + self._record_itemized(self.stats.excluded_roots, filepath) + continue + + if should_skip_path(filepath, include_hidden=self.include_hidden): self.skipped_count += 1 + self._record_itemized(self.stats.skipped_roots, filepath) continue try: + # Symlink gate (DA-002 #8): record, never traverse — a link to + # /home would otherwise walk the HOST filesystem into the catalog + if filepath.is_symlink(): + self._record_symlink(filepath, outfile, archive_path) + continue if filepath.is_file(): - self._process_file(filepath, outfile, archive_path) + self._process_file(filepath, outfile, archive_path, archive_depth) elif filepath.is_dir(): - self._scan_directory(filepath, outfile, archive_path) + self._scan_directory(filepath, outfile, archive_path, archive_depth) except (PermissionError, OSError) as e: self.error_count += 1 self.log(f"Error accessing {filepath}: {e}") - def _process_file(self, filepath: Path, outfile, archive_path: Path | None = None): + def _record_symlink(self, filepath: Path, outfile, archive_path: Path | None): + self.stats.symlinks += 1 + try: + target = os.readlink(filepath) + except OSError: + target = None + metadata = { + "path": sanitize_for_json(str(filepath.absolute())), + "name": sanitize_for_json(filepath.name), + "extension": filepath.suffix.lower(), + "category": SYMLINK_CATEGORY, + "size_bytes": 0, + "size_mb": 0.0, + "symlink_target": sanitize_for_json(target) if target else None, + "parent_dir": sanitize_for_json(str(filepath.parent)), + "scan_timestamp": datetime.now().isoformat(), + "in_archive": bool(archive_path), + "archive_path": sanitize_for_json(str(archive_path)) if archive_path else None, + } + self.stats.categories[SYMLINK_CATEGORY] += 1 + if outfile is not None: + outfile.write(json.dumps(metadata, ensure_ascii=False) + "\n") + if self.checkpoint: + self.checkpoint.mark_scanned(filepath) + + def _process_file( + self, filepath: Path, outfile, archive_path: Path | None = None, archive_depth: int = 0 + ): """Process a single file, handling archives recursively.""" try: metadata = self._extract_metadata(filepath, archive_path) - outfile.write(json.dumps(metadata, ensure_ascii=False) + "\n") - outfile.flush() + corrupt = metadata["category"] == CORRUPT_CATEGORY + hardlink_dup = bool(metadata.get("hardlink_dup")) + + if outfile is not None: + outfile.write(json.dumps(metadata, ensure_ascii=False) + "\n") + outfile.flush() self.file_count += 1 + category = metadata["category"] or "Unclassified" + self.stats.categories[category] += 1 + self.stats.extensions[metadata["extension"] or "(none)"] += 1 + size = metadata["size_bytes"] or 0 + self.stats.claimed_bytes += size + if corrupt: + self.stats.corrupt_entries += 1 + else: + self.stats.total_bytes += size + if category in GNSS_CATEGORIES: + self.stats.gnss_files += 1 + if ( + self._fs_capacity + and self.stats.claimed_bytes > self._fs_capacity + and not self.stats.metadata_inconsistent + ): + self.stats.metadata_inconsistent = True + self.log( + "Sum of claimed file sizes exceeds filesystem capacity — " + "directory metadata is inconsistent (probable corruption)", + level="WARNING", + ) + if self.progress and self.task is not None: - self.progress.update(self.task, advance=1, description=f"[cyan]Scanning... ({self.file_count} files)") + self.progress.update( + self.task, advance=1, description=f"[cyan]Scanning... ({self.file_count} files)" + ) - if self.on_classified is not None: + if self.on_classified is not None and not corrupt: try: self.on_classified(metadata) except Exception as e: self.error_count += 1 self.log(f"on_classified callback error for {filepath}: {e}", level="ERROR") - if self.archive_handler.is_archive(filepath): - self.log(f"Found archive: {filepath}. Extracting...") - temp_dir = self.archive_handler.extract(filepath) - if temp_dir: - self.log(f"Successfully extracted to: {temp_dir}") - self._scan_directory(temp_dir, outfile, archive_path=filepath) - shutil.rmtree(temp_dir, ignore_errors=True) - self.log(f"Cleaned up temporary directory: {temp_dir}") + # Never open a corrupt direntry; never re-extract a cross-linked one + if self.archive_handler.is_archive(filepath) and not corrupt and not hardlink_dup: + self.stats.archives_seen += 1 + if archive_depth >= self.max_archive_depth: + self.stats.archives_depth_capped += 1 + self.log( + f"Archive depth cap ({self.max_archive_depth}) reached, not extracting: {filepath}", + level="WARNING", + ) else: - self.log(f"Failed to extract archive: {filepath}", level="WARNING") + self.log(f"Found archive: {filepath}. Extracting...") + temp_dir = self.archive_handler.extract(filepath) + if temp_dir: + self.stats.archives_extracted += 1 + self.log(f"Successfully extracted to: {temp_dir}") + self._scan_directory( + temp_dir, + outfile, + archive_path=filepath, + archive_depth=archive_depth + 1, + ) + shutil.rmtree(temp_dir, ignore_errors=True) + self.log(f"Cleaned up temporary directory: {temp_dir}") + else: + self.log(f"Failed to extract archive: {filepath}", level="WARNING") if self.checkpoint: self.checkpoint.mark_scanned(filepath) @@ -153,26 +356,98 @@ def _process_file(self, filepath: Path, outfile, archive_path: Path | None = Non def _extract_metadata(self, filepath: Path, archive_path: Path | None = None) -> dict: """Extract file metadata.""" - stat = filepath.stat() - category = self.classifier.classify(filepath) + suspect = is_suspect_name(filepath.name) + try: + stat = filepath.stat() + size: int | None = stat.st_size + modified = datetime.fromtimestamp(stat.st_mtime).isoformat() + created = datetime.fromtimestamp(stat.st_ctime).isoformat() + inode_key = (stat.st_dev, stat.st_ino) if stat.st_ino else None + except OSError: + # Corrupt entries may not even stat + suspect = True + size = None + modified = None + created = None + inode_key = None + + corrupt_reason = None + if suspect: + corrupt_reason = "undecodable_name" + elif self._fs_capacity is not None and size is not None and size > self._fs_capacity: + # A single direntry claiming more bytes than the filesystem can + # hold is a corrupt FAT chain (DA-002 #1) — never read it + corrupt_reason = "oversize_direntry" + + if corrupt_reason: + category: str | None = CORRUPT_CATEGORY + else: + category = self.classifier.classify(filepath) + + hardlink_dup = False + if inode_key is not None and corrupt_reason is None: + if inode_key in self._seen_inodes: + hardlink_dup = True + self.stats.hardlink_dups += 1 + else: + self._seen_inodes.add(inode_key) data = { "path": sanitize_for_json(str(filepath.absolute())), - "name": filepath.name, + "name": sanitize_for_json(filepath.name), "extension": filepath.suffix.lower(), "category": category, - "size_bytes": stat.st_size, - "size_mb": round(stat.st_size / (1024 * 1024), 2), - "modified": datetime.fromtimestamp(stat.st_mtime).isoformat(), - "created": datetime.fromtimestamp(stat.st_ctime).isoformat(), + "size_bytes": size, + "size_mb": round(size / (1024 * 1024), 2) if size is not None else None, + "modified": modified, + "created": created, "parent_dir": sanitize_for_json(str(filepath.parent)), "depth": len(filepath.relative_to(self.root).parts) if not archive_path else None, "scan_timestamp": datetime.now().isoformat(), "in_archive": bool(archive_path), "archive_path": sanitize_for_json(str(archive_path)) if archive_path else None, } + if corrupt_reason: + data["corrupt_reason"] = corrupt_reason + if hardlink_dup: + data["hardlink_dup"] = True return data + def survey_verdict(self) -> tuple[str, list[str]]: + """One-line wipe/keep verdict + disclosure warnings (DA-003).""" + warnings = [] + if self.stats.metadata_inconsistent or self.stats.corrupt_entries: + warnings.append( + f"filesystem metadata inconsistent — {self.stats.corrupt_entries} corrupt " + "direntries; sizes are not trustworthy" + ) + if self.skipped_count and not self.include_hidden: + roots = ", ".join(self.stats.skipped_roots[:5]) + warnings.append( + f"{self.skipped_count} hidden/system entries were NOT surveyed " + f"(first roots: {roots}) — re-run with --include-hidden for full coverage" + ) + if self.stats.symlinks: + warnings.append(f"{self.stats.symlinks} symlinks recorded but not followed") + if self.stats.archives_seen and not self.stats.archives_extracted: + warnings.append( + f"{self.stats.archives_seen} archives present but not opened — " + "GNSS files inside archives would not be counted" + ) + if self.error_count: + warnings.append(f"{self.error_count} entries could not be read") + + if self.stats.gnss_files: + verdict = ( + f"{self.stats.gnss_files} GNSS-classified files — DO NOT wipe; " + "run a full scan and excavate first" + ) + elif self.stats.metadata_inconsistent: + verdict = "corrupt filesystem — verdict unreliable, inspect manually before wiping" + else: + verdict = "no GNSS payload detected — safe-to-wipe candidate (human confirms)" + return verdict, warnings + def _print_summary(self): """Print final statistics""" elapsed = time.time() - self.start_time @@ -182,12 +457,44 @@ def _print_summary(self): console.print("\n" + "=" * 60) console.print("[bold green]Scan Complete![/bold green]") console.print(f"[bold]Files processed:[/bold] {self.file_count:,}") - console.print(f"[bold]Files skipped:[/bold] {self.skipped_count:,}") + console.print(f"[bold]Total size:[/bold] {self.stats.total_bytes / (1024**2):,.1f} MB") + console.print(f"[bold]Files skipped (hidden/system):[/bold] {self.skipped_count:,}") + if self.stats.skipped_roots: + console.print( + f"[dim] skipped roots (first {min(len(self.stats.skipped_roots), 10)}): " + + ", ".join(self.stats.skipped_roots[:10]) + + "[/dim]" + ) + if self.stats.excluded_count: + console.print(f"[bold]Excluded by pattern:[/bold] {self.stats.excluded_count:,}") + if self.stats.symlinks: + console.print( + f"[bold]Symlinks (recorded, not followed):[/bold] {self.stats.symlinks:,}" + ) + if self.stats.corrupt_entries: + console.print( + f"[bold red]Corrupt direntries:[/bold red] {self.stats.corrupt_entries:,}" + ) + if self.stats.hardlink_dups: + console.print( + f"[bold]Hardlink/cross-link duplicates:[/bold] {self.stats.hardlink_dups:,}" + ) + if self.stats.archives_depth_capped: + console.print( + f"[bold yellow]Archives past depth cap (not opened):[/bold yellow] " + f"{self.stats.archives_depth_capped:,}" + ) + if self.stats.metadata_inconsistent: + console.print( + "[bold red]⚠ filesystem metadata inconsistent with capacity — " + "probable corruption[/bold red]" + ) console.print(f"[bold yellow]Errors:[/bold yellow] {self.error_count}") console.print(f"[bold]Time elapsed:[/bold] {elapsed_str}") console.print(f"[bold]Rate:[/bold] {rate:.1f} files/sec") - console.print(f"[bold green]Results:[/bold green] {self.output_file}") - console.print(f"[bold blue]Log:[/bold blue] {self.log_file}") + if self.output_file is not None: + console.print(f"[bold green]Results:[/bold green] {self.output_file}") + console.print(f"[bold blue]Log:[/bold blue] {self.log_file}") console.print("=" * 60 + "\n") self.log("=" * 60) self.log("Scan Complete!") @@ -196,4 +503,4 @@ def _print_summary(self): self.log(f"Errors: {self.error_count}") self.log(f"Time elapsed: {elapsed_str}") self.log(f"Rate: {rate:.1f} files/sec") - self.log("=" * 60) \ No newline at end of file + self.log("=" * 60) diff --git a/tools/drive-archaeologist/src/drive_archaeologist/utils/paths.py b/tools/drive-archaeologist/src/drive_archaeologist/utils/paths.py index b053af0..9944e18 100644 --- a/tools/drive-archaeologist/src/drive_archaeologist/utils/paths.py +++ b/tools/drive-archaeologist/src/drive_archaeologist/utils/paths.py @@ -43,16 +43,23 @@ } -def should_skip_path(path: Path) -> bool: +def should_skip_path(path: Path, include_hidden: bool = False) -> bool: """ Determine if a path should be skipped during scanning. Args: path: Path to check + include_hidden: When True, do not skip hidden (dot-prefixed) or + system entries. Skipping them silently hides real content — + a drive whose only files live in .Trash-1000 would otherwise + survey as empty (DA-002 finding #7). Returns: True if path should be skipped, False otherwise """ + if include_hidden: + return False + # Skip system directories for part in path.parts: if part in SYSTEM_DIRECTORIES: @@ -66,19 +73,31 @@ def should_skip_path(path: Path) -> bool: return False -def sanitize_for_json(text: str) -> str: +def is_suspect_name(name: str) -> bool: """ - Sanitize string for JSON output. + Detect filenames that indicate filesystem corruption (DA-002 finding #2). - Args: - text: String to sanitize + Corrupt FAT directory entries surface as mojibake: bytes that do not + decode as UTF-8 (Python exposes them as lone surrogates) or embedded + C0 control characters. Such entries must never be opened or extracted. + """ + try: + name.encode("utf-8") + except UnicodeEncodeError: + return True + return any(ord(c) < 0x20 or ord(c) == 0x7F for c in name) - Returns: - Sanitized string safe for JSON + +def sanitize_for_json(text: str) -> str: + """ + Sanitize a string so it can be written to a UTF-8 JSONL stream. + + Paths from corrupt filesystems can contain lone surrogates (undecodable + bytes); writing those to a UTF-8 file raises UnicodeEncodeError and the + record is lost. backslashreplace keeps the raw byte values visible + (e.g. ``\\udcff``) so corrupt names stay forensically identifiable. """ - # Replace backslashes with forward slashes for cross-platform consistency - # (but keep original paths in output for user clarity) - return text + return text.encode("utf-8", errors="backslashreplace").decode("utf-8") def safe_filename(name: str) -> str: diff --git a/tools/drive-archaeologist/tests/test_hardening.py b/tools/drive-archaeologist/tests/test_hardening.py new file mode 100644 index 0000000..46aa65e --- /dev/null +++ b/tools/drive-archaeologist/tests/test_hardening.py @@ -0,0 +1,285 @@ +""" +DA-002 scanner-hardening tests: corrupt-FAT gates, symlink non-traversal, +exclude globs, archive depth cap, clobber guard, skip itemization. +""" + +import json +import os +import zipfile +from pathlib import Path + +import pytest +from drive_archaeologist.classifier import Classifier +from drive_archaeologist.scanner import CORRUPT_CATEGORY, SYMLINK_CATEGORY, DeepScanner +from drive_archaeologist.utils.paths import is_suspect_name, sanitize_for_json, should_skip_path + + +def read_jsonl(path: Path) -> list[dict]: + return [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines()] + + +def run_scan(root: Path, out: Path, **kwargs) -> DeepScanner: + scanner = DeepScanner(root, output_file=out, **kwargs) + scanner.scan() + return scanner + + +# --- capacity sanity gate (finding #1) --------------------------------------- + + +def test_oversize_direntry_classified_corrupt_and_not_extracted(tmp_path): + root = tmp_path / "drive" + root.mkdir() + # a zip whose claimed size exceeds the injected fs capacity — must be + # flagged corrupt and never opened + bomb = root / "huge.zip" + with zipfile.ZipFile(bomb, "w") as zf: + zf.writestr("inner.txt", "x" * 2048) + out = tmp_path / "out.jsonl" + scanner = run_scan(root, out, fs_capacity_bytes=100) + records = {r["name"]: r for r in read_jsonl(out)} + assert records["huge.zip"]["category"] == CORRUPT_CATEGORY + assert records["huge.zip"]["corrupt_reason"] == "oversize_direntry" + assert "inner.txt" not in records # never extracted + assert scanner.stats.corrupt_entries == 1 + assert scanner.stats.archives_extracted == 0 + + +def test_claimed_sum_over_capacity_sets_inconsistent_flag(tmp_path): + root = tmp_path / "drive" + root.mkdir() + (root / "a.bin").write_bytes(b"x" * 600) + (root / "b.bin").write_bytes(b"y" * 600) + out = tmp_path / "out.jsonl" + # each file fits capacity, the sum does not + scanner = run_scan(root, out, fs_capacity_bytes=1000) + assert scanner.stats.metadata_inconsistent is True + + +def test_sane_tree_not_flagged(tmp_path): + root = tmp_path / "drive" + root.mkdir() + (root / "a.txt").write_text("hello") + out = tmp_path / "out.jsonl" + scanner = run_scan(root, out) + assert scanner.stats.metadata_inconsistent is False + assert scanner.stats.corrupt_entries == 0 + + +# --- mojibake / undecodable names (finding #2) -------------------------------- + + +def test_is_suspect_name(): + assert is_suspect_name("bad\x01name.txt") + assert is_suspect_name("bad\udcffname") # lone surrogate from undecodable bytes + assert not is_suspect_name("ALGO0010.22O") + assert not is_suspect_name("normál-ünïcode.txt") + + +def test_undecodable_filename_recorded_not_opened(tmp_path): + root = tmp_path / "drive" + root.mkdir() + fd = os.open(os.path.join(str(root).encode(), b"bad\xff.zip"), os.O_CREAT | os.O_WRONLY) + os.write(fd, b"not really a zip") + os.close(fd) + out = tmp_path / "out.jsonl" + scanner = run_scan(root, out) + records = read_jsonl(out) # must not crash on the surrogate path + assert len(records) == 1 + assert records[0]["category"] == CORRUPT_CATEGORY + assert records[0]["corrupt_reason"] == "undecodable_name" + assert scanner.stats.archives_seen == 0 # suspect entries are never opened + + +def test_sanitize_for_json_handles_surrogates(): + out = sanitize_for_json("bad\udcffname") + assert "\\udcff" in out # byte value preserved, string now UTF-8-safe + out.encode("utf-8") # must not raise + assert sanitize_for_json("clean") == "clean" + + +# --- symlink gate (finding #8) ------------------------------------------------- + + +def test_symlink_recorded_never_traversed(tmp_path): + outside = tmp_path / "outside" + outside.mkdir() + (outside / "secret.txt").write_text("host filesystem file") + root = tmp_path / "drive" + root.mkdir() + (root / "escape").symlink_to(outside) + (root / "normal.txt").write_text("ok") + out = tmp_path / "out.jsonl" + scanner = run_scan(root, out) + records = {r["name"]: r for r in read_jsonl(out)} + assert "secret.txt" not in records # did NOT walk through the link + assert records["escape"]["category"] == SYMLINK_CATEGORY + assert records["escape"]["symlink_target"] == str(outside) + assert scanner.stats.symlinks == 1 + + +def test_symlink_loop_is_harmless(tmp_path): + root = tmp_path / "drive" + root.mkdir() + (root / "loop").symlink_to(root) + (root / "file.txt").write_text("data") + out = tmp_path / "out.jsonl" + scanner = run_scan(root, out) + names = [r["name"] for r in read_jsonl(out)] + assert names.count("file.txt") == 1 # no re-walk through the loop + assert scanner.stats.symlinks == 1 + + +# --- exclude globs (finding #5) ------------------------------------------------ + + +def test_exclude_glob_skips_subtree(tmp_path): + root = tmp_path / "drive" + (root / "keep").mkdir(parents=True) + (root / "junk").mkdir() + (root / "keep" / "a.txt").write_text("a") + (root / "junk" / "b.txt").write_text("b") + out = tmp_path / "out.jsonl" + scanner = run_scan(root, out, excludes=["junk"]) + names = [r["name"] for r in read_jsonl(out)] + assert "a.txt" in names + assert "b.txt" not in names + assert scanner.stats.excluded_count == 1 + assert scanner.stats.excluded_roots # itemized + + +# --- hidden/system skip visibility (finding #7) -------------------------------- + + +def test_hidden_skipped_by_default_but_itemized(tmp_path): + root = tmp_path / "drive" + (root / ".Trash-1000").mkdir(parents=True) + (root / ".Trash-1000" / "movie.mkv").write_text("x") + out = tmp_path / "out.jsonl" + scanner = run_scan(root, out) + assert read_jsonl(out) == [] + assert scanner.skipped_count == 1 + assert any(".Trash-1000" in r for r in scanner.stats.skipped_roots) + + +def test_include_hidden_scans_trash(tmp_path): + root = tmp_path / "drive" + (root / ".Trash-1000").mkdir(parents=True) + (root / ".Trash-1000" / "movie.mkv").write_text("x") + out = tmp_path / "out.jsonl" + scanner = run_scan(root, out, include_hidden=True) + names = [r["name"] for r in read_jsonl(out)] + assert "movie.mkv" in names + assert scanner.skipped_count == 0 + + +def test_should_skip_path_include_hidden_flag(): + assert should_skip_path(Path("/d/.hidden")) + assert not should_skip_path(Path("/d/.hidden"), include_hidden=True) + assert should_skip_path(Path("/d/$RECYCLE.BIN/f")) + assert not should_skip_path(Path("/d/$RECYCLE.BIN/f"), include_hidden=True) + + +# --- archive depth cap (finding #9) --------------------------------------------- + + +def _nested_zip(root: Path) -> Path: + inner = root / "inner.zip" + with zipfile.ZipFile(inner, "w") as zf: + zf.writestr("deep.txt", "bottom") + outer = root / "outer.zip" + with zipfile.ZipFile(outer, "w") as zf: + zf.write(inner, "inner.zip") + inner.unlink() + return outer + + +def test_archive_depth_cap(tmp_path): + root = tmp_path / "drive" + root.mkdir() + _nested_zip(root) + out = tmp_path / "out.jsonl" + scanner = run_scan(root, out, max_archive_depth=1) + names = [r["name"] for r in read_jsonl(out)] + assert "inner.zip" in names # cataloged + assert "deep.txt" not in names # but not extracted past the cap + assert scanner.stats.archives_depth_capped == 1 + + +def test_archive_depth_zero_never_extracts(tmp_path): + root = tmp_path / "drive" + root.mkdir() + zip_path = root / "top.zip" + with zipfile.ZipFile(zip_path, "w") as zf: + zf.writestr("member.txt", "x") + out = tmp_path / "out.jsonl" + scanner = run_scan(root, out, max_archive_depth=0) + names = [r["name"] for r in read_jsonl(out)] + assert "top.zip" in names + assert "member.txt" not in names + assert scanner.stats.archives_extracted == 0 + + +# --- clobber guard (finding #10) ------------------------------------------------- + + +def test_existing_output_refused_without_force(tmp_path): + root = tmp_path / "drive" + root.mkdir() + (root / "a.txt").write_text("a") + out = tmp_path / "out.jsonl" + out.write_text("precious previous survey\n") + with pytest.raises(FileExistsError): + DeepScanner(root, output_file=out).scan() + assert out.read_text() == "precious previous survey\n" # untouched + + +def test_existing_output_overwritten_with_force(tmp_path): + root = tmp_path / "drive" + root.mkdir() + (root / "a.txt").write_text("a") + out = tmp_path / "out.jsonl" + out.write_text("old\n") + run_scan(root, out, force=True) + assert [r["name"] for r in read_jsonl(out)] == ["a.txt"] + + +# --- cross-link / hardlink duplicates (finding #3) -------------------------------- + + +def test_hardlink_duplicate_marked_not_reextracted(tmp_path): + root = tmp_path / "drive" + root.mkdir() + first = root / "a_data.zip" + with zipfile.ZipFile(first, "w") as zf: + zf.writestr("member.txt", "x") + os.link(first, root / "b_link.zip") + out = tmp_path / "out.jsonl" + scanner = run_scan(root, out) + records = read_jsonl(out) + dups = [r for r in records if r.get("hardlink_dup")] + assert len(dups) == 1 + assert scanner.stats.hardlink_dups == 1 + # the shared inode was extracted exactly once + assert sum(1 for r in records if r["name"] == "member.txt") == 1 + + +# --- classifier RINEX fallback ----------------------------------------------------- + + +def test_classifier_rinex_regex_fallback(): + c = Classifier() + assert c.classify(Path("PPPP0010.23o")) == "GNSS Data" # year past profile list + assert c.classify(Path("algo1150.05d")) == "GNSS Data" # Hatanaka, legacy year + assert c.classify(Path("site001a.99n")) == "GNSS Data" + assert c.classify(Path("ALGO0010.22O")) == "GNSS Data" # still via ext map + assert c.classify(Path("notrinex.23x")) is None + assert c.classify(Path("toolongname0010.23o")) is None + + +def test_classifier_leica_raw_fallback(): + c = Classifier() + assert c.classify(Path("STATION_2019.m00")) == "GNSS Raw (Leica)" + assert c.classify(Path("$IZWLIHF.m00")) == "GNSS Raw (Leica)" # recycle-bin stub keeps ext + assert c.classify(Path("file.m1")) is None + assert c.classify(Path("movie.m4v")) == "Video" # ext map wins before fallback diff --git a/tools/drive-archaeologist/tests/test_survey.py b/tools/drive-archaeologist/tests/test_survey.py new file mode 100644 index 0000000..74ad5a2 --- /dev/null +++ b/tools/drive-archaeologist/tests/test_survey.py @@ -0,0 +1,105 @@ +""" +DA-003 survey-mode tests: stats-only walk, wipe/keep verdict, disclosures. +""" + +import zipfile + +from click.testing import CliRunner +from drive_archaeologist.cli import main +from drive_archaeologist.scanner import DeepScanner + + +def make_media_tree(root): + (root / "Movies").mkdir(parents=True) + (root / "Movies" / "film.mkv").write_bytes(b"m" * 100) + (root / "Movies" / "film.srt").write_text("subs") + (root / "song.mp3").write_bytes(b"a" * 50) + + +def test_stats_only_writes_nothing(tmp_path, monkeypatch): + root = tmp_path / "drive" + root.mkdir() + make_media_tree(root) + monkeypatch.chdir(tmp_path) # any accidental default output would land here + scanner = DeepScanner(root, stats_only=True) + scanner.scan() + assert scanner.output_file is None + leftovers = [p for p in tmp_path.iterdir() if p.name != "drive"] + assert leftovers == [] # no jsonl, no log, no checkpoint + assert scanner.file_count == 3 + assert scanner.stats.categories["Video"] == 1 + assert scanner.stats.categories["Audio"] == 1 + + +def test_verdict_safe_on_pure_media(tmp_path): + root = tmp_path / "drive" + root.mkdir() + make_media_tree(root) + scanner = DeepScanner(root, stats_only=True, include_hidden=True) + scanner.scan() + verdict, warnings = scanner.survey_verdict() + assert "safe-to-wipe candidate" in verdict + assert scanner.stats.gnss_files == 0 + + +def test_verdict_do_not_wipe_on_gnss(tmp_path): + root = tmp_path / "drive" + root.mkdir() + make_media_tree(root) + (root / "PPPP0010.23o").write_text("rinex obs") # only the regex fallback catches .23o + scanner = DeepScanner(root, stats_only=True, include_hidden=True) + scanner.scan() + verdict, _ = scanner.survey_verdict() + assert "DO NOT wipe" in verdict + assert scanner.stats.gnss_files == 1 + + +def test_verdict_unreliable_on_corrupt_fs(tmp_path): + root = tmp_path / "drive" + root.mkdir() + (root / "a.bin").write_bytes(b"x" * 600) + (root / "b.bin").write_bytes(b"y" * 600) + scanner = DeepScanner(root, stats_only=True, fs_capacity_bytes=1000) + scanner.scan() + verdict, warnings = scanner.survey_verdict() + assert "verdict unreliable" in verdict + assert any("inconsistent" in w for w in warnings) + + +def test_verdict_disclosures(tmp_path): + root = tmp_path / "drive" + (root / ".Trash-1000").mkdir(parents=True) + (root / ".Trash-1000" / "hidden.mkv").write_text("x") + (root / "visible.txt").write_text("v") + with zipfile.ZipFile(root / "bundle.zip", "w") as zf: + zf.writestr("member.txt", "m") + (root / "link").symlink_to(root / "visible.txt") + + scanner = DeepScanner(root, stats_only=True, include_hidden=False, max_archive_depth=0) + scanner.scan() + _, warnings = scanner.survey_verdict() + joined = " | ".join(warnings) + assert "NOT surveyed" in joined # hidden skip disclosed + assert "symlinks" in joined + assert "archives present but not opened" in joined + + +def test_survey_cli_end_to_end(tmp_path): + root = tmp_path / "drive" + root.mkdir() + make_media_tree(root) + runner = CliRunner() + result = runner.invoke(main, ["survey", str(root)]) + assert result.exit_code == 0, result.output + assert "Verdict:" in result.output + assert "safe-to-wipe candidate" in result.output + + +def test_survey_cli_gnss_verdict(tmp_path): + root = tmp_path / "drive" + root.mkdir() + (root / "ALGO0010.22O").write_text("obs") + runner = CliRunner() + result = runner.invoke(main, ["survey", str(root)]) + assert result.exit_code == 0, result.output + assert "DO NOT wipe" in result.output