scanner/layer1_secrets.py

"""
Layer 1A — Secrets Scanner (TruffleHog)
Scans current files AND full git history. --verify flag confirms live secrets.
"""
from __future__ import annotations
import subprocess
import json
import math
import os
from pathlib import Path
from typing import List, Optional
from .models import Finding, Severity, Confidence, ScanDomain

SECRET_TYPE_MAP = {
    "OpenAI": "OPENAI_KEY", "Anthropic": "ANTHROPIC_KEY", "Stripe": "STRIPE_KEY",
    "AWS": "AWS_KEY", "GitHub": "GITHUB_TOKEN", "Supabase": "SUPABASE_KEY",
    "Twilio": "TWILIO_KEY", "SendGrid": "SENDGRID_KEY", "Resend": "RESEND_KEY",
    "Clerk": "CLERK_KEY", "Mapbox": "MAPBOX_KEY", "JWT": "JWT_SECRET",
    "PrivateKey": "PRIVATE_KEY", "MongoDB": "DB_CONNECTION_STRING",
    "PostgreSQL": "DB_CONNECTION_STRING", "MySQL": "DB_CONNECTION_STRING",
}

PLACEHOLDER_PATTERNS = [
    "your_", "YOUR_", "example", "EXAMPLE", "replace_", "REPLACE",
    "xxxxxxx", "XXXXXXXXX", "dummy", "DUMMY", "fake", "FAKE", "test123",
    "password123", "secret123", "changeme", "CHANGEME",
]

# Browser-safe publishable key prefixes — these are designed to be exposed to the client
_PUBLISHABLE_KEY_INDICATORS = [
    "pk_test_", "pk_live_",                # Stripe publishable keys
    "PUBLISHABLE_KEY", "publishable_key",  # Clerk, generic publishable keys
    "_ANON_KEY", "anon_key",               # Supabase anon keys (designed for client)
    "NEXT_PUBLIC_", "VITE_", "REACT_APP_", # Client-exposed prefix indicators
]

def _shannon_entropy(data: str) -> float:
    """Calculate Shannon entropy of a string. High entropy (>3.5) suggests real secrets."""
    if not data:
        return 0.0
    freq: dict[str, int] = {}
    for c in data:
        freq[c] = freq.get(c, 0) + 1
    length = len(data)
    return -sum((count / length) * math.log2(count / length) for count in freq.values())

def _has_char_diversity(value: str) -> bool:
    """Check if a value has mixed character classes (digits + letters). Low diversity = placeholder."""
    has_upper = any(c.isupper() for c in value)
    has_lower = any(c.islower() for c in value)
    has_digit = any(c.isdigit() for c in value)
    classes = sum([has_upper, has_lower, has_digit])
    return classes >= 2

def is_placeholder(value: str) -> bool:
    return any(p in value for p in PLACEHOLDER_PATTERNS) or len(value) < 8

def _is_publishable_key(value: str, context_line: str = "") -> bool:
    """Returns True if the value or its surrounding context indicates a browser-safe publishable key."""
    combined = value + " " + context_line
    return any(indicator in combined for indicator in _PUBLISHABLE_KEY_INDICATORS)

def is_in_comment(content: str, start_idx: int) -> bool:
    """Check if the given character index lies within a single-line or block comment."""
    before = content[:start_idx]
    
    # Check if inside a block comment /* ... */
    last_open_block = before.rfind("/*")
    last_close_block = before.rfind("*/")
    if last_open_block != -1 and last_open_block > last_close_block:
        return True
        
    # Check if inside a single-line comment (//, #, or --)
    line_start = before.rfind("\n") + 1
    line_before_match = before[line_start:]
    
    for marker in ("//", "#", "--"):
        marker_idx = line_before_match.find(marker)
        if marker_idx != -1:
            # Avoid matching URLs (e.g. https://) as comments
            if marker == "//":
                if marker_idx >= 5 and line_before_match[marker_idx-5:marker_idx] in ("http:", "ttps:"):
                    continue
            return True
    return False

