secretscan.py

#!/usr/bin/env python3
"""
Multi-tool secret scanner for either:

  - a directory of bare (mirror) git repos, or
  - a filesystem directory such as Manspider's downloaded loot.

In git mode, runs four scanners over each repo's FULL git history
(all branches/refs) and collects results:

  - gitleaks     git history scan  -> JSON report file
  - betterleaks  git history scan  -> JSON report file
  - trufflehog   bare-repo scan    -> JSONL on stdout
  - titus        git history scan  -> datastore + `report` JSON

In filesystem mode, the same tools scan a normal directory tree:

  - gitleaks     dir scan
  - betterleaks  dir scan
  - trufflehog   filesystem scan
  - titus        file/directory scan

Each target's raw per-tool output is preserved, and everything is folded into a
normalized findings.csv plus a summary.csv / summary.json.

The repos directory should contain `git clone --mirror` clones: they hold every
branch, tag and ref but no working tree. That is the ideal input for secret
scanning (a normal clone would only cover the default branch), so we scan the
bare repos directly and pass each tool its all-history mode.

Usage:
    python3 secretscan.py                       # scan every repo, default dirs
    python3 secretscan.py --limit 5             # first 5 repos (smoke test)
    python3 secretscan.py --repo example_repo   # one repo by name
    python3 secretscan.py --workers 8           # 8 repos in parallel
    python3 secretscan.py --verify              # enable live secret verification
    python3 secretscan.py --tools gitleaks,titus
    python3 secretscan.py --mode filesystem --target-dir ~/.manspider/loot \
        --titus-extract all --archive-depth 2

Raw per-tool outputs contain UNREDACTED secrets for remediation. findings.csv
previews are redacted by default (use --no-redact for full values). Treat the
output directory as sensitive: do not commit it.
"""

from __future__ import annotations

import argparse
import base64
import concurrent.futures
import csv
import datetime as dt
import hashlib
import json
import logging
import os
import shlex
import shutil
import subprocess
import sys
import threading
from collections import Counter
from dataclasses import dataclass, field
from pathlib import Path

DEFAULT_FILESYSTEM_TARGET = "~/.manspider/loot"
ALL_TOOLS = ("gitleaks", "betterleaks", "trufflehog", "titus")
SEVERITY_ORDER = {"": 0, "info": 1, "low": 2, "medium": 3, "high": 4, "critical": 5}
CSV_FORMULA_PREFIXES = ("=", "+", "-", "@", "\t", "\r")

log = logging.getLogger("secret_scan")

try:
    import readline
except ImportError:  # pragma: no cover - readline is unavailable on some platforms
    readline = None


# --------------------------------------------------------------------------- #
# Data model
# --------------------------------------------------------------------------- #
@dataclass
class Finding:
    repo: str
    tool: str
    rule: str = ""
    severity: str = ""
    verified: str = ""        # true/false/status/"" depending on tool + --verify
    file: str = ""
    line: str = ""
    commit: str = ""
    secret: str = ""          # full secret (redacted later for the CSV preview)


@dataclass
class RepoResult:
    repo: str
    commits: int
    status: str = "ok"                       # ok | partial | skipped-empty | error
    tool_counts: dict = field(default_factory=dict)   # tool -> int or status sentinel
    findings: list = field(default_factory=list)
    max_severity: str = ""


# --------------------------------------------------------------------------- #
# Small helpers
# --------------------------------------------------------------------------- #
def redact(secret: str, keep: int = 4) -> str:
    if not secret:
        return ""
    s = secret.replace("\n", "\\n").replace("\r", "")
    if len(s) <= keep * 2:
        return "*" * len(s)
    return f"{s[:keep]}...{s[-keep:]} (len={len(secret)})"


def b64_decode(s: str) -> str:
    if not s:
        return ""
    try:
        return base64.b64decode(s).decode("utf-8", "replace")
    except Exception:
        return s


def normalize_verification(value) -> str:
    """Normalize tool-specific validation results without discarding uncertainty."""
    if isinstance(value, bool):
        return str(value).lower()
    status = str(value or "").strip().lower()
    if status in {"active", "confirmed", "true", "valid", "verified"}:
        return "true"
    if status in {"denied", "false", "inactive", "invalid", "revoked"}:
        return "false"
    return status


def csv_safe(value) -> str:
    """Prevent spreadsheet applications from evaluating report cells as formulas."""
    text = str(value)
    return "'" + text if text.startswith(CSV_FORMULA_PREFIXES) else text


