#!/usr/bin/env python3 """ VulscanNervePinch.py — General-purpose, resumable, cost-controlled vulnerability TRIAGE engine. PHILOSOPHY ---------- This is not a one-shot scanner. It is a *triage and routing* engine: 1. RANK — cheaply score every code file by how likely it is to contain each vulnerability class (no API calls). Language-aware. 2. RUN — send the top-N files ONE AT A TIME to Claude for focused review. Each finding is tagged with OWASP/CWE and — crucially — a *validation pointer* (the endpoint / parameter / file to aim a tool at) plus the *downstream tools* relevant to that class (e.g. sqlmap for SQLi, dalfox for XSS, osv-scanner for vulnerable deps). 3. REPORT— compile findings ranked by severity x effort, with a per-finding "validation playbook": exactly which tool to run and where. The engine is GENERAL-PURPOSE. Three reusable registries below define everything; one clearly-marked TARGET_PROFILE block holds the only application-specific bits. To audit a different project, you edit the TARGET_PROFILE and (optionally) add language signals — the engine is untouched. - LANGUAGE_PROFILES : file extension -> language, with the right "input source" patterns per language (PHP $_GET, JS req.query, etc.) - VULN_CLASSES : each class bundles detection signals (per language), OWASP + CWE mapping, severity baseline, AND the relevant validation tools. ONE place to see what we look for. - TOOLS : the downstream scanner registry (SAST/DAST/deps/secrets/ IaC) — name, category, what it targets, install hint, invocation template. Easy to extend with new tools. See the whole catalog any time: python VulscanNervePinch.py catalog QUICK START ----------- export ANTHROPIC_API_KEY="sk-ant-..." python VulscanNervePinch.py catalog # see what we look for + tools python VulscanNervePinch.py rank --repo /path/to/your/repo # score files (FREE) python VulscanNervePinch.py status # queue + spend python VulscanNervePinch.py run --top 10 --dry-run # preview + cost estimate python VulscanNervePinch.py run --top 10 # analyze (resumes automatically) python VulscanNervePinch.py report # ranked report + validation playbook """ import argparse import hashlib import json import os import re import sys import time from datetime import datetime, timezone from pathlib import Path try: import requests except ImportError: sys.exit("ERROR: 'requests' is required. pip install requests") # ═══════════════════════════════════════════════════════════════════════════ # CONFIG — models, pricing, batch, paths # ═══════════════════════════════════════════════════════════════════════════ # Friendly alias -> (API model string, $/Mtok input, $/Mtok output). # Verified 2026-06-17; VERIFY before a big run. MODELS = { "opus-4-8": ("claude-opus-4-8", 5.00, 25.00), "sonnet-4-6": ("claude-sonnet-4-6", 3.00, 15.00), "haiku-4-5": ("claude-haiku-4-5-20251001", 1.00, 5.00), } DEFAULT_MODEL = "opus-4-8" DEFAULT_TOP_N = 10 MAX_OUTPUT_TOKENS = 4096 DEFAULT_MAX_COST = None # ── TARGET REPO ────────────────────────────────────────────────────── # Set this to the local path of the repo you want to audit, or pass # --repo on every command. The placeholder forces an explicit choice. DEFAULT_REPO = None # ── OUTPUT DIRECTORY ───────────────────────────────────────────────── # Where manifests, findings, and reports are written. Override with # --output or change this default. DEFAULT_OUTPUT_DIR = os.path.expanduser("~/Downloads/Vulnerability_Scanner_Output") MANIFEST_NAME = "manifest.json" FINDINGS_SUBDIR = "findings" REPORT_NAME = "SECURITY_AUDIT_REPORT.md" # tests/ excluded: test helpers rarely harbor deployable vulns, and including # them inflates the queue. Remove "tests"/"test"/"Tests" if you want coverage. EXCLUDE_DIRS = {"vendor", "node_modules", ".git", "tests", "test", "Tests", "dist", "build", "coverage", ".github"} EXCLUDE_FILE_PATTERNS = [r"\.min\.js$", r"-min\.js$", r"\.min\.css$", r"\.lock$", r"\.map$", r"composer\.lock", r"package-lock\.json"] MAX_FILE_BYTES = 200_000 API_URL = "https://api.anthropic.com/v1/messages" API_VERSION = "2023-06-01" RETRY_ATTEMPTS = 3 REQUEST_TIMEOUT = 180 INTER_REQUEST_SLEEP = 1.0 # ═══════════════════════════════════════════════════════════════════════════ # ENGINE REGISTRY 1 — LANGUAGE PROFILES (general-purpose) # ═══════════════════════════════════════════════════════════════════════════ # Map file extensions to a language, and define the "user input source" pattern # for each language. Input-reachability is the strongest cross-language signal # that a file is attacker-facing, so getting it right per language matters. LANGUAGE_PROFILES = { "php": {"ext": {".php", ".phtml", ".inc", ".tpl"}, "input": r"\$_(GET|POST|REQUEST|COOKIE|SERVER|FILES)"}, "js": {"ext": {".js", ".jsx", ".ts", ".tsx", ".mjs"}, "input": r"req\.(query|body|params|cookies|headers)|location\.(search|hash)"}, "python": {"ext": {".py"}, "input": r"request\.(args|form|values|json|GET|POST|data|cookies)|sys\.argv"}, "ruby": {"ext": {".rb", ".erb"}, "input": r"params\[|request\.(GET|POST|params)|ARGV"}, "go": {"ext": {".go"}, "input": r"r\.(URL\.Query|FormValue|PostFormValue|Form|Header)|os\.Args"}, "java": {"ext": {".java"}, "input": r"getParameter|getHeader|getQueryString|@RequestParam|@PathVariable"}, "c_cpp": {"ext": {".c", ".cpp", ".cc", ".h", ".hpp"}, "input": r"argv|getenv|scanf|gets|read\(|recv\("}, "shell": {"ext": {".sh", ".bash"}, "input": r"\$\{?[1-9@*]|read\s+\w"}, "sql": {"ext": {".sql"}, "input": r""}, } # Extension -> profile name (built once from the table above). EXT_TO_PROFILE = {ext: name for name, spec in LANGUAGE_PROFILES.items() for ext in spec["ext"]} CODE_EXTENSIONS = set(EXT_TO_PROFILE.keys()) # ═══════════════════════════════════════════════════════════════════════════ # ENGINE REGISTRY 2 — DOWNSTREAM TOOLS (general-purpose) # ═══════════════════════════════════════════════════════════════════════════ # The scanner doesn't run these — it ROUTES findings to them. Each entry tells a # human (or a later automation step) exactly how to validate a finding. Add new # tools here; reference them from VULN_CLASSES[...]["tools"]. Placeholders like # , , , get noted in the report so you fill them in. # # category: SAST (static) | DAST (dynamic, needs a running target you OWN) | # DEPS (dependency/supply-chain) | SECRETS | IAC (infra-as-code/containers) TOOLS = { # --- DAST: dynamic, require a running instance you control --------------- "sqlmap": {"name": "sqlmap", "category": "DAST", "targets": "live endpoint + parameter", "install": "pipx install sqlmap (or apt install sqlmap)", "invocation": "sqlmap -u '?=1' -p --batch --risk=2 --level=3", "note": "Confirms/exploits SQLi dynamically. Your own instance only."}, "ghauri": {"name": "ghauri", "category": "DAST", "targets": "live endpoint + parameter", "install": "pipx install ghauri", "invocation": "ghauri -u '?=1' -p --batch", "note": "Faster SQLi alternative to sqlmap."}, "dalfox": {"name": "dalfox", "category": "DAST", "targets": "live endpoint + parameter", "install": "go install github.com/hahwul/dalfox/v2@latest", "invocation": "dalfox url '?=test'", "note": "XSS scanner/verifier."}, "commix": {"name": "commix", "category": "DAST", "targets": "live endpoint + parameter", "install": "pipx install commix", "invocation": "commix -u '' --data='=1'", "note": "Command-injection confirmation."}, "tplmap": {"name": "tplmap", "category": "DAST", "targets": "live endpoint + parameter", "install": "git clone github.com/epinna/tplmap", "invocation": "tplmap -u '?=*'", "note": "Server-side template injection (SSTI)."}, "ssrfmap": {"name": "SSRFmap", "category": "DAST", "targets": "request file + param", "install": "git clone github.com/swisskyrepo/SSRFmap", "invocation": "python ssrfmap.py -r request.txt -p -m readfiles", "note": "SSRF exploitation against your own instance."}, "zap": {"name": "OWASP ZAP", "category": "DAST", "targets": "whole web app (base URL)", "install": "snap install zaproxy (or Docker: zaproxy/zap-stable)", "invocation": "zap-baseline.py -t '' (baseline) / full scan in GUI", "note": "Broad web DAST: auth, XSS, headers, access control."}, "nuclei": {"name": "nuclei", "category": "DAST", "targets": "base URL", "install": "go install github.com/projectdiscovery/nuclei/v3/cmd/nuclei@latest", "invocation": "nuclei -u '' -tags ", "note": "Huge community template library; known-CVE + misconfig checks."}, "nikto": {"name": "Nikto", "category": "DAST", "targets": "base URL", "install": "apt install nikto", "invocation": "nikto -h ''", "note": "Web-server misconfig + dated-software checks."}, "ffuf": {"name": "ffuf", "category": "DAST", "targets": "base URL + wordlist", "install": "go install github.com/ffuf/ffuf/v2@latest", "invocation": "ffuf -u '/FUZZ' -w wordlist.txt", "note": "Content/parameter discovery + path-traversal fuzzing."}, "burp": {"name": "Burp Suite", "category": "DAST", "targets": "whole web app (proxy)", "install": "portswigger.net/burp (Community is free)", "invocation": "manual — proxy traffic, Repeater/Intruder on ", "note": "Manual verification of access-control / IDOR / logic flaws."}, # --- SAST: static, run against the repo --------------------------------- "semgrep": {"name": "Semgrep", "category": "SAST", "targets": "repo path", "install": "pipx install semgrep", "invocation": "semgrep --config p/ (e.g. p/php, p/security-audit)", "note": "Multi-language rule-based SAST; corroborates code findings."}, "bandit": {"name": "Bandit", "category": "SAST", "targets": "python repo", "install": "pipx install bandit", "invocation": "bandit -r ", "note": "Python SAST."}, "brakeman": {"name": "Brakeman", "category": "SAST", "targets": "rails repo", "install": "gem install brakeman", "invocation": "brakeman ", "note": "Ruby on Rails SAST."}, "gosec": {"name": "gosec", "category": "SAST", "targets": "go repo", "install": "go install github.com/securego/gosec/v2/cmd/gosec@latest", "invocation": "gosec ./...", "note": "Go SAST."}, "psalm": {"name": "Psalm", "category": "SAST", "targets": "php repo", "install": "composer require --dev vimeo/psalm", "invocation": "vendor/bin/psalm --taint-analysis", "note": "PHP taint analysis — strong for tracing input->sink in PHP."}, "codeql": {"name": "CodeQL", "category": "SAST", "targets": "repo (multi-lang)", "install": "github.com/github/codeql-cli-binaries", "invocation": "codeql database create db --language= && codeql database analyze", "note": "Deep semantic dataflow; best-in-class cross-file taint tracking."}, # --- DEPS: dependency / supply chain ------------------------------------ "osv-scanner": {"name": "osv-scanner", "category": "DEPS", "targets": "repo / lockfiles", "install": "go install github.com/google/osv-scanner/cmd/osv-scanner@latest", "invocation": "osv-scanner --recursive ", "note": "Cross-ecosystem known-CVE scanning (the general-purpose choice)."}, "trivy": {"name": "Trivy", "category": "DEPS", "targets": "repo / image / fs", "install": "apt install trivy", "invocation": "trivy fs / trivy image ", "note": "Deps + container + IaC in one tool."}, "grype": {"name": "Grype", "category": "DEPS", "targets": "repo / SBOM / image", "install": "brew install grype (or install script)", "invocation": "grype dir:", "note": "Dependency CVE scanner; pairs with Syft for SBOM."}, # --- SECRETS ------------------------------------------------------------ "gitleaks": {"name": "gitleaks", "category": "SECRETS", "targets": "repo + git history", "install": "github.com/gitleaks/gitleaks releases", "invocation": "gitleaks detect --source ", "note": "Secrets across full git history."}, "trufflehog": {"name": "TruffleHog", "category": "SECRETS", "targets": "repo + git history", "install": "pipx install trufflehog (or binary)", "invocation": "trufflehog filesystem ", "note": "Secrets with live-credential verification."}, # --- IAC / containers --------------------------------------------------- "checkov": {"name": "Checkov", "category": "IAC", "targets": "IaC / Dockerfiles / k8s", "install": "pipx install checkov", "invocation": "checkov -d ", "note": "Terraform/CloudFormation/k8s/Dockerfile misconfig."}, "hadolint": {"name": "hadolint", "category": "IAC", "targets": "Dockerfile", "install": "github.com/hadolint/hadolint releases", "invocation": "hadolint ", "note": "Dockerfile best-practice + security linting."}, } # ═══════════════════════════════════════════════════════════════════════════ # ENGINE REGISTRY 3 — VULNERABILITY CLASSES (general-purpose) # ═══════════════════════════════════════════════════════════════════════════ # THE heart of the scanner. Each class declares, in ONE place: # - what it is (description), its OWASP category and CWE id(s) # - severity_baseline (used when the model is unsure) # - weight (how much a signal hit boosts a file's rank) # - signals: detection regexes, keyed by language profile (+ "generic" always) # - tools: the downstream validators relevant to this class (keys into TOOLS) # This is what makes it obvious "where each vuln type is looked for and which # tools are relevant." Add a class, or add a language to a class, right here. VULN_CLASSES = { "sqli": { "name": "SQL Injection", "owasp": "A03:2021 Injection", "cwe": ["CWE-89"], "severity_baseline": "High", "weight": 5, "description": "User input concatenated into a SQL query instead of parameterized.", "signals": { "php": [r"sql(Statement|Query|Insert)\s*\([^)]*(\.|\$_(GET|POST|REQUEST))", r"mysqli_query|->query\(", r"SELECT .*\$"], "python": [r"cursor\.execute\([^,)]*%|\.execute\(f[\"']|\.raw\(", r"SELECT .*\{|SELECT .*%"], "java": [r"createStatement\(\)|executeQuery\([^?]*\+", r"\"SELECT .*\" *\+"], "go": [r"db\.(Query|Exec)\([^,)]*\+|fmt\.Sprintf\([^)]*SELECT"], "generic": [r"(SELECT|INSERT|UPDATE|DELETE) .*(\+|\$\{|%s|format\()"], }, "tools": ["sqlmap", "ghauri", "semgrep", "psalm", "codeql"], "validation_hint": "Aim sqlmap/ghauri at the live endpoint+parameter in the pointer, on YOUR instance.", }, "command_injection": { "name": "OS Command Injection", "owasp": "A03:2021 Injection", "cwe": ["CWE-78"], "severity_baseline": "Critical", "weight": 5, "description": "User input reaches a shell/OS command execution sink.", "signals": { "php": [r"\b(exec|shell_exec|system|passthru|popen|proc_open)\s*\(", r"`[^`]*\$"], "python": [r"os\.(system|popen)\(|subprocess\.(call|run|Popen)\([^)]*shell\s*=\s*True"], "js": [r"child_process|exec\(|execSync\(|spawn\("], "ruby": [r"`[^`]*#\{|system\(|%x\(|Open3\."], "go": [r"exec\.Command\("], "java": [r"Runtime\.getRuntime\(\)\.exec|ProcessBuilder"], "generic": [r"\bsystem\s*\(|\bexec\w*\s*\("], }, "tools": ["commix", "semgrep", "nuclei"], "validation_hint": "commix against the endpoint+param in the pointer (your instance).", }, "xss": { "name": "Cross-Site Scripting", "owasp": "A03:2021 Injection", "cwe": ["CWE-79"], "severity_baseline": "Medium", "weight": 4, "description": "User input rendered into HTML/JS without output encoding.", "signals": { "php": [r"(echo|print)\b[^;]*\$_(GET|POST|REQUEST)"], "js": [r"innerHTML\s*=|document\.write\(|dangerouslySetInnerHTML|\.html\(\s*\w"], "python": [r"render_template_string\(|mark_safe\(|\|\s*safe"], "generic": [r"innerHTML|document\.write"], }, "tools": ["dalfox", "zap", "semgrep"], "validation_hint": "dalfox/ZAP against the reflecting endpoint+param; for stored XSS, trace the render sink.", }, "ssrf_file_xxe": { "name": "SSRF / File Disclosure / XXE", "owasp": "A10:2021 SSRF / A05", "cwe": ["CWE-918", "CWE-22", "CWE-611"], "severity_baseline": "High", "weight": 4, "description": "Outbound request, file path, or XML parse influenced by user input.", "signals": { "php": [r"(curl_init|curl_setopt|file_get_contents)\s*\([^)]*\$", r"(include|require)(_once)?\s*\([^)]*\$", r"(fopen|readfile|file)\s*\([^)]*\$_(GET|POST|REQUEST)", r"simplexml_load|DOMDocument|loadXML"], "js": [r"axios\.(get|post)\([^)]*req\.|fetch\([^)]*req\.|require\([^)]*req\.", r"fs\.(readFile|createReadStream)\([^)]*req\."], "python": [r"requests\.(get|post)\([^)]*request\.|urlopen\([^)]*request\.", r"open\([^)]*request\.|etree\.parse|lxml"], "generic": [r"file_get_contents|readfile|loadXML|XMLReader|DOMParser"], }, "tools": ["ssrfmap", "ffuf", "nuclei", "semgrep"], "validation_hint": "SSRF: SSRFmap with a saved request. Path traversal: ffuf the param with ../ payloads. XXE: trace the XML parser config.", }, "access_control": { "name": "Broken Access Control / IDOR", "owasp": "A01:2021 Broken Access Control", "cwe": ["CWE-639", "CWE-862", "CWE-285"], "severity_baseline": "High", "weight": 5, "description": "Object reference or action not gated by an authorization check.", "signals": { "php": [r"\$_(GET|POST|REQUEST)\[['\"](id|uid|user_id|account|order|doc|record|object)", r"->(find|get|load|fetch)\w*\(\s*\$_(GET|POST|REQUEST)"], "js": [r"req\.params\.(id|userId|pid)|findById\(\s*req\."], "python": [r"objects\.get\(\s*id\s*=\s*request|get_object_or_404\([^)]*request"], "generic": [], }, # NOTE: also boosted by the "handles input + touches data + no authz check" # heuristic in score_file(), using TARGET_PROFILE["authz_check_pattern"]. "tools": ["burp", "zap", "nuclei"], "validation_hint": "Manual: in Burp, replay the request as a lower-priv user / swap the object ID. ZAP access-control testing.", }, "auth_session": { "name": "Authentication / Session Weakness", "owasp": "A07:2021 Auth Failures", "cwe": ["CWE-287", "CWE-384", "CWE-352"], "severity_baseline": "High", "weight": 4, "description": "Weak session handling, missing CSRF, OAuth/token validation flaws.", "signals": { "php": [r"\$_SESSION|session_|setcookie|oauth|jwt|->login\(", r"password_verify|password_hash"], "js": [r"jsonwebtoken|jwt\.(sign|verify)|express-session|passport"], "python": [r"session\[|flask_login|jwt\.(encode|decode)|django\.contrib\.auth"], "generic": [r"oauth|jwt|session|csrf"], }, "tools": ["zap", "nuclei", "semgrep", "burp"], "validation_hint": "ZAP for session/CSRF; manually test token scope and redirect-URI validation in Burp.", }, "deserialization": { "name": "Insecure Deserialization", "owasp": "A08:2021 Integrity Failures", "cwe": ["CWE-502"], "severity_baseline": "High", "weight": 4, "description": "Untrusted data deserialized into objects (object injection).", "signals": { "php": [r"\bunserialize\s*\("], "python": [r"pickle\.loads|yaml\.load\([^)]*\)|marshal\.loads"], "java": [r"readObject\(|ObjectInputStream"], "ruby": [r"Marshal\.load|YAML\.load"], "generic": [r"unserialize|pickle\.loads|readObject"], }, "tools": ["semgrep", "codeql"], "validation_hint": "Manual gadget-chain analysis (no auto-exploit generation). Confirm the source is attacker-controllable.", }, "crypto": { "name": "Cryptographic Failure", "owasp": "A02:2021 Cryptographic Failures", "cwe": ["CWE-327", "CWE-330"], "severity_baseline": "Medium", "weight": 3, "description": "Weak hashing, weak randomness, broken cipher modes, hardcoded keys.", "signals": { "generic": [r"\b(md5|sha1)\s*\(", r"\b(mt_rand|rand)\s*\(", r"MCRYPT_MODE_ECB|\bECB\b", r"\bDES\b|RC4"], "python": [r"random\.(random|randint)\(|hashlib\.(md5|sha1)\("], "js": [r"Math\.random\(|createHash\(['\"](md5|sha1)"], }, "tools": ["semgrep"], "validation_hint": "Static confirmation usually suffices; assess whether the weak primitive guards a security boundary.", }, "secrets": { "name": "Hardcoded Secret", "owasp": "A05:2021 Security Misconfiguration", "cwe": ["CWE-798"], "severity_baseline": "High", "weight": 3, "description": "Credentials, API keys, or tokens committed in source/config.", "signals": { "generic": [r"(password|passwd|secret|api[_-]?key|token)\s*[=:]\s*['\"][^'\" ]{6,}"], }, "tools": ["gitleaks", "trufflehog"], "validation_hint": "gitleaks/trufflehog across git history; rotate any live credential immediately.", }, "ssti": { "name": "Server-Side Template Injection", "owasp": "A03:2021 Injection", "cwe": ["CWE-1336"], "severity_baseline": "High", "weight": 4, "description": "User input flows into a server-side template engine.", "signals": { "python": [r"render_template_string\(|Template\([^)]*request|Environment\("], "js": [r"\.compile\([^)]*req\.|new Function\([^)]*req\."], "php": [r"->render\([^)]*\$_(GET|POST|REQUEST)|createTemplate"], "generic": [r"render_template_string|createTemplate"], }, "tools": ["tplmap", "semgrep"], "validation_hint": "tplmap against the endpoint+param in the pointer (your instance).", }, # --- Tooling-only classes: no per-file code signal, but the catalog and the # report should still route to the right scanner. ---------------------- "vulnerable_deps": { "name": "Vulnerable / Outdated Dependency", "owasp": "A06:2021 Vulnerable Components", "cwe": ["CWE-1104", "CWE-937"], "severity_baseline": "Medium", "weight": 0, "description": "Known-CVE versions in dependency manifests (detected by tooling, not per-file review).", "signals": {}, "tools": ["osv-scanner", "trivy", "grype"], "validation_hint": "Run osv-scanner/trivy on the repo; confirm the vulnerable code path is reachable in this app's usage.", }, "misconfiguration": { "name": "Security Misconfiguration", "owasp": "A05:2021 Security Misconfiguration", "cwe": ["CWE-16"], "severity_baseline": "Medium", "weight": 0, "description": "Insecure infra/container/IaC settings, debug flags, exposed services.", "signals": {}, "tools": ["checkov", "hadolint", "trivy", "nuclei", "nikto"], "validation_hint": "checkov/hadolint on IaC + Dockerfiles; nikto/nuclei against the running service.", }, } # ═══════════════════════════════════════════════════════════════════════════ # TARGET PROFILE — the ONLY application-specific block (edit to retarget) # ═══════════════════════════════════════════════════════════════════════════ # Everything above is reusable. To audit a different project, edit THIS block: # - dir_multipliers : path fragments that are high-yield in THIS codebase # - authz_check_pattern : how THIS app gates access (used by the access-control # heuristic: input + data access + NO authz check = boost) # - safe_glossary : project-specific safe wrappers, told to the model so it # doesn't false-positive on them (clawing back cross-file context) TARGET_PROFILE = { # ── Change "name" and the settings below to fit your target repo. ── # These are GENERIC defaults that work reasonably for most web apps. # For a specific project, tune dir_multipliers to its high-risk paths, # set authz_check_pattern to its authorization gate function(s), and # fill safe_glossary with its known-safe wrappers so the AI model # doesn't false-positive on them. "name": "Generic", "dir_multipliers": { "controllers": 1.7, "api": 1.8, "apis": 1.8, "routes": 1.8, "auth": 1.8, "oauth": 1.8, "admin": 1.7, "middleware": 1.5, "handlers": 1.7, "views": 1.5, "endpoints": 1.8, "src": 1.2, }, "authz_check_pattern": r"(authorize|isAuthenticated|checkPermission|requireAuth|canAccess|has_permission|@login_required|@require_role)", "safe_glossary": """\ No project-specific safe patterns configured. If this is a known codebase, edit TARGET_PROFILE["safe_glossary"] in VulscanNervePinch.py to list sanitization wrappers, escaping helpers, and authorization gates so the AI model skips them instead of false-positive flagging. If a sink may be sanitized by a helper in another file you can't see, say so and lower your confidence rather than guessing. """, } INPUT_REACHABILITY_MULTIPLIER = 1.8 # applied when a file takes user input # ═══════════════════════════════════════════════════════════════════════════ # Ranking (cheap, no API calls) # ═══════════════════════════════════════════════════════════════════════════ def profile_for(path: str) -> str: return EXT_TO_PROFILE.get(Path(path).suffix.lower(), "generic") def iter_code_files(repo: Path): for root, dirs, files in os.walk(repo): dirs[:] = [d for d in dirs if d not in EXCLUDE_DIRS] for name in files: abspath = Path(root) / name rel = abspath.relative_to(repo).as_posix() if abspath.suffix.lower() not in CODE_EXTENSIONS: continue if any(re.search(p, rel) for p in EXCLUDE_FILE_PATTERNS): continue try: if abspath.stat().st_size > MAX_FILE_BYTES: continue except OSError: continue yield rel, abspath def score_file(rel: str, text: str): """Heuristic score using language-appropriate + generic signals. Returns (priority, suspected_classes, explanation).""" profile = profile_for(rel) input_pat = LANGUAGE_PROFILES.get(profile, {}).get("input", "") suspected, signals, class_score = [], {}, 0.0 for cls, spec in VULN_CLASSES.items(): if not spec["signals"]: continue # tooling-only class pats = spec["signals"].get(profile, []) + spec["signals"].get("generic", []) hits = sum(len(re.findall(p, text, re.IGNORECASE)) for p in pats) if hits: suspected.append(cls) signals[cls] = hits class_score += spec["weight"] * hits # Access-control heuristic (general): takes input + touches data + no authz check. takes_input = bool(input_pat and re.search(input_pat, text)) touches_data = bool(re.search(r"sql(Statement|Query|Insert)|\.execute\(|db\.(Query|Exec)|objects\.get", text)) has_authz = bool(re.search(TARGET_PROFILE["authz_check_pattern"], text)) if takes_input and touches_data and not has_authz: class_score += VULN_CLASSES["access_control"]["weight"] if "access_control" not in suspected: suspected.append("access_control") signals["access_control_gap"] = 1 signals["access_control_gap_reason"] = "input + data access, no authz check found" dir_mult = 1.0 for frag, mult in TARGET_PROFILE["dir_multipliers"].items(): if frag in rel: dir_mult = max(dir_mult, mult) input_mult = INPUT_REACHABILITY_MULTIPLIER if takes_input else 1.0 priority = round(class_score * dir_mult * input_mult, 2) explanation = {"profile": profile, "class_signal_hits": signals, "dir_multiplier": dir_mult, "input_reachability": takes_input, "input_multiplier": input_mult, "raw_class_score": round(class_score, 2)} return priority, suspected, explanation def file_hash(text: str) -> str: return hashlib.sha256(text.encode("utf-8", "ignore")).hexdigest()[:16] def cmd_rank(args): if not args.repo: sys.exit("ERROR: No target repo specified.\n" " Use: python VulscanNervePinch.py rank --repo /path/to/your/repo\n" " Or set DEFAULT_REPO near the top of VulscanNervePinch.py.") repo = Path(args.repo).expanduser().resolve() if not repo.is_dir(): sys.exit(f"ERROR: repo not found: {repo}") manifest = load_manifest(args) print(f"==> Ranking files in {repo} ...") seen = requeued = 0 for rel, abspath in iter_code_files(repo): seen += 1 try: text = abspath.read_text("utf-8", "ignore") except OSError: continue h = file_hash(text) priority, suspected, explanation = score_file(rel, text) existing = manifest["files"].get(rel) if existing: if existing.get("content_hash") == h: existing.update(priority_score=priority, suspected_classes=suspected, signals=explanation) continue existing.update(status="pending", content_hash=h, priority_score=priority, suspected_classes=suspected, signals=explanation, report_md=None, report_json=None, findings_count=0, input_tokens=0, output_tokens=0, cost_usd=0.0) requeued += 1 continue manifest["files"][rel] = { "priority_score": priority, "suspected_classes": suspected, "signals": explanation, "size_bytes": abspath.stat().st_size, "content_hash": h, "status": "pending", "analyzed_at": None, "model": None, "findings_count": 0, "report_md": None, "report_json": None, "input_tokens": 0, "output_tokens": 0, "cost_usd": 0.0, } manifest["repo_path"] = str(repo) manifest["last_ranked"] = now_iso() recompute_totals(manifest) save_manifest(args, manifest) print(f" Scanned {seen} code files. {manifest['totals']['pending']} pending, " f"{manifest['totals']['analyzed']} analyzed" + (f", {requeued} re-queued (changed)." if requeued else ".")) print(f" Manifest: {manifest_path(args)}") # ═══════════════════════════════════════════════════════════════════════════ # Per-file analysis # ═══════════════════════════════════════════════════════════════════════════ SYSTEM_PROMPT = """\ You are a senior application security engineer performing a focused, single-file code review for a coordinated vulnerability-disclosure effort on open source software. You review ONE file at a time, deliberately, for depth. Find real, exploitable vulnerabilities. Prioritize the suspected classes given, but report anything genuine — your full security knowledge is in scope (CSRF, logic flaws, race conditions, SSTI, mass assignment, etc.), not only the hints. For EACH finding assess: - severity: Critical|High|Medium|Low|Info (CVSS-aligned) - effort: Trivial(<1hr)|Low(1-4hr)|Medium(1-2d)|High(>2d) to PATCH - confidence: high|medium|low (lower it when cross-file context is missing) - owasp + cwe: best-matching OWASP Top 10 category and CWE id(s) - validation_pointer: the CONCRETE place to aim a downstream tool — the likely endpoint/route, the parameter name, and/or the function — so a human can verify it on their own instance. Be as specific as the code allows. Optimize the reader's attention toward HIGH-severity, LOW-effort issues. Output ONLY a JSON object (no prose, no fences) matching exactly: { "file_summary": "", "findings": [ { "title": "", "vuln_class": "", "severity": "<...>", "effort": "<...>", "confidence": "<...>", "owasp": "", "cwe": ["CWE-89"], "lines": "", "what": "", "why_exploitable": " sink -> trust boundary; observed behavior only>", "validation_pointer": "", "evidence": "", "proposed_fix": "", "missing_context_caveat": "" } ] } Return "findings": [] if none. """ USER_TEMPLATE = """\ {glossary} SUSPECTED CLASSES (from heuristic pre-ranking — confirm or dismiss): {suspected} LANGUAGE PROFILE: {profile} FILE: {path} ``` {code} ```""" def severity_points(s): return {"critical":5,"high":4,"medium":3,"low":2,"info":1}.get(s.lower(),1) def effort_points(e): return {"trivial":4,"low":3,"medium":2,"high":1}.get(e.lower(),1) def priority_score(s,e): return severity_points(s) * effort_points(e) def call_claude(api_key, model_string, system, user_msg): payload = {"model": model_string, "max_tokens": MAX_OUTPUT_TOKENS, "system": system, "messages": [{"role": "user", "content": user_msg}]} headers = {"x-api-key": api_key, "anthropic-version": API_VERSION, "content-type": "application/json"} for attempt in range(1, RETRY_ATTEMPTS + 1): try: r = requests.post(API_URL, headers=headers, json=payload, timeout=REQUEST_TIMEOUT) if r.status_code in (429, 529): wait = int(r.headers.get("retry-after", 15)) print(f" rate-limited; waiting {wait}s ({attempt}/{RETRY_ATTEMPTS})") time.sleep(wait); continue r.raise_for_status() data = r.json() text = "".join(b.get("text","") for b in data.get("content",[]) if b.get("type")=="text") u = data.get("usage", {}) return text, u.get("input_tokens",0), u.get("output_tokens",0), data.get("stop_reason") except requests.exceptions.HTTPError as e: if r.status_code < 500: print(f" client error {r.status_code}: {e} — not retrying") return None, 0, 0, None print(f" server error {r.status_code}: {e} ({attempt}/{RETRY_ATTEMPTS})") time.sleep(5) except requests.exceptions.Timeout: print(f" timeout ({attempt}/{RETRY_ATTEMPTS})"); time.sleep(5) except Exception as e: print(f" error: {e} ({attempt}/{RETRY_ATTEMPTS})"); time.sleep(5) return None, 0, 0, None def parse_findings(text): if not text: return None try: return json.loads(text.strip()) except json.JSONDecodeError: pass cleaned = re.sub(r"^```(?:json)?|```$", "", text.strip(), flags=re.MULTILINE).strip() try: return json.loads(cleaned) except json.JSONDecodeError: m = re.search(r"\{.*\}", cleaned, re.DOTALL) if m: try: return json.loads(m.group(0)) except json.JSONDecodeError: return None return None def estimate_tokens(text): return len(text) // 4 def cmd_run(args): if not args.repo: sys.exit("ERROR: No target repo specified.\n" " Use: python VulscanNervePinch.py run --repo /path/to/your/repo\n" " Or set DEFAULT_REPO near the top of VulscanNervePinch.py.") repo = Path(args.repo).expanduser().resolve() if args.top < 1: sys.exit("ERROR: --top must be >= 1.") if args.max_cost is not None and args.max_cost <= 0: sys.exit("ERROR: --max-cost must be a positive dollar amount.") manifest = load_manifest(args) if not manifest["files"]: sys.exit("No ranked files. Run `rank` first.") alias = args.model or manifest.get("model_default") or DEFAULT_MODEL if alias not in MODELS: sys.exit(f"Unknown model '{alias}'. Options: {', '.join(MODELS)}") model_string, price_in, price_out = MODELS[alias] if args.retry_errors: for m in manifest["files"].values(): if m["status"] == "error": m["status"] = "pending" save_manifest(args, manifest) pending = [(rel, m) for rel, m in manifest["files"].items() if m["status"] == "pending"] pending.sort(key=lambda kv: kv[1]["priority_score"], reverse=True) batch = pending[: args.top] if not batch: print("Nothing pending. Run `report`."); return if args.dry_run: print(f"DRY RUN — {alias}, top {args.top} of {len(pending)} pending:\n") est_in = est_out = 0 for rel, m in batch: try: code = (repo / rel).read_text("utf-8", "ignore") except OSError: code = "" est_in += estimate_tokens(SYSTEM_PROMPT + TARGET_PROFILE["safe_glossary"] + code) + 120 est_out += MAX_OUTPUT_TOKENS // 3 print(f" [{m['priority_score']:>6.1f}] {rel} ({', '.join(m['suspected_classes']) or '—'})") cost = est_in/1e6*price_in + est_out/1e6*price_out print(f"\n ~{est_in:,} in + ~{est_out:,} out tokens → ~${cost:,.2f} (rough)") return api_key = os.environ.get("ANTHROPIC_API_KEY") if not api_key: sys.exit("ERROR: set ANTHROPIC_API_KEY.") findings_dir = Path(output_dir(args)) / FINDINGS_SUBDIR findings_dir.mkdir(parents=True, exist_ok=True) max_cost = args.max_cost if args.max_cost is not None else DEFAULT_MAX_COST print(f"==> Analyzing {len(batch)} files with {alias} (${price_in}/${price_out} per Mtok)\n") run_cost = 0.0 for i, (rel, m) in enumerate(batch, 1): if max_cost is not None: if run_cost >= max_cost: print(f"\n Cost cap ${max_cost:.2f} reached (${run_cost:.2f}). Remaining stay pending.") break # Estimate next file's cost; skip if it would exceed the cap. try: est_size = (repo / rel).stat().st_size except OSError: est_size = 4000 est_in = est_size // 4 + estimate_tokens(SYSTEM_PROMPT) + 120 est_out = MAX_OUTPUT_TOKENS // 3 est_cost = est_in / 1e6 * price_in + est_out / 1e6 * price_out if run_cost + est_cost > max_cost: print(f"\n Next file ~${est_cost:.3f} would exceed cap ${max_cost:.2f} " f"(spent ${run_cost:.2f}). Stopping early.") break try: code = (repo / rel).read_text("utf-8", "ignore") except OSError: m["status"] = "error"; continue user_msg = USER_TEMPLATE.format(glossary=TARGET_PROFILE["safe_glossary"], suspected=", ".join(m["suspected_classes"]) or "none", profile=m["signals"].get("profile", profile_for(rel)), path=rel, code=code) print(f" [{i}/{len(batch)}] {rel} (priority {m['priority_score']})", flush=True) text, tin, tout, stop_reason = call_claude(api_key, model_string, SYSTEM_PROMPT, user_msg) cost = tin/1e6*price_in + tout/1e6*price_out run_cost += cost parsed = parse_findings(text) if parsed is None: m.update(status="error", input_tokens=tin, output_tokens=tout, cost_usd=round(cost,4)) # Save raw response for debugging. raw_path = findings_dir / f"{re.sub(r'[^\w.-]', '_', rel)}.raw.txt" raw_path.write_text(text or "(empty response)", encoding="utf-8") if stop_reason == "max_tokens": print(f" ⚠ response truncated (hit {MAX_OUTPUT_TOKENS} token limit) — " f"consider increasing MAX_OUTPUT_TOKENS or splitting the file. " f"Raw saved to {raw_path.name}") else: print(f" ⚠ unparseable response — saved to {raw_path.name}") save_manifest(args, manifest); continue findings = parsed.get("findings", []) for f in findings: f["priority_score"] = priority_score(f.get("severity","Info"), f.get("effort","High")) cls = VULN_CLASSES.get(f.get("vuln_class", ""), {}) f["recommended_tools"] = cls.get("tools", []) if not f.get("owasp"): f["owasp"] = cls.get("owasp", "") if not f.get("cwe"): f["cwe"] = cls.get("cwe", []) findings.sort(key=lambda f: f.get("priority_score",0), reverse=True) md_path, json_path = write_finding_files(findings_dir, rel, m, parsed, findings, alias) m.update(status="analyzed", analyzed_at=now_iso(), model=alias, findings_count=len(findings), report_md=str(md_path), report_json=str(json_path), input_tokens=tin, output_tokens=tout, cost_usd=round(cost,4)) flag = " ⚑" if any(f.get("severity","").lower() in ("critical","high") for f in findings) else "" print(f" {len(findings)} finding(s), ${cost:.3f}{flag}") save_manifest(args, manifest) time.sleep(INTER_REQUEST_SLEEP) recompute_totals(manifest) save_manifest(args, manifest) t = manifest["totals"] print(f"\n==> Run done. This run: ${run_cost:.2f}. " f"Totals: {t['analyzed']} analyzed / {t['pending']} pending, cumulative ${t['cost_usd']:.2f}") def render_tool_commands(tool_ids, pointer): """Render ready-to-run validation commands for a finding's tools.""" out = [] for tid in tool_ids: t = TOOLS.get(tid) if not t: continue out.append(f" - **{t['name']}** ({t['category']}, targets {t['targets']}): " f"`{t['invocation']}`") if pointer: out.append(f" - _Aim at:_ {pointer}") return out def write_finding_files(findings_dir, rel, meta, parsed, findings, alias): slug = re.sub(r"[^\w.-]", "_", rel) json_path = findings_dir / f"{slug}.json" md_path = findings_dir / f"{slug}.md" json_path.write_text(json.dumps({ "file": rel, "priority_score": meta["priority_score"], "suspected_classes": meta["suspected_classes"], "model": alias, "analyzed_at": now_iso(), "file_summary": parsed.get("file_summary",""), "findings": findings}, indent=2), encoding="utf-8") lines = [f"# Security review: `{rel}`", "", f"- **Model:** {alias}", f"- **Heuristic priority:** {meta['priority_score']} " f"(suspected: {', '.join(meta['suspected_classes']) or '—'})", f"- **Summary:** {parsed.get('file_summary','')}", ""] if not findings: lines.append("_No findings (single-file scope)._") for n, f in enumerate(findings, 1): lines += [ f"## {n}. {f.get('title','(untitled)')}", "", f"- **Class:** {f.get('vuln_class','?')} | **OWASP:** {f.get('owasp','?')} | " f"**CWE:** {', '.join(f.get('cwe',[]))}", f"- **Severity:** {f.get('severity','?')} | **Effort:** {f.get('effort','?')} | " f"**Confidence:** {f.get('confidence','?')} | **Priority:** {f.get('priority_score','?')}", f"- **Lines:** {f.get('lines','?')}", "", f"**What:** {f.get('what','')}", "", f"**Why exploitable:** {f.get('why_exploitable','')}", "", f"**Validation pointer:** {f.get('validation_pointer','(none)')}", "", "**Validate with:**"] + render_tool_commands(f.get("recommended_tools",[]), f.get("validation_pointer","")) + [ "", f"**Evidence:**", "```", f.get("evidence",""), "```", "", f"**Proposed fix:** {f.get('proposed_fix','')}", ""] if f.get("missing_context_caveat"): lines += [f"> ⚠ Cross-file caveat: {f['missing_context_caveat']}", ""] md_path.write_text("\n".join(lines), encoding="utf-8") return md_path, json_path # ═══════════════════════════════════════════════════════════════════════════ # Catalog / Status / Report # ═══════════════════════════════════════════════════════════════════════════ def cmd_catalog(args): """Print the full detection + tooling registry — what we look for and how to validate.""" print("=" * 70) print(" VULNERABILITY CLASS REGISTRY (what we look for, where, and with what)") print("=" * 70) for cid, c in VULN_CLASSES.items(): langs = [p for p in c["signals"]] if c["signals"] else ["(tooling-only — no per-file signal)"] tools = ", ".join(TOOLS[t]["name"] for t in c["tools"] if t in TOOLS) print(f"\n* {c['name']} [{cid}]") print(f" OWASP: {c['owasp']} CWE: {', '.join(c['cwe'])}") print(f" Baseline severity: {c['severity_baseline']} Rank weight: {c['weight']}") print(f" {c['description']}") print(f" Detected in: {', '.join(langs)}") print(f" Validate with: {tools}") print(f" How: {c['validation_hint']}") print("\n" + "=" * 70) print(" DOWNSTREAM TOOL REGISTRY (grouped by category)") print("=" * 70) by_cat = {} for tid, t in TOOLS.items(): by_cat.setdefault(t["category"], []).append((tid, t)) for cat in ["SAST", "DAST", "DEPS", "SECRETS", "IAC"]: if cat not in by_cat: continue print(f"\n[{cat}]") for tid, t in by_cat[cat]: print(f" * {t['name']:<14} targets {t['targets']}") print(f" install: {t['install']}") print(f" run: {t['invocation']}") print(f" {t['note']}") print(f"\nLanguages with detection signals: {', '.join(sorted(LANGUAGE_PROFILES.keys()))}") print(f"Target profile: {TARGET_PROFILE['name']} " f"(edit TARGET_PROFILE to retarget; the engine is reusable)") def cmd_status(args): manifest = load_manifest(args) if not manifest["files"]: sys.exit("No manifest yet. Run `rank` first.") if args.top < 1: sys.exit("ERROR: --top must be >= 1.") recompute_totals(manifest) t = manifest["totals"] print(f"Repo: {manifest.get('repo_path','?')}") print(f"Files: {t['files_total']} total — analyzed {t['analyzed']}, " f"pending {t['pending']}, errors {t['errors']}") print(f"Spend: ${t['cost_usd']:.2f} ({t['input_tokens']:,} in / {t['output_tokens']:,} out)") print(f"Findings: {sum(m['findings_count'] for m in manifest['files'].values())}") pending = [(rel, m) for rel, m in manifest["files"].items() if m["status"] == "pending"] pending.sort(key=lambda kv: kv[1]["priority_score"], reverse=True) n = args.top or 10 print(f"\nNext {min(n, len(pending))} by priority:") for rel, m in pending[:n]: print(f" [{m['priority_score']:>6.1f}] {rel} ({', '.join(m['suspected_classes']) or '—'})") def cmd_report(args): manifest = load_manifest(args) all_f = [] for rel, meta in manifest["files"].items(): jp = meta.get("report_json") if not jp or not Path(jp).exists(): continue for f in json.loads(Path(jp).read_text()).get("findings", []): f["_file"] = rel; all_f.append(f) all_f.sort(key=lambda f: f.get("priority_score", 0), reverse=True) sev_c, eff_c, owasp_c = {}, {}, {} for f in all_f: sev_c[f.get("severity","?")] = sev_c.get(f.get("severity","?"),0)+1 eff_c[f.get("effort","?")] = eff_c.get(f.get("effort","?"),0)+1 owasp_c[f.get("owasp","?")] = owasp_c.get(f.get("owasp","?"),0)+1 o = [f"# Security Audit Report", "", f"_Generated {now_iso()} from {manifest.get('repo_path','?')}_", f"_Spend so far: ${manifest['totals']['cost_usd']:.2f} across " f"{manifest['totals']['analyzed']} files; {manifest['totals']['pending']} pending._", "", "## Summary", "", f"- **{len(all_f)}** findings across {manifest['totals']['analyzed']} analyzed files.", ""] o += ["| Severity | Count | | Effort | Count |", "|---|---|---|---|---|"] so, eo = ["Critical","High","Medium","Low","Info"], ["Trivial","Low","Medium","High"] for i in range(max(len(so), len(eo))): s = so[i] if i < len(so) else ""; e = eo[i] if i < len(eo) else "" o.append(f"| {s} | {sev_c.get(s,'') if s else ''} | | {e} | {eff_c.get(e,'') if e else ''} |") o += ["", "## OWASP Top 10 coverage", "", "| OWASP category | Findings |", "|---|---|"] for cat, n in sorted(owasp_c.items(), key=lambda kv: -kv[1]): o.append(f"| {cat} | {n} |") o.append("\n_Note: A05 Misconfiguration and A06 Vulnerable Components are assessed by the " "dependency/IaC/secrets tools (see each finding's validation block), " "not by per-file code review._") fix_first = [f for f in all_f if f.get("priority_score",0) >= 12] o += ["", "## Fix First — high severity, low effort (Priority >= 12)", ""] if fix_first: o += ["| Score | Sev | Effort | OWASP | Finding | File |", "|---|---|---|---|---|---|"] for f in fix_first: o.append(f"| {f.get('priority_score')} | {f.get('severity')} | {f.get('effort')} | " f"{f.get('owasp','')} | {f.get('title')} | `{f['_file']}` |") else: o.append("_None yet._") o += ["", "## All findings (by priority)", "", "| Score | Sev | Effort | Conf | Class | Finding | File |", "|---|---|---|---|---|---|---|"] for f in all_f: o.append(f"| {f.get('priority_score')} | {f.get('severity')} | {f.get('effort')} | " f"{f.get('confidence')} | {f.get('vuln_class')} | {f.get('title')} | `{f['_file']}` |") o += ["", "## Detailed findings + validation playbook", ""] for f in all_f: o += [f"### [{f.get('priority_score')}] {f.get('title')} — `{f['_file']}`", f"*{f.get('severity')} severity · {f.get('effort')} effort · {f.get('confidence')} confidence · " f"{f.get('owasp','')} · {', '.join(f.get('cwe',[]))} · lines {f.get('lines')}*", "", f"**What:** {f.get('what')}", "", f"**Why exploitable:** {f.get('why_exploitable')}", "", f"**Validation pointer:** {f.get('validation_pointer','(none)')}", "", "**Validate with:**"] + render_tool_commands(f.get("recommended_tools",[]), f.get("validation_pointer","")) + [ "", f"**Proposed fix:** {f.get('proposed_fix')}", ""] if f.get("missing_context_caveat"): o.append(f"> ⚠ {f['missing_context_caveat']}\n") o.append("---") o += ["", "## Scoring method", "", "Each finding's **priority score = severity × effort-to-fix**, designed to surface " "high-impact, easy-to-patch issues first.", "", "| Severity | Points | | Effort to fix | Points |", "|---|---|---|---|---|", "| Critical | 5 | | Trivial (<1hr) | 4 |", "| High | 4 | | Low (1-4hr) | 3 |", "| Medium | 3 | | Medium (1-2d) | 2 |", "| Low | 2 | | High (>2d) | 1 |", "| Info | 1 | | | |", "", "A Critical + Trivial finding scores 5×4 = 20 (fix first). " "An Info + High-effort finding scores 1×1 = 1 (low priority).", "", "## Method & caveats", "", "Findings are single-file static analysis, AI-assisted, pending live validation on an " "instance you OWN. Per-file scope can miss cross-file sanitization — heed `low` confidence " "and missing-context flags. Use the validation playbook to confirm before disclosure. " "Run `VulscanNervePinch.py catalog` to see the full detection + tooling registry."] report_path = Path(output_dir(args)) / REPORT_NAME report_path.write_text("\n".join(o), encoding="utf-8") print(f"==> Report: {report_path} ({len(all_f)} findings, {len(fix_first)} Fix-First)") # ═══════════════════════════════════════════════════════════════════════════ # Manifest helpers + CLI # ═══════════════════════════════════════════════════════════════════════════ def output_dir(args): return Path(args.output).expanduser().resolve() def manifest_path(args): return output_dir(args) / MANIFEST_NAME def load_manifest(args): p = manifest_path(args) if p.exists(): return json.loads(p.read_text()) return {"repo_path": None, "created": now_iso(), "model_default": args.model or DEFAULT_MODEL, "files": {}, "totals": {}} def save_manifest(args, manifest): p = manifest_path(args); p.parent.mkdir(parents=True, exist_ok=True) tmp = p.with_suffix(".tmp") tmp.write_text(json.dumps(manifest, indent=2), encoding="utf-8") tmp.replace(p) def recompute_totals(manifest): fs = manifest["files"].values() manifest["totals"] = { "files_total": len(manifest["files"]), "analyzed": sum(1 for m in fs if m["status"]=="analyzed"), "pending": sum(1 for m in fs if m["status"]=="pending"), "errors": sum(1 for m in fs if m["status"]=="error"), "input_tokens": sum(m["input_tokens"] for m in fs), "output_tokens": sum(m["output_tokens"] for m in fs), "cost_usd": round(sum(m["cost_usd"] for m in fs), 4)} def now_iso(): return datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC") def build_parser(): p = argparse.ArgumentParser(description="General-purpose per-file vulnerability triage engine.") common = argparse.ArgumentParser(add_help=False) common.add_argument("--repo", default=DEFAULT_REPO) common.add_argument("--output", default=DEFAULT_OUTPUT_DIR) common.add_argument("--model", default=None, help=f"alias: {', '.join(MODELS)}") sub = p.add_subparsers(dest="cmd", required=True) sp = sub.add_parser("catalog", parents=[common], help="print detection + tool registry") sp.set_defaults(func=cmd_catalog) sp = sub.add_parser("rank", parents=[common], help="score/queue files (no API cost)") sp.set_defaults(func=cmd_rank) sp = sub.add_parser("status", parents=[common], help="show queue + spend") sp.add_argument("--top", type=int, default=10) sp.set_defaults(func=cmd_status) sp = sub.add_parser("run", parents=[common], help="analyze the next N files") sp.add_argument("--top", type=int, default=DEFAULT_TOP_N, help=f"analyze the top N files (default {DEFAULT_TOP_N}, must be >= 1)") sp.add_argument("--max-cost", type=float, default=None) sp.add_argument("--dry-run", action="store_true") sp.add_argument("--retry-errors", action="store_true", help="re-queue files marked 'error' so they are retried") sp.set_defaults(func=cmd_run) sp = sub.add_parser("report", parents=[common], help="compile ranked report + playbook") sp.set_defaults(func=cmd_report) return p if __name__ == "__main__": args = build_parser().parse_args() try: args.func(args) except KeyboardInterrupt: print("\n\nInterrupted. Progress from completed files was saved automatically.") sys.exit(1)