#!/usr/bin/env python3
"""Gitleaks secret scanning agent.

Wraps the Gitleaks CLI to scan git repositories, directories, or
specific commits for hardcoded secrets, API keys, tokens, and
credentials. Parses JSON output into structured findings.
"""
import argparse
import json
import os
import subprocess
import sys
from datetime import datetime, timezone


def find_gitleaks_binary():
    """Locate the gitleaks binary on the system."""
    custom_path = os.environ.get("GITLEAKS_PATH")
    if custom_path and os.path.isfile(custom_path):
        return custom_path
    for name in ["gitleaks", "gitleaks.exe"]:
        for directory in os.environ.get("PATH", "").split(os.pathsep):
            full_path = os.path.join(directory, name)
            if os.path.isfile(full_path):
                return full_path
    print("[!] gitleaks binary not found. Install: https://github.com/gitleaks/gitleaks",
          file=sys.stderr)
    sys.exit(1)


def run_detect(gitleaks_bin, target, config=None, baseline=None,
               log_opts=None, no_git=False, verbose=False):
    """Run gitleaks detect on a repository or directory."""
    cmd = [gitleaks_bin, "detect"]
    if no_git:
        cmd.extend(["--no-git", "--source", target])
    else:
        cmd.extend(["--source", target])
    cmd.extend(["--report-format", "json", "--report-path", "/dev/stdout"])
    if config:
        cmd.extend(["--config", config])
    if baseline:
        cmd.extend(["--baseline-path", baseline])
    if log_opts:
        cmd.extend(["--log-opts", log_opts])
    if verbose:
        cmd.append("--verbose")
    cmd.extend(["--exit-code", "0"])

    print(f"[*] Running: {' '.join(cmd)}")
    result = subprocess.run(
        cmd,
        capture_output=True,
        text=True,
        timeout=600,
    )
    if result.returncode not in (0, 1):
        print(f"[!] Gitleaks error (exit {result.returncode}): {result.stderr}",
              file=sys.stderr)
    return result.stdout, result.stderr, result.returncode


def run_protect(gitleaks_bin, target, config=None, staged=False, verbose=False):
    """Run gitleaks protect for pre-commit scanning."""
    cmd = [gitleaks_bin, "protect", "--source", target]
    cmd.extend(["--report-format", "json", "--report-path", "/dev/stdout"])
    if staged:
        cmd.append("--staged")
    if config:
        cmd.extend(["--config", config])
    if verbose:
        cmd.append("--verbose")
    cmd.extend(["--exit-code", "0"])

    print(f"[*] Running protect mode: {' '.join(cmd)}")
    result = subprocess.run(
        cmd,
        capture_output=True,
        text=True,
        timeout=300,
    )
    return result.stdout, result.stderr, result.returncode


def parse_findings(raw_json):
    """Parse gitleaks JSON output into structured findings."""
    findings = []
    if not raw_json or not raw_json.strip():
        return findings
    try:
        results = json.loads(raw_json)
    except json.JSONDecodeError:
        return findings
    if not isinstance(results, list):
        results = [results]
    for item in results:
        secret_val = item.get("Secret", "")
        findings.append({
            "rule_id": item.get("RuleID", "unknown"),
            "description": item.get("Description", ""),
            "secret": secret_val[:8] + "..." if secret_val else "",
            "file": item.get("File", ""),
            "line": item.get("StartLine", 0),
            "commit": (item.get("Commit", "") or "")[:12],
            "author": item.get("Author", ""),
            "email": item.get("Email", ""),
            "date": item.get("Date", ""),
            "message": (item.get("Message", "") or "")[:80],
            "entropy": item.get("Entropy", 0),
            "fingerprint": item.get("Fingerprint", ""),
            "tags": item.get("Tags", []),
        })
    return findings