def positive_int(value: str) -> int:
    parsed = int(value)
    if parsed < 1:
        raise argparse.ArgumentTypeError("must be at least 1")
    return parsed


def nonnegative_int(value: str) -> int:
    parsed = int(value)
    if parsed < 0:
        raise argparse.ArgumentTypeError("must be 0 or greater")
    return parsed


def path_is_within(path: str, parent: str) -> bool:
    """Return whether path resolves to parent or one of its descendants."""
    try:
        path = os.path.realpath(path)
        parent = os.path.realpath(parent)
        return os.path.commonpath((path, parent)) == parent
    except ValueError:  # Different drives on Windows.
        return False


def record_error(path: str, message: str) -> None:
    with open(path, "w", encoding="utf-8") as f:
        f.write(message.rstrip() + "\n")


def harden_output_permissions(root: str) -> None:
    """Ensure scanner artifacts remain accessible only to the current user."""
    for directory, dirnames, filenames in os.walk(root):
        os.chmod(directory, 0o700)
        for dirname in dirnames:
            path = os.path.join(directory, dirname)
            if not os.path.islink(path):
                os.chmod(path, 0o700)
        for filename in filenames:
            path = os.path.join(directory, filename)
            if not os.path.islink(path):
                os.chmod(path, 0o600)


def run(cmd, timeout, stdout_path=None):
    """Run a command. If stdout_path is given, stream stdout to that file.
    Returns (returncode, stderr_text). Raises subprocess.TimeoutExpired."""
    stdout_target = (
        open(stdout_path, "w", encoding="utf-8")
        if stdout_path
        else subprocess.PIPE
    )
    try:
        proc = subprocess.run(
            cmd,
            stdout=stdout_target,
            stderr=subprocess.PIPE,
            text=True,
            errors="replace",
            timeout=timeout,
        )
        return proc.returncode, (proc.stderr or "")
    finally:
        if stdout_path:
            stdout_target.close()


def git_commit_count(repo_path: str) -> int:
    out = subprocess.run(
        ["git", "-C", repo_path, "rev-list", "--all", "--count"],
        capture_output=True,
        text=True,
        errors="replace",
        timeout=60,
    )
    if out.returncode != 0:
        raise RuntimeError(out.stderr.strip() or "git rev-list failed")
    return int(out.stdout.strip() or "0")


def file_count(path: str) -> int:
    if os.path.isfile(path):
        return 1
    count = 0

    def raise_walk_error(error):
        raise error

    for _, _, files in os.walk(path, onerror=raise_walk_error):
        count += len(files)
    return count


def safe_name(name: str) -> str:
    cleaned = "".join(c if c.isalnum() or c in "._-" else "_" for c in name)
    cleaned = cleaned.strip("._-") or "target"
    if cleaned != name:
        suffix = hashlib.sha256(name.encode("utf-8")).hexdigest()[:8]
        return f"{cleaned}-{suffix}"
    return cleaned


def path_completion_candidates(text: str) -> list[str]:
    """Return filesystem matches while preserving a leading ~ or relative path."""
    # Let the next completion operate inside the home directory without replacing
    # a shorthand such as "~" or "~user" with a relative-looking username.
    if text.startswith("~") and not any(separator in text for separator in {os.sep, "/"}):
        expanded_home = os.path.expanduser(text)
        if expanded_home != text and os.path.isdir(expanded_home):
            return [text + os.sep]
        return []

    expanded = os.path.expanduser(text)
    search_dir = os.path.dirname(expanded) or "."
    prefix = os.path.basename(expanded)
    display_dir = os.path.dirname(text)
    try:
        names = sorted(name for name in os.listdir(search_dir) if name.startswith(prefix))
    except OSError:
        return []

    matches = []
    for name in names:
        candidate = os.path.join(display_dir, name) if display_dir else name
        expanded_candidate = os.path.join(search_dir, name)
        if os.path.isdir(expanded_candidate):
            candidate += os.sep
        matches.append(candidate)
    return matches