def is_commented_in_file(content: str, val: str, file_path: str) -> bool:
    """Verify if all occurrences of the secret in the file content are commented out."""
    idx = 0
    occurrences = []
    while True:
        idx = content.find(val, idx)
        if idx == -1:
            break
        occurrences.append(idx)
        idx += len(val)
        
    if not occurrences:
        return False
        
    return all(is_in_comment(content, o) for o in occurrences)

def is_valid_secret(val: str, file_path: Optional[str] = None, repo_path: Optional[str] = None, secret_type: Optional[str] = None) -> bool:
    """
    Validates a potential secret string.
    Returns True if it passes entropy and formatting checks (likely a real secret).
    Returns False if it is a false positive.
    """
    val = val.strip().strip("'\"`")
    
    # 1. Suspiciously short check
    if len(val) < 16:
        return False
        
    # 2. Repeated characters check
    if len(set(val)) == 1:
        return False
        
    # 3. Common placeholder check (case-insensitive)
    PLACEHOLDERS = [
        "your-secret-here", "changeme", "placeholder", 
        "example", "test", "dummy", "fake", "todo"
    ]
    val_lower = val.lower()
    if any(p in val_lower for p in PLACEHOLDERS):
        return False
        
    # 4. Version or semver check
    import re
    if re.match(r"^v?\d+\.\d+(?:\.\d+)?(?:-[a-zA-Z0-9.]+)?$", val):
        return False
        
    # 5. Test files filter
    if file_path:
        f_lower = file_path.lower().replace("\\", "/").lstrip("/")
        parts = f_lower.split("/")
        if len(parts) >= 2 and parts[0] == "sandbox_unzip":
            f_lower = "/".join(parts[2:])
        from .tier0_pre_filter import classify_file_role
        role = classify_file_role(f_lower)
        if role == "test":
            return False

            
    # 6. Comments check
    if file_path and repo_path:
        fpath = os.path.join(repo_path, file_path)
        if os.path.exists(fpath):
            try:
                content = Path(fpath).read_text(encoding="utf-8", errors="replace")
                if is_commented_in_file(content, val, file_path):
                    return False
            except Exception:
                pass
                
    # 7. Shannon entropy threshold
    # If it is a known high-confidence secret type (from TruffleHog or regex rules),
    # use a lower threshold of 2.8 to avoid false negatives (e.g. AWS access keys or simple DB passwords).
    threshold = 2.8 if secret_type and secret_type != "Unknown" else 3.5
    if _shannon_entropy(val) <= threshold:
        return False
        
    return True

