securehash.py

"""
SecureHash — File Integrity Monitoring Tool

Computes and verifies SHA-256 checksums across directory trees to detect
unauthorized modifications, additions, or deletions. Designed for security
auditing and change detection workflows.

Usage:
    python securehash.py baseline ./target_dir        # Create baseline snapshot
    python securehash.py verify ./target_dir           # Verify against baseline
    python securehash.py watch ./target_dir --interval 30  # Continuous monitoring

Author: Faruk Cehajic (mrceha)
License: MIT
"""

from __future__ import annotations

import argparse
import hashlib
import json
import os
import sys
import time
from dataclasses import dataclass, field, asdict
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional


# -- Constants ----------------------------------------------------------------

BASELINE_FILE = ".securehash.json"
HASH_ALGORITHM = "sha256"
READ_CHUNK_SIZE = 8192
VERSION = "1.2.0"


# -- Data Structures ----------------------------------------------------------

@dataclass
class FileRecord:
    """Immutable snapshot of a single file's state."""
    path: str
    sha256: str
    size: int
    modified: float


@dataclass
class BaselineManifest:
    """Container for a full directory baseline snapshot."""
    version: str
    algorithm: str
    created_at: str
    root_directory: str
    file_count: int
    records: dict[str, dict] = field(default_factory=dict)


@dataclass
class IntegrityReport:
    """Results of a verification pass against a stored baseline."""
    modified: list[str] = field(default_factory=list)
    added: list[str] = field(default_factory=list)
    removed: list[str] = field(default_factory=list)
    unchanged: int = 0

    @property
    def is_clean(self) -> bool:
        return not (self.modified or self.added or self.removed)

    @property
    def total_changes(self) -> int:
        return len(self.modified) + len(self.added) + len(self.removed)


# -- Core Functions -----------------------------------------------------------

def compute_file_hash(filepath: Path) -> Optional[str]:
    """
    Compute the SHA-256 digest of a file using chunked reads to
    handle arbitrarily large files without excessive memory usage.

    Returns None if the file cannot be read (permissions, broken symlinks).
    """
    hasher = hashlib.new(HASH_ALGORITHM)
    try:
        with open(filepath, "rb") as fh:
            while chunk := fh.read(READ_CHUNK_SIZE):
                hasher.update(chunk)
    except (PermissionError, OSError):
        return None
    return hasher.hexdigest()


def scan_directory(root: Path, exclude: Optional[list[str]] = None) -> list[FileRecord]:
    """
    Recursively walk a directory tree, computing hash records for every
    regular file. Skips symlinks and unreadable entries gracefully.

    Args:
        root: Top-level directory to scan.
        exclude: Optional list of directory or file names to skip.

    Returns:
        Sorted list of FileRecord instances.
    """
    exclude = set(exclude or [])
    exclude.add(BASELINE_FILE)
    records = []

    for dirpath, dirnames, filenames in os.walk(root, followlinks=False):
        # Prune excluded directories in-place to prevent os.walk from descending
        dirnames[:] = [d for d in dirnames if d not in exclude]

        for filename in filenames:
            if filename in exclude:
                continue

            filepath = Path(dirpath) / filename

            if filepath.is_symlink():
                continue

            digest = compute_file_hash(filepath)
            if digest is None:
                continue

            stat = filepath.stat()
            relative = str(filepath.relative_to(root))

            records.append(FileRecord(
                path=relative,
                sha256=digest,
                size=stat.st_size,
                modified=stat.st_mtime,
            ))

    records.sort(key=lambda r: r.path)
    return records


def create_baseline(root: Path, exclude: Optional[list[str]] = None) -> BaselineManifest:
    """
    Scan the target directory and produce a baseline manifest that can
    be serialized to disk for future verification.
    """
    records = scan_directory(root, exclude)
    manifest = BaselineManifest(
        version=VERSION,
        algorithm=HASH_ALGORITHM,
        created_at=datetime.now(timezone.utc).isoformat(),
        root_directory=str(root.resolve()),
        file_count=len(records),
        records={r.path: asdict(r) for r in records},
    )
    return manifest


def save_baseline(manifest: BaselineManifest, root: Path) -> Path:
    """Write the baseline manifest to the target directory as JSON."""
    output_path = root / BASELINE_FILE
    with open(output_path, "w", encoding="utf-8") as fh:
        json.dump(asdict(manifest), fh, indent=2, ensure_ascii=False)
    return output_path


def load_baseline(root: Path) -> BaselineManifest:
    """
    Load a previously saved baseline from the target directory.

    Raises:
        FileNotFoundError: If no baseline exists in the directory.
        ValueError: If the baseline file is malformed.
    """
    baseline_path = root / BASELINE_FILE
    if not baseline_path.exists():
        raise FileNotFoundError(
            f"No baseline found at {baseline_path}. "
            f"Run 'securehash baseline {root}' first."
        )

    with open(baseline_path, "r", encoding="utf-8") as fh:
        data = json.load(fh)

    if "records" not in data or "version" not in data:
        raise ValueError(f"Malformed baseline file: {baseline_path}")

    return BaselineManifest(**data)