def prompt_for_path(prompt: str) -> str:
    previous_completer = previous_delims = None
    if readline is not None:
        previous_completer = readline.get_completer()
        previous_delims = readline.get_completer_delims()
        def complete_path(text, state):
            matches = path_completion_candidates(text)
            return matches[state] if state < len(matches) else None

        readline.set_completer(complete_path)
        # Paths may contain spaces, so complete against the whole input line.
        readline.set_completer_delims("\t\n")
        if "libedit" in (getattr(readline, "__doc__", "") or ""):
            readline.parse_and_bind("bind ^I rl_complete")
        else:
            readline.parse_and_bind("tab: complete")

    try:
        while True:
            value = input(prompt).strip()
            if value:
                return os.path.abspath(os.path.expanduser(value))
            print("Please enter a path.")
    finally:
        if readline is not None:
            readline.set_completer(previous_completer)
            readline.set_completer_delims(previous_delims)


# --------------------------------------------------------------------------- #
# Per-tool runners. Each returns (count|sentinel, list[Finding]).
# Sentinels: "ERR", "TIMEOUT". Real errors are logged; the run continues.
# --------------------------------------------------------------------------- #
def scan_gitleaks(tool_bin, repo, repo_path, out_dir, timeout, verify, archive_depth=0,
                  source_type="git"):
    """gitleaks / betterleaks share a CLI (betterleaks is a gitleaks fork)."""
    out_json = os.path.join(out_dir, f"{tool_bin}.json")
    err_file = os.path.join(out_dir, f"{tool_bin}.err")
    subcommand = "git" if source_type == "git" else "dir"
    cmd = [
        tool_bin, subcommand, repo_path,
        "-f", "json",
        "-r", out_json,
        "--no-banner",
        "--exit-code", "0",          # "leaks found" must not look like an error
    ]
    if source_type == "git":
        # All refs; gitleaks already does this for bare repos, explicit for safety.
        cmd.append("--log-opts=--all")
    if archive_depth:                # traverse into archives (zip etc.) up to this depth (default 0 = off)
        cmd.append(f"--max-archive-depth={archive_depth}")
    if tool_bin == "betterleaks" and verify:
        cmd.append("--validation")
    try:
        rc, stderr = run(cmd, timeout)
    except subprocess.TimeoutExpired as exc:
        record_error(err_file, f"scan timed out after {exc.timeout} seconds")
        log.error("%s/%s: timed out after %s seconds", repo, tool_bin, exc.timeout)
        return "TIMEOUT", []

    if rc != 0:                      # with --exit-code 0, any nonzero is a genuine scan error
        record_error(err_file, stderr or f"scan exited with status {rc}")
        log.error("%s/%s: scan error rc=%s (see %s)", repo, tool_bin, rc, err_file)
        return "ERR", []
    if not os.path.exists(out_json):
        record_error(err_file, stderr or "scan completed without producing a report")
        log.error("%s/%s: no report produced (rc=%s)", repo, tool_bin, rc)
        return "ERR", []
    try:
        with open(out_json, encoding="utf-8") as f:
            data = json.load(f) or []      # gitleaks writes `null`, not `[]`, when clean
        if not isinstance(data, list):
            raise ValueError("JSON report root must be a list")
    except (json.JSONDecodeError, ValueError) as exc:
        record_error(err_file, f"unparseable JSON report: {exc}\n{stderr}")
        log.error("%s/%s: unparseable JSON report", repo, tool_bin)
        return "ERR", []

    findings = []
    for r in data:
        findings.append(Finding(
            repo=repo, tool=tool_bin,
            rule=r.get("RuleID", ""),
            verified=normalize_verification(
                r.get("ValidationStatus", r.get("Validation", ""))
            ),
            file=r.get("File", ""),
            line=str(r.get("StartLine", "")),
            commit=r.get("Commit", ""),
            secret=r.get("Secret", "") or r.get("Match", ""),
        ))
    return len(findings), findings


