scanner/layer7_validator.py

"""
Layer 7 — Tiered Validation
Tier 0: SQLite feedback store lookup (instant)
Tier 1: NVIDIA fast model (8B) — kills 70% of noise
Tier 2: NVIDIA smart model (70B) with tools — evidence-cited verdicts
Tier 3: Joern CPG (only ~3% of findings reach here)

PRODUCTION FIXES:
- Respects L4 taint_confirmed=True flag (skips AI re-validation for AST-confirmed findings)
- Uses validated verdict parsing with strict enum mapping (no hallucinated verdicts)
- Marks LLM-omitted verdicts for manual review instead of fail-open REAL
- Properly handles "needs_investigation" as MEDIUM-confidence REAL (not needs_human_review loop)
"""

from __future__ import annotations
import json
import re
import subprocess
import tempfile
import os
from typing import List, Optional
from pathlib import Path
from .models import Finding, Confidence, ScanDomain
from .nvidia_client import complete_fast_json, complete_smart_json
from .layer8_feedback import FeedbackStore, record_verdict

def get_file_imports_context(repo_path: str, file_path: str) -> str:
    """Read the first 35 lines of a file to extract import statements."""
    full_path = Path(repo_path) / file_path
    if not full_path.exists():
        return ""
    try:
        lines = full_path.read_text(encoding="utf-8", errors="ignore").splitlines()
        import_lines = []
        for line in lines[:35]:
            stripped = line.strip()
            if not stripped:
                continue
            if (
                stripped.startswith(("import ", "from ", "const ", "var ", "let "))
                and any(keyword in stripped for keyword in ("require(", "import ", "from "))
            ) or stripped.startswith(("using ", "package ", "import ")):
                import_lines.append(line)
        if import_lines:
            return "File Imports:\n" + "\n".join(import_lines) + "\n"
    except Exception:
        pass
    return ""