def run_secrets_scan(repo_path: str, file_roles: dict) -> List[Finding]:
    findings: List[Finding] = []
    
    # For large repositories or resource-constrained environments, running TruffleHog as a subprocess
    # will exceed container memory limits (512MB on Render) and crash the server.
    # We gracefully bypass TruffleHog and use our high-speed, memory-safe custom regex scanner instead.
    file_count = len(file_roles)
    max_trufflehog_files = int(os.environ.get("VIBESEC_MAX_TRUFFLEHOG_FILES", "10000"))
    is_hf = os.environ.get("SPACE_ID", "") != ""
    low_mem = os.environ.get("VIBESEC_LOW_MEMORY", "false").lower() == "true" and not is_hf
    
    # We allow overriding this check via environment variables if desired
    bypass_trufflehog = file_count > max_trufflehog_files or (low_mem and file_count > 300)
    force_trufflehog = os.environ.get("VIBESEC_FORCE_TRUFFLEHOG", "false").lower() == "true"

    if bypass_trufflehog and not force_trufflehog:
        bypass_reason = f"large repository ({file_count} files)" if file_count > max_trufflehog_files else f"low-memory container with {file_count} files"
        print(f"  -> [L1 Secrets] {bypass_reason.capitalize()} detected. Bypassing TruffleHog to prevent OOM. Running custom regex scanner.")
        findings.extend(_regex_secrets_scan(repo_path, file_roles))
        findings.extend(_scan_env_files(repo_path, file_roles))
        return findings

    production_files = {p for p, r in file_roles.items() if r == "production"}

    # Write a temporary ignore file for TruffleHog to avoid scanning massive directories and binaries
    ignore_file_path = os.path.join(repo_path, ".trufflehog-ignore")
    ignore_patterns = [
        r"node_modules",
        r"\.git",
        r"dist",
        r"build",
        r"\.venv",
        r"venv",
        r"public",
        r"static",
        r"\.png$",
        r"\.jpg$",
        r"\.jpeg$",
        r"\.gif$",
        r"\.zip$",
        r"\.pdf$",
        r"\.svg$",
        r"package-lock\.json$",
        r"yarn\.lock$",
        r"pnpm-lock\.yaml$",
        r"bun\.lockb$",
        r"\.map$"
    ]
    try:
        with open(ignore_file_path, "w", encoding="utf-8") as f:
            f.write("\n".join(ignore_patterns))
    except Exception:
        pass

    # Check if this is a Git repository containing history (i.e. has .git folder)
    git_dir = os.path.join(repo_path, ".git")
    is_git_repo = os.path.exists(git_dir)

    # A shallow clone (cloned with --depth=1) has no real git history to scan.
    # Running TruffleHog in git mode on a shallow clone is extremely memory-intensive
    # and causes OOM crashes on resource-constrained environments (like 512MB Render instances)
    # without providing any additional security value over a filesystem scan.
    is_shallow = os.path.exists(os.path.join(git_dir, "shallow"))

    # Allow overriding git history scan via environment variables if desired
    force_fs = os.environ.get("VIBESEC_FORCE_FILESYSTEM_SCAN", "false").lower() == "true"
    enable_git_scan = os.environ.get("VIBESEC_ENABLE_GIT_SCAN", "false").lower() == "true"

    if is_git_repo and not is_shallow and (enable_git_scan or not force_fs):
        # Generate safe URI for the Git repo clone
        repo_uri = Path(repo_path).as_uri()
        print(f"  -> [L1 Secrets] Git history detected. Running TruffleHog in memory-safe Git mode on: {repo_uri}")
        cmd = [
            "trufflehog", "git", repo_uri,
            "--json",
            "--no-update",
            "--max-depth=50",
            "--trust-local-git-config",
        ]
    else:
        # Fall back to standard filesystem scan on active files
        scan_reason = "Shallow clone detected (no history)" if is_shallow else "Local files / FS fallback"
        print(f"  -> [L1 Secrets] {scan_reason}. Running TruffleHog in filesystem mode.")
        cmd = [
            "trufflehog", "filesystem", repo_path,
            "--json",
            "--no-update",
            f"--exclude-paths={ignore_file_path}",
        ]

    try:
        env = {
            **os.environ,
            "GOMEMLIMIT": "2GiB",
        }
        result = subprocess.run(cmd, capture_output=True, text=True, timeout=300, env=env)
        # TruffleHog exits with non-zero on findings — parse stdout regardless of returncode
        stdout = result.stdout or ""
        if stdout.strip():
            for line in stdout.strip().split("\n"):
                if not line.strip():
                    continue
                try:
                    item = json.loads(line)
                except json.JSONDecodeError:
                    continue
                findings.extend(_parse_trufflehog_finding(item, production_files, repo_path))
        else:
            # TruffleHog produced no output — may have crashed silently. Run regex fallback.
            print("  !! [L1 Secrets warning] TruffleHog produced no output. Falling back to regex-based secret scanning.")
            findings.extend(_regex_secrets_scan(repo_path, file_roles))
    except (subprocess.TimeoutExpired, FileNotFoundError, OSError) as e:
        # TruffleHog not installed or crashed — fall back to regex-based scan
        print(f"  !! [L1 Secrets warning] TruffleHog execution failed: {e}. Falling back to regex-based secret scanning.")
        findings.extend(_regex_secrets_scan(repo_path, file_roles))

    # Clean up temporary ignore file
    try:
        if os.path.exists(ignore_file_path):
            os.remove(ignore_file_path)
    except Exception:
        pass

    # Also scan .env files explicitly (TruffleHog may miss committed env files)
    findings.extend(_scan_env_files(repo_path, file_roles))

    return findings