def format_summary(findings, target):
    """Print human-readable summary of secrets found."""
    print(f"\n{'='*60}")
    print(f"  Gitleaks Secret Scan Report")
    print(f"{'='*60}")
    print(f"  Target    : {target}")
    print(f"  Secrets   : {len(findings)}")

    if not findings:
        print(f"  Status    : CLEAN - No secrets detected")
        print(f"{'='*60}")
        return

    by_rule = {}
    for f in findings:
        rule = f["rule_id"]
        by_rule.setdefault(rule, []).append(f)

    print(f"\n  Findings by Rule:")
    for rule, items in sorted(by_rule.items(), key=lambda x: -len(x[1])):
        print(f"    {rule:40s}: {len(items)} occurrence(s)")

    by_file = {}
    for f in findings:
        by_file.setdefault(f["file"], []).append(f)

    print(f"\n  Affected Files: {len(by_file)}")
    for filepath, items in sorted(by_file.items(), key=lambda x: -len(x[1]))[:10]:
        print(f"    {filepath} ({len(items)} secret(s))")

    print(f"\n  Top Findings:")
    for f in findings[:15]:
        print(f"    [{f['rule_id']:30s}] {f['file']}:{f['line']} "
              f"(commit: {f['commit']}, secret: {f['secret']})")


def main():
    parser = argparse.ArgumentParser(
        description="Gitleaks secret scanning agent"
    )
    parser.add_argument("--target", required=True,
                        help="Repository path or directory to scan")
    parser.add_argument("--mode", choices=["detect", "protect"], default="detect",
                        help="Scan mode: detect (full history) or protect (pre-commit)")
    parser.add_argument("--config", help="Path to custom .gitleaks.toml config")
    parser.add_argument("--baseline", help="Path to baseline file for known findings")
    parser.add_argument("--log-opts", help="Git log options (e.g., '--since=2026-01-01')")
    parser.add_argument("--no-git", action="store_true",
                        help="Scan files without git history")
    parser.add_argument("--staged", action="store_true",
                        help="Only scan staged changes (protect mode)")
    parser.add_argument("--output", "-o", help="Output JSON report path")
    parser.add_argument("--verbose", "-v", action="store_true")
    args = parser.parse_args()

    gitleaks_bin = find_gitleaks_binary()
    print(f"[*] Using gitleaks: {gitleaks_bin}")

    if args.mode == "protect":
        raw_json, stderr, exit_code = run_protect(
            gitleaks_bin, args.target, args.config, args.staged, args.verbose
        )
    else:
        raw_json, stderr, exit_code = run_detect(
            gitleaks_bin, args.target, args.config, args.baseline,
            args.log_opts, args.no_git, args.verbose
        )

    findings = parse_findings(raw_json)
    format_summary(findings, args.target)

    report = {
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "tool": "gitleaks",
        "target": args.target,
        "mode": args.mode,
        "secrets_found": len(findings),
        "findings": findings,
        "risk_level": (
            "CRITICAL" if len(findings) > 10
            else "HIGH" if len(findings) > 0
            else "LOW"
        ),
    }

    if args.output:
        with open(args.output, "w") as f:
            json.dump(report, f, indent=2)
        print(f"\n[+] Report saved to {args.output}")
    elif args.verbose:
        print(json.dumps(report, indent=2))

    if findings:
        print(f"\n[!] {len(findings)} secret(s) detected - remediation required")
    else:
        print(f"\n[+] No secrets detected")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Gitleaks Secret Scanning Pipeline Script

Runs Gitleaks scans, manages baselines, evaluates findings,
and generates remediation reports.

Usage:
    python process.py --repo-path /path/to/repo --scan-type detect
    python process.py --repo-path . --scan-type protect --staged
    python process.py --repo-path . --baseline .gitleaks-baseline.json --output report.json