def scan_trufflehog(repo, repo_path, out_dir, timeout, verify, archive_depth=0,
                    source_type="git"):
    out_jsonl = os.path.join(out_dir, "trufflehog.jsonl")
    err_file = os.path.join(out_dir, "trufflehog.err")
    if source_type == "git":
        cmd = ["trufflehog", "git", "--bare", "--json", "--no-update"]
    else:
        cmd = ["trufflehog", "filesystem", "--json", "--no-update"]
    if not verify:
        cmd.append("--no-verification")
    if archive_depth:
        cmd += ["--archive-max-depth", str(archive_depth)]
    cmd.append(Path(repo_path).resolve().as_uri() if source_type == "git" else repo_path)
    try:
        rc, stderr = run(cmd, timeout, stdout_path=out_jsonl)
    except subprocess.TimeoutExpired as exc:
        record_error(err_file, f"scan timed out after {exc.timeout} seconds")
        log.error("%s/trufflehog: timed out after %s seconds", repo, exc.timeout)
        return "TIMEOUT", []

    if stderr:
        record_error(err_file, stderr)
    if rc != 0:
        if not stderr:
            record_error(err_file, f"scan exited with status {rc}")
        log.error("%s/trufflehog: exit %s (see %s)", repo, rc, err_file)
        return "ERR", []

    findings = []
    invalid_lines = 0
    with open(out_jsonl, encoding="utf-8", errors="replace") as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            try:
                r = json.loads(line)
            except json.JSONDecodeError:
                invalid_lines += 1
                continue
            if "DetectorName" not in r:
                continue
            source_data = (r.get("SourceMetadata") or {}).get("Data", {}) or {}
            git = source_data.get("Git", {}) or {}
            fs = source_data.get("Filesystem", {}) or {}
            findings.append(Finding(
                repo=repo, tool="trufflehog",
                rule=r.get("DetectorName", ""),
                verified=normalize_verification(r.get("Verified", "")),
                file=git.get("file", "") or fs.get("file", ""),
                line=str(git.get("line", "") or fs.get("line", "")),
                commit=git.get("commit", ""),
                secret=r.get("Redacted") or r.get("Raw", ""),
            ))
    if invalid_lines:
        log.warning("%s/trufflehog: ignored %d non-JSON output lines", repo, invalid_lines)
    return len(findings), findings


def scan_titus(repo, repo_path, out_dir, timeout, verify, max_file_size=None, extract=None,
               source_type="git"):
    ds_path = os.path.join(out_dir, "titus.ds")
    report_json = os.path.join(out_dir, "titus.report.json")
    err_file = os.path.join(out_dir, "titus.err")
    if os.path.exists(ds_path):                       # ensure a clean, non-incremental scan
        shutil.rmtree(ds_path, ignore_errors=True)

    scan_cmd = ["titus", "scan", repo_path, "--output", ds_path, "-q"]
    if source_type == "git":
        scan_cmd.append("--git")
    if verify:
        scan_cmd.append("--validate")
    if max_file_size:                                 # raise/remove the 10MB default to scan large blobs
        scan_cmd += ["--max-file-size", str(max_file_size)]
    if extract:                                       # extract+scan inside archives/office docs/PDFs (e.g. "all")
        scan_cmd += ["--extract", extract]
    try:
        rc, stderr = run(scan_cmd, timeout)
    except subprocess.TimeoutExpired as exc:
        record_error(err_file, f"scan timed out after {exc.timeout} seconds")
        log.error("%s/titus: scan timed out after %s seconds", repo, exc.timeout)
        return "TIMEOUT", []
    if rc != 0:
        record_error(err_file, stderr or f"scan exited with status {rc}")
        log.error("%s/titus: scan exit %s (see %s)", repo, rc, err_file)
        return "ERR", []

    report_cmd = ["titus", "report", "--datastore", ds_path,
                  "--format", "json", "--color", "never"]
    try:
        rc, stderr = run(report_cmd, timeout, stdout_path=report_json)
    except subprocess.TimeoutExpired as exc:
        record_error(err_file, f"report timed out after {exc.timeout} seconds")
        log.error("%s/titus: report timed out after %s seconds", repo, exc.timeout)
        return "TIMEOUT", []
    if rc != 0:
        record_error(err_file, stderr or f"report exited with status {rc}")
        log.error("%s/titus: report exit %s", repo, rc)
        return "ERR", []

    try:
        with open(report_json, encoding="utf-8") as f:
            data = json.load(f) or []
        if not isinstance(data, list):
            raise ValueError("JSON report root must be a list")
    except (json.JSONDecodeError, FileNotFoundError, ValueError) as exc:
        record_error(err_file, f"unparseable JSON report: {exc}\n{stderr}")
        log.error("%s/titus: unparseable JSON report", repo)
        return "ERR", []

    findings = []
    for finding in data:
        sev = (finding.get("Score") or {}).get("SuggestedSeverity", "")
        rule = finding.get("RuleID", "")
        for m in finding.get("Matches", []):
            src = (m.get("Location") or {}).get("Source", {}).get("Start", {})
            snippet = b64_decode((m.get("Snippet") or {}).get("Matching", ""))
            validation = m.get("ValidationResult") or m.get("validation_result") or {}
            findings.append(Finding(
                repo=repo, tool="titus",
                rule=m.get("RuleName", "") or rule,
                severity=str(sev).lower(),
                verified=normalize_verification(
                    validation.get("status", validation.get("Status", ""))
                ),
                file=m.get("file_path", ""),
                line=str(src.get("Line", "")),
                commit="",                       # titus report exposes file_path, not commit
                secret=snippet,
            ))
    return len(findings), findings