def verify_integrity(root: Path, exclude: Optional[list[str]] = None) -> IntegrityReport:
    """
    Compare the current state of a directory against its stored baseline.

    Returns an IntegrityReport detailing all modifications, additions,
    and deletions detected since the baseline was created.
    """
    baseline = load_baseline(root)
    current_records = scan_directory(root, exclude)

    report = IntegrityReport()
    current_map = {r.path: r for r in current_records}
    baseline_paths = set(baseline.records.keys())
    current_paths = set(current_map.keys())

    # Detect removed files
    for path in sorted(baseline_paths - current_paths):
        report.removed.append(path)

    # Detect added files
    for path in sorted(current_paths - baseline_paths):
        report.added.append(path)

    # Detect modified files (hash comparison only — timestamps are unreliable)
    for path in sorted(baseline_paths & current_paths):
        stored_hash = baseline.records[path]["sha256"]
        current_hash = current_map[path].sha256
        if stored_hash != current_hash:
            report.modified.append(path)
        else:
            report.unchanged += 1

    return report


# -- Display Helpers ----------------------------------------------------------

def _color(text: str, code: int) -> str:
    """Apply ANSI color code if stdout is a terminal."""
    if not sys.stdout.isatty():
        return text
    return f"\033[{code}m{text}\033[0m"


def print_report(report: IntegrityReport) -> None:
    """Format and print a verification report to stdout."""
    if report.is_clean:
        print(_color("\n✓ Integrity verified — no changes detected.", 32))
        print(f"  {report.unchanged} file(s) checked.\n")
        return

    print(_color(f"\n⚠ Integrity check failed — {report.total_changes} change(s) detected.\n", 31))

    if report.modified:
        print(_color(f"  Modified ({len(report.modified)}):", 33))
        for path in report.modified:
            print(f"    ~ {path}")

    if report.added:
        print(_color(f"  Added ({len(report.added)}):", 36))
        for path in report.added:
            print(f"    + {path}")

    if report.removed:
        print(_color(f"  Removed ({len(report.removed)}):", 31))
        for path in report.removed:
            print(f"    - {path}")

    print(f"\n  {report.unchanged} file(s) unchanged.\n")


# -- CLI Interface ------------------------------------------------------------

def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="securehash",
        description="File integrity monitoring via SHA-256 baseline comparison.",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=(
            "Examples:\n"
            "  securehash baseline ./my_project\n"
            "  securehash verify ./my_project\n"
            "  securehash watch ./my_project --interval 60\n"
            "  securehash baseline ./src --exclude __pycache__ .git node_modules"
        ),
    )
    parser.add_argument(
        "--version", action="version", version=f"%(prog)s {VERSION}",
    )
    subparsers = parser.add_subparsers(dest="command", required=True)

    # baseline subcommand
    bp = subparsers.add_parser("baseline", help="Create a new integrity baseline.")
    bp.add_argument("directory", type=Path, help="Target directory to snapshot.")
    bp.add_argument(
        "--exclude", nargs="*", default=[], metavar="NAME",
        help="Directory or file names to exclude from scanning.",
    )

    # verify subcommand
    vp = subparsers.add_parser("verify", help="Verify directory against its baseline.")
    vp.add_argument("directory", type=Path, help="Target directory to verify.")
    vp.add_argument(
        "--exclude", nargs="*", default=[], metavar="NAME",
        help="Directory or file names to exclude from scanning.",
    )

    # watch subcommand
    wp = subparsers.add_parser("watch", help="Continuously monitor for changes.")
    wp.add_argument("directory", type=Path, help="Target directory to watch.")
    wp.add_argument(
        "--interval", type=int, default=30,
        help="Seconds between verification passes (default: 30).",
    )
    wp.add_argument(
        "--exclude", nargs="*", default=[], metavar="NAME",
        help="Directory or file names to exclude from scanning.",
    )

    return parser


def main() -> int:
    parser = build_parser()
    args = parser.parse_args()

    target = args.directory.resolve()
    if not target.is_dir():
        print(f"Error: '{target}' is not a valid directory.", file=sys.stderr)
        return 1

    if args.command == "baseline":
        manifest = create_baseline(target, args.exclude)
        output = save_baseline(manifest, target)
        print(f"Baseline created: {output}")
        print(f"  Files indexed: {manifest.file_count}")
        print(f"  Algorithm:     {manifest.algorithm}")
        return 0

    elif args.command == "verify":
        try:
            report = verify_integrity(target, args.exclude)
        except FileNotFoundError as exc:
            print(f"Error: {exc}", file=sys.stderr)
            return 1
        print_report(report)
        return 0 if report.is_clean else 2

    elif args.command == "watch":
        print(f"Watching '{target}' every {args.interval}s. Press Ctrl+C to stop.\n")
        try:
            while True:
                try:
                    report = verify_integrity(target, args.exclude)
                    timestamp = datetime.now().strftime("%H:%M:%S")
                    if report.is_clean:
                        print(f"[{timestamp}] ✓ Clean — {report.unchanged} files verified.")
                    else:
                        print(f"[{timestamp}] ⚠ {report.total_changes} change(s) detected!")
                        print_report(report)
                except FileNotFoundError:
                    print("No baseline found. Creating initial baseline...")
                    manifest = create_baseline(target, args.exclude)
                    save_baseline(manifest, target)
                    print(f"  Baseline created ({manifest.file_count} files).")
                time.sleep(args.interval)
        except KeyboardInterrupt:
            print("\nStopped.")
            return 0

    return 0


if __name__ == "__main__":
    raise SystemExit(main())