def _parse_trufflehog_finding(item: dict, production_files: set, repo_path: str = "") -> List[Finding]:
    findings = []
    source_data = item.get("SourceMetadata") or {}
    data_block = source_data.get("Data") or {}
    source_meta_fs = data_block.get("Filesystem") or {}
    source_meta_git = data_block.get("Git") or {}

    file_path = source_meta_fs.get("file") or source_meta_git.get("file", "")
    line = source_meta_fs.get("line") or source_meta_git.get("line", 0)
    commit = source_meta_git.get("commit", "")
    email = source_meta_git.get("email", "")

    detector = item.get("DetectorName", "Unknown")
    raw = item.get("Raw", "")
    verified = item.get("Verified", False)

    rel_path = file_path
    if rel_path:
        # Clean up absolute paths or relative formatting
        rel_path = os.path.basename(rel_path) if "/" not in rel_path and "\\" not in rel_path else rel_path
        rel_path = rel_path.replace("\\", "/")

    if not is_valid_secret(raw, file_path=rel_path, repo_path=repo_path, secret_type=detector):
        return []

    # Filter out browser-safe publishable keys
    context_line = ""
    if rel_path and line and repo_path:
        fpath = os.path.join(repo_path, rel_path)
        if os.path.exists(fpath):
            try:
                lines = Path(fpath).read_text(encoding="utf-8", errors="replace").splitlines()
                if 1 <= line <= len(lines):
                    context_line = lines[line - 1]
            except Exception:
                pass
    if _is_publishable_key(raw, context_line):
        return []

    rel_path = file_path
    if rel_path:
        # Clean up absolute paths or relative formatting
        rel_path = os.path.basename(rel_path) if "/" not in rel_path and "\\" not in rel_path else rel_path
        rel_path = rel_path.replace("\\", "/")

    severity = Severity.CRITICAL if verified else Severity.HIGH
    confidence = Confidence.HIGH if verified else Confidence.MEDIUM

    secret_type = "OTHER"
    for k, v in SECRET_TYPE_MAP.items():
        if k.lower() in detector.lower():
            secret_type = v
            break

    title = f"{'Live ' if verified else ''}Secret Exposed: {detector}"
    if commit:
        title += f" (Commit: {commit[:7]})"

    desc_loc = f"source code file '{rel_path}'" if rel_path in production_files else "repository"
    if commit:
        desc_loc += f" in historical commit {commit[:8]}"

    description = f"{'Verified live ' if verified else ''}API key/secret found in {desc_loc}. Detector: {detector}"
    
    if commit:
        explanation = (
            f"A {'verified live' if verified else 'potential'} {detector} secret was found in the repository history "
            f"in commit {commit} (authored by {email or 'unknown'}). "
            f"{'This key is currently active and provides unauthorized access.' if verified else 'Verify this is not a real credential.'}"
        )
    else:
        explanation = (
            f"A {'verified live' if verified else 'potential'} {detector} secret was found in active files. "
            f"{'This key is currently active and provides unauthorized access.' if verified else 'Verify this is not a real credential.'}"
        )

    findings.append(Finding(
        title=title,
        description=description,
        severity=severity,
        confidence=confidence,
        domain=ScanDomain.SECRETS,
        check_id=f"TRUFFLEHOG_{detector.upper().replace(' ', '_')}",
        check_category="secrets",
        file_path=rel_path or None,
        line_number=line or None,
        explanation=explanation,
        suggested_fix="Revoke this key immediately, rotate it, and use environment variables instead. Never commit secrets to version control. If found in history, rotate it immediately and rewrite history if possible.",
        validity="active" if verified else "unknown",
        secret_type=secret_type,
        raw_secret_value=raw[:8] + "..." if raw else None,
    ))
    return findings