# --------------------------------------------------------------------------- #
# Orchestration
# --------------------------------------------------------------------------- #
def scan_one_target(target_name, target_path, out_root, tools, timeout, verify, opts=None,
                    source_type="git", display_name=None, output_key=None) -> RepoResult:
    opts = opts or {}
    repo_path = target_path
    repo = display_name or (
        target_name[:-4]
        if source_type == "git" and target_name.endswith(".git")
        else target_name
    )

    try:
        units = git_commit_count(repo_path) if source_type == "git" else file_count(repo_path)
    except Exception as exc:
        log.error("%s: unable to inspect target: %s", repo, exc)
        return RepoResult(repo=repo, commits=0, status="error")
    result = RepoResult(repo=repo, commits=units)
    if source_type == "git" and units == 0:
        result.status = "skipped-empty"
        log.info("SKIP %s (0 commits)", repo)
        return result
    if source_type == "filesystem" and units == 0:
        result.status = "skipped-empty"
        log.info("SKIP %s (0 files)", repo)
        return result

    raw_dir_name = "per_repo" if source_type == "git" else "per_target"
    out_dir = os.path.join(out_root, raw_dir_name, output_key or safe_name(target_name))
    os.makedirs(out_dir, exist_ok=True)
    unit_name = "commits" if source_type == "git" else "files"
    log.info("SCAN %s (%d %s)", repo, units, unit_name)

    for tool in tools:
        if tool in opts.get("missing_tools", ()):
            result.tool_counts[tool] = "MISSING"
            continue
        try:
            if tool in ("gitleaks", "betterleaks"):
                count, finds = scan_gitleaks(tool, repo, repo_path, out_dir, timeout, verify,
                                             archive_depth=opts.get("archive_depth", 0),
                                             source_type=source_type)
            elif tool == "trufflehog":
                count, finds = scan_trufflehog(repo, repo_path, out_dir, timeout, verify,
                                               archive_depth=opts.get("archive_depth", 0),
                                               source_type=source_type)
            elif tool == "titus":
                count, finds = scan_titus(repo, repo_path, out_dir, timeout, verify,
                                          max_file_size=opts.get("titus_max_file_size"),
                                          extract=opts.get("titus_extract"),
                                          source_type=source_type)
            else:
                continue
        except Exception:                    # one tool/repo blowing up must not abort the sweep
            log.exception("%s/%s: unexpected error", repo, tool)
            count, finds = "ERR", []
        result.tool_counts[tool] = count
        result.findings.extend(finds)
        for fnd in finds:
            severity = fnd.severity.lower()
            if SEVERITY_ORDER.get(severity, 0) > SEVERITY_ORDER.get(result.max_severity, 0):
                result.max_severity = severity

    ran_ok = [t for t in tools if isinstance(result.tool_counts.get(t), int)]
    if not ran_ok:
        result.status = "error"        # every tool errored or timed out -> no coverage
    elif len(ran_ok) < len(tools):
        result.status = "partial"      # at least one tool failed/timed out -> incomplete
    return result


def discover_repos(repos_dir):
    entries = []
    for name in sorted(os.listdir(repos_dir)):
        path = os.path.join(repos_dir, name)
        if name == ".git" or not os.path.isdir(path):
            continue
        # A .git suffix alone is not evidence of a repository; crawler output
        # can contain arbitrary directories with that name.
        if os.path.exists(os.path.join(path, "HEAD")):
            entries.append(name)
    return entries


def detect_source_type(path):
    """Infer whether a directory is a container of Git repos or ordinary files."""
    if os.path.isdir(path) and discover_repos(path):
        return "git"
    return "filesystem"


def discover_filesystem_targets(target_dir, split_targets=False, target_name=None):
    target_dir = os.path.abspath(os.path.expanduser(target_dir))
    if not split_targets:
        label = target_name or os.path.basename(target_dir.rstrip(os.sep)) or target_dir
        return [(label, target_dir)]

    entries = []
    for name in sorted(os.listdir(target_dir)):
        path = os.path.join(target_dir, name)
        if os.path.isdir(path) or os.path.isfile(path):
            entries.append((name, path))
    return entries


