""" TruffleHog Secret Scanner Integration ===================================== Secure scanning for git repositories and code to detect: - Hardcoded secrets - API keys - Credentials - Tokens """ import json import logging import subprocess from dataclasses import dataclass from datetime import datetime logger = logging.getLogger(__name__) @dataclass class SecretFinding: """TruffleHog finding.""" detector_type: str detector_name: str source_type: str source: str secret: str # Partial redacted in production secret_hash: str raw: str raw_v2: str verified: bool verified_type: str time_found: str commit: str file_path: str line: int column: int # ─── TRUFFLEHOG SCANNER ─────────────────────────────────────────── class TruffleHogScanner: """Scanner using TruffleHog CLI.""" def __init__(self, verbose: bool = False): self._available = self._check_available() self.verbose = verbose def _check_available(self) -> bool: """Check if TruffleHog is installed.""" try: result = subprocess.run(["trufflehog", "--version"], capture_output=True, timeout=5) return result.returncode == 0 except (subprocess.TimeoutExpired, FileNotFoundError): return False def scan_directory(self, path: str, **kwargs) -> list[SecretFinding]: """ Scan a directory for secrets. Args: path: Directory path to scan **kwargs: Additional trufflehog arguments Returns: List of findings """ if not self._available: return self._stub_scan(path) cmd = ["trufflehog", "filesystem", path, "--json"] # TruffleHog v3 uses different flags if kwargs.get("include"): cmd.extend(["--include-paths", kwargs["include"]]) if kwargs.get("exclude"): cmd.extend(["--exclude-paths", kwargs["exclude"]]) if kwargs.get("max_depth"): cmd.extend(["--max-decode-depth", str(kwargs["max_depth"])]) if self.verbose: cmd.append("-v") try: result = subprocess.run( cmd, capture_output=True, text=True, timeout=300, # 5 minutes max ) if result.returncode != 0 and "no bugs found" not in result.stderr.lower(): logger.warning(f"TruffleHog scan warning: {result.stderr}") findings = [] for line in result.stdout.split("\n"): if line.strip(): try: data = json.loads(line) finding = SecretFinding( detector_type=data.get("DetectorType", ""), detector_name=data.get("DetectorName", ""), source_type=data.get("SourceType", ""), source=data.get("SourceMetadata", {}).get("Data", {}).get("Filesystem", {}).get("File", ""), secret=data.get("Raw", ""), secret_hash=data.get("Redacted", ""), raw=data.get("Raw", ""), raw_v2=data.get("RawV2", ""), verified=data.get("Verified", False), verified_type=data.get("VerificationStatus", ""), time_found=datetime.utcnow().isoformat(), commit=data.get("SourceMetadata", {}).get("Data", {}).get("Git", {}).get("Commit", ""), file_path=data.get("SourceMetadata", {}) .get("Data", {}) .get("Filesystem", {}) .get("File", ""), line=data.get("SourceMetadata", {}).get("Data", {}).get("Git", {}).get("Line", 0), column=data.get("SourceMetadata", {}).get("Data", {}).get("Git", {}).get("Column", 0), ) findings.append(finding) except json.JSONDecodeError: continue return findings except subprocess.TimeoutExpired: logger.error("TruffleHog scan timed out") return [] def scan_git_repo(self, url: str, **kwargs) -> list[SecretFinding]: """ Scan a Git repository for secrets. Args: url: Repository URL **kwargs: Additional trufflehog arguments Returns: List of findings """ if not self._available: return self._stub_scan(url) cmd = ["trufflehog", "git", url, "--json", "--unvalidated"] if kwargs.get("branch"): cmd.extend(["--branch", kwargs["branch"]]) if self.verbose: cmd.append("-v") try: result = subprocess.run( cmd, capture_output=True, text=True, timeout=600, # 10 minutes for git clone + scan ) findings = [] for line in result.stdout.split("\n"): if line.strip(): try: data = json.loads(line) findings.append( SecretFinding( detector_type=data.get("DetectorType", ""), detector_name=data.get("DetectorName", ""), source_type=data.get("SourceType", ""), source=data.get("SourceMetadata", {}).get("Data", {}).get("Git", {}).get("Repo", ""), secret=data.get("Raw", ""), secret_hash=data.get("Redacted", ""), raw=data.get("Raw", ""), raw_v2=data.get("RawV2", ""), verified=data.get("Verified", False), verified_type=data.get("VerificationStatus", ""), time_found=datetime.utcnow().isoformat(), commit=data.get("SourceMetadata", {}).get("Data", {}).get("Git", {}).get("Commit", ""), file_path=data.get("SourceMetadata", {}).get("Data", {}).get("Git", {}).get("File", ""), line=data.get("SourceMetadata", {}).get("Data", {}).get("Git", {}).get("Line", 0), column=data.get("SourceMetadata", {}).get("Data", {}).get("Git", {}).get("Column", 0), ) ) except json.JSONDecodeError: continue return findings except subprocess.TimeoutExpired: logger.error("TruffleHog git scan timed out") return [] def _stub_scan(self, path: str) -> list[SecretFinding]: """Stub scan when TruffleHog unavailable.""" return [ SecretFinding( detector_type="stub", detector_name="stub_detector", source_type="stub", source=path, secret="", secret_hash="stub", raw="", raw_v2="", verified=False, verified_type="", time_found=datetime.utcnow().isoformat(), commit="", file_path=path, line=0, column=0, ) ] def filter_findings(self, findings: list[SecretFinding], min_severity: int = 0) -> list[SecretFinding]: """ Filter findings by severity/verified status. Args: findings: List of findings min_severity: Minimum severity (0-100) Returns: Filtered findings """ return [f for f in findings if f.verified or (not f.verified and min_severity == 0)] # ─── GLOBAL SINGLETON ───────────────────────────────────────────── _scanner: TruffleHogScanner | None = None def get_scanner() -> TruffleHogScanner: """Get global TruffleHog scanner instance.""" global _scanner if _scanner is None: _scanner = TruffleHogScanner() return _scanner def scan_directory(path: str, **kwargs) -> list[SecretFinding]: """Scan a directory for secrets.""" scanner = get_scanner() return scanner.scan_directory(path, **kwargs) def scan_git_repo(url: str, **kwargs) -> list[SecretFinding]: """Scan a Git repository for secrets.""" scanner = get_scanner() return scanner.scan_git_repo(url, **kwargs)