def _scan_env_files(repo_path: str, file_roles: dict) -> List[Finding]:
    """Specifically hunt for committed .env files — extremely common in vibe-coded apps."""
    findings = []
    env_patterns = [".env", ".env.local", ".env.production", ".env.development", ".env.staging"]
    
    for root, dirs, files in os.walk(repo_path):
        dirs[:] = [d for d in dirs if d not in {"node_modules", ".git", ".next", "dist"}]
        for fname in files:
            if fname in env_patterns or fname.endswith(".env"):
                fpath = os.path.join(root, fname)
                rel = os.path.relpath(fpath, repo_path)
                try:
                    content = Path(fpath).read_text(encoding="utf-8", errors="replace")
                except Exception:
                    continue
                for i, line in enumerate(content.splitlines(), 1):
                    if "=" in line and not line.startswith("#"):
                        key, _, val = line.partition("=")
                        if val:
                            # Clean the value by removing inline comments, quotes, and spacing
                            val_clean = val.split("#", 1)[0].split("//", 1)[0].strip().strip("'\"`")
                            if is_valid_secret(val_clean, file_path=rel, repo_path=repo_path, secret_type="env_var"):
                                findings.append(Finding(
                                    title=f"Secret in Committed .env File: {key.strip()}",
                                    description=f"Environment file '{fname}' is committed to the repository and contains what appears to be a real secret value.",
                                    severity=Severity.CRITICAL,
                                    confidence=Confidence.HIGH,
                                    domain=ScanDomain.SECRETS,
                                    check_id="COMMITTED_ENV_FILE",
                                    check_category="secrets",
                                    file_path=rel,
                                    line_number=i,
                                    explanation=f"The file {rel} is committed to git and contains the variable {key.strip()} with a non-placeholder value. This exposes credentials to anyone with repo access.",
                                    suggested_fix=f"Add {fname} to .gitignore immediately. Remove from git history using `git filter-branch` or BFG Repo-Cleaner. Rotate the exposed credential.",
                                ))
    return findings