def assign_target_labels(targets, source_type):
    """Add unique report labels and collision-free raw-output directory keys."""
    labels = [
        name[:-4] if source_type == "git" and name.endswith(".git") else name
        for name, _ in targets
    ]
    counts = Counter(labels)
    used_keys = set()
    assigned = []
    for (name, path), label in zip(targets, labels):
        output_key = safe_name(name)
        if output_key in used_keys:
            suffix = hashlib.sha256(name.encode("utf-8")).hexdigest()[:12]
            output_key = f"{output_key}-{suffix}"
            counter = 2
            while output_key in used_keys:
                output_key = f"{safe_name(name)}-{suffix}-{counter}"
                counter += 1
        used_keys.add(output_key)
        assigned.append((name, path, name if counts[label] > 1 else label, output_key))
    return assigned


def write_reports(out_root, results, tools, redact_secrets, target_label="repo",
                  unit_label="commits"):
    findings_csv = os.path.join(out_root, "findings.csv")
    summary_csv = os.path.join(out_root, "summary.csv")
    summary_json = os.path.join(out_root, "summary.json")

    with open(findings_csv, "w", newline="", encoding="utf-8") as f:
        w = csv.writer(f)
        w.writerow([target_label, "tool", "rule", "severity", "verified",
                    "file", "line", "commit", "secret_preview"])
        for res in sorted(results, key=lambda r: r.repo):
            for fnd in res.findings:
                secret = redact(fnd.secret) if redact_secrets else fnd.secret.replace("\n", "\\n")
                row = [fnd.repo, fnd.tool, fnd.rule, fnd.severity, fnd.verified,
                       fnd.file, fnd.line, fnd.commit, secret]
                w.writerow(csv_safe(value) for value in row)

    with open(summary_csv, "w", newline="", encoding="utf-8") as f:
        w = csv.writer(f)
        w.writerow([target_label, unit_label, *tools, "total", "max_severity", "status"])
        for res in sorted(results, key=lambda r: r.repo):
            counts = [res.tool_counts.get(t, "-") for t in tools]
            total = sum(c for c in counts if isinstance(c, int))
            row = [res.repo, res.commits, *counts, total, res.max_severity, res.status]
            w.writerow(csv_safe(value) for value in row)

    totals = {t: 0 for t in tools}
    for res in results:
        for t in tools:
            c = res.tool_counts.get(t)
            if isinstance(c, int):
                totals[t] += c
    scanned_key = "repos_scanned" if target_label == "repo" else "targets_scanned"
    partial_key = "repos_partial" if target_label == "repo" else "targets_partial"
    skipped_key = "repos_skipped_empty" if target_label == "repo" else "targets_skipped_empty"
    errored_key = "repos_errored" if target_label == "repo" else "targets_errored"
    summary = {
        "generated": dt.datetime.now().astimezone().isoformat(timespec="seconds"),
        scanned_key: sum(1 for r in results if r.status in {"ok", "partial"}),
        partial_key: sum(1 for r in results if r.status == "partial"),
        skipped_key: sum(1 for r in results if r.status == "skipped-empty"),
        errored_key: sum(1 for r in results if r.status == "error"),
        "total_findings_by_tool": totals,
        "total_findings": sum(totals.values()),
    }
    with open(summary_json, "w", encoding="utf-8") as f:
        json.dump(summary, f, indent=2)
        f.write("\n")
    return summary


def log_completion_summary(out_root, summary, source_type, no_redact=False):
    """Print a practical handoff showing where results are and what to inspect."""
    target_kind = "repos" if source_type == "git" else "targets"
    scanned = summary[f"{target_kind}_scanned"]
    partial = summary[f"{target_kind}_partial"]
    skipped = summary[f"{target_kind}_skipped_empty"]
    errored = summary[f"{target_kind}_errored"]
    total_findings = summary["total_findings"]
    raw_dir_name = "per_repo" if source_type == "git" else "per_target"

    findings_csv = os.path.join(out_root, "findings.csv")
    summary_csv = os.path.join(out_root, "summary.csv")
    summary_json = os.path.join(out_root, "summary.json")
    scan_log = os.path.join(out_root, "scan.log")
    raw_dir = os.path.join(out_root, raw_dir_name)

    finding_status = (
        f"{total_findings} finding(s) reported"
        if total_findings
        else "No findings reported (this does not guarantee the data is secret-free)"
    )
    preview_status = "full secret values" if no_redact else "redacted secret previews"

    log.info("=" * 72)
    log.info("SCAN COMPLETE")
    log.info("RESULTS SAVED TO:")
    log.info("  %s", out_root)
    log.info("Result: %s", finding_status)
    log.info("Coverage: %d scanned, %d partial, %d errored, %d empty/skipped",
             scanned, partial, errored, skipped)
    log.info("OPEN THIS FIRST:")
    log.info("  %s", summary_csv)
    log.info("REPORT FILES:")
    log.info("  Findings CSV (%s): %s", preview_status, findings_csv)
    log.info("  Machine-readable summary: %s", summary_json)
    log.info("  Full run log: %s", scan_log)
    log.info("RAW SCANNER OUTPUT (UNREDACTED; TREAT AS SENSITIVE):")
    log.info("  %s/", raw_dir)
    log.info("QUICK VIEW COMMANDS:")
    log.info("  column -s, -t < %s", shlex.quote(summary_csv))
    log.info("  less %s", shlex.quote(findings_csv))
    log.info("  find %s -maxdepth 2 -type f -print", shlex.quote(raw_dir))
    log.info("=" * 72)