"""

import argparse
import json
import os
import subprocess
import sys
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Optional


@dataclass
class SecretFinding:
    rule_id: str
    description: str
    file: str
    line: int
    commit: str
    author: str
    date: str
    secret: str
    entropy: float
    match: str
    tags: list = field(default_factory=list)
    is_new: bool = True


@dataclass
class ScanResult:
    findings: list = field(default_factory=list)
    new_findings: list = field(default_factory=list)
    baseline_findings: list = field(default_factory=list)
    commits_scanned: int = 0
    scan_duration: float = 0.0
    error: str = ""


def run_gitleaks(repo_path: str, scan_type: str = "detect",
                 baseline_path: Optional[str] = None,
                 commit_range: Optional[str] = None,
                 staged: bool = False,
                 config_path: Optional[str] = None) -> dict:
    """Execute Gitleaks and return JSON results."""
    cmd = ["gitleaks", scan_type, "--source", repo_path,
           "--report-format", "json", "--report-path", "/dev/stdout"]

    if baseline_path and os.path.exists(baseline_path):
        cmd.extend(["--baseline-path", baseline_path])

    if commit_range:
        cmd.extend(["--log-opts", commit_range])

    if staged:
        cmd.append("--staged")

    if config_path:
        cmd.extend(["--config", config_path])

    cmd.append("--verbose")

    try:
        proc = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            timeout=300
        )

        findings = []
        if proc.stdout.strip():
            try:
                findings = json.loads(proc.stdout)
            except json.JSONDecodeError:
                pass

        return {
            "findings": findings if isinstance(findings, list) else [],
            "exit_code": proc.returncode,
            "stderr": proc.stderr
        }

    except subprocess.TimeoutExpired:
        return {"findings": [], "exit_code": -1, "stderr": "Scan timed out after 300s"}
    except FileNotFoundError:
        return {"findings": [], "exit_code": -1,
                "stderr": "gitleaks not found. Install from https://github.com/gitleaks/gitleaks"}


def parse_findings(raw_findings: list) -> list:
    """Parse raw Gitleaks JSON findings into SecretFinding objects."""
    findings = []
    for f in raw_findings:
        redacted_secret = redact_secret(f.get("Secret", ""))
        findings.append(SecretFinding(
            rule_id=f.get("RuleID", "unknown"),
            description=f.get("Description", ""),
            file=f.get("File", ""),
            line=f.get("StartLine", 0),
            commit=f.get("Commit", "")[:8],
            author=f.get("Author", ""),
            date=f.get("Date", ""),
            secret=redacted_secret,
            entropy=f.get("Entropy", 0.0),
            match=f.get("Match", "")[:100],
            tags=f.get("Tags", [])
        ))
    return findings


def redact_secret(secret: str) -> str:
    """Redact a secret, showing only first 4 and last 4 characters."""
    if len(secret) <= 12:
        return "*" * len(secret)
    return secret[:4] + "..." + secret[-4:]


def load_baseline(baseline_path: str) -> set:
    """Load baseline fingerprints for comparison."""
    if not os.path.exists(baseline_path):
        return set()

    with open(baseline_path, "r") as f:
        try:
            baseline = json.load(f)
        except json.JSONDecodeError:
            return set()

    fingerprints = set()
    for entry in baseline:
        fp = f"{entry.get('RuleID', '')}:{entry.get('File', '')}:{entry.get('Commit', '')}"
        fingerprints.add(fp)

    return fingerprints


def classify_findings(findings: list, baseline_fingerprints: set) -> tuple:
    """Separate findings into new and baseline (pre-existing)."""
    new_findings = []
    baseline_findings = []

    for f in findings:
        fp = f"{f.rule_id}:{f.file}:{f.commit}"
        if fp in baseline_fingerprints:
            f.is_new = False
            baseline_findings.append(f)
        else:
            f.is_new = True
            new_findings.append(f)

    return new_findings, baseline_findings


def generate_report(scan_result: ScanResult, repo_path: str) -> dict:
    """Generate a structured scan report."""
    rule_summary = {}
    for f in scan_result.findings:
        rule_summary[f.rule_id] = rule_summary.get(f.rule_id, 0) + 1

    author_summary = {}
    for f in scan_result.new_findings:
        author_summary[f.author] = author_summary.get(f.author, 0) + 1

    return {
        "report_metadata": {
            "repository": repo_path,
            "scan_date": datetime.now(timezone.utc).isoformat(),
            "duration_seconds": scan_result.scan_duration,
            "commits_scanned": scan_result.commits_scanned
        },
        "summary": {
            "total_findings": len(scan_result.findings),
            "new_findings": len(scan_result.new_findings),
            "baseline_findings": len(scan_result.baseline_findings),
            "unique_rules_triggered": len(rule_summary),
            "rules_breakdown": rule_summary,
            "authors_with_new_findings": author_summary
        },
        "quality_gate": {
            "passed": len(scan_result.new_findings) == 0,
            "blocking_count": len(scan_result.new_findings)
        },
        "new_findings": [
            {
                "rule_id": f.rule_id,
                "description": f.description,
                "file": f.file,
                "line": f.line,
                "commit": f.commit,
                "author": f.author,
                "date": f.date,
                "secret_preview": f.secret,
                "entropy": f.entropy
            }
            for f in scan_result.new_findings
        ],
        "remediation_steps": [
            f"1. Rotate the {f.rule_id} credential found in {f.file}"
            for f in scan_result.new_findings
        ]
    }


def main():
    parser = argparse.ArgumentParser(description="Gitleaks Secret Scanning Pipeline")
    parser.add_argument("--repo-path", required=True, help="Path to git repository")
    parser.add_argument("--scan-type", default="detect", choices=["detect", "protect"],
                        help="Scan type: detect (history) or protect (staged/pre-commit)")
    parser.add_argument("--baseline", default=None, help="Path to baseline JSON file")
    parser.add_argument("--commit-range", default=None,
                        help="Git log range (e.g., HEAD~10..HEAD)")
    parser.add_argument("--staged", action="store_true",
                        help="Scan only staged changes (for pre-commit)")
    parser.add_argument("--config", default=None, help="Path to .gitleaks.toml config")
    parser.add_argument("--output", default="gitleaks-report.json", help="Output report path")
    parser.add_argument("--fail-on-findings", action="store_true",
                        help="Exit non-zero on new findings")
    parser.add_argument("--create-baseline", action="store_true",
                        help="Generate baseline from current findings")
    args = parser.parse_args()

    repo_path = os.path.abspath(args.repo_path)
    start_time = datetime.now(timezone.utc)

    print(f"[*] Running Gitleaks {args.scan_type} on {repo_path}")

    raw_result = run_gitleaks(
        repo_path,
        scan_type=args.scan_type,
        baseline_path=args.baseline,
        commit_range=args.commit_range,
        staged=args.staged,
        config_path=args.config
    )

    if raw_result["exit_code"] == -1:
        print(f"[ERROR] {raw_result['stderr']}")
        sys.exit(2)

    scan_result = ScanResult()
    scan_result.findings = parse_findings(raw_result["findings"])
    scan_result.scan_duration = (datetime.now(timezone.utc) - start_time).total_seconds()

    if args.baseline:
        baseline_fps = load_baseline(args.baseline)
        scan_result.new_findings, scan_result.baseline_findings = classify_findings(
            scan_result.findings, baseline_fps
        )
    else:
        scan_result.new_findings = scan_result.findings
        scan_result.baseline_findings = []

    if args.create_baseline:
        baseline_path = os.path.join(repo_path, ".gitleaks-baseline.json")
        with open(baseline_path, "w") as f:
            json.dump(raw_result["findings"], f, indent=2)
        print(f"[*] Baseline created: {baseline_path}")
        print(f"    Contains {len(raw_result['findings'])} findings")
        return

    report = generate_report(scan_result, repo_path)

    output_path = os.path.abspath(args.output)
    with open(output_path, "w") as f:
        json.dump(report, f, indent=2)
    print(f"[*] Report: {output_path}")

    print(f"\n[*] Total: {len(scan_result.findings)} | "
          f"New: {len(scan_result.new_findings)} | "
          f"Baseline: {len(scan_result.baseline_findings)}")

    if scan_result.new_findings:
        print("\n[!] New secrets detected:")
        for f in scan_result.new_findings:
            print(f"  [{f.rule_id}] {f.file}:{f.line} (commit: {f.commit}, author: {f.author})")
            print(f"    Secret: {f.secret}")

    if report["quality_gate"]["passed"]:
        print("\n[PASS] No new secrets detected.")
    else:
        print(f"\n[FAIL] {len(scan_result.new_findings)} new secrets found. Rotate immediately.")

    if args.fail_on_findings and not report["quality_gate"]["passed"]:
        sys.exit(1)


if __name__ == "__main__":
    main()