1088 lines
60 KiB
Python
1088 lines
60 KiB
Python
#!/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
|
||
# <URL>, <param>, <repo>, <file> 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 '<URL>?<param>=1' -p <param> --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 '<URL>?<param>=1' -p <param> --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 '<URL>?<param>=test'",
|
||
"note": "XSS scanner/verifier."},
|
||
"commix": {"name": "commix", "category": "DAST", "targets": "live endpoint + parameter",
|
||
"install": "pipx install commix",
|
||
"invocation": "commix -u '<URL>' --data='<param>=1'",
|
||
"note": "Command-injection confirmation."},
|
||
"tplmap": {"name": "tplmap", "category": "DAST", "targets": "live endpoint + parameter",
|
||
"install": "git clone github.com/epinna/tplmap",
|
||
"invocation": "tplmap -u '<URL>?<param>=*'",
|
||
"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 <param> -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 '<URL>' (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 '<URL>' -tags <tag>",
|
||
"note": "Huge community template library; known-CVE + misconfig checks."},
|
||
"nikto": {"name": "Nikto", "category": "DAST", "targets": "base URL",
|
||
"install": "apt install nikto",
|
||
"invocation": "nikto -h '<URL>'",
|
||
"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 '<URL>/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 <URL>",
|
||
"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/<ruleset> <repo> (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 <repo>",
|
||
"note": "Python SAST."},
|
||
"brakeman": {"name": "Brakeman", "category": "SAST", "targets": "rails repo",
|
||
"install": "gem install brakeman",
|
||
"invocation": "brakeman <repo>",
|
||
"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=<lang> && 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 <repo>",
|
||
"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 <repo> / trivy image <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:<repo>",
|
||
"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 <repo>",
|
||
"note": "Secrets across full git history."},
|
||
"trufflehog": {"name": "TruffleHog", "category": "SECRETS", "targets": "repo + git history",
|
||
"install": "pipx install trufflehog (or binary)",
|
||
"invocation": "trufflehog filesystem <repo>",
|
||
"note": "Secrets with live-credential verification."},
|
||
# --- IAC / containers ---------------------------------------------------
|
||
"checkov": {"name": "Checkov", "category": "IAC", "targets": "IaC / Dockerfiles / k8s",
|
||
"install": "pipx install checkov",
|
||
"invocation": "checkov -d <repo>",
|
||
"note": "Terraform/CloudFormation/k8s/Dockerfile misconfig."},
|
||
"hadolint": {"name": "hadolint", "category": "IAC", "targets": "Dockerfile",
|
||
"install": "github.com/hadolint/hadolint releases",
|
||
"invocation": "hadolint <Dockerfile>",
|
||
"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": "<one sentence>",
|
||
"findings": [
|
||
{
|
||
"title": "<concise>",
|
||
"vuln_class": "<sqli|command_injection|xss|ssrf_file_xxe|access_control|auth_session|deserialization|crypto|secrets|ssti|other>",
|
||
"severity": "<...>", "effort": "<...>", "confidence": "<...>",
|
||
"owasp": "<e.g. A03:2021 Injection>", "cwe": ["CWE-89"],
|
||
"lines": "<approx>",
|
||
"what": "<the vulnerability>",
|
||
"why_exploitable": "<data flow: input source -> sink -> trust boundary; observed behavior only>",
|
||
"validation_pointer": "<endpoint/route + parameter + function to aim a tool at>",
|
||
"evidence": "<minimal code excerpt>",
|
||
"proposed_fix": "<concrete corrected pattern or diff>",
|
||
"missing_context_caveat": "<name any unseen code your confidence depends on, else ''>"
|
||
}
|
||
]
|
||
}
|
||
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) |