From 63f33687f8f091be754b3fe41cb69fba76091034 Mon Sep 17 00:00:00 2001 From: choom Date: Mon, 12 Oct 2026 04:01:36 +0000 Subject: [PATCH] Upload files to "/" Add VulscanNervePinch vulnerability-triage engine --- VulscanNervePinch.py | 1088 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 1088 insertions(+) create mode 100644 VulscanNervePinch.py diff --git a/VulscanNervePinch.py b/VulscanNervePinch.py new file mode 100644 index 0000000..1d7a8fb --- /dev/null +++ b/VulscanNervePinch.py @@ -0,0 +1,1088 @@ +#!/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) \ No newline at end of file