def main(argv=None):
    p = argparse.ArgumentParser(description="Run gitleaks, betterleaks, trufflehog and titus over git mirrors or filesystem output directories.")
    p.add_argument("--mode", choices=("git", "filesystem"), default=None,
                   help="input type; inferred for an interactively entered path when omitted")
    p.add_argument("--repos-dir", default=None,
                   help="folder of bare/mirror repos; prompted if omitted")
    p.add_argument("--target-dir", default=None,
                   help=f"filesystem mode target (default: {DEFAULT_FILESYSTEM_TARGET})")
    p.add_argument("--target-name", default=None,
                   help="filesystem mode label for reports (default: target basename)")
    p.add_argument("--split-targets", action="store_true",
                   help="filesystem mode: scan each immediate child of --target-dir separately")
    p.add_argument("--out", default=None, help="output dir (default: ./secret_scan_results_<ts>)")
    p.add_argument("--tools", default=",".join(ALL_TOOLS), help="comma list of tools to run")
    p.add_argument("--workers", type=positive_int, default=4, help="targets scanned in parallel")
    p.add_argument("--timeout", type=positive_int, default=900,
                   help="per-tool, per-target timeout (seconds)")
    p.add_argument("--limit", type=nonnegative_int, default=0,
                   help="scan only the first N targets")
    p.add_argument("--repo", default=None, help="git mode: scan a single repo by name")
    p.add_argument("--verify", action="store_true", help="enable live secret verification (network calls)")
    p.add_argument("--no-redact", action="store_true", help="store full secrets in findings.csv preview column")
    p.add_argument("--titus-max-file-size", type=nonnegative_int, default=0,
                   help="titus max file size in BYTES (0 = titus default 10MB; raise to scan large blobs)")
    p.add_argument("--titus-extract", default=None,
                   help="titus archive/document extraction: 'all' or e.g. 'xlsx,docx,pdf,zip'")
    p.add_argument("--archive-depth", type=nonnegative_int, default=0,
                   help="gitleaks/betterleaks/trufflehog: traverse into archives up to this depth (0 = off)")
    args = p.parse_args(argv)

    if args.mode == "git" and (args.target_dir or args.target_name or args.split_targets):
        p.error("--target-dir, --target-name, and --split-targets require --mode filesystem")
    if args.mode == "filesystem" and (args.repos_dir or args.repo):
        p.error("--repos-dir and --repo require --mode git")

    prompted_path = None
    if args.mode:
        source_type = args.mode
    elif args.repos_dir or args.repo:
        if args.target_dir or args.target_name or args.split_targets:
            p.error("cannot combine Git and filesystem target options")
        source_type = "git"
    elif args.target_dir or args.target_name or args.split_targets:
        source_type = "filesystem"
    else:
        prompted_path = prompt_for_path("Path to repos/files directory: ")
        if not (os.path.isdir(prompted_path) or os.path.isfile(prompted_path)):
            sys.exit(f"scan path not found: {prompted_path}")
        source_type = detect_source_type(prompted_path)

    if source_type == "git":
        repos_dir = (
            os.path.abspath(os.path.expanduser(args.repos_dir))
            if args.repos_dir
            else prompted_path or prompt_for_path("Path to repos directory: ")
        )
        if not os.path.isdir(repos_dir):
            sys.exit(f"repos dir not found: {repos_dir}")
        target_label = "repo"
        unit_label = "commits"
    else:
        filesystem_path = prompted_path or args.target_dir or DEFAULT_FILESYSTEM_TARGET
        repos_dir = os.path.abspath(os.path.expanduser(filesystem_path))
        if not (os.path.isdir(repos_dir) or os.path.isfile(repos_dir)):
            sys.exit(f"filesystem target not found: {repos_dir}")
        if args.split_targets and not os.path.isdir(repos_dir):
            sys.exit("--split-targets requires --target-dir to be a directory")
        target_label = "target"
        unit_label = "files"

    tools = list(dict.fromkeys(t.strip() for t in args.tools.split(",") if t.strip()))
    unknown_tools = [t for t in tools if t not in ALL_TOOLS]
    if unknown_tools:
        p.error(f"unknown tool(s): {', '.join(unknown_tools)}; choose from {', '.join(ALL_TOOLS)}")
    missing = [t for t in tools if not shutil.which(t)]
    for t in missing:
        log.warning("tool not on PATH, marking coverage incomplete: %s", t)
    if len(missing) == len(tools):
        sys.exit("no usable tools found on PATH")

    ts = dt.datetime.now().strftime("%Y%m%d_%H%M%S")
    out_root = os.path.abspath(args.out or f"secret_scan_results_{ts}")
    if source_type == "filesystem" and os.path.isdir(repos_dir) and path_is_within(out_root, repos_dir):
        sys.exit("output dir must be outside the filesystem target to prevent self-scanning")
    if os.path.islink(out_root):
        sys.exit(f"refusing to use a symlink as the output dir: {out_root}")
    if os.path.exists(out_root) and not os.path.isdir(out_root):
        sys.exit(f"output path is not a directory: {out_root}")
    if os.path.isdir(out_root) and os.listdir(out_root):
        sys.exit(f"output dir is not empty: {out_root}")

    # Raw reports can contain live credentials. This umask is inherited by tools.
    os.umask(0o077)
    os.makedirs(out_root, mode=0o700, exist_ok=True)
    os.chmod(out_root, 0o700)

    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s %(levelname)s %(message)s",
        handlers=[logging.FileHandler(os.path.join(out_root, "scan.log")),
                  logging.StreamHandler()],
    )

    if source_type == "git":
        targets = [(r, os.path.join(repos_dir, r)) for r in discover_repos(repos_dir)]
        if args.repo:
            want = {args.repo, args.repo + ".git"}
            targets = [t for t in targets if t[0] in want]
    else:
        targets = discover_filesystem_targets(repos_dir, args.split_targets, args.target_name)
    if args.limit:
        targets = targets[: args.limit]
    if not targets:
        sys.exit("no targets matched")
    targets = assign_target_labels(targets, source_type)

    log.info("Scanning %d %s with tools=%s workers=%d verify=%s -> %s",
             len(targets), "repos" if source_type == "git" else "targets",
             tools, args.workers, args.verify, out_root)

    results = []
    interrupted = False
    lock = threading.Lock()
    try:
        with concurrent.futures.ThreadPoolExecutor(max_workers=args.workers) as ex:
            opts = {
                "titus_max_file_size": args.titus_max_file_size,
                "titus_extract": args.titus_extract,
                "archive_depth": args.archive_depth,
                "missing_tools": frozenset(missing),
            }
            futs = {
                ex.submit(scan_one_target, name, path, out_root, tools,
                          args.timeout, args.verify, opts, source_type, label, output_key): (name, label)
                for name, path, label, output_key in targets
            }
            done = 0
            for fut in concurrent.futures.as_completed(futs):
                try:
                    res = fut.result()
                except Exception:
                    name, label = futs[fut]
                    log.exception("%s: unexpected target failure", name)
                    res = RepoResult(repo=label, commits=0, status="error")
                with lock:
                    results.append(res)
                    done += 1
                if res.status == "ok":
                    log.info("[%d/%d] done %s -> %s", done, len(targets), res.repo,
                             {t: res.tool_counts.get(t) for t in tools})
    except KeyboardInterrupt:
        interrupted = True
        log.warning("interrupted - writing partial results for %d targets", len(results))

    summary = write_reports(out_root, results, tools, redact_secrets=not args.no_redact,
                            target_label=target_label, unit_label=unit_label)
    harden_output_permissions(out_root)

    log_completion_summary(out_root, summary, source_type, args.no_redact)

    if interrupted:
        return 130
    if any(result.status in {"partial", "error"} for result in results):
        return 2
    return 0


if __name__ == "__main__":
    sys.exit(main())