evandrodevbr
c17192c9fe
Copia local (nao fork, nao submodule) de safedeps, sca-audit, secrets-scan, code-review-security, api-security-review, web-security-review, cve-triage, patch-prioritization, dependency-scanning e hermaguard -- commit exato de origem e licenca de cada uma em skills/NOTICE.md. Elimina o passo de clone externo pro pilar de seguranca do wf-security-review: git clone deste repo ja traz tudo funcional. Corrige tambem o comando de instalacao pro Claude Code nos dois READMEs e no docs/SKILLS.md: Claude Code exige um diretorio por skill (skills/<nome>/SKILL.md), nao aceita os .md soltos que o comando antigo copiava -- confirmado testando a instalacao real.
240 lines
10 KiB
Python
240 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Hermaguard Locate — hypothesis-driven fault localization pre-pass.
|
|
|
|
SHERLOC-style (arXiv 2606.24820): SHERLOC pairs a reasoning LLM with compact
|
|
repo tools (grep, AST, git blame) to produce actionable fault diagnoses —
|
|
not just file paths but the diagnostic context a repair agent needs.
|
|
Its SWE-Bench numbers (84.3% acc@1 Lite) show the localization pass is
|
|
worth doing BEFORE adversarial review.
|
|
|
|
This tool is the COMPACT-REPO-TOOLS layer, fully deterministic and stdlib-only:
|
|
1. Churn analysis — git diff --numstat (fallback: plain diff parsing)
|
|
2. Hotspot scoring — hunk-level concentration of added lines
|
|
3. Signal mapping — per-class keyword/pattern hits (SQL, auth, races,
|
|
deserialization, async, I/O, config) mapped to the
|
|
hermaguard coverage classes
|
|
4. Hypothesis list — top-N ranked suspicion points with WHY each was
|
|
flagged, ready for the review agents to attack
|
|
|
|
The reasoning loop stays with the review agents (SKILL.md Phase 1): they
|
|
receive the hypothesis list as their starting point and attack from known
|
|
weak points instead of re-deriving context from scratch.
|
|
|
|
Usage:
|
|
hermaguard-locate.py --diff <diff-file-or-'-'> --repo <root> [--top N]
|
|
hermaguard-locate.py --files a.py,b.py --repo <root>
|
|
# reads a unified diff from stdin when --diff -
|
|
|
|
Output JSON: {churn, hotspots, signals, hypothesis, top_files, stats}
|
|
|
|
Exit 0 always on successful analysis; 2 on input errors.
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
VERSION = "1.0.0"
|
|
|
|
# Signal map: coverage class -> list of (label, regex) triggers.
|
|
# These are deliberately conservative: they flag *suspicion*, not findings.
|
|
SIGNAL_MAP = [
|
|
("CWE-89", "sql-injection", [r"\b(f-string|format|%)[^\n]*(SELECT|INSERT|UPDATE|DELETE)", r"\bcur(?:sor)?\.execute\(f", r"\.query\(f[^)]*"]),
|
|
("CWE-79", "xss", [r"dangerouslySetInnerHTML", r"innerHTML\s*=", r"v-html", r"\.html\("]),
|
|
("CWE-22", "path-traversal", [r"os\.path\.join\([^)]*(user|filename|name)", r"open\(f?['\"](?!.*\.\.)", r"Path\([^)]*user"]),
|
|
("CWE-78", "command-injection", [r"os\.system\(f", r"subprocess\.(run|Popen|call)\([^)]*shell\s*=\s*True", r"subprocess\.(run|Popen|call)\(f"]),
|
|
("CWE-502", "deserialization", [r"pickle\.load", r"yaml\.load\(", r"loads?\([^)]*\)\s*#?\s*.*(untrusted|user|request)", r"eval\(f?"]),
|
|
("CWE-798", "hardcoded-creds", [r"(password|passwd|secret|api_key|apikey|token)\s*=\s*['\"][^'\"]{6,}['\"]"]),
|
|
("CWE-362", "race-toctou", [r"os\.path\.exists|os\.access", r"(thread|Thread|asyncio\.gather|await asyncio\.(sleep|wait))", r"check[^\n]*then[^\n]*(write|delete|move)", r"\.balance\s*[+\-]=|balance\s*[+\-]=.*if\s"]),
|
|
("HG-ASYNC", "async-gap", [r"await\s+\w+\([^)]*\)\s*\n(?!.*except)", r"\.catch\(", r"async\s+\w+\([^)]*\)\s*\{[^}]*$"]),
|
|
("CWE-200", "info-exposure", [r"traceback\.format_exc", r"print\([^)]*(token|password|secret|key)", r"return\s+str\(e\)"]),
|
|
("CWE-918", "ssrf", [r"requests\.(get|post|put|delete)\([^)]*(url|user)", r"urllib\.request", r"fetch\([^)]*(url|req)"]),
|
|
("CWE-476", "null-deref", [r"\.fetchone\(\)[^\n]*\[0\]", r"row\[0\]", r"\[0\]\s*(\.|$)", r"\bget\([^)]*\)\[0\]"]),
|
|
("CWE-434", "unrestricted-upload", [r"\.save\([^)]*(filename|file|name)", r"upload", r"content_type\s*="]),
|
|
("CWE-306", "missing-auth", [r"@app\.route|@router\.(get|post|put|delete)", r"def\s+\w+\([^)]*request\)\s*:"]) ,
|
|
("HG-PARTIAL-WRITE", "partial-write", [r"\.write\([^)]*\)\s*\n(?!.*(flush|close|commit))", r"(update|insert).*(where|SET)", r"without\s+(transaction|commit)"]),
|
|
("CWE-400", "resource-exhaustion", [r"while\s+True", r"recursive", r"\.read\(\)\s*$", r"all\(|any\(", r"\*\s*1000|\*\s*1024"]),
|
|
("CWE-190", "integer-overflow", [r"\+=\s*1" , r"max_|min_", r"len\([^)]*\)\s*\+"]),
|
|
]
|
|
|
|
CLASS_SIGNALS: dict = {}
|
|
for _cls, _label, _pats in SIGNAL_MAP:
|
|
CLASS_SIGNALS.setdefault(_cls, []).extend((_label, re.compile(p, re.IGNORECASE)) for p in _pats)
|
|
|
|
# Files never worth hot-spotting (vendored/generated/binary-ish)
|
|
IGNORED_SUFFIXES = {".pyc", ".min.js", ".map", ".lock", ".svg", ".png", ".jpg", ".woff", ".woff2"}
|
|
|
|
|
|
def git_numstat(repo: str, files: list) -> dict:
|
|
"""Churn per file via git diff --numstat. Returns {path: {added, deleted}}."""
|
|
if not files:
|
|
return {}
|
|
try:
|
|
r = subprocess.run(
|
|
["git", "-C", repo, "diff", "--numstat", "--"] + list(files),
|
|
capture_output=True, text=True, timeout=30)
|
|
if r.returncode != 0:
|
|
return {}
|
|
out = {}
|
|
for line in r.stdout.splitlines():
|
|
parts = line.split("\t")
|
|
if len(parts) == 3 and parts[0].isdigit():
|
|
out[parts[2]] = {"added": int(parts[0]), "deleted": int(parts[1])}
|
|
return out
|
|
except (subprocess.SubprocessError, OSError):
|
|
return {}
|
|
|
|
|
|
def parse_diff(diff_text: str) -> list:
|
|
"""Parse a unified diff into per-file added-line spans.
|
|
Returns [{file, hunks: [{start, added_lines: [line_no,...]}]}]"""
|
|
files_out = []
|
|
cur_file = None
|
|
cur_hunk = None
|
|
for raw in diff_text.splitlines():
|
|
if raw.startswith("+++ b/"):
|
|
cur_file = raw[6:]
|
|
files_out.append({"file": cur_file, "hunks": []})
|
|
cur_hunk = None
|
|
elif raw.startswith("@@") and cur_file is not None:
|
|
m = re.search(r"\+(\d+)(?:,(\d+))?", raw)
|
|
if m:
|
|
cur_hunk = {"start": int(m.group(1)), "added_lines": []}
|
|
files_out[-1]["hunks"].append(cur_hunk)
|
|
elif cur_hunk is not None:
|
|
if raw.startswith("+") and not raw.startswith("+++"):
|
|
cur_hunk["added_lines"].append(cur_hunk["start"] + len(cur_hunk["added_lines"]))
|
|
elif raw.startswith("-") and not raw.startswith("---"):
|
|
pass # deletions shift nothing for added-line numbering
|
|
else:
|
|
cur_hunk["start"] += 1 # context line advances the new-file line
|
|
return [f for f in files_out if f["hunks"]]
|
|
|
|
|
|
def scan_signals(file_path: str, text: str) -> list:
|
|
"""Return [{class_id, label, line, snippet}] for every signal hit."""
|
|
hits = []
|
|
if any(file_path.endswith(s) for s in IGNORED_SUFFIXES):
|
|
return hits
|
|
for cls, entries in CLASS_SIGNALS.items():
|
|
for label, rx in entries:
|
|
for m in rx.finditer(text):
|
|
line_no = text[:m.start()].count("\n") + 1
|
|
snippet = text.splitlines()[line_no - 1].strip()[:120]
|
|
hits.append({"class_id": cls, "label": label,
|
|
"line": line_no, "snippet": snippet})
|
|
return hits
|
|
|
|
|
|
def score_file(path: str, churn: dict, hits: list) -> dict:
|
|
"""Hotspot score: churn concentration + unique signal classes."""
|
|
c = churn.get(path, {"added": 0, "deleted": 0})
|
|
classes = {h["class_id"] for h in hits}
|
|
score = min(c["added"] // 5, 5) + min(len(classes) * 2, 10)
|
|
if c["added"] == 0 and not hits:
|
|
score = 0
|
|
suspicion = "HIGH" if score >= 8 else ("MEDIUM" if score >= 4 else "LOW")
|
|
return {
|
|
"path": path,
|
|
"added": c["added"],
|
|
"deleted": c["deleted"],
|
|
"signal_hits": len(hits),
|
|
"signal_classes": sorted(classes),
|
|
"score": score,
|
|
"suspicion": suspicion,
|
|
"top_signals": sorted(hits, key=lambda h: -h["line"])[:5],
|
|
}
|
|
|
|
|
|
def build_hypotheses(scored: list, top_n: int) -> list:
|
|
"""Ranked attack list with explicit WHY notes for the review agents."""
|
|
hyps = []
|
|
for s in sorted(scored, key=lambda x: -x["score"])[:top_n]:
|
|
why = []
|
|
if s["added"] >= 10:
|
|
why.append(f"{s['added']} added lines — large change surface")
|
|
if s["signal_classes"]:
|
|
why.append("signal classes: " + ", ".join(s["signal_classes"]))
|
|
if s["deleted"] > s["added"]:
|
|
why.append("net deletion — check for removed guards")
|
|
hyps.append({
|
|
"file": s["path"],
|
|
"suspicion": s["suspicion"],
|
|
"why": why,
|
|
"attack_from": [h["snippet"] for h in s["top_signals"]],
|
|
})
|
|
return hyps
|
|
|
|
|
|
def analyze(files: list, repo: str, top_n: int) -> dict:
|
|
churn = git_numstat(repo, files)
|
|
scored = []
|
|
for f in files:
|
|
p = Path(f)
|
|
if not p.is_file():
|
|
continue
|
|
try:
|
|
text = p.read_text(errors="replace")
|
|
except OSError:
|
|
continue
|
|
hits = scan_signals(f, text)
|
|
scored.append(score_file(f, churn, hits))
|
|
scored = [s for s in scored if s["score"] > 0]
|
|
hypotheses = build_hypotheses(scored, top_n)
|
|
return {
|
|
"tool": "hermaguard-locate",
|
|
"version": VERSION,
|
|
"hypotheses": hypotheses,
|
|
"top_files": [h["file"] for h in hypotheses],
|
|
"stats": {
|
|
"files_scanned": len(files),
|
|
"files_flagged": len(scored),
|
|
"signal_hits_total": sum(s["signal_hits"] for s in scored),
|
|
},
|
|
}
|
|
|
|
|
|
def main():
|
|
p = argparse.ArgumentParser(description="Hermaguard locate pre-pass — fault localization")
|
|
p.add_argument("--diff", help="Unified diff file, or '-' for stdin")
|
|
p.add_argument("--files", help="Comma-separated file paths")
|
|
p.add_argument("--repo", default=".", help="Repo root for git churn (default: cwd)")
|
|
p.add_argument("--top", type=int, default=5, help="Top-N hypotheses (default: 5)")
|
|
p.add_argument("--json", dest="json_out", help="Write output JSON to path")
|
|
args = p.parse_args()
|
|
|
|
if not args.diff and not args.files:
|
|
p.error("--diff or --files required")
|
|
|
|
if args.diff == "-":
|
|
diff_text = sys.stdin.read()
|
|
elif args.diff:
|
|
diff_text = Path(args.diff).read_text()
|
|
else:
|
|
diff_text = ""
|
|
|
|
if args.files:
|
|
files = [f.strip() for f in args.files.split(",") if f.strip()]
|
|
elif diff_text:
|
|
parsed = parse_diff(diff_text)
|
|
files = [f["file"] for f in parsed]
|
|
else:
|
|
p.error("no files resolved")
|
|
|
|
out = analyze(files, args.repo, args.top)
|
|
if args.json_out:
|
|
Path(args.json_out).write_text(json.dumps(out, indent=2))
|
|
print(f"locate: {out['stats']['files_flagged']}/{len(files)} files flagged — {args.json_out}")
|
|
else:
|
|
print(json.dumps(out, indent=2))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|