def get_resolved_dependencies_context(indexer, file_path: str, code_content: str) -> str:
    """Finds imported variables/functions in code_content, resolves their source files, and returns their bodies."""
    resolved_context = []
    # Find all words that could be functions/variables
    words = set(re.findall(r"\b[a-zA-Z_]\w*\b", code_content))
    if not words:
        return ""
        
    full_path = indexer.repo_path / file_path
    if not full_path.exists():
        return ""
        
    try:
        file_lines = full_path.read_text(encoding="utf-8", errors="ignore").splitlines()
        import_map = {}
        for line in file_lines[:60]: # Scan first 60 lines for imports
            line = line.strip()
            m1 = re.search(r"import\s+\{([^}]+)\}\s+from\s+['\"]([^'\"]+)['\"]", line)
            m2 = re.search(r"(?:const|let|var)\s+\{([^}]+)\}\s*=\s*require\(['\"]([^'\"]+)['\"]\)", line)
            m3 = re.search(r"from\s+([\w.]+)\s+import\s+([^\n]+)", line)
            
            if m1:
                names, module = m1.group(1), m1.group(2)
                for n in names.split(","):
                    name = n.strip().split(" as ")[0].strip()
                    if name:
                        import_map[name] = module
            elif m2:
                names, module = m2.group(1), m2.group(2)
                for n in names.split(","):
                    name = n.strip().split(" as ")[0].strip()
                    if name:
                        import_map[name] = module
            elif m3:
                module, names = m3.group(1), m3.group(2)
                for n in names.split(","):
                    name = n.strip().split(" as ")[0].strip()
                    if name:
                        import_map[name] = module
                        
        resolved_files = {}
        for word in words:
            if word in import_map:
                module_path = import_map[word]
                target_file = indexer._resolve_import_to_file(file_path, module_path)
                if target_file and target_file != file_path:
                    resolved_files.setdefault(target_file, []).append(word)
                    
        for target_file, target_words in resolved_files.items():
            funcs = indexer.function_defs.get(target_file, [])
            for f in funcs:
                if f.get("name") in target_words:
                    start = f.get("start_line", 1)
                    end = f.get("end_line", start + 30)
                    body = indexer.get_file_context(target_file, (start + end) // 2, window=(end - start) // 2 + 5)
                    resolved_context.append(
                        f"--- Resolved Dependency '{f['name']}' from '{target_file}' ---\n"
                        f"{body}\n"
                    )
    except Exception as e:
        print(f"      [GCI Warning] Error resolving dependencies for {file_path}: {e}")
        
    return "\n".join(resolved_context)

TIER1_SYSTEM = """You are a security triage assistant. Given a security finding and surrounding code, quickly determine if this is a false positive.

Respond ONLY with one of: real | false_positive | needs_investigation

Rules:  
- "false_positive" if: finding is in a test/fixture/mock file, code is clearly commented out, the "dangerous" pattern is in a string literal not actual code, the variable is hardcoded (not user-controlled)  
- "needs_investigation" if: you cannot determine without tracing data flow across files  
- "real" if: the issue is clearly present in production code and user data likely reaches the dangerous operation

File role tag will be provided. Test/fixture/vendor files that reach this point are almost always false positives."""

def validate_findings(
    findings: List[Finding],
    feedback_store: "FeedbackStore",
    indexer,
) -> List[Finding]:
    validated: List[Finding] = []
    indexer.l7_degraded = False
    
    if getattr(indexer, "is_training_app", False):
        print("  [Layer 7 Validation] Intentionally vulnerable training app detected. Bypassing dual-model AI validation engine to retain all findings.")
        for finding in findings:
            finding.is_false_positive = False
            finding.confidence = Confidence.HIGH
        return findings

    triage_ready_findings = []
    
    # 1. First Pass: Handle lockfiles, feedback store, and taint confirmations instantly
    for finding in findings:
        if finding.is_false_positive:
            validated.append(finding)
            continue
            
        LOCKFILES = ['bun.lock', 'package-lock.json', 'yarn.lock', 'poetry.lock', 'Pipfile.lock']
        if finding.file_path and any(lf in finding.file_path for lf in LOCKFILES):
            finding.confidence = Confidence.LOW
            finding.category = "deps-lockfile"
            finding.verdict = "FALSE_POSITIVE"
            finding.is_false_positive = True
            validated.append(finding)
            continue
            
        if getattr(finding, "taint_confirmed", False) is True:
            finding.verdict = "REAL"
            finding.validation_reason = "Confirmed by Layer 4 taint analysis"
            finding.is_false_positive = False
            finding.confidence = Confidence.HIGH
            validated.append(finding)
            continue
            
        if feedback_store.is_known_false_positive(finding):
            finding.is_false_positive = True
            finding.false_positive_reason = "Previously confirmed false positive (feedback store)"
            validated.append(finding)
            continue
            
        triage_ready_findings.append(finding)

    if not triage_ready_findings:
        return validated

    # 2. Second Pass: Batch Triage via AI validation engine
    # We chunk findings into batches of 5 to prevent token context overhead, avoid timeouts, and ensure hallucination-free triage
    batch_size = 5
    finding_batches = [triage_ready_findings[i:i + batch_size] for i in range(0, len(triage_ready_findings), batch_size)]
    
    print(f"  [Layer 7 Validation] Gathered {len(triage_ready_findings)} findings for AI verification. Executing single-shot Batch Triage across {len(finding_batches)} batches...")
    
    for batch_idx, batch in enumerate(finding_batches, 1):
        findings_payload = []
        findings_map = {}
        
        for idx, f in enumerate(batch, 1):
            findings_map[str(idx)] = f
            
            # Retrieve rich call-graph context
            func_info = indexer.get_enclosing_function_info(f.file_path or "", f.line_number or 1)
            imports_ctx = get_file_imports_context(indexer.repo_path, f.file_path or "")
            body_ctx = ""
            
            if func_info:
                body_ctx = (
                    f"Enclosing Function: {func_info['name']} (Lines {func_info['start_line']}-{func_info['end_line']})\n"
                    f"Callers (Incoming calls): {', '.join(func_info['callers']) or 'None'}\n"
                    f"Callees (Outgoing calls): {', '.join(func_info['callees']) or 'None'}\n"
                    f"Function Body:\n{func_info['body']}"
                )
            else:
                full_file_path = Path(indexer.repo_path) / (f.file_path or "")
                is_small_file = False
                file_lines_count = 0
                if full_file_path.exists():
                    try:
                        file_lines_count = len(full_file_path.read_text(encoding="utf-8", errors="ignore").splitlines())
                        if file_lines_count <= 150:
                            is_small_file = True
                    except Exception:
                        pass
                
                if is_small_file:
                    file_content = indexer.get_file_context(f.file_path or "", file_lines_count // 2, window=file_lines_count // 2 + 5)
                    body_ctx = f"Full File Content:\n{file_content}"
                else:
                    try:
                        lines = full_file_path.read_text(encoding="utf-8", errors="ignore").splitlines()
                        start = max(0, (f.line_number or 1) - 40)
                        end = min(len(lines), (f.line_number or 1) + 15)
                        numbered = [f"{i+1}: {lines[i]}" for i in range(start, end)]
                        body_ctx = f"Surrounding Code (Lines {start+1}-{end}):\n" + "\n".join(numbered)
                    except Exception:
                        body_ctx = indexer.get_file_context(f.file_path or "", f.line_number or 1, window=25)
            
            gci_ctx = get_resolved_dependencies_context(indexer, f.file_path or "", body_ctx)
            
            # Resolve route middleware context if function is enclosing route handler
            route_mw_ctx = ""
            if func_info:
                mw_info = indexer.get_route_middleware_context(func_info['name'])
                if mw_info and mw_info.get("routes"):
                    route_mw_ctx = "Route Middleware Guards:\n"
                    for r in mw_info["routes"]:
                        route_mw_ctx += f"- Route {r.get('method', 'ANY')} {r.get('path')} protected by: {', '.join(r.get('middleware', [])) or 'None'}\n"
                    if mw_info.get("middleware"):
                        route_mw_ctx += "Middleware Implementations:\n"
                        for mw, mw_details in mw_info["middleware"].items():
                            route_mw_ctx += f"  Middleware: {mw} in {mw_details.get('file')}\n  Body:\n{mw_details.get('body')}\n"
            
            raw_context = ""
            if imports_ctx:
                raw_context += imports_ctx + "\n"
            if body_ctx:
                raw_context += body_ctx + "\n"
            if route_mw_ctx:
                raw_context += route_mw_ctx + "\n"
            if gci_ctx:
                raw_context += gci_ctx + "\n"
                
            # Truncate each line in context to max 200 characters to prevent huge token overhead/crash on minified lines
            truncated_lines = []
            for line in raw_context.splitlines():
                if len(line) > 200:
                    truncated_lines.append(line[:200] + " ... [TRUNCATED]")
                else:
                    truncated_lines.append(line)
            context = "\n".join(truncated_lines)
            
            file_role = indexer.file_roles.get(f.file_path or "", "unknown")
            findings_payload.append({
                "id": idx,
                "title": f.title,
                "check_id": f.check_id,
                "file": f.file_path,
                "line": f.line_number,
                "file_role": file_role,
                "code_context": context
            })
            
        # Build prompt with strict completeness instructions
        system = """You are a senior security engineer doing precise false positive triage.
Given a list of security findings with their code context, evaluate if each finding is:
- "false_positive" if:
  1. The code does NOT use user-controlled/network-provided input, but rather uses hardcoded/static values, mock data, or static config constants.
  2. The input is sanitized, escaped, type-cast (e.g. parseInt), or validated before reaching the dangerous operation (e.g. DOMPurify.sanitize, regex validation).
  3. The finding is in a test, mock, fixture, build script, local dev tool, migration, database seed file, or non-production environment config.
  4. The code containing the issue is commented out, disabled, or unreachable in the production flow.
  5. The vulnerability check is for a hardcoded URL, but it only contains localhost, 127.0.0.1, internal test hostnames, or standard public URLs (like github.com).
  6. The vulnerability check is for a missing credit/quota validation, but the code is client-side code (HTML, CSS, UI component) where such checks are impossible to implement.
  7. The vulnerability check is for Supabase RLS bypass, but the code is a database seeding script, backend service-role administrator helper, or runs in a local migrations context.
  8. The finding is for missing authentication, authorization, or access control, but the 'Route Middleware Guards' and 'Middleware Implementations' context indicates the route endpoint is protected by guarding middleware (e.g., verifySession, isAdmin, authenticateToken, checkRole).
- "real" if:
  1. The vulnerability is in production code, handles dynamic, untrusted user-controlled input (e.g. from req.query, req.body, URL parameters, external API payloads), and does not apply any visible sanitization or validation before passing it to the sink (e.g. innerHTML, eval, child_process.exec, database query).
- "low_confidence" if:
  1. You cannot confidently determine whether the finding is real or a false positive without tracing data flow across multiple files.

CRITICAL INSTRUCTIONS:
- You MUST process and return a verdict for EVERY single finding ID in the input list. Do not skip or omit any IDs. The JSON keys in your response must exactly correspond to the finding IDs provided (e.g. "1", "2").
- Default to "false_positive" if there is clear evidence of safe context, static configurations, proper sanitization libraries, or test code.
- You MUST respond ONLY with a valid JSON object mapping each finding ID to a triage verdict dictionary, exactly like this format:
{
  "1": {"verdict": "false_positive", "confidence": "high", "reason": "Surrounding code uses DOMPurify.sanitize to clean the user input before inserting into innerHTML"},
  "2": {"verdict": "real", "confidence": "high", "reason": "User input from req.query.name is directly concatenated into database query string without parameterization"}
}
Do not include any markdown styling (like ```json), notes, explanations, or text outside of the JSON object itself."""

        user_prompt = f"Here is the list of findings to validate:\n\n{json.dumps(findings_payload, indent=2)}"
        
        # Query primary LLM (NVIDIA smart model 70B for high quality triage)
        verdicts = {}
        llm_is_degraded = False
        try:
            print(f"    [LLM Batch {batch_idx}] Querying primary triage model (70B) for {len(batch)} findings...")
            verdicts = complete_smart_json(system, user_prompt, max_tokens=2048)
        except Exception as e:
            print(f"    [LLM Batch {batch_idx} ERROR] Primary batch query failed: {e}. Retrying findings individually with smart model...")
            # Retrying each finding in the batch individually to isolate any context/token limit issues
            for f_payload in findings_payload:
                str_id = str(f_payload["id"])
                single_prompt = f"Here is the single finding to validate:\n\n{json.dumps([f_payload], indent=2)}"
                try:
                    single_verdicts = complete_smart_json(system, single_prompt, max_tokens=512)
                    if single_verdicts and str_id in single_verdicts:
                        verdicts[str_id] = single_verdicts[str_id]
                except Exception as single_err:
                    print(f"      [LLM Finding {str_id} ERROR] Individual smart query failed: {single_err}. Retrying with fast model...")
                    try:
                        single_verdicts = complete_fast_json(system, single_prompt, max_tokens=512)
                        if single_verdicts and str_id in single_verdicts:
                            verdicts[str_id] = single_verdicts[str_id]
                    except Exception as fast_err:
                        print(f"      [LLM Finding {str_id} ERROR] Individual fast query failed: {fast_err}. Retrying with secondary Cloudflare AI...")
                        try:
                            from .nvidia_client import complete_cloudflare_ai_json
                            single_verdicts = complete_cloudflare_ai_json(system, single_prompt)
                            if single_verdicts and str_id in single_verdicts:
                                verdicts[str_id] = single_verdicts[str_id]
                        except Exception as cf_err:
                            print(f"      [LLM Finding {str_id} ERROR] Secondary Cloudflare AI failed: {cf_err}. Bypassing AI verification for this finding.")
            
            if not verdicts:
                print(f"    [LLM Batch {batch_idx} ERROR] All individual query fallbacks failed. Activating Safe Triage fallback...")
                llm_is_degraded = True
                
        # Parse verdicts and apply to findings
        for str_id, f in findings_map.items():
            verdict_data = {}
            raw_verdict_str = None

            # Strict verdict extraction: validate structure and enum values
            if verdicts and str_id in verdicts:
                raw_verdict_data = verdicts[str_id]
                # Validate structure: must have "verdict" key
                if isinstance(raw_verdict_data, dict) and "verdict" in raw_verdict_data:
                    raw_verdict_str = raw_verdict_data.get("verdict", "")
                    raw_confidence_str = raw_verdict_data.get("confidence", "")
                    raw_reason_str = raw_verdict_data.get("reason", "")

                    # Map only known enum values; ignore hallucinated values
                    VERDICT_MAP = {
                        "real": "real",
                        "false_positive": "false_positive",
                        "needs_investigation": "needs_investigation",
                        "low_confidence": "low_confidence",
                        "low": "low_confidence",
                        "manual_review_needed": "needs_investigation",
                    }
                    VALID_CONFIDENCE_MAP = {
                        "high": Confidence.HIGH,
                        "medium": Confidence.MEDIUM,
                        "low": Confidence.LOW,
                    }

                    mapped_verdict = VERDICT_MAP.get(raw_verdict_str.lower().strip(), "")
                    if mapped_verdict:
                        verdict_data["verdict"] = mapped_verdict
                        verdict_data["confidence"] = VALID_CONFIDENCE_MAP.get(
                            raw_confidence_str.lower().strip(),
                            Confidence.MEDIUM
                        )
                        verdict_data["reason"] = raw_reason_str
                    else:
                        # LLM returned invalid/unknown verdict string — flag for manual review, don't fail open
                        print(f"    [Layer 7 Validation WARNING] LLM returned invalid verdict '{raw_verdict_str}' for finding {str_id}. Flagging for manual review.")
                        verdict_data["verdict"] = "needs_investigation"
                        verdict_data["confidence"] = Confidence.LOW
                        verdict_data["reason"] = f"Invalid LLM verdict '{raw_verdict_str}' — manual review required"
                else:
                    # Structure invalid (missing "verdict" key) — flag for manual review
                    print(f"    [Layer 7 Validation WARNING] LLM returned malformed verdict for finding {str_id}. Flagging for manual review.")
                    verdict_data["verdict"] = "needs_investigation"
                    verdict_data["confidence"] = Confidence.LOW
                    verdict_data["reason"] = "Malformed LLM response — manual review required"

            verdict_str = verdict_data.get("verdict", "")
            confidence = verdict_data.get("confidence", Confidence.MEDIUM)
            reason_str = verdict_data.get("reason", "")

            if llm_is_degraded:
                # SAFE TRIAGE FALLBACK: Keep findings active (REAL) but set low confidence and manual verify badge
                indexer.l7_degraded = True
                f.confidence = Confidence.LOW
                f.validation_reason = "Triage engine degraded. Manual verification recommended."
                f.is_false_positive = False
                validated.append(f)
            elif not verdict_str:
                # LLM HALLUCINATION/MISSING ID FALLBACK: Mark for manual review, NOT fail-open REAL.
                # Keeping a finding as REAL at low confidence without validation is a false-positive factory.
                # Flagging it for manual review ensures human eyes catch LLM misses without polluting the FP rate.
                print(f"    [Layer 7 Validation WARNING] LLM omitted verdict for finding ID: {str_id} ({f.title}). Flagging for manual review.")
                f.confidence = Confidence.LOW
                f.validation_reason = "AI Triage: verdict omitted by triage engine. Manual verification required."
                f.is_false_positive = False
                f.requires_llm_gate = True  # Flag so downstream knows this needs human review
                validated.append(f)
            elif verdict_str == "false_positive":
                f.is_false_positive = True
                f.false_positive_reason = f"AI Triage: {reason_str}"
                validated.append(f)
            elif verdict_str == "real":
                # Route all dataflow/taint findings from previous layers to Joern for verification
                is_taint_check = (
                    f.domain in {ScanDomain.SAST, ScanDomain.LLM_VULNS, ScanDomain.FRONTEND}
                    and not (f.check_id and f.check_id.startswith("VS-"))
                    and f.check_category not in {"secrets", "deps", "config", "auth"}
                )
                
                import shutil
                has_joern = shutil.which("joern") is not None

                if is_taint_check and has_joern:
                    tier3_result = _tier3_joern(f, indexer)
                    if tier3_result is not None:
                        if not tier3_result:
                            # Joern CPG confirmed no taint flow exists! This is a False Positive!
                            f.is_false_positive = True
                            f.false_positive_reason = "Tier 3 (Joern CPG): Bypassed AI 'real' verdict — no taint path confirmed in CPG"
                        else:
                            # Taint path is confirmed!
                            f.is_false_positive = False
                            f.confidence = Confidence.HIGH
                            f.validation_reason = f"AI Verified & Joern CPG Confirmed: {reason_str}" if reason_str else "AI Verified & Joern CPG Confirmed"
                    else:
                        f.is_false_positive = False
                        f.confidence = confidence
                        f.validation_reason = f"AI Verified (Joern bypass): {reason_str}" if reason_str else "AI Verified (Joern bypass)"
                else:
                    f.is_false_positive = False
                    f.confidence = confidence
                    f.validation_reason = f"AI Verified: {reason_str}" if reason_str else "AI Verified"
                validated.append(f)
            elif verdict_str == "needs_investigation":
                # Fix: "needs_investigation" (LLM can't determine) means moderate confidence REAL.
                # Do NOT route to expensive Joern analysis — that's for "not exploitable" verdicts.
                # Route to L4 dataflow agent if available, otherwise keep as flagged finding.
                f.confidence = Confidence.MEDIUM if not f.confidence else f.confidence
                f.validation_reason = f"AI Triage (Uncertain): {reason_str}" if reason_str else "AI Triage: needs manual investigation"
                # Don't mark as FP — this is an uncertain REAL, not a confirmed FP
                f.is_false_positive = False
                f.requires_llm_gate = True
                validated.append(f)
            elif verdict_str == "low_confidence":
                # LLM says low confidence — route to Joern Tier 3 for definitive answer
                f.confidence = Confidence.LOW
                f.validation_reason = f"AI Triage (Low Conf): {reason_str}" if reason_str else "Low confidence — Joern CPG analysis"

                is_taint_check = (
                    f.domain in {ScanDomain.SAST, ScanDomain.LLM_VULNS, ScanDomain.FRONTEND}
                    and not (f.check_id and f.check_id.startswith("VS-"))
                    and f.check_category not in {"secrets", "deps", "config", "auth"}
                )

                import shutil
                has_joern = shutil.which("joern") is not None

                if is_taint_check and has_joern:
                    tier3_result = _tier3_joern(f, indexer)
                    if tier3_result is not None:
                        f.is_false_positive = not tier3_result
                        if not tier3_result:
                            f.false_positive_reason = "Tier 3 (Joern CPG): no taint path confirmed"
                        else:
                            f.confidence = Confidence.MEDIUM
                            f.validation_reason = "Tier 3 (Joern CPG): taint path confirmed"
                validated.append(f)
            else:
                # Unknown verdict — keep as uncertain, flag for review
                f.confidence = Confidence.LOW
                f.validation_reason = f"Unknown verdict '{verdict_str}' — manual review required."
                f.is_false_positive = False
                f.requires_llm_gate = True
                validated.append(f)

            # Record for feedback learning
            f.file_role = indexer.file_roles.get(f.file_path or "", "unknown")
            record_verdict(f.check_id or "", f.file_role, is_false_positive=f.is_false_positive)

        # Reclaim memory and cooperative yield to let the CPU breathe and prevent OOM/timeouts on Render 512MB
        import gc
        import time
        gc.collect()
        time.sleep(1.0)

    # Reclaim memory
    import gc
    gc.collect()
    
    return validated

def _tier3_joern(finding: Finding, indexer) -> bool | None:
    """Run Joern CPG analysis on files relevant to this finding. Returns True=exploitable, False=FP, None=error."""
    if not finding.file_path:
        return None
    
    # Fast check if Joern is installed to avoid spawning a failing subprocess on low-memory environments
    import shutil
    if not shutil.which("joern"):
        return None
        
    # Strip dangerous characters to prevent Joern script injection
    clean_file_path = finding.file_path.replace('"', '').replace('\\', '/').replace('\n', '').replace('\r', '')
    
    # Ensure scan ID is only alphanumeric/safe
    clean_scan_id = re.sub(r'[^a-zA-Z0-9_-]', '', finding.id) if hasattr(finding, 'id') and finding.id else "scan_tmp"
    
    # Dynamically customize the target sink regex based on the specific type of vulnerability
    check_id = (finding.check_id or "").lower()
    if any(k in check_id for k in ["prompt-injection", "prompt_injection", "llm-security", "ai-agent"]):
        sink_regex = ".*create.*|.*generate.*|.*complete.*|.*chat.*|.*predict.*|.*invoke.*"
    elif any(k in check_id for k in ["xss", "inner_html", "inner-html"]):
        sink_regex = ".*innerHTML.*|.*dangerouslySetInnerHTML.*|.*write.*|.*html.*"
    elif any(k in check_id for k in ["sql", "db", "query"]):
        sink_regex = ".*query.*|.*execute.*|.*find.*|.*select.*|.*update.*|.*insert.*|.*delete.*"
    elif any(k in check_id for k in ["command", "eval", "rce", "exec", "code-execution"]):
        sink_regex = ".*exec.*|.*eval.*|.*run.*|.*system.*|.*spawn.*|.*popen.*"
    elif any(k in check_id for k in ["path-traversal", "file"]):
        sink_regex = ".*read.*|.*write.*|.*open.*|.*file.*"
    elif any(k in check_id for k in ["ssrf", "xxe"]):
        sink_regex = ".*fetch.*|.*get.*|.*request.*|.*parse.*|.*post.*"
    else:
        sink_regex = ".*exec.*|.*eval.*|.*query.*|.*create.*|.*write.*|.*read.*|.*fetch.*"
    
    temp_script_path = None
    try:
        # Write scripting commands to a cross-platform temporary file instead of Unix-only /dev/stdin
        with tempfile.NamedTemporaryFile(mode="w", suffix=".sc", delete=False) as f:
            if hasattr(indexer, "cpg_path") and indexer.cpg_path and os.path.exists(indexer.cpg_path):
                # Query global CPG, filtering parameters by filename
                f.write(f"""
importCpg("{indexer.cpg_path}")
println(cpg.call.where(_.name("{sink_regex}")).reachableByFlows(cpg.parameter.where(_.filename.contains("{clean_file_path}"))).toList.length)
""")
            else:
                # Fallback to local on-demand compilation
                f.write(f"""
importCode("{indexer.repo_path}/{clean_file_path}", "scan_{clean_scan_id[:8]}")
println(cpg.call.where(_.name("{sink_regex}")).reachableByFlows(cpg.parameter).toList.length)
""")
            temp_script_path = f.name

        result = subprocess.run(
            ["joern", "-J-Xmx4g", "--script", temp_script_path],
            capture_output=True, text=True, timeout=120
        )
        output = result.stdout.strip()
        if output and output.isdigit():
            return int(output) > 0
    except (subprocess.TimeoutExpired, FileNotFoundError):
        pass
    finally:
        if temp_script_path and os.path.exists(temp_script_path):
            try:
                os.remove(temp_script_path)
            except Exception:
                pass
    return None