def _regex_secrets_scan(repo_path: str, file_roles: dict) -> List[Finding]:
    """Fallback when TruffleHog is not installed."""
    import re
    findings = []
    
    # Prefix-based detection with entropy scoring for robust secret identification.
    # Each entry: (regex_pattern, label, secret_type, min_entropy)
    # min_entropy: minimum Shannon entropy threshold to consider the match a real secret.
    SECRET_REGEXES = [
        # OpenAI — legacy sk-XXXX and modern sk-proj-XXXX formats
        (r"sk-[a-zA-Z0-9]{12,}", "OpenAI API Key", "OPENAI_KEY", 3.5),
        (r"sk-proj-[a-zA-Z0-9_-]{20,}", "OpenAI Project API Key", "OPENAI_KEY", 3.0),
        # Anthropic
        (r"sk-ant-[a-zA-Z0-9_-]{20,}", "Anthropic API Key", "ANTHROPIC_KEY", 3.0),
        # Stripe
        (r"sk_live_[a-zA-Z0-9]{24,}", "Stripe Live Secret Key", "STRIPE_KEY", 3.5),
        (r"rk_live_[a-zA-Z0-9]{24,}", "Stripe Restricted Key", "STRIPE_KEY", 3.5),
        # AWS
        (r"AKIA[0-9A-Z]{16}", "AWS Access Key", "AWS_KEY", 3.0),
        # GitHub — PAT, OAuth, fine-grained, app
        (r"ghp_[a-zA-Z0-9]{36}", "GitHub Personal Token", "GITHUB_TOKEN", 3.0),
        (r"gho_[a-zA-Z0-9]{36}", "GitHub OAuth Token", "GITHUB_TOKEN", 3.0),
        (r"github_pat_[a-zA-Z0-9_]{22,}", "GitHub Fine-Grained PAT", "GITHUB_TOKEN", 3.0),
        # GitLab
        (r"glpat-[a-zA-Z0-9_-]{20,}", "GitLab PAT", "GITLAB_TOKEN", 3.0),
        # JWT
        (r"eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9\.[a-zA-Z0-9_-]+\.[a-zA-Z0-9_-]+", "JWT Token", "JWT_SECRET", 3.5),
        # DB connection strings
        (r"(?:mongodb|postgresql|mysql|redis)://[^\s\"']+:[^\s\"'@]+@", "DB Connection String", "DB_CONNECTION_STRING", 2.5),
        # Supabase service role
        (r"service_role['\"]?\s*[:=]\s*['\"]?eyJ[a-zA-Z0-9_-]{100,}", "Supabase Service Role Key", "SUPABASE_KEY", 3.5),
        # Client-exposed secrets (NEXT_PUBLIC_, VITE_, REACT_APP_)
        (r"NEXT_PUBLIC_[A-Z_]*(?:KEY|SECRET|TOKEN)['\"]?\s*=\s*['\"]?[a-zA-Z0-9_-]{20,}", "Client-exposed Secret (NEXT_PUBLIC_)", "CLIENT_EXPOSED_SECRET", 3.5),
        (r"VITE_[A-Z_]*(?:KEY|SECRET|TOKEN)['\"]?\s*=\s*['\"]?[a-zA-Z0-9_-]{20,}", "Client-exposed Secret (VITE_)", "CLIENT_EXPOSED_SECRET", 3.5),
        (r"REACT_APP_[A-Z_]*(?:KEY|SECRET|TOKEN)['\"]?\s*=\s*['\"]?[a-zA-Z0-9_-]{20,}", "Client-exposed Secret (REACT_APP_)", "CLIENT_EXPOSED_SECRET", 3.5),
    ]

    production_files = {p for p, r in file_roles.items() if r in {"production", "config"}}
    
    for rel_path in production_files:
        fpath = os.path.join(repo_path, rel_path)
        try:
            # Skip files larger than 1MB to prevent OOM
            if os.path.getsize(fpath) > 1024 * 1024:
                continue
            content = Path(fpath).read_text(encoding="utf-8", errors="replace")
        except Exception:
            continue
        for pattern, name, stype, min_entropy in SECRET_REGEXES:
            for m in re.finditer(pattern, content):
                val = m.group(0)
                # Extract the secret portion (after the prefix) for entropy scoring
                # For patterns like sk-proj-XXXX, the entropy-relevant part is XXXX
                secret_body = val.split("-", 2)[-1] if "-" in val else val
                if "=" in secret_body:
                    secret_body = secret_body.split("=", 1)[-1].strip().strip("'\"`")
                else:
                    secret_body = secret_body.strip().strip("'\"`")

                if not is_valid_secret(secret_body, file_path=rel_path, repo_path=repo_path, secret_type=stype):
                    continue

                entropy = _shannon_entropy(secret_body)
                if entropy < min_entropy:
                    continue  # Too low entropy — likely a placeholder or structured constant

                # Filter out browser-safe publishable keys
                line_num = content[:m.start()].count("\n") + 1
                context_line = content.splitlines()[line_num - 1] if line_num <= len(content.splitlines()) else ""
                if _is_publishable_key(val, context_line):
                    continue  # Publishable keys are designed to be client-exposed

                # CRITICAL: Filter commented-out matches to reduce false positives.
                # Matches in comments (single-line //, #, or block /* */) are almost never real secrets.
                if is_in_comment(content, m.start()):
                    continue
                
                # Determine confidence based on entropy + character diversity
                if entropy > 4.0 and _has_char_diversity(secret_body):
                    confidence = Confidence.HIGH
                else:
                    confidence = Confidence.MEDIUM
                
                findings.append(Finding(
                    title=f"Potential Secret: {name}",
                    description=f"Pattern matching {name} found in {rel_path}",
                    severity=Severity.HIGH,
                    confidence=confidence,
                    domain=ScanDomain.SECRETS,
                    check_id=f"REGEX_{stype}",
                    check_category="secrets",
                    file_path=rel_path,
                    line_number=line_num,
                    explanation=f"A value matching the pattern for {name} was found (entropy: {entropy:.2f}). If this is a real credential, it is exposed.",
                    suggested_fix="Move this value to environment variables and never commit credentials.",
                    secret_type=stype,
                ))
    return findings