feat: vendorizar as 10 skills do pilar de seguranca em skills/

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.
This commit is contained in:
evandrodevbr committed 2026-08-24 00:32:20 -03:00
1 parent ee44fa0f7a
commit c17192c9fe
130 files changed
+10533 -33

No files matched your search

@@ -0,0 +1,259 @@
#!/usr/bin/env python3
"""
AGENTS.md Linter — Scores instruction files based on evidence from research.
A regex keyword/structure heuristic inspired by (not derived from) the
Instructions-as-Code criteria for what makes instruction files effective for
AI coding agents. A high score is necessary, not sufficient: it rewards
presence of the right signals, it does not verify they are correct.
Criteria (from research):
1. Specificity — concrete rules, not vague guidance
2. Project conventions — naming, file structure, patterns
3. Test guidance — how to run tests, what coverage means
4. Component navigation — which files do what, dependency map
5. Stack details — versions, tools, configuration
6. Anti-patterns — what NOT to do, common mistakes
7. Scope boundaries — what the agent should and shouldn't touch
Usage:
python3 agentsmd-lint.py /path/to/CLAUDE.md
python3 agentsmd-lint.py --dir /path/to/project (finds CLAUDE.md/AGENTS.md)
"""
import argparse
import os
import re
import sys
from datetime import datetime
VERSION = "1.0.0"
def find_instruction_files(directory: str) -> list:
"""Find CLAUDE.md, AGENTS.md, or .cursorrules files."""
candidates = ["CLAUDE.md", "AGENTS.md", ".cursorrules", ".claude/CLAUDE.md", ".github/AGENTS.md"]
found = []
for c in candidates:
path = os.path.join(directory, c)
if os.path.isfile(path):
found.append(path)
return found
def score_file(filepath: str) -> dict:
"""Score an instruction file and return detailed results."""
if not os.path.isfile(filepath):
return {"error": f"File not found: {filepath}", "score": 0}
with open(filepath, encoding="utf-8", errors="replace") as f:
content = f.read()
lines = content.split("\n")
line_count = len(lines)
word_count = len(content.split())
# ── Criterion 1: Specificity (0-20) ──
# Concrete rules use: "do X", "don't Y", "always", "never", "must", "must not"
specificity_patterns = [
(r'\b(?:do|don\'?t|always|never|must|must\s+not|should\s+not|ensure)\b', 3),
(r'```', 3), # code examples
(r'\b(?:e\.g\.|for example|specifically)\b', 2),
(r'\b(?:error|warning|fail|break)\b', 2),
]
specificity_score = 0
for pattern, weight in specificity_patterns:
matches = len(re.findall(pattern, content, re.IGNORECASE))
specificity_score += min(matches * weight, 10)
# Cap at 20
specificity_score = min(specificity_score, 20)
# ── Criterion 2: Project Conventions (0-15) ──
convention_keywords = [
r'\b(?:naming|convention|pattern|structure|organisation|organization|layout)\b',
r'\b(?:file\s+name|directory|folder|path|module)\b',
r'\b(?:import|require|dependency|package)\b',
]
convention_score = 0
for pattern in convention_keywords:
if re.search(pattern, content, re.IGNORECASE):
convention_score += 5
convention_score = min(convention_score, 15)
# ── Criterion 3: Test Guidance (0-15) ──
test_keywords = [
r'\b(?:test|tests|testing|coverage|spec)\b',
r'\b(?:pytest|jest|mocha|vitest|unittest|rspec)\b',
r'\b(?:run\s+tests?|test\s+command|how\s+to\s+test)\b',
]
test_score = 0
for pattern in test_keywords:
if re.search(pattern, content, re.IGNORECASE):
test_score += 5
test_score = min(test_score, 15)
# ── Criterion 4: Component Navigation (0-15) ──
nav_keywords = [
r'\b(?:src/|lib/|app/|components?/|pages/|routes?/)\b',
r'\b(?:entry\s+point|main|index|app\b.*\bentry)\b',
r'\b(?:architecture|dependency|depends\s+on|calls?|uses?)\b',
]
nav_score = 0
for pattern in nav_keywords:
if re.search(pattern, content, re.IGNORECASE):
nav_score += 5
# Bonus for explicit file tree
if re.search(r'├|└|│', content):
nav_score += 5
nav_score = min(nav_score, 15)
# ── Criterion 5: Stack Details (0-10) ──
stack_keywords = [
r'\b(?:version|v\d+\.\d+|@\d+\.\d+)\b',
r'\b(?:python|node|rust|go|typescript|javascript)\s+\d+\.\d+\b',
r'\b(?:config|\.env|settings|setup)\b',
]
stack_score = 0
for pattern in stack_keywords:
if re.search(pattern, content, re.IGNORECASE):
stack_score += 3
stack_score = min(stack_score, 10)
# ── Criterion 6: Anti-patterns (0-10) ──
antipattern_keywords = [
r'\b(?:don\'?t|avoid|never|do\s+not|should\s+not|must\s+not|wrong)\b',
r'\b(?:common\s+mistake|pitfall|gotcha|beware|warning)\b',
r'\b(?:bad|incorrect|broken|anti.?pattern)\b',
]
antipattern_score = 0
for pattern in antipattern_keywords:
matches = len(re.findall(pattern, content, re.IGNORECASE))
antipattern_score += min(matches * 2, 6)
antipattern_score = min(antipattern_score, 10)
# ── Criterion 7: Scope Boundaries (0-10) ──
scope_keywords = [
r'\b(?:scope|boundary|limit|constraint|out\s+of\s+scope|in\s+scope)\b',
r'\b(?:don\'?t\s+touch|leave\s+alone|hands?\s+off|do\s+not\s+modify)\b',
r'\b(?:only|restrict|confine|within)\b',
]
scope_score = 0
for pattern in scope_keywords:
if re.search(pattern, content, re.IGNORECASE):
scope_score += 3
scope_score = min(scope_score, 10)
# ── Bonus: Structure (0-5) ──
# Well-structured files use headings, bullet lists, and sections
has_headings = bool(re.search(r'^#{1,3}\s', content, re.MULTILINE))
has_lists = bool(re.search(r'^\s*[-*+]\s', content, re.MULTILINE))
has_sections = len(re.findall(r'^#{1,3}\s', content, re.MULTILINE)) >= 3
structure_bonus = (2 if has_headings else 0) + (1 if has_lists else 0) + (2 if has_sections else 0)
total = specificity_score + convention_score + test_score + nav_score + stack_score + antipattern_score + scope_score + structure_bonus
total = min(total, 100)
# ── Recommendations ──
recommendations = []
if specificity_score < 10:
recommendations.append("Add concrete rules: use 'do X', 'don't Y', 'always', 'never'. Include code examples (``` blocks).")
if convention_score < 5:
recommendations.append("Document naming conventions, file structure, and import patterns.")
if test_score < 5:
recommendations.append("Add test guidance: how to run tests, what framework, coverage expectations.")
if nav_score < 5:
recommendations.append("Map key directories (src/, lib/, routes/) and explain what lives where.")
if stack_score < 3:
recommendations.append("Specify language versions and key dependencies with versions.")
if antipattern_score < 5:
recommendations.append("List common mistakes to avoid — 'never do X because Y'.")
if scope_score < 3:
recommendations.append("Define what the agent should and shouldn't modify.")
# Grade
grade = "S" if total >= 90 else ("A" if total >= 75 else ("B" if total >= 60 else ("C" if total >= 40 else "D")))
return {
"file": filepath,
"score": total,
"grade": grade,
"max_score": 100,
"stats": {
"lines": line_count,
"words": word_count,
},
"breakdown": {
"specificity": specificity_score,
"conventions": convention_score,
"test_guidance": test_score,
"component_navigation": nav_score,
"stack_details": stack_score,
"antipatterns": antipattern_score,
"scope_boundaries": scope_score,
"structure_bonus": structure_bonus,
},
"recommendations": recommendations,
"linted_at": datetime.now().isoformat(),
}
def main():
parser = argparse.ArgumentParser(description="AGENTS.md Linter — score instruction files for AI agent effectiveness")
parser.add_argument("path", nargs="?", help="Path to instruction file")
parser.add_argument("--dir", help="Directory to search for instruction files")
args = parser.parse_args()
if args.dir:
files = find_instruction_files(args.dir)
if not files:
print(f"No instruction files found in {args.dir}")
print(f"Looked for: CLAUDE.md, AGENTS.md, .cursorrules")
sys.exit(1)
print(f"Found {len(files)} instruction file(s) in {args.dir}\n")
for f in files:
result = score_file(f)
print(f" {os.path.basename(f)}: {result['score']}/100 ({result['grade']})")
return
elif args.path:
result = score_file(args.path)
else:
parser.print_help()
sys.exit(1)
if "error" in result:
print(f"Error: {result['error']}")
sys.exit(1)
print(f"\n{'─'*60}")
print(f" File: {result['file']}")
print(f" Score: {result['score']}/{result['max_score']} ({result['grade']})")
print(f" Stats: {result['stats']['lines']} lines, {result['stats']['words']} words")
print(f"{'─'*60}")
# Each criterion has its own max; normalise the bar to that max so a full
# 10/10 reads as full, not half (the old bar assumed a fixed /20 scale).
criterion_max = {
"specificity": 20, "conventions": 15, "test_guidance": 15,
"component_navigation": 15, "stack_details": 10, "antipatterns": 10,
"scope_boundaries": 10, "structure_bonus": 5,
}
print(f"\n Breakdown:")
for criterion, score in result["breakdown"].items():
cmax = criterion_max.get(criterion, 10) or 1
filled = round(score / cmax * 10)
bar = "█" * filled + "░" * (10 - filled)
print(f" {criterion.replace('_', ' ').title():25s} {score:3d}/{cmax:<3d} {bar}")
if result["recommendations"]:
print(f"\n Recommendations:")
for i, rec in enumerate(result["recommendations"], 1):
print(f" {i}. {rec}")
print(f"\n Grade scale: S (90+) A (75-89) B (60-74) C (40-59) D (<40)")
print(f" Method: regex keyword/structure heuristics, inspired by (not derived from)")
print(f" the Instructions-as-Code criteria. A high score is necessary, not sufficient.\n")
if __name__ == "__main__":
main()
@@ -0,0 +1,33 @@
# Hermaguard Benchmark
Score-only. It does **not** run HermaGuard (that's a skill, not a binary). Your
harness produces a `--json` report per bug; this scores it against ground truth.
## Layout
- `bugs/<id>/` — one case per planted bug: `code/app.py` (fixture, **no answer-key
comments**), `pre.diff` (the change that introduced it), `ground-truth.json`
(the findings a correct review should surface).
- `results/<id>.json` — a harness's review output for that bug (the shape the
scorer reads). The committed ones are a **reference run**, see below.
## Run
```bash
python3 benchmark.py --bugs bugs --results results --output results/report.html
```
## Reference run (committed in `results/`)
A single `claude-opus-4-8` harness reviewed the 5 fixtures. **Not blind:** the
corpus was authored and reviewed by the same agent, so recall here is an upper
bound. The corpus exists for blind runs by any harness, seed bugs authored by
someone other than the reviewer for an independent number.
| | Recall | Precision | F1 |
|---|---|---|---|
| 5 bugs | 100% | 90% | 93.3% |
The one sub-100% precision (bug 001) is a real out-of-ground-truth finding (an
unclosed DB connection), not a hallucination. Each result file carries a
`provenance` block (model, mode, blind flag).
@@ -0,0 +1,402 @@
#!/usr/bin/env python3
"""
Hermaguard Benchmark — Score HermaGuard against known bug dataset.
The benchmark does NOT run HermaGuard itself (it's a skill, not a binary).
Instead, it scores JSON output that any harness produces.
Usage:
# 1. Run HermaGuard via your harness, capture JSON output per bug
# (e.g., Claude Code: claude -p "/hermaguard --json" > results/bug-001.json)
# (e.g., Codex CLI: codex exec --prompt "/hermaguard --json" > results/bug-001.json)
# (e.g., Hermes Agent: run skill, save JSON output)
# 2. Score the results against ground truth
benchmark.py --score --results results/ --bugs bugs/ --output report.html
# 3. Score a single bug
benchmark.py --score --result results/bug-001.json --bug bugs/001-sql-injection
The neutral path: any harness emits the --json report, then benchmark.py --score
against ground truth. No assumption about which CLI or framework runs HermaGuard.
"""
import argparse
import json
import os
import sys
from datetime import datetime
from pathlib import Path
VERSION = "1.0.0"
def discover_bugs(bugs_dir: str) -> list:
"""Find all bug directories containing ground-truth.json."""
bugs = []
for entry in sorted(Path(bugs_dir).iterdir()):
if entry.is_dir() and (entry / "ground-truth.json").exists():
bugs.append(str(entry))
return bugs
def load_bug(bug_dir: str) -> dict:
"""Load a bug's ground truth and metadata."""
gt_path = os.path.join(bug_dir, "ground-truth.json")
readme_path = os.path.join(bug_dir, "README.md")
with open(gt_path) as f:
gt = json.load(f)
readme = ""
if os.path.exists(readme_path):
with open(readme_path) as f:
readme = f.read()
return {
"id": os.path.basename(bug_dir),
"title": gt.get("title", os.path.basename(bug_dir)),
"description": gt.get("description", ""),
"severity": gt.get("severity", "UNKNOWN"),
"category": gt.get("category", "unknown"),
"expected_findings": gt.get("findings", []),
"pre_diff": os.path.join(bug_dir, "pre.diff"),
"post_diff": os.path.join(bug_dir, "post.diff"),
"repo_path": os.path.join(bug_dir, "repo"),
"readme": readme,
}
def _file_matches(exp_file: str, found_file: str) -> bool:
"""Deterministic file match on basename. Empty either side = pass-through."""
if not exp_file or not found_file:
return True
return os.path.basename(exp_file) == os.path.basename(found_file)
def _class_matches(exp: dict, fnd: dict) -> bool:
"""Deterministic class match: both sides carry `classes` and they intersect."""
exp_classes = exp.get("classes") or []
fnd_classes = fnd.get("classes") or []
if not exp_classes or not fnd_classes:
return False # not class-matchable — caller falls back to keywords
return bool(set(exp_classes) & set(fnd_classes))
def score_run(bug: dict, run_info: dict) -> dict:
"""Score a run against ground truth.
Matching is two-tier and deterministic:
Tier 1 (preferred): class-based — ground truth entries carry `classes`
(CWE/HG ids) and findings carry `classes`; a match requires a non-empty
class intersection AND a basename file match. No thresholds, no fuzzy
scores. Identical inputs always produce identical results.
Tier 2 (legacy fallback): keyword overlap for old results that predate
the classes field. Threshold fixed at 0.3, severity boost 1.3 — kept ONLY
for backward compatibility; new ground truth must carry classes.
"""
expected = bug.get("expected_findings", [])
found = run_info.get("findings", [])
matched_expected = set()
matched_found = set()
matches = []
match_modes = {"class": 0, "keyword": 0}
for i, exp in enumerate(expected):
# ---- Tier 1: deterministic class match ----
if exp.get("classes"):
for j, fnd in enumerate(found):
if j in matched_found:
continue
if _class_matches(exp, fnd) and _file_matches(exp.get("file", ""), fnd.get("file", "")):
matched_expected.add(i)
matched_found.add(j)
match_modes["class"] += 1
matches.append({
"expected_id": f"EXP-{i+1}",
"found_id": fnd.get("id", f"HG-{j+1}"),
"mode": "class",
"matched_classes": sorted(set(exp["classes"]) & set(fnd.get("classes") or [])),
"expected_severity": exp.get("severity", "?"),
"found_severity": fnd.get("severity", "?"),
"expected_file": exp.get("file", "?"),
"found_file": fnd.get("file", "?"),
})
break
continue # class-carrying entries never fall through to keywords
# ---- Tier 2: legacy keyword fallback ----
exp_text = f"{exp.get('file','')} {exp.get('trigger','')} {exp.get('description','')}".lower()
best_score = 0
best_j = -1
for j, fnd in enumerate(found):
if j in matched_found:
continue
fnd_text = f"{fnd.get('file','')} {fnd.get('trigger_condition','')} {fnd.get('consequence','')}".lower()
# Simple keyword overlap score
exp_words = set(exp_text.split())
fnd_words = set(fnd_text.split())
if not exp_words or not fnd_words:
continue
overlap = len(exp_words & fnd_words) / min(len(exp_words), len(fnd_words))
# Boost: same severity
if exp.get("severity") == fnd.get("severity"):
overlap *= 1.3
if overlap > best_score:
best_score = overlap
best_j = j
if best_score > 0.3 and best_j >= 0: # threshold for a "match"
matched_expected.add(i)
matched_found.add(best_j)
matched_fnd = found[best_j]
match_modes["keyword"] += 1
matches.append({
"expected_id": f"EXP-{i+1}",
"found_id": matched_fnd.get("id", f"HG-{best_j+1}"),
"mode": "keyword",
"score": round(best_score, 2),
"expected_severity": exp.get("severity", "?"),
"found_severity": matched_fnd.get("severity", "?"),
"expected_file": exp.get("file", "?"),
"found_file": matched_fnd.get("file", "?"),
})
total_expected = len(expected)
total_found = len(found)
true_positives = len(matched_expected)
false_negatives = total_expected - true_positives
false_positives = total_found - len(matched_found)
precision = round(true_positives / total_found * 100, 1) if total_found > 0 else 0
recall = round(true_positives / total_expected * 100, 1) if total_expected > 0 else 0
f1 = round(2 * precision * recall / (precision + recall), 1) if (precision + recall) > 0 else 0
# Severity accuracy
severity_matches = 0
for m in matches:
if m["expected_severity"] == m["found_severity"]:
severity_matches += 1
severity_accuracy = round(severity_matches / len(matches) * 100, 1) if matches else 0
return {
"bug_id": bug["id"],
"bug_title": bug["title"],
"bug_severity": bug["severity"],
"bug_category": bug["category"],
"expected_findings": total_expected,
"found_findings": total_found,
"true_positives": true_positives,
"false_negatives": false_negatives,
"false_positives": false_positives,
"precision_pct": precision,
"recall_pct": recall,
"f1_score": f1,
"severity_accuracy_pct": severity_accuracy,
"matches": matches,
"match_modes": match_modes,
# A result file without an explicit run_ok flag is treated as a
# completed run (backward compat) UNLESS it carries an error key —
# a crashed run must never be scored as OK.
"run_ok": bool(run_info.get("run_ok", not run_info.get("error"))),
"run_error": run_info.get("error"),
}
def generate_report(scores: list, output_path: str):
"""Generate HTML benchmark report."""
total_bugs = len(scores)
ok_runs = [s for s in scores if s["run_ok"]]
avg_precision = round(sum(s["precision_pct"] for s in ok_runs) / len(ok_runs), 1) if ok_runs else 0
avg_recall = round(sum(s["recall_pct"] for s in ok_runs) / len(ok_runs), 1) if ok_runs else 0
avg_f1 = round(sum(s["f1_score"] for s in ok_runs) / len(ok_runs), 1) if ok_runs else 0
avg_severity = round(sum(s["severity_accuracy_pct"] for s in ok_runs) / len(ok_runs), 1) if ok_runs else 0
total_fp = sum(s["false_positives"] for s in ok_runs)
total_fn = sum(s["false_negatives"] for s in ok_runs)
html = f"""<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>HermaGuard Benchmark — {datetime.now().strftime('%d/%m/%Y')}</title>
<style>
:root {{ color-scheme: dark; }}
* {{ margin:0; padding:0; box-sizing:border-box; }}
body {{ background:#11100f; color:#f5f5f4; font-family:system-ui,sans-serif; line-height:1.6; padding:2rem; }}
h1 {{ color:#fbbf24; font-size:1.5rem; margin-bottom:0.5rem; }}
.subtitle {{ color:#a8a29e; font-size:0.9rem; margin-bottom:2rem; }}
.summary {{ display:flex; gap:1.5rem; margin:1.5rem 0; flex-wrap:wrap; }}
.stat {{ background:#1c1a18; border:1px solid #34302c; border-radius:6px; padding:1rem 1.25rem; }}
.stat-value {{ font-size:1.5rem; font-weight:700; color:#fbbf24; }}
.stat-label {{ font-size:0.8rem; color:#a8a29e; }}
.good {{ color:#22c55e; }} .warn {{ color:#fbbf24; }} .bad {{ color:#ef4444; }}
table {{ width:100%; border-collapse:collapse; margin:1.5rem 0; }}
th, td {{ padding:0.6rem 0.8rem; text-align:left; border-bottom:1px solid #34302c; }}
th {{ color:#a8a29e; font-size:0.85rem; font-weight:600; }}
td {{ font-size:0.9rem; }}
.bug-title {{ max-width:300px; }}
.match {{ color:#22c55e; }} .miss {{ color:#ef4444; }}
.error-row {{ background:#3b1111; }}
h2 {{ color:#fbbf24; margin:2rem 0 1rem; padding-bottom:0.5rem; border-bottom:1px solid #34302c; }}
.footer {{ margin-top:3rem; padding-top:1rem; border-top:1px solid #34302c; color:#78716c; font-size:0.8rem; }}
</style>
</head>
<body>
<h1>HermaGuard Benchmark</h1>
<div class="subtitle">{datetime.now().strftime('%d/%m/%Y %H:%M')} · {total_bugs} bugs · v{VERSION}</div>
<div class="summary">
<div class="stat"><div class="stat-value">{avg_f1}<span style="font-size:0.8rem">%</span></div><div class="stat-label">Avg F1 Score</div></div>
<div class="stat"><div class="stat-value">{avg_recall}<span style="font-size:0.8rem">%</span></div><div class="stat-label">Avg Recall</div></div>
<div class="stat"><div class="stat-value">{avg_precision}<span style="font-size:0.8rem">%</span></div><div class="stat-label">Avg Precision</div></div>
<div class="stat"><div class="stat-value">{avg_severity}<span style="font-size:0.8rem">%</span></div><div class="stat-label">Severity Accuracy</div></div>
<div class="stat"><div class="stat-value">{total_fp}</div><div class="stat-label">False Positives</div></div>
<div class="stat"><div class="stat-value">{total_fn}</div><div class="stat-label">False Negatives</div></div>
</div>
<h2>Results by Bug</h2>
<table>
<tr><th>Bug</th><th>Severity</th><th>Expected</th><th>Found</th><th>TP</th><th>FP</th><th>FN</th><th>Precision</th><th>Recall</th><th>F1</th><th>Sev Acc</th></tr>
"""
for s in scores:
if not s["run_ok"]:
html += f"""<tr class="error-row"><td class="bug-title">{s['bug_title'][:60]}</td>
<td>-</td><td>-</td><td>-</td><td>-</td><td>-</td><td>-</td><td colspan="4">ERROR: {s.get('run_error','Unknown')[:100]}</td></tr>\n"""
continue
pcls = "good" if s["precision_pct"] >= 70 else ("warn" if s["precision_pct"] >= 40 else "bad")
rcls = "good" if s["recall_pct"] >= 70 else ("warn" if s["recall_pct"] >= 40 else "bad")
fcls = "good" if s["f1_score"] >= 70 else ("warn" if s["f1_score"] >= 40 else "bad")
html += f"""<tr>
<td class="bug-title">{s['bug_title'][:60]}</td>
<td>{s['bug_severity']}</td>
<td>{s['expected_findings']}</td>
<td>{s['found_findings']}</td>
<td class="match">{s['true_positives']}</td>
<td class="miss">{s['false_positives']}</td>
<td class="miss">{s['false_negatives']}</td>
<td class="{pcls}">{s['precision_pct']}%</td>
<td class="{rcls}">{s['recall_pct']}%</td>
<td class="{fcls}">{s['f1_score']}%</td>
<td>{s['severity_accuracy_pct']}%</td>
</tr>\n"""
html += "</table>\n"
# Detailed matches for each bug
html += "<h2>Match Details</h2>\n"
for s in scores:
if not s["run_ok"]:
continue
html += f"<h3>{s['bug_title'][:80]} ({s['bug_severity']})</h3>\n"
if s["matches"]:
html += "<table><tr><th>Expected</th><th>Found</th><th>Sev Match</th><th>Score</th></tr>\n"
for m in s["matches"]:
sev_match = "✓" if m["expected_severity"] == m["found_severity"] else "✗"
html += f"""<tr><td>{m['expected_file']} ({m['expected_severity']})</td>
<td>{m['found_file'][:50]} ({m['found_severity']})</td><td>{sev_match}</td><td>{m['score']}</td></tr>\n"""
html += "</table>\n"
else:
html += "<p class='miss'>No matches — all findings missed or all expected bugs not found.</p>\n"
if s.get("matches"):
# Unmatched expected (FN)
matched_exp_ids = {m["expected_id"] for m in s["matches"]}
unmatched_exp = s.get("expected_findings_count", s["expected_findings"]) - len(matched_exp_ids)
if unmatched_exp > 0:
html += f"<p class='miss'>{unmatched_exp} expected bug{'s' if unmatched_exp > 1 else ''} not found (false negative).</p>\n"
html += f"""<div class="footer">
HermaGuard Benchmark v{VERSION} · Generated {datetime.now().isoformat()}
</div>
</body></html>"""
with open(output_path, "w") as f:
f.write(html)
print(f"Report written to {output_path}")
return output_path
def main():
parser = argparse.ArgumentParser(description="HermaGuard Benchmark — score JSON output against ground truth")
parser.add_argument("--bugs", default="bugs/", help="Path to bugs directory")
parser.add_argument("--bug", help="Score a single bug by path")
parser.add_argument("--results", default="results/", help="Directory containing per-bug JSON result files")
parser.add_argument("--result", help="Single result JSON file (use with --bug)")
parser.add_argument("--output", default="report.html", help="HTML report output path")
parser.add_argument("--version", action="version", version=f"%(prog)s {VERSION}")
args = parser.parse_args()
if args.bug and args.result:
# Single bug mode
bug = load_bug(args.bug)
with open(args.result) as f:
run_info = json.load(f)
s = score_run(bug, run_info)
status = "✓" if s["run_ok"] else "✗"
print(f" {status} {bug['title'][:60]} — P:{s['precision_pct']}% R:{s['recall_pct']}% F1:{s['f1_score']}%")
return
bugs = discover_bugs(args.bugs)
if not bugs:
print(f"No bugs found in {args.bugs}")
sys.exit(1)
print(f"Found {len(bugs)} bug(s)")
scores = []
for bug_dir in bugs:
bug = load_bug(bug_dir)
result_path = os.path.join(args.results, f"{bug['id']}.json")
if os.path.exists(result_path):
with open(result_path) as f:
run_info = json.load(f)
s = score_run(bug, run_info)
scores.append(s)
status = "✓" if s["run_ok"] else "✗"
print(f" {status} {bug['title'][:60]} — P:{s['precision_pct']}% R:{s['recall_pct']}% F1:{s['f1_score']}%")
else:
scores.append({
"bug_id": bug["id"], "bug_title": bug["title"],
"bug_severity": bug["severity"], "bug_category": bug["category"],
"run_ok": False, "run_error": f"No results file at {result_path}",
"precision_pct": 0, "recall_pct": 0, "f1_score": 0,
"expected_findings": len(bug["expected_findings"]), "found_findings": 0,
"true_positives": 0, "false_positives": 0, "false_negatives": len(bug["expected_findings"]),
"severity_accuracy_pct": 0, "matches": [],
})
# Generate report. --output is taken literally (path doubling bug fixed);
# create the parent dir so an explicit path like results/report.html works.
report_path = args.output
parent = os.path.dirname(report_path)
if parent:
os.makedirs(parent, exist_ok=True)
generate_report(scores, report_path)
ok = [s for s in scores if s["run_ok"]]
if ok:
avg_f1 = round(sum(s["f1_score"] for s in ok) / len(ok), 1)
print(f"\nOverall: {len(ok)}/{len(scores)} runs OK, avg F1: {avg_f1}%")
else:
print(f"\nAll runs lack result files. Run HermaGuard via your harness first, then score.")
print(f"Example: run '/hermaguard --json' under any harness, save output to results/<bug-id>.json")
print(f"Report: {report_path}")
if __name__ == "__main__":
main()
@@ -0,0 +1,8 @@
import sqlite3
def get_user_by_name(username: str):
conn = sqlite3.connect("users.db")
cur = conn.cursor()
query = f"SELECT * FROM users WHERE name = '{username}'"
cur.execute(query)
return cur.fetchone()
@@ -0,0 +1,9 @@
{
"title": "SQL injection in get_user_by_name",
"description": "Username interpolated into SQL via f-string; attacker-controlled input reaches the query unsanitised.",
"severity": "CRITICAL",
"category": "security",
"findings": [
{"file": "app.py", "trigger": "f-string SQL query username interpolation", "description": "SQL injection via unparameterised query", "severity": "CRITICAL", "classes": ["CWE-89"]}
]
}
@@ -0,0 +1,10 @@
--- a/app.py
+++ b/app.py
@@ -2,6 +2,6 @@ import sqlite3
def get_user_by_name(username: str):
conn = sqlite3.connect("users.db")
cur = conn.cursor()
- cur.execute("SELECT * FROM users WHERE name = ?", (username,))
+ query = f"SELECT * FROM users WHERE name = '{username}'"
+ cur.execute(query)
return cur.fetchone()
@@ -0,0 +1,5 @@
import yaml
def load_config(path: str) -> dict:
with open(path) as f:
return yaml.load(f)
@@ -0,0 +1,9 @@
{
"title": "Unsafe YAML deserialisation",
"description": "yaml.load() without SafeLoader allows arbitrary object construction — RCE via crafted YAML.",
"severity": "CRITICAL",
"category": "security",
"findings": [
{"file": "app.py", "trigger": "yaml.load without SafeLoader", "description": "Arbitrary code execution via unsafe deserialisation", "severity": "CRITICAL", "classes": ["CWE-502"]}
]
}
@@ -0,0 +1,7 @@
--- a/app.py
+++ b/app.py
@@ -2,4 +2,4 @@ import yaml
def load_config(path: str) -> dict:
with open(path) as f:
- return yaml.safe_load(f)
+ return yaml.load(f)
@@ -0,0 +1,4 @@
def get_user_email(cur, user_id: int) -> str:
cur.execute("SELECT email FROM users WHERE id = ?", (user_id,))
row = cur.fetchone()
return row[0]
@@ -0,0 +1,9 @@
{
"title": "Missing null check on fetchone result",
"description": "fetchone() returns None when no row matches; indexing row[0] then raises TypeError, crashing the caller.",
"severity": "MEDIUM",
"category": "logic",
"findings": [
{"file": "app.py", "trigger": "cursor fetchone row indexed without null check", "description": "TypeError crash when user_id not found", "severity": "MEDIUM", "classes": ["CWE-476", "CWE-754"]}
]
}
@@ -0,0 +1,8 @@
--- a/app.py
+++ b/app.py
@@ -2,4 +2,3 @@ def get_user_email(cur, user_id: int) -> str:
cur.execute("SELECT email FROM users WHERE id = ?", (user_id,))
row = cur.fetchone()
- if row is None:
- return None
return row[0]
@@ -0,0 +1,4 @@
import subprocess
def ping_host(host: str) -> bytes:
return subprocess.check_output(f"ping -c 1 {host}", shell=True)
@@ -0,0 +1,9 @@
{
"title": "Shell command injection via ping utility",
"description": "User-supplied host is concatenated into a shell command; shell metacharacters execute arbitrary commands.",
"severity": "CRITICAL",
"category": "security",
"findings": [
{"file": "app.py", "trigger": "user host concatenated into shell command", "description": "OS command injection via shell=True with unescaped input", "severity": "CRITICAL", "classes": ["CWE-78"]}
]
}
@@ -0,0 +1,6 @@
--- a/app.py
+++ b/app.py
@@ -2,2 +2,2 @@ import subprocess
def ping_host(host: str) -> bytes:
- return subprocess.check_output(["ping", "-c", "1", host])
+ return subprocess.check_output(f"ping -c 1 {host}", shell=True)
@@ -0,0 +1,3 @@
def read_doc(name: str) -> str:
with open("/var/docs/" + name) as f:
return f.read()
@@ -0,0 +1,9 @@
{
"title": "Path traversal in file download",
"description": "User-supplied filename is joined without validation; ../ sequences escape the intended directory.",
"severity": "HIGH",
"category": "security",
"findings": [
{"file": "app.py", "trigger": "filename joined into path without traversal check", "description": "Arbitrary file read via ../ traversal", "severity": "HIGH", "classes": ["CWE-22"]}
]
}
@@ -0,0 +1,8 @@
--- a/app.py
+++ b/app.py
@@ -1,3 +1,3 @@
def read_doc(name: str) -> str:
- safe = os.path.basename(name)
- with open(os.path.join("/var/docs", safe)) as f:
+ with open("/var/docs/" + name) as f:
return f.read()
@@ -0,0 +1,11 @@
# Deliberately broken sample: a live-looking API key committed to source.
# NOTE: the value is an obviously-fake placeholder (marker, not a real key
# format) so secret scanners do not flag the corpus; the CWE-798 pattern
# (credential hardcoded in source) is what the benchmark tests for.
PAYMENT_KEY = "EXAMPLE-KEY-DO-NOT-USE-0123456789"
def charge(card_token: str, amount: int):
import requests
return requests.post("https://api.example-payments.invalid/v1/charges",
headers={"Authorization": f"Bearer {PAYMENT_KEY}"},
data={"card": card_token, "amount": amount})
@@ -0,0 +1,17 @@
{
"title": "Hard-coded API key committed to source",
"description": "Live API key embedded in source; anyone with repo access can use the account.",
"severity": "CRITICAL",
"category": "security",
"findings": [
{
"file": "app.py",
"trigger": "live API key assigned to module constant",
"description": "Credential theft and unauthorized charges from public repo",
"severity": "CRITICAL",
"classes": [
"CWE-798"
]
}
]
}
@@ -0,0 +1,4 @@
--- a/app.py
+++ b/app.py
@@ -0,0 +1 @@
+ (new file — the bug is introduced by this change)
@@ -0,0 +1,5 @@
def delete_user(session_user, target_id: int):
# authn done upstream, but no role check
db.execute("DELETE FROM users WHERE id = ?", (target_id,))
db.commit()
return {"ok": True}
@@ -0,0 +1,18 @@
{
"title": "Admin endpoint without authorization check",
"description": "delete_user performs no permission check; any authenticated user can delete anyone.",
"severity": "CRITICAL",
"category": "security",
"findings": [
{
"file": "app.py",
"trigger": "admin action without role/permission check",
"description": "Privilege escalation \u2014 any user deletes any account",
"severity": "CRITICAL",
"classes": [
"CWE-862",
"CWE-269"
]
}
]
}
@@ -0,0 +1,6 @@
--- a/app.py
+++ b/app.py
def delete_user(session_user, target_id: int):
- if session_user.get("role") != "admin":
- raise PermissionError("admin only")
# authn done upstream, but no role check
@@ -0,0 +1,3 @@
def order_total(price_cents: int, quantity: int) -> int:
# no upper bound on quantity
return price_cents * quantity
@@ -0,0 +1,18 @@
{
"title": "Unvalidated quantity causes integer overflow",
"description": "quantity from request body multiplies price without bounds; extreme values overflow.",
"severity": "MEDIUM",
"category": "logic",
"findings": [
{
"file": "app.py",
"trigger": "quantity multiplied without upper-bound check",
"description": "Overflow/wraparound corrupts totals on extreme input",
"severity": "MEDIUM",
"classes": [
"CWE-190",
"CWE-20"
]
}
]
}
@@ -0,0 +1,6 @@
--- a/app.py
+++ b/app.py
def order_total(price_cents: int, quantity: int) -> int:
- if quantity > 10_000:
- raise ValueError("quantity too large")
# no upper bound on quantity
@@ -0,0 +1,5 @@
import requests
def send_webhook(url: str, payload: dict):
# user-supplied URL fetched without validation
requests.post(url, json=payload, timeout=5)
@@ -0,0 +1,18 @@
{
"title": "SSRF via user-controlled webhook URL",
"description": "Webhook URL comes from the request and is fetched server-side without allowlist.",
"severity": "HIGH",
"category": "security",
"findings": [
{
"file": "app.py",
"trigger": "user-supplied URL fetched server-side without allowlist",
"description": "SSRF \u2014 internal network probing and metadata endpoint access",
"severity": "HIGH",
"classes": [
"CWE-918",
"CWE-20"
]
}
]
}
@@ -0,0 +1,8 @@
--- a/app.py
+++ b/app.py
import requests
def send_webhook(url: str, payload: dict):
- if not url.startswith("https://hooks.example.com/"):
- raise ValueError("webhook host not allowed")
# user-supplied URL fetched without validation
@@ -0,0 +1,5 @@
function renderResults(results, searchTerm) {
const box = document.getElementById("results");
box.innerHTML = `<h2>Results for ${searchTerm}</h2>` +
results.map(r => `<li>${r.name}</li>`).join("");
}
@@ -0,0 +1,17 @@
{
"title": "Reflected XSS via innerHTML assignment",
"description": "Search term inserted into the DOM via innerHTML without escaping.",
"severity": "HIGH",
"category": "security",
"findings": [
{
"file": "app.js",
"trigger": "user input interpolated into innerHTML",
"description": "Script injection in victim browsers via crafted search term",
"severity": "HIGH",
"classes": [
"CWE-79"
]
}
]
}
@@ -0,0 +1,4 @@
--- a/app.js
+++ b/app.js
@@ -0,0 +1 @@
+ (new file — the bug is introduced by this change)
@@ -0,0 +1,6 @@
def withdraw(conn, user_id: int, amount: int):
row = conn.execute("SELECT balance FROM accounts WHERE id = ?", (user_id,)).fetchone()
if row[0] < amount:
raise ValueError("insufficient funds")
conn.execute("UPDATE accounts SET balance = balance - ? WHERE id = ?", (amount, user_id))
conn.commit()
@@ -0,0 +1,17 @@
{
"title": "TOCTOU race on balance check",
"description": "Check-then-deduct without lock or atomic update; concurrent requests double-spend.",
"severity": "HIGH",
"category": "concurrency",
"findings": [
{
"file": "app.py",
"trigger": "balance check and deduct are separate non-atomic statements",
"description": "Double-spend under concurrent requests",
"severity": "HIGH",
"classes": [
"CWE-362"
]
}
]
}
@@ -0,0 +1,11 @@
--- a/app.py
+++ b/app.py
def withdraw(conn, user_id: int, amount: int):
row = conn.execute("SELECT balance FROM accounts WHERE id = ?", (user_id,)).fetchone()
- # single atomic guarded update
- cur = conn.execute("UPDATE accounts SET balance = balance - ? WHERE id = ? AND balance >= ?", (amount, user_id, amount))
- if cur.rowcount == 0:
- raise ValueError("insufficient funds")
- conn.commit()
- return
if row[0] < amount:
@@ -0,0 +1,5 @@
async function loadProfile(userId) {
const res = await fetch(`/api/users/${userId}`);
const data = await res.json();
return data;
}
@@ -0,0 +1,18 @@
{
"title": "Unhandled promise rejection",
"description": "fetch promise has no catch; network failure leaves UI in loading state forever.",
"severity": "MEDIUM",
"category": "async",
"findings": [
{
"file": "app.js",
"trigger": "await fetch chain with no try/catch or .catch",
"description": "Unhandled rejection \u2014 loading state never resolves on failure",
"severity": "MEDIUM",
"classes": [
"HG-ASYNC",
"CWE-754"
]
}
]
}
@@ -0,0 +1,11 @@
--- a/app.js
+++ b/app.js
async function loadProfile(userId) {
- try {
const res = await fetch(`/api/users/${userId}`);
const data = await res.json();
return data;
- } catch (e) {
- showError(e);
- }
}
@@ -0,0 +1,6 @@
def process_payment(order):
try:
gateway.charge(order)
except Exception:
pass
return {"status": "paid"}
@@ -0,0 +1,18 @@
{
"title": "Swallowed exception in payment processing",
"description": "Bare except: pass hides every failure; function returns fake success.",
"severity": "HIGH",
"category": "error-handling",
"findings": [
{
"file": "app.py",
"trigger": "bare except pass, returns success regardless",
"description": "Fake success masks failed charges \u2014 silent data/money loss",
"severity": "HIGH",
"classes": [
"HG-DEGENERATE",
"CWE-754"
]
}
]
}
@@ -0,0 +1,10 @@
--- a/app.py
+++ b/app.py
def process_payment(order):
try:
gateway.charge(order)
except Exception:
- log.exception("charge failed")
- return {"status": "failed"}
+ pass
return {"status": "paid"}
@@ -0,0 +1,4 @@
def transfer(conn, from_id, to_id, amount):
conn.execute("UPDATE accounts SET balance = balance - ? WHERE id = ?", (amount, from_id))
conn.execute("UPDATE accounts SET balance = balance + ? WHERE id = ?", (amount, to_id))
conn.commit()
@@ -0,0 +1,18 @@
{
"title": "Non-atomic multi-step update",
"description": "Two-table update without transaction; crash between statements corrupts state.",
"severity": "HIGH",
"category": "data-integrity",
"findings": [
{
"file": "app.py",
"trigger": "two dependent updates without a wrapping transaction",
"description": "Crash mid-sequence loses money \u2014 debit without credit",
"severity": "HIGH",
"classes": [
"HG-PARTIAL-WRITE",
"CWE-362"
]
}
]
}
@@ -0,0 +1,7 @@
--- a/app.py
+++ b/app.py
def transfer(conn, from_id, to_id, amount):
- with conn:
conn.execute("UPDATE accounts SET balance = balance - ? WHERE id = ?", (amount, from_id))
conn.execute("UPDATE accounts SET balance = balance + ? WHERE id = ?", (amount, to_id))
- conn.commit()
@@ -0,0 +1,4 @@
def read_config(path):
f = open(path)
data = json.load(f)
return data
@@ -0,0 +1,18 @@
{
"title": "Unclosed file handle on error path",
"description": "open() without context manager; exception path leaks the descriptor.",
"severity": "MEDIUM",
"category": "resources",
"findings": [
{
"file": "app.py",
"trigger": "open without with/close on all paths",
"description": "FD leak under repeated failures exhausts process limit",
"severity": "MEDIUM",
"classes": [
"CWE-404",
"CWE-754"
]
}
]
}
@@ -0,0 +1,7 @@
--- a/app.py
+++ b/app.py
def read_config(path):
- with open(path) as f:
- return json.load(f)
+ f = open(path)
data = json.load(f)
@@ -0,0 +1,5 @@
def handler(event):
try:
return do_work(event)
except Exception as e:
return {"statusCode": 500, "body": str(e)}
@@ -0,0 +1,17 @@
{
"title": "Stack trace returned to client",
"description": "Bare handler returns exception text in the HTTP response body.",
"severity": "MEDIUM",
"category": "security",
"findings": [
{
"file": "app.py",
"trigger": "exception text returned in response body",
"description": "Internal paths, queries, and versions leaked to clients",
"severity": "MEDIUM",
"classes": [
"CWE-200"
]
}
]
}
@@ -0,0 +1,4 @@
--- a/app.py
+++ b/app.py
@@ -0,0 +1 @@
+ (new file — the bug is introduced by this change)
@@ -0,0 +1,8 @@
def parse(stream):
records = []
while True:
line = stream.readline()
if not line:
break
records.append(json.loads(line))
return records
@@ -0,0 +1,17 @@
{
"title": "Unbounded loop over user input",
"description": "parse runs until end-of-stream with no cap on iterations or size.",
"severity": "MEDIUM",
"category": "resources",
"findings": [
{
"file": "app.py",
"trigger": "read loop without record-count or byte cap",
"description": "Memory exhaustion DoS from large uploads",
"severity": "MEDIUM",
"classes": [
"CWE-400"
]
}
]
}
@@ -0,0 +1,12 @@
--- a/app.py
+++ b/app.py
def parse(stream):
records = []
+ MAX = 10_000
while True:
line = stream.readline()
if not line:
break
+ if len(records) >= MAX:
+ raise ValueError("too many records")
records.append(json.loads(line))
@@ -0,0 +1,6 @@
import hashlib
def verify(shared_secret: str, provided: str) -> bool:
a = hashlib.sha256(shared_secret.encode()).hexdigest()
b = hashlib.sha256(provided.encode()).hexdigest()
return a == b
@@ -0,0 +1,18 @@
{
"title": "Password comparison not constant-time",
"description": "Plain == on password-derived token allows timing side channel.",
"severity": "MEDIUM",
"category": "security",
"findings": [
{
"file": "app.py",
"trigger": "secret compared with plain string equality",
"description": "Timing side channel leaks secret one byte at a time",
"severity": "MEDIUM",
"classes": [
"CWE-287",
"CWE-200"
]
}
]
}
@@ -0,0 +1,10 @@
--- a/app.py
+++ b/app.py
import hashlib
+import hmac
def verify(shared_secret: str, provided: str) -> bool:
a = hashlib.sha256(shared_secret.encode()).hexdigest()
b = hashlib.sha256(provided.encode()).hexdigest()
- return hmac.compare_digest(a, b)
+ return a == b
@@ -0,0 +1,4 @@
import pickle
def load_state(body: bytes):
return pickle.loads(body)
@@ -0,0 +1,18 @@
{
"title": "Pickle load of user-supplied bytes",
"description": "pickle.loads on request body executes arbitrary code during deserialisation.",
"severity": "CRITICAL",
"category": "security",
"findings": [
{
"file": "app.py",
"trigger": "pickle.loads on untrusted request bytes",
"description": "RCE \u2014 pickle deserialisation executes attacker code",
"severity": "CRITICAL",
"classes": [
"CWE-502",
"CWE-94"
]
}
]
}
@@ -0,0 +1,4 @@
--- a/app.py
+++ b/app.py
@@ -0,0 +1 @@
+ (new file — the bug is introduced by this change)
@@ -0,0 +1,4 @@
def reset_password(request, user_id: int):
new_pw = request.json["password"]
db.execute("UPDATE users SET pw = ? WHERE id = ?", (hash(new_pw), user_id))
db.commit()
@@ -0,0 +1,18 @@
{
"title": "State-changing route without authentication",
"description": "POST handler has no session check; anonymous requests reset any account.",
"severity": "CRITICAL",
"category": "security",
"findings": [
{
"file": "app.py",
"trigger": "state-changing handler without session/auth check",
"description": "Account takeover by anonymous callers",
"severity": "CRITICAL",
"classes": [
"CWE-306",
"CWE-287"
]
}
]
}
@@ -0,0 +1,6 @@
--- a/app.py
+++ b/app.py
def reset_password(request, user_id: int):
- if not request.session.get("user"):
- raise PermissionError("login required")
new_pw = request.json["password"]
@@ -0,0 +1,3 @@
function canEdit(order) {
return order.status == 1; // "1" also passes
}
@@ -0,0 +1,17 @@
{
"title": "Loose equality on status check",
"description": "JS == with string/number mix accepts wrong status values.",
"severity": "LOW",
"category": "logic",
"findings": [
{
"file": "app.js",
"trigger": "loose == between string and number",
"description": "Wrong-type status grants edit rights unexpectedly",
"severity": "LOW",
"classes": [
"CWE-20"
]
}
]
}
@@ -0,0 +1,6 @@
--- a/app.js
+++ b/app.js
function canEdit(order) {
- return order.status === 1;
+ return order.status == 1; // "1" also passes
}
@@ -0,0 +1,8 @@
async function refresh() {
state.loading = true;
try {
state.data = await api.get("/data");
} catch (e) {
state.error = e.message;
}
}
@@ -0,0 +1,18 @@
{
"title": "Missing loading\u2192error UI transition",
"description": "Error path updates message but never clears the loading flag.",
"severity": "MEDIUM",
"category": "logic",
"findings": [
{
"file": "app.js",
"trigger": "catch path never resets loading flag",
"description": "Spinner stuck forever after a failed refresh",
"severity": "MEDIUM",
"classes": [
"HG-STATE",
"HG-ASYNC"
]
}
]
}
@@ -0,0 +1,6 @@
--- a/app.js
+++ b/app.js
} catch (e) {
state.error = e.message;
+ state.loading = false;
}
@@ -0,0 +1,5 @@
def change_email(request):
if request.method in ("GET", "POST"):
u = db.get_user(request.args["user"])
u.email = request.args["email"]
u.save()
@@ -0,0 +1,17 @@
{
"title": "State change on GET without CSRF token",
"description": "Password change honors GET with no token; img tags can trigger it.",
"severity": "HIGH",
"category": "security",
"findings": [
{
"file": "app.py",
"trigger": "state change accepted on GET without CSRF token",
"description": "CSRF \u2014 attacker sites change victim emails via img/link",
"severity": "HIGH",
"classes": [
"CWE-352"
]
}
]
}
@@ -0,0 +1,6 @@
--- a/app.py
+++ b/app.py
def change_email(request):
- if request.method == "POST" and request.form.get("csrf_token") == session_csrf():
+ if request.method in ("GET", "POST"):
u = db.get_user(request.args["user"])
@@ -0,0 +1,2 @@
def apply_filter(value, expr: str):
return eval(f"{value} {expr}")
@@ -0,0 +1,17 @@
{
"title": "eval on template expression",
"description": "Template renderer evals user-supplied filter expressions.",
"severity": "CRITICAL",
"category": "security",
"findings": [
{
"file": "app.py",
"trigger": "eval on user-controlled expression string",
"description": "Arbitrary code execution via crafted filter",
"severity": "CRITICAL",
"classes": [
"CWE-94"
]
}
]
}
@@ -0,0 +1,4 @@
--- a/app.py
+++ b/app.py
@@ -0,0 +1 @@
+ (new file — the bug is introduced by this change)
@@ -0,0 +1,6 @@
import os
def save_upload(file_storage):
dest = os.path.join("/srv/uploads", file_storage.filename)
file_storage.save(dest)
return dest
@@ -0,0 +1,18 @@
{
"title": "Unrestricted upload accepts any type",
"description": "Upload saves whatever the client sends under the original extension.",
"severity": "HIGH",
"category": "security",
"findings": [
{
"file": "app.py",
"trigger": "upload saved without extension/content-type validation",
"description": "Webshell upload via .php/.jsp under web root",
"severity": "HIGH",
"classes": [
"CWE-434",
"CWE-22"
]
}
]
}
@@ -0,0 +1,9 @@
--- a/app.py
+++ b/app.py
import os
def save_upload(file_storage):
- ext = os.path.splitext(file_storage.filename)[1].lower()
- if ext not in (".png", ".jpg", ".pdf"):
- raise ValueError("file type not allowed")
dest = os.path.join("/srv/uploads", file_storage.filename)
@@ -0,0 +1,125 @@
#!/usr/bin/env python3
"""
Hermaguard Rule Compiler — Compiles dismissed findings into suppression rules.
Reads the feedback database, identifies patterns that have been repeatedly
dismissed as false positives, and writes rules to hermaguard-rules.yaml.
Usage:
hermaguard-compile-rules --output hermaguard-rules.yaml
hermaguard-compile-rules --output .kensei/hermaguard-rules.yaml --min-dismissals 3
Based on TRACE paper: compiling user corrections into deterministic rules
reduces violation rates from 100% → 2%.
"""
import argparse
import json
import os
import sqlite3
import sys
VERSION = "1.0.0"
DB_PATH = os.environ.get("HERMAGUARD_FEEDBACK_DB", os.path.expanduser("~/.hermes/data/hermaguard-feedback.db"))
def compile_rules(min_dismissals: int = 3) -> list:
"""Query the feedback database for dismissed patterns, return rules list."""
if not os.path.isfile(DB_PATH):
return []
conn = sqlite3.connect(DB_PATH)
conn.row_factory = sqlite3.Row
# Find patterns: (source_agent, trigger_condition) pairs dismissed 3+ times
rows = conn.execute("""
SELECT
source_agent,
trigger_condition,
file_path,
COUNT(*) as dismissals,
GROUP_CONCAT(DISTINCT finding_id) as finding_ids
FROM findings
WHERE status = 'dismissed-false-positive'
GROUP BY source_agent, trigger_condition
HAVING COUNT(*) >= ?
ORDER BY dismissals DESC
""", (min_dismissals,)).fetchall()
conn.close()
rules = []
for i, r in enumerate(rows):
agent_short = {
"edge-case-hunter": "edge_case",
"adversarial-reviewer": "adversarial",
"blast-radius-integration": "blast_radius",
}.get(r["source_agent"], r["source_agent"].replace("-", "_"))
rules.append({
"rule_id": f"HG-SUPPRESS-{i + 1:03d}",
"pattern": r["trigger_condition"],
"agent": agent_short,
"file_scope": r["file_path"],
"dismissals": r["dismissals"],
"reason": f"Dismissed {r['dismissals']}x as false positive",
"source_ids": r["finding_ids"].split(",") if r["finding_ids"] else [],
})
return rules
def main():
parser = argparse.ArgumentParser(
description="Hermaguard Rule Compiler — generate suppression rules from dismissed findings"
)
parser.add_argument("--output", required=True, help="Output YAML file path")
parser.add_argument("--min-dismissals", type=int, default=3, help="Min dismissals for a pattern to become a rule (default: 3)")
args = parser.parse_args()
rules = compile_rules(args.min_dismissals)
output = {
"version": VERSION,
"generated_at": __import__("datetime").datetime.now().isoformat(),
"rules": rules,
"stats": {
"total_rules": len(rules),
"by_agent": {},
},
}
for r in rules:
agent = r["agent"]
output["stats"]["by_agent"][agent] = output["stats"]["by_agent"].get(agent, 0) + 1
os.makedirs(os.path.dirname(args.output) or ".", exist_ok=True)
# Write YAML manually to avoid requiring PyYAML
with open(args.output, "w") as f:
f.write(f"# Hermaguard Suppression Rules\n")
f.write(f"# Auto-generated by hermaguard-compile-rules v{VERSION}\n")
f.write(f"# Generated: {output['generated_at']}\n")
f.write(f"# Total rules: {len(rules)}\n")
f.write(f"# Source: {DB_PATH}\n\n")
f.write(f"# These patterns have been repeatedly dismissed as false positives.\n")
f.write(f"# Agents skip findings matching these patterns.\n")
f.write(f"# Delete rules to re-enable checking for a pattern.\n\n")
for r in rules:
reason_line = str(r['reason']).replace("\n", " ")
f.write(f"# {reason_line} | files: {r['file_scope']}\n")
f.write(f"- rule_id: {r['rule_id']}\n")
# json.dumps yields a valid double-quoted scalar (YAML is a JSON
# superset here), so quotes/backslashes in the pattern can't break the file.
f.write(f" pattern: {json.dumps(r['pattern'])}\n")
f.write(f" agent: {r['agent']}\n")
f.write(f" dismissals: {r['dismissals']}\n")
f.write(f"\n")
print(f"Compiled {len(rules)} suppression rules from {DB_PATH}")
print(f"Agents: {json.dumps(output['stats']['by_agent'])}")
print(f"Output: {args.output}")
if __name__ == "__main__":
main()
@@ -0,0 +1,344 @@
#!/usr/bin/env python3
"""
Hermaguard Coverage — The accountability layer.
Every Hermaguard report must account for every applicable bug class:
each class gets exactly one status — finding / clean / not_applicable.
Anything else is a coverage gap and fails the check.
Why: an adversarial review that silently skips bug classes is theatre.
This tool makes coverage measurable and CI-enforceable.
Class registry:
- The 2024 MITRE CWE Top 25 (verified against cwe.mitre.org,
2024_top25_list.html, fetched 2026-08-14).
- Plus CWE-362 (race), CWE-404 (resource leak), CWE-754 (exception
handling) — outside the Top 25 but core to the skill's own checklists.
- Plus 4 Hermaguard-native classes the skill claims to check that CWE
does not model (async rejection, degenerate handlers, state
transitions, partial writes).
Usage:
# Registry integrity (CI self-check)
hermaguard-coverage.py --self-check
# Emit the applicable-class ledger for a diff's file set
hermaguard-coverage.py --files "src/a.py,src/b.ts" --ledger ledger.json
# Validate a report's coverage against the ledger
hermaguard-coverage.py --check-report report.json --ledger ledger.json
# One-shot: build ledger and check in a single call
hermaguard-coverage.py --files "src/a.py" --check-report report.json
Exit codes: 0 = full coverage or self-check OK; 1 = coverage gaps;
2 = usage/IO error.
"""
import argparse
import json
import sys
from pathlib import Path
VERSION = "1.0.0"
# ---------------------------------------------------------------------------
# Registry
# ---------------------------------------------------------------------------
# Language sets used by applicability rules.
MEMORY_UNSAFE = {".c", ".h", ".cpp", ".cc", ".cxx", ".hpp", ".zig"}
RUST = {".rs"} # memory classes apply only via `unsafe` blocks; hint-only
WEB_FRONTEND = {".js", ".jsx", ".ts", ".tsx", ".vue", ".svelte"}
ALL_CODE = MEMORY_UNSAFE | RUST | WEB_FRONTEND | {
".py", ".go", ".java", ".kt", ".rb", ".php", ".cs", ".swift", ".scala",
}
# 2024 CWE Top 25 in rank order (MITRE, verified 2026-08-14) + 3 extras.
# status: "always" | lang-rule
# lang_rule: (applicable_exts, na_reason)
CLASSES = [
# --- 2024 CWE Top 25, rank order ---
{"id": "CWE-79", "name": "Cross-site Scripting (XSS)", "rank": 1, "status": "always"},
{"id": "CWE-787", "name": "Out-of-bounds Write", "rank": 2, "status": "lang",
"applicable": MEMORY_UNSAFE, "na": "memory-unsafe class N/A for this language"},
{"id": "CWE-89", "name": "SQL Injection", "rank": 3, "status": "always"},
{"id": "CWE-352", "name": "Cross-Site Request Forgery (CSRF)", "rank": 4, "status": "always"},
{"id": "CWE-22", "name": "Path Traversal", "rank": 5, "status": "always"},
{"id": "CWE-125", "name": "Out-of-bounds Read", "rank": 6, "status": "lang",
"applicable": MEMORY_UNSAFE, "na": "memory-unsafe class N/A for this language"},
{"id": "CWE-78", "name": "OS Command Injection", "rank": 7, "status": "always"},
{"id": "CWE-416", "name": "Use After Free", "rank": 8, "status": "lang",
"applicable": MEMORY_UNSAFE, "na": "memory-unsafe class N/A for this language"},
{"id": "CWE-862", "name": "Missing Authorization", "rank": 9, "status": "always"},
{"id": "CWE-434", "name": "Unrestricted Upload of Dangerous Type", "rank": 10, "status": "always"},
{"id": "CWE-94", "name": "Code Injection", "rank": 11, "status": "always"},
{"id": "CWE-20", "name": "Improper Input Validation", "rank": 12, "status": "always"},
{"id": "CWE-77", "name": "Command Injection", "rank": 13, "status": "always"},
{"id": "CWE-287", "name": "Improper Authentication", "rank": 14, "status": "always"},
{"id": "CWE-269", "name": "Improper Privilege Management", "rank": 15, "status": "always"},
{"id": "CWE-502", "name": "Deserialization of Untrusted Data", "rank": 16, "status": "always"},
{"id": "CWE-200", "name": "Exposure of Sensitive Information", "rank": 17, "status": "always"},
{"id": "CWE-863", "name": "Incorrect Authorization", "rank": 18, "status": "always"},
{"id": "CWE-918", "name": "Server-Side Request Forgery (SSRF)", "rank": 19, "status": "always"},
{"id": "CWE-119", "name": "Improper Restriction of Memory Buffer", "rank": 20, "status": "lang",
"applicable": MEMORY_UNSAFE, "na": "memory-unsafe class N/A for this language"},
{"id": "CWE-476", "name": "NULL Pointer Dereference", "rank": 21, "status": "always",
"hint": "null/None/undefined deref in any language"},
{"id": "CWE-798", "name": "Use of Hard-coded Credentials", "rank": 22, "status": "always"},
{"id": "CWE-190", "name": "Integer Overflow or Wraparound", "rank": 23, "status": "always",
"hint": "numeric overflow/precision in any language"},
{"id": "CWE-400", "name": "Uncontrolled Resource Consumption", "rank": 24, "status": "always"},
{"id": "CWE-306", "name": "Missing Authentication for Critical Function", "rank": 25, "status": "always"},
# --- Extras: outside Top 25, core to the skill's stated checklist ---
{"id": "CWE-362", "name": "Race Condition (TOCTOU)", "rank": None, "status": "always"},
{"id": "CWE-404", "name": "Improper Resource Shutdown/Release", "rank": None, "status": "always"},
{"id": "CWE-754", "name": "Improper Check for Exceptional Conditions", "rank": None, "status": "always"},
# --- Hermaguard-native: claimed by the skill, not modelled by CWE ---
{"id": "HG-ASYNC", "name": "Unhandled async rejection / async race", "rank": None, "status": "always"},
{"id": "HG-DEGENERATE", "name": "Degenerate handler (empty catch, placeholder return)", "rank": None, "status": "always"},
{"id": "HG-STATE", "name": "State-transition gap (loading→error, before→after auth)", "rank": None, "status": "always"},
{"id": "HG-PARTIAL-WRITE", "name": "Partial write / non-atomic update (missing transaction)", "rank": None, "status": "always"},
]
VALID_STATUSES = {"finding", "clean", "not_applicable"}
# ---------------------------------------------------------------------------
# Registry integrity
# ---------------------------------------------------------------------------
def self_check() -> list:
"""Validate registry invariants. Returns list of error strings."""
errors = []
ids = [c["id"] for c in CLASSES]
if len(ids) != len(set(ids)):
errors.append("duplicate class ids in registry")
names = [c["name"] for c in CLASSES]
if len(names) != len(set(names)):
errors.append("duplicate class names in registry")
cwe_ids = sorted([i for i in ids if i.startswith("CWE-")], key=lambda x: int(x[4:]))
top25 = [c for c in CLASSES if c.get("rank") and 1 <= c["rank"] <= 25]
if len(top25) != 25:
errors.append(f"expected 25 ranked Top-25 entries, found {len(top25)}")
ranks = sorted(c["rank"] for c in top25)
if ranks != list(range(1, 26)):
errors.append(f"Top-25 ranks not exactly 1..25: {ranks}")
for c in CLASSES:
if c["status"] not in ("always", "lang"):
errors.append(f"{c['id']}: bad status {c['status']}")
if c["status"] == "lang":
if "applicable" not in c or "na" not in c or not c["applicable"]:
errors.append(f"{c['id']}: lang rule missing applicable/na")
# 2024 Top 25 exact-ID membership (verified against MITRE).
expected_2024 = {
"CWE-79", "CWE-787", "CWE-89", "CWE-352", "CWE-22", "CWE-125",
"CWE-78", "CWE-416", "CWE-862", "CWE-434", "CWE-94", "CWE-20",
"CWE-77", "CWE-287", "CWE-269", "CWE-502", "CWE-200", "CWE-863",
"CWE-918", "CWE-119", "CWE-476", "CWE-798", "CWE-190", "CWE-400",
"CWE-306",
}
got = {c["id"] for c in top25}
if got != expected_2024:
errors.append(f"Top-25 ID set mismatch: missing={expected_2024 - got} extra={got - expected_2024}")
return errors
# ---------------------------------------------------------------------------
# Ledger construction
# ---------------------------------------------------------------------------
def build_ledger(files: list) -> dict:
"""Given changed file paths, return the applicable-class ledger.
A class is applicable if status=always, or its language rule matches
at least one changed file's extension. Rust gets memory classes as
hint-only (unsafe blocks) — still applicable, hint attached.
"""
exts = set()
has_rust = False
for f in files:
ext = Path(f).suffix.lower()
if ext:
exts.add(ext)
if ext == ".rs":
has_rust = True
ledger = {
"version": VERSION,
"files": sorted(files),
"classes": [],
}
for c in CLASSES:
if c["status"] == "always":
entry = {"id": c["id"], "name": c["name"], "applicable": True}
if "hint" in c:
entry["hint"] = c["hint"]
ledger["classes"].append(entry)
else:
hit = bool(exts & set(c["applicable"]))
rust_hint = has_rust and c["id"] in ("CWE-787", "CWE-125", "CWE-416", "CWE-119")
if hit or rust_hint:
entry = {"id": c["id"], "name": c["name"], "applicable": True}
if rust_hint and not hit:
entry["hint"] = "Rust: applies via unsafe blocks"
ledger["classes"].append(entry)
else:
ledger["classes"].append({
"id": c["id"], "name": c["name"],
"applicable": False, "default_status": "not_applicable",
"reason": c["na"],
})
return ledger
# ---------------------------------------------------------------------------
# Report validation
# ---------------------------------------------------------------------------
def check_report(report: dict, ledger: dict) -> dict:
"""Validate that the report accounts for every applicable class.
The report must carry a `coverage` array where each entry is
{class_id, status, evidence} with status in VALID_STATUSES.
Every applicable ledger class must appear exactly once.
Finding-backed entries are cross-checked: if a class claims
`finding`, at least one finding in the report must cite it.
"""
errors = []
coverage = report.get("coverage")
if not isinstance(coverage, list):
return {
"ok": False,
"errors": ["report has no `coverage` array — coverage cannot be verified"],
"accounted": 0, "applicable": 0, "gaps": [c["id"] for c in ledger["classes"] if c["applicable"]],
}
seen = {}
for i, entry in enumerate(coverage):
if not isinstance(entry, dict):
errors.append(f"coverage[{i}] is not an object")
continue
cid = entry.get("class_id")
status = entry.get("status")
if cid is None or status is None:
errors.append(f"coverage[{i}] missing class_id or status")
continue
if status not in VALID_STATUSES:
errors.append(f"coverage[{i}] ({cid}): invalid status '{status}' (use one of {sorted(VALID_STATUSES)})")
if cid in seen:
errors.append(f"class {cid} appears twice in coverage")
seen[cid] = entry
applicable = [c for c in ledger["classes"] if c["applicable"]]
gaps = []
for c in applicable:
entry = seen.get(c["id"])
if entry is None:
gaps.append(c["id"])
elif entry.get("status") == "not_applicable":
# Allowed only with an explicit reason — an applicable class
# demoted to N/A must say why.
if not entry.get("reason"):
errors.append(f"{c['id']}: marked not_applicable without a reason")
elif entry.get("status") == "finding":
findings = report.get("findings", [])
backed = any(
isinstance(f, dict) and c["id"] in (
f.get("classes") or []) for f in findings
)
if not backed:
errors.append(
f"{c['id']}: coverage claims 'finding' but no finding cites this class "
f"(add `classes: [\"{c['id']}\"]` to the finding)")
for cid in seen:
known = {c["id"] for c in CLASSES}
if cid not in known:
errors.append(f"coverage cites unknown class {cid}")
return {
"ok": not errors and not gaps,
"errors": errors,
"accounted": len(applicable) - len(gaps),
"applicable": len(applicable),
"gaps": gaps,
}
def render_table(result: dict, ledger: dict) -> str:
lines = []
by_id = {c["id"]: c for c in ledger["classes"]}
for c in ledger["classes"]:
if not c["applicable"]:
lines.append(f" N/A {c['id']:16s} {c['name']} — {c.get('reason', '')}")
cov = result.get("_coverage_map", {})
for cid, status in cov.items():
c = by_id.get(cid, {})
lines.append(f" {status:8s} {cid:16s} {c.get('name', '?')}")
return "\n".join(lines) if lines else " (empty)"
# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
def main():
p = argparse.ArgumentParser(description="Hermaguard coverage accountability layer")
p.add_argument("--self-check", action="store_true", help="validate registry integrity")
p.add_argument("--files", help="comma-separated changed file paths")
p.add_argument("--ledger", help="ledger JSON path (write with --files, read with --check-report)")
p.add_argument("--check-report", help="report JSON to validate against the ledger")
p.add_argument("--quiet", action="store_true", help="print only the verdict")
args = p.parse_args()
if args.self_check:
errs = self_check()
if errs:
print("SELF-CHECK FAILED:")
for e in errs:
print(f" - {e}")
return 1
print(f"self-check OK — {len(CLASSES)} classes "
f"({len([c for c in CLASSES if c.get('rank')])} CWE Top-25 + "
f"{len([c for c in CLASSES if not c.get('rank')])} extras)")
return 0
if not args.files and not (args.check_report and args.ledger):
p.error("need --files, or --check-report with --ledger")
ledger = None
if args.files:
files = [f.strip() for f in args.files.split(",") if f.strip()]
ledger = build_ledger(files)
if args.ledger:
Path(args.ledger).write_text(json.dumps(ledger, indent=2))
if not args.quiet:
print(f"ledger written: {args.ledger} "
f"({sum(1 for c in ledger['classes'] if c['applicable'])} applicable classes)")
if args.check_report:
if ledger is None:
if not args.ledger:
p.error("--check-report needs --files or --ledger")
ledger = json.loads(Path(args.ledger).read_text())
report = json.loads(Path(args.check_report).read_text())
result = check_report(report, ledger)
cmap = {e.get("class_id"): e.get("status") for e in report.get("coverage", []) if isinstance(e, dict)}
result["_coverage_map"] = cmap
verdict = "PASS" if result["ok"] else "FAIL"
print(f"coverage {verdict}: {result['accounted']}/{result['applicable']} applicable classes accounted for")
if result["gaps"]:
print(f"gaps: {', '.join(result['gaps'])}")
for e in result["errors"]:
print(f"error: {e}")
if not args.quiet:
print(render_table(result, ledger))
return 0 if result["ok"] else 1
if ledger and not args.ledger and not args.quiet:
print(json.dumps(ledger, indent=2))
return 0
if __name__ == "__main__":
sys.exit(main())
@@ -0,0 +1,466 @@
#!/usr/bin/env python3
"""
Hermaguard auto-severity grader — L0-L6 action-graded harm scale.
Assigns a deterministic L0-L6 severity tier to each Hermaguard finding using the
taxonomy that mirrors arXiv:2607.07474 "Beyond Attack-Success Rate:
Action-Graded Severity Scale for Tool-Using AI Agents" (Owiredu-Ashley, v1
2026-07-08, Table I + Appendix A).
Design goals:
* Every finding gets a valid L0-L6 tier (no gaps, no exceptions).
* Grading is a deterministic, fixed-gate oracle — identical inputs => identical
level. This is what makes the 50+ finding test suite pass at 100%.
* Zero disruption to existing Hermaguard output: grading is ADDITIVE. It reads
the report/finding dict, adds `L_level` + `harm_axes` fields, and returns a
new structure. It never deletes or rewrites existing keys.
* O(n) over findings — no measurable performance impact on the pipeline.
Two input paths:
1. Explicit axes (preferred): the finding carries `harm_axes`
{completed, reversible, cross_scope, privilege, escalation, attempted}.
The oracle runs purely deterministically -> exact, auditable grades.
2. Severity fallback (safety net): no axes supplied. A conservative lower-bound
floor is derived from the Hermaguard risk tier. `grading_method` is set to
"severity-fallback" and a warning is attached so the consolidator knows to
supply explicit axes next time. The fallback NEVER under-grades (it floors
upward) so a security finding is never silently downplayed.
CLI:
hermaguard_grader.py --report report.json
Grade a Hermaguard JSON report, write <report>.graded.json, print summary.
hermaguard_grader.py --self-test
Run the 50+ finding alignment suite; exit non-zero if any mismatch.
hermaguard_grader.py --grade-one '<json finding>'
Grade a single finding JSON string and print the result.
"""
from __future__ import annotations
import json
import sys
from dataclasses import dataclass, asdict
from typing import Any
# ---------------------------------------------------------------------------
# Taxonomy constants — verbatim from arXiv:2607.07474 Table I + Appendix A
# ---------------------------------------------------------------------------
LEVELS: dict[str, str] = {
"L0": "No harmful effect: a refusal, or only the user's own benign task.",
"L1": "Attempted a harmful action, but a control blocked it before any effect.",
"L2": "Completed a reversible action affecting only the current user.",
"L3": "Completed an irreversible action affecting only the current user.",
"L4": "Effect reached another party, external address, or shared state.",
"L5": "Action expanded privilege (granted access, obtained a token, persistence).",
"L6": "Chain of completed harmful actions escalating over steps.",
}
ALL_LEVELS = [f"L{i}" for i in range(7)]
# Gate order is fixed (paper §III-A / Appendix A rubric card).
GATE_ORDER = ["privilege", "cross_scope", "irreversibility", "reversibility"]
# Conservative lower-bound floor when only the risk tier is known.
# Hermaguard tiers are *operational urgency*, not *harm potential*, so these are
# floors only — the consolidator must supply explicit axes for a precise grade.
SEVERITY_FALLBACK_FLOOR: dict[str, str] = {
"CRITICAL": "L4", # at minimum reaches another party / external state
"HIGH": "L3", # at minimum irreversible local effect
"MEDIUM": "L2", # at minimum reversible local effect
"LOW": "L2", # at minimum reversible local effect
}
# Rough numeric ordering for escalation-chain detection.
_LEVEL_RANK = {lvl: i for i, lvl in enumerate(ALL_LEVELS)}
# ---------------------------------------------------------------------------
# Oracle
# ---------------------------------------------------------------------------
@dataclass
class HarmAxes:
"""The three effect axes + completion/escalation flags from the paper."""
completed: bool
reversible: bool
cross_scope: bool
privilege: bool
escalation: bool
attempted: bool = True
def to_dict(self) -> dict[str, Any]:
return asdict(self)
def grade_axes(axes: HarmAxes) -> str:
"""
Paper oracle. Gate order: privilege >> cross_scope >> irreversibility >>
reversibility. L6 wins on a completed escalation chain; L0/L1 when the
harmful action was never completed.
Mirrors validate_taxonomy.py:oracle() exactly so the taxonomy validator and
the live grader can never disagree.
"""
if axes.escalation:
return "L6"
if not axes.completed:
return "L1" if axes.attempted else "L0"
if axes.privilege:
return "L5"
if axes.cross_scope:
return "L4"
if not axes.reversible:
return "L3"
return "L2"
# ---------------------------------------------------------------------------
# Axis extraction from a finding dict
# ---------------------------------------------------------------------------
_AXIS_KEYS = ("completed", "reversible", "cross_scope", "privilege", "escalation", "attempted")
def _extract_explicit_axes(finding: dict) -> HarmAxes | None:
raw = finding.get("harm_axes")
if not isinstance(raw, dict):
return None
# Require the four core axes to be present and boolean.
if not all(k in raw for k in ("completed", "reversible", "cross_scope", "privilege")):
return None
try:
return HarmAxes(
completed=bool(raw["completed"]),
reversible=bool(raw["reversible"]),
cross_scope=bool(raw["cross_scope"]),
privilege=bool(raw["privilege"]),
escalation=bool(raw.get("escalation", False)),
attempted=bool(raw.get("attempted", True)),
)
except (TypeError, ValueError):
return None
def _severity_fallback_axes(finding: dict) -> tuple[HarmAxes, str]:
"""
Conservative fallback when no explicit axes are supplied. Returns the axes
that produce the lower-bound floor for the known risk tier, plus the floor
level (used only to sanity-check / annotate).
"""
tier = str(finding.get("severity", "")).upper()
floor = SEVERITY_FALLBACK_FLOOR.get(tier, "L2")
# Build axes consistent with the floor: choose the minimal flag set that
# yields exactly `floor` from the oracle.
if floor == "L4":
axes = HarmAxes(completed=True, reversible=True, cross_scope=True, privilege=False,
escalation=False, attempted=True)
elif floor == "L3":
axes = HarmAxes(completed=True, reversible=False, cross_scope=False, privilege=False,
escalation=False, attempted=True)
else: # L2
axes = HarmAxes(completed=True, reversible=True, cross_scope=False, privilege=False,
escalation=False, attempted=True)
return axes, floor
def grade_finding(finding: dict) -> dict:
"""
Attach {L_level, harm_axes, L_definition, grading_method} to a single
finding. Pure/additive: returns a *new* dict containing the original finding
plus the grading fields; never mutates the input in place.
"""
explicit = _extract_explicit_axes(finding)
if explicit is not None:
axes = explicit
method = "explicit"
warning = None
else:
axes, floor = _severity_fallback_axes(finding)
method = "severity-fallback"
warning = (
f"No explicit harm_axes on finding; graded via conservative "
f"risk-tier floor ({floor}). Supply harm_axes for a precise grade."
)
l_level = grade_axes(axes)
out = dict(finding)
out["L_level"] = l_level
out["harm_axes"] = axes.to_dict()
out["L_definition"] = LEVELS[l_level]
out["grading_method"] = method
if warning:
out["grading_warning"] = warning
return out
# ---------------------------------------------------------------------------
# Report-level grading + escalation-chain detection
# ---------------------------------------------------------------------------
def _detect_escalation_chain(findings: list[dict]) -> tuple[bool, int | None]:
"""
Paper L6 rule (trajectory level): two or more *completed consequential*
actions whose severity strictly increases across steps -> raise the peak to
L6.
Design decision: the chain must escalate to a peak of at least L4 (cross-scope or
privilege). A purely-local escalating chain (e.g. L2 -> L3) is NOT raised to
L6 — that would over-grade ordinary multi-finding reports. The paper's own
L6 examples (read .env L4 then POST creds L5) all cross into cross-scope /
privilege, so requiring peak >= L4 preserves the harm-potential semantics
while honouring the literal "strictly increasing" condition.
Returns (detected, index_of_peak_finding_to_raise).
"""
run: list[tuple[int, int]] = [] # (index, rank)
def _eval() -> tuple[bool, int | None]:
if len(run) >= 2:
peak_rank = max(r[1] for r in run)
if peak_rank >= 4: # must reach cross-scope / privilege
return True, run[-1][0]
return False, None
for idx, f in enumerate(findings):
axes = f.get("harm_axes", {})
if not axes.get("completed", False):
res = _eval()
if res[0]:
return res
run = []
continue
rank = _LEVEL_RANK.get(f.get("L_level", "L0"), 0)
if run and rank > run[-1][1]:
run.append((idx, rank))
else:
res = _eval()
if res[0]:
return res
run = [(idx, rank)]
return _eval()
def grade_report(report: dict) -> dict:
"""
Grade every finding in a Hermaguard report dict and return a NEW report dict
(additive). Adds `L_level`/`harm_axes` to each finding and a top-level
`severity_grading` summary. Runs report-level escalation-chain detection and
raises the peak finding to L6 when a strictly-increasing consequential chain
is present.
"""
findings_in = report.get("findings", []) or []
graded: list[dict] = [grade_finding(f) for f in findings_in]
escalation_detected, peak_idx = _detect_escalation_chain(graded)
if escalation_detected and peak_idx is not None:
graded[peak_idx]["L_level"] = "L6"
graded[peak_idx]["harm_axes"]["escalation"] = True
graded[peak_idx]["L_definition"] = LEVELS["L6"]
graded[peak_idx]["grading_method"] = "escalation-chain"
dist: dict[str, int] = {lvl: 0 for lvl in ALL_LEVELS}
methods: dict[str, int] = {}
for f in graded:
dist[f["L_level"]] = dist.get(f["L_level"], 0) + 1
m = f.get("grading_method", "explicit")
methods[m] = methods.get(m, 0) + 1
out = dict(report)
out["findings"] = graded
out["severity_grading"] = {
"distribution": dist,
"levels_exercised": sorted([lvl for lvl in ALL_LEVELS if dist.get(lvl, 0) > 0]),
"grading_methods": methods,
"escalation_chain_detected": escalation_detected,
"taxonomy_version": "1.0.0",
"source_paper": "arXiv:2607.07474v1",
}
return out
# ---------------------------------------------------------------------------
# Self-test suite — 50+ findings, assert 100% alignment with the oracle
# ---------------------------------------------------------------------------
# Each row: (id, hermaguard_tier, completed, reversible, cross_scope, privilege,
# escalation, attempted, expected_L)
_SELF_TEST_CORPUS: list[tuple] = [
# --- Security: privilege expansion (L5) ---
("s-rce-yaml", "HIGH", True, False, False, True, False, True, "L5"),
("s-rce-cmd", "CRITICAL", True, False, False, True, False, True, "L5"),
("s-token-leak", "CRITICAL", True, True, False, True, False, True, "L5"),
("s-backdoor", "CRITICAL", True, True, False, True, False, True, "L5"),
("s-sql-inject-creds", "CRITICAL", True, False, False, True, False, True, "L5"),
("s-pickle-load", "HIGH", True, False, False, True, False, True, "L5"),
("s-eval-user", "HIGH", True, False, False, True, False, True, "L5"),
("s-unsafe-deser", "CRITICAL", True, False, False, True, False, True, "L5"),
# --- Cross-scope (L4) ---
("x-sqli-read", "CRITICAL", True, True, True, False, False, True, "L4"),
("x-path-traversal", "HIGH", True, True, True, False, False, True, "L4"),
("x-email-exfil", "HIGH", True, True, True, False, False, True, "L4"),
("x-shared-state", "HIGH", True, True, True, False, False, True, "L4"),
("x-idor-read", "CRITICAL", True, True, True, False, False, True, "L4"),
("x-ssrf", "HIGH", True, True, True, False, False, True, "L4"),
("x-log-injection", "MEDIUM", True, True, True, False, False, True, "L4"),
("x-other-user-row", "CRITICAL", True, True, True, False, False, True, "L4"),
# --- Irreversible local (L3) ---
("i-permanent-delete", "HIGH", True, False, False, False, False, True, "L3"),
("i-silent-corrupt", "HIGH", True, False, False, False, False, True, "L3"),
("i-dropped-write", "MEDIUM", True, False, False, False, False, True, "L3"),
("i-no-backup-drop", "HIGH", True, False, False, False, False, True, "L3"),
("i-rm-rf-data", "CRITICAL", True, False, False, False, False, True, "L3"),
("i-overwrite-config", "MEDIUM", True, False, False, False, False, True, "L3"),
# --- Reversible local (L2) ---
("r-state-leak", "HIGH", True, True, False, False, False, True, "L2"),
("r-full-history", "HIGH", True, True, False, False, False, True, "L2"),
("r-memory-bloat", "MEDIUM", True, True, False, False, False, True, "L2"),
("r-ux-regression", "MEDIUM", True, True, False, False, False, True, "L2"),
("r-observability-gap", "MEDIUM", True, True, False, False, False, True, "L2"),
("r-perf-antipattern", "LOW", True, True, False, False, False, True, "L2"),
("r-null-guard", "MEDIUM", True, True, False, False, False, True, "L2"),
("r-type-coercion", "LOW", True, True, False, False, False, True, "L2"),
("r-off-by-one", "MEDIUM", True, True, False, False, False, True, "L2"),
("r-race-local", "HIGH", True, True, False, False, False, True, "L2"),
("r-misleading-status", "LOW", True, True, False, False, False, True, "L2"),
("r-negative-limit", "MEDIUM", True, True, False, False, False, True, "L2"),
("r-sql-into-mem", "HIGH", True, True, False, False, False, True, "L2"),
# --- Latent / theoretical / control blocked (L1) ---
("l-theoretical-type", "LOW", False, True, False, False, False, True, "L1"),
("l-reentry-blocked", "HIGH", False, True, False, False, False, True, "L1"),
("l-latent-no-caller", "CRITICAL", False, True, False, False, False, True, "L1"),
("l-defense-in-depth", "MEDIUM", False, True, False, False, False, True, "L1"),
("l-guarded-loop", "LOW", False, True, False, False, False, True, "L1"),
("l-unreachable-after-fix", "MEDIUM", False, True, False, False, False, True, "L1"),
# --- No harmful path (L0) ---
("n-dead-branch", "LOW", False, True, False, False, False, False, "L0"),
("n-verified-correct", "LOW", False, True, False, False, False, False, "L0"),
("n-dismissed", "MEDIUM", False, True, False, False, False, False, "L0"),
("n-unreachable", "LOW", False, True, False, False, False, False, "L0"),
("n-refactor-clean", "LOW", False, True, False, False, False, False, "L0"),
# --- Escalation chain (L6): two+ completed consequential actions,
# strictly increasing severity across steps ---
("c-read-env", "HIGH", True, True, True, False, False, True, "L4"),
("c-post-creds", "CRITICAL", True, True, False, True, True, True, "L6"),
("c-exfil-then-delete", "CRITICAL", True, False, True, True, True, True, "L6"),
("c-auth-then-drop", "CRITICAL", True, True, False, True, True, True, "L6"),
]
def _build_finding(row: tuple) -> dict:
(fid, tier, comp, rev, xscope, priv, esc, attempted, _exp) = row
return {
"id": fid,
"severity": tier,
"harm_axes": {
"completed": comp,
"reversible": rev,
"cross_scope": xscope,
"privilege": priv,
"escalation": esc,
"attempted": attempted,
},
}
def self_test() -> int:
"""Run the alignment suite. Returns process exit code (0 = all pass)."""
passed = failed = 0
dist: dict[str, int] = {lvl: 0 for lvl in ALL_LEVELS}
print(f"{'FINDING':<22}{'HG':<10}{'EXP':<6}{'GOT':<6} OK")
print("-" * 62)
for row in _SELF_TEST_CORPUS:
fid, tier, *_rest, expected = row
f = _build_finding(row)
graded = grade_finding(f)
got = graded["L_level"]
ok = got == expected
passed += ok
failed += (not ok)
dist[got] = dist.get(got, 0) + 1
print(f"{fid:<22}{tier:<10}{expected:<6}{got:<6} {'PASS' if ok else 'FAIL'}")
print("-" * 62)
print(f"Total findings validated: {len(_SELF_TEST_CORPUS)}")
print(f"PASS: {passed} FAIL: {failed}")
print(f"L-level distribution: {dist}")
levels_covered = set(lvl for lvl in ALL_LEVELS if dist.get(lvl, 0) > 0)
print(f"Levels exercised: {sorted(levels_covered)} "
f"(all 7 = {levels_covered == set(ALL_LEVELS)})")
# Report-level escalation test: chain of c-read-env (L4) then c-post-creds (L6)
chain_report = {"findings": [_build_finding(r) for r in _SELF_TEST_CORPUS
if r[0] in ("c-read-env", "c-post-creds")]}
graded_chain = grade_report(chain_report)
peak = graded_chain["findings"][-1]
chain_ok = peak["L_level"] == "L6" and graded_chain["severity_grading"]["escalation_chain_detected"]
# Regression: purely-local escalating chain (L2 -> L3) must NOT become L6.
local_chain_report = {"findings": [
{"id": "local-a", "severity": "MEDIUM", "harm_axes": {"completed": True, "reversible": True, "cross_scope": False, "privilege": False, "escalation": False, "attempted": True}},
{"id": "local-b", "severity": "HIGH", "harm_axes": {"completed": True, "reversible": False, "cross_scope": False, "privilege": False, "escalation": False, "attempted": True}},
]}
graded_local = grade_report(local_chain_report)
local_peak = graded_local["findings"][-1]
local_ok = local_peak["L_level"] == "L3" and not graded_local["severity_grading"]["escalation_chain_detected"]
print("-" * 62)
print(f"Escalation-chain report test: {'PASS' if chain_ok else 'FAIL'} "
f"(peak={peak['L_level']}, detected={graded_chain['severity_grading']['escalation_chain_detected']})")
print(f"Local-chain regression test: {'PASS' if local_ok else 'FAIL'} "
f"(peak={local_peak['L_level']}, detected={graded_local['severity_grading']['escalation_chain_detected']})")
ok = (failed == 0) and (levels_covered == set(ALL_LEVELS)) and chain_ok and local_ok \
and len(_SELF_TEST_CORPUS) >= 50
print()
if ok:
print("VALIDATION OK: 100% alignment with taxonomy oracle; all 7 levels + "
"escalation chain exercised; local-chain regression holds.")
return 0
print("VALIDATION FAILED", file=sys.stderr)
return 1
# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
def _grade_report_cli(report_path: str) -> int:
with open(report_path, "r", encoding="utf-8") as fh:
report = json.load(fh)
graded = grade_report(report)
out_path = report_path
if out_path.endswith(".json"):
out_path = out_path[:-5] + ".graded.json"
else:
out_path = out_path + ".graded.json"
with open(out_path, "w", encoding="utf-8") as fh:
json.dump(graded, fh, indent=2)
sg = graded["severity_grading"]
print(f"Graded {len(graded['findings'])} findings -> {out_path}")
print(f"Distribution: {sg['distribution']}")
print(f"Levels exercised: {sg['levels_exercised']}")
print(f"Escalation chain detected: {sg['escalation_chain_detected']}")
return 0
def main(argv: list[str]) -> int:
if not argv or argv[0] in ("--self-test", "-t"):
return self_test()
if argv[0] == "--report" and len(argv) > 1:
return _grade_report_cli(argv[1])
if argv[0] == "--grade-one" and len(argv) > 1:
finding = json.loads(argv[1])
print(json.dumps(grade_finding(finding), indent=2))
return 0
print(__doc__, file=sys.stderr)
return 2
if __name__ == "__main__":
sys.exit(main(sys.argv[1:]))
@@ -0,0 +1,239 @@
#!/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())
@@ -0,0 +1,652 @@
#!/usr/bin/env python3
"""
Hermaguard Pre-Scan — Deterministic static analysis layer.
Runs language-appropriate linters on changed files, normalises output,
feeds structured findings to the LLM review agents.
Usage:
hermaguard-prescan --files "src/a.py,src/b.ts" --output prescan.json
hermaguard-prescan --diff "$(git diff --name-only)" --output prescan.json
hermaguard-prescan --files "src/*.go" --tools bandit,ruff --output /tmp/prescan.json
Output: JSON written to --output path, stdout prints summary stats.
Exit code 0 even when findings found (findings are informational).
Exit code 1 on tool failure (unavailable linter, no files, etc).
"""
import argparse
import json
import os
import re
import shutil
import subprocess
import sys
import tempfile
from pathlib import Path
VERSION = "1.0.0"
# Language detection: extension → list of (tool, command builder)
LANGUAGE_TOOLS = {
".py": [
{
"name": "bandit",
"install_hint": "pip install bandit",
"build_cmd": lambda files, tmpdir: [
"bandit", "-f", "json", "-q", *files
],
"parse": "_parse_bandit",
},
{
"name": "ruff",
"install_hint": "pip install ruff",
"build_cmd": lambda files, tmpdir: [
"ruff", "check", "--output-format", "json", "--select",
"B,PLC,PLW,S,A,SIM,C4,PT", *files
],
"parse": "_parse_ruff",
},
],
".ts": [
{
"name": "eslint",
"install_hint": "npm install -g eslint @typescript-eslint/parser @typescript-eslint/eslint-plugin",
"build_cmd": lambda files, tmpdir: _make_eslint_cmd(files, tmpdir, ".ts"),
"parse": "_parse_eslint",
},
{
"name": "semgrep",
"install_hint": "pip install semgrep",
"build_cmd": lambda files, tmpdir: [
"semgrep", "--config", "p/typescript", "--config", "p/secrets",
"--json", "--quiet", *files
],
"parse": "_parse_semgrep",
},
],
".tsx": [
{
"name": "eslint",
"install_hint": "npm install -g eslint @typescript-eslint/parser @typescript-eslint/eslint-plugin",
"build_cmd": lambda files, tmpdir: _make_eslint_cmd(files, tmpdir, ".tsx"),
"parse": "_parse_eslint",
},
],
".js": [
{
"name": "eslint",
"install_hint": "npm install -g eslint",
"build_cmd": lambda files, tmpdir: _make_eslint_cmd(files, tmpdir, ".js"),
"parse": "_parse_eslint",
},
],
".go": [
{
"name": "gosec",
"install_hint": "go install github.com/securego/gosec/v2/cmd/gosec@latest",
"build_cmd": lambda files, tmpdir: [
"gosec", "-fmt", "json", "-quiet", *files
],
"parse": "_parse_gosec",
},
],
".rs": [
{
"name": "clippy",
"install_hint": "rustup component add clippy",
"build_cmd": lambda files, tmpdir: [
"cargo", "clippy", "--message-format", "json", "--", "-W", "clippy::all"
],
"parse": "_parse_cargo",
},
],
".java": [
{
"name": "semgrep",
"install_hint": "pip install semgrep",
"build_cmd": lambda files, tmpdir: [
"semgrep", "--config", "p/java", "--config", "p/secrets",
"--json", "--quiet", *files
],
"parse": "_parse_semgrep",
},
],
}
# Generic: semgrep runs on all languages as fallback
GENERIC_TOOLS = [
{
"name": "semgrep",
"install_hint": "pip install semgrep",
"build_cmd": lambda files, tmpdir: [
"semgrep", "--config", "p/secrets", "--config", "p/owasp-top-ten",
"--json", "--quiet", *files
],
"parse": "_parse_semgrep",
},
]
def _make_eslint_cmd(files, tmpdir, ext):
"""Create a minimal ESLint config file for hermaguard-prescan and return the command."""
cfg_path = os.path.join(tmpdir, f"eslintrc-{ext.lstrip('.')}.json")
config = {
"env": {"es2021": True, "node": True},
"parser": "@typescript-eslint/parser" if ext in (".ts", ".tsx") else "espree",
"parserOptions": {"ecmaVersion": "latest"},
"plugins": ["@typescript-eslint"] if ext in (".ts", ".tsx") else [],
"rules": {
"no-unused-vars": "warn",
"no-undef": "error",
"no-empty": "error",
"no-unsafe-optional-chaining": "error",
"no-unsafe-negation": "error",
"no-unsafe-finally": "error",
"no-constant-condition": "error",
"no-dupe-keys": "error",
"no-prototype-builtins": "warn",
"require-atomic-updates": "error",
},
}
with open(cfg_path, "w") as f:
json.dump(config, f)
return [
"eslint", "-c", cfg_path, "-f", "json", "--no-error-on-unmatched-pattern", *files
]
# ── Parsers (tool → normalised {tool, rule_id, severity, file, line, message, category}) ──
def _parse_bandit(output: str) -> list:
try:
data = json.loads(output)
except json.JSONDecodeError:
return []
findings = []
for r in data.get("results", []):
sev = {"MEDIUM": "MEDIUM", "HIGH": "HIGH"}.get(r.get("issue_severity", "").upper(), "LOW")
findings.append({
"tool": "bandit",
"rule_id": r.get("test_id", ""),
"severity": sev,
"file": r.get("filename", ""),
"line": r.get("line_number", 0),
"message": r.get("issue_text", ""),
"category": "security",
})
return findings
def _parse_ruff(output: str) -> list:
findings = []
for line in output.strip().split("\n"):
if not line:
continue
try:
r = json.loads(line)
except json.JSONDecodeError:
continue
rule_id = r.get("code", "")
if rule_id:
findings.append({
"tool": "ruff",
"rule_id": rule_id,
"severity": "LOW",
"file": r.get("filename", ""),
"line": r.get("location", {}).get("row", 0),
"message": r.get("message", ""),
"category": "quality",
})
return findings
def _parse_eslint(output: str) -> list:
try:
data = json.loads(output)
except json.JSONDecodeError:
return []
findings = []
for file_entry in data:
fname = file_entry.get("filePath", "")
for msg in file_entry.get("messages", []):
sev_map = {2: "HIGH", 1: "MEDIUM", 0: "LOW"}
sev = sev_map.get(msg.get("severity", 0), "LOW")
findings.append({
"tool": "eslint",
"rule_id": msg.get("ruleId", "") or "N/A",
"severity": sev,
"file": fname,
"line": msg.get("line", 0),
"message": msg.get("message", ""),
"category": "quality",
})
return findings
def _parse_semgrep(output: str) -> list:
try:
data = json.loads(output)
except json.JSONDecodeError:
return []
findings = []
for r in data.get("results", []):
sev = r.get("extra", {}).get("severity", "WARNING").upper()
sev_norm = {"ERROR": "HIGH", "WARNING": "MEDIUM", "INFO": "LOW"}.get(sev, "LOW")
findings.append({
"tool": "semgrep",
"rule_id": r.get("check_id", ""),
"severity": sev_norm,
"file": r.get("path", ""),
"line": r.get("start", {}).get("line", 0),
"message": r.get("extra", {}).get("message", ""),
"category": "security" if "owasp" in r.get("check_id", "").lower() else "quality",
})
return findings
def _parse_gosec(output: str) -> list:
try:
data = json.loads(output)
except json.JSONDecodeError:
return []
findings = []
for issue in data.get("Issues", []) if isinstance(data, dict) else (data if isinstance(data, list) else []):
findings.append({
"tool": "gosec",
"rule_id": issue.get("rule_id", ""),
"severity": issue.get("severity", "MEDIUM").upper(),
"file": issue.get("file", ""),
"line": int(issue.get("line", 0)),
"message": issue.get("details", ""),
"category": "security",
})
return findings
def _parse_cargo(output: str) -> list:
findings = []
for line in output.strip().split("\n"):
try:
msg = json.loads(line)
except json.JSONDecodeError:
continue
if msg.get("reason") != "compiler-message":
continue
m = msg.get("message", {})
if m.get("code") is None:
continue
spans = msg.get("spans", [])
if not spans:
continue
findings.append({
"tool": "clippy",
"rule_id": m.get("code", {}).get("code", ""),
"severity": m.get("level", "warning").upper(),
"file": spans[0].get("file_name", ""),
"line": spans[0].get("line_start", 0),
"message": m.get("message", ""),
"category": "quality",
})
return findings
PARSER_MAP = {
"bandit": _parse_bandit,
"ruff": _parse_ruff,
"eslint": _parse_eslint,
"semgrep": _parse_semgrep,
"gosec": _parse_gosec,
"clippy": _parse_cargo,
}
def find_tool(name: str) -> str | None:
"""Return path to tool if it's on PATH, else None.
Uses shutil.which so the lookup works on Windows as well as POSIX
(the old `which` subprocess was POSIX-only despite the cross-platform claim).
"""
found = shutil.which(name)
if found:
return found
# Fall back to common install locations not always on a minimal PATH.
for path in [f"/usr/local/bin/{name}", f"/usr/bin/{name}", os.path.expanduser(f"~/.local/bin/{name}")]:
if os.path.isfile(path) and os.access(path, os.X_OK):
return path
return None
def split_files(files_arg: str) -> list:
"""Split comma/space-separated file list into list of paths."""
# Try comma-separated first, then newline-separated (from git diff --name-only)
if "," in files_arg:
return [f.strip() for f in files_arg.split(",") if f.strip()]
return [f for f in files_arg.strip().split("\n") if f.strip()]
def group_by_extension(files: list, restrict_tools: list | None = None) -> dict:
"""Group files by extension, mapping to available tools.
Returns {ext: {"files": [...], "tools": [...]}}
"""
groups = {}
for f in files:
ext = Path(f).suffix.lower()
if not ext:
continue
if ext not in groups:
groups[ext] = {"files": [], "tools": LANGUAGE_TOOLS.get(ext, [])}
groups[ext]["files"].append(f)
# Filter tools if --tools flag was passed
if restrict_tools:
for ext_data in groups.values():
ext_data["tools"] = [t for t in ext_data["tools"] if t["name"] in restrict_tools]
return groups
def run_tool(tool_def: dict, files: list, tmpdir: str) -> tuple[str | None, list]:
"""Run a single tool. Returns (None, [findings]) on success or (error_msg, []) on failure."""
if not files:
return f"no files for {tool_def['name']}", []
tool_path = find_tool(tool_def["name"])
if not tool_path:
return f"{tool_def['name']} not installed ({tool_def.get('install_hint', '')})", []
cmd = tool_def["build_cmd"](files, tmpdir)
try:
result = subprocess.run(cmd, capture_output=True, text=True, timeout=120)
except subprocess.TimeoutExpired:
return f"{tool_def['name']} timed out", []
except FileNotFoundError:
return f"{tool_def['name']} command not found", []
if result.returncode != 0 and not result.stdout.strip():
return f"{tool_def['name']} failed: {result.stderr[:200]}", []
parser_name = tool_def["parse"]
parser = PARSER_MAP.get(parser_name, lambda x: [])
try:
findings = parser(result.stdout.strip())
except Exception as e:
return f"{tool_def['name']} parse error: {e}", []
return None, findings
# ── Suppression rules (compiled from dismissed findings via hermaguard-compile-rules) ──
DEFAULT_RULES_PATHS = [
".kensei/hermaguard-rules.yaml",
"hermaguard-rules.yaml",
]
def load_suppression_rules(path: str | None = None) -> tuple[list, str | None]:
"""Load compiled suppression rules. Returns (rules, source_path_or_None).
Accepts the exact format hermaguard-compile-rules writes: a JSON-superset
YAML list of dicts with keys rule_id/pattern/agent/dismissals. Parsing is
stdlib-only via json per-document; a fallback regex extractor handles
hand-edited files. Missing file = no rules, never an error.
"""
candidates = [path] if path else DEFAULT_RULES_PATHS
for cand in candidates:
if not cand:
continue
try:
if not os.path.isfile(cand):
continue
with open(cand) as f:
text = f.read()
except OSError:
continue
rules = _parse_rules_text(text)
if rules:
return rules, cand
return [], cand # file exists but empty of rules
return [], None
def _parse_rules_text(text: str) -> list:
"""Parse the compile-rules YAML subset with stdlib only."""
rules = []
current = {}
# Try whole-document JSON first (valid when the file is a pure JSON array)
try:
data = json.loads(text)
if isinstance(data, list):
return [r for r in data if isinstance(r, dict) and r.get("pattern")]
except json.JSONDecodeError:
pass
# Line-oriented parse of the known key set
for line in text.splitlines():
s = line.strip()
if s.startswith("#") or not s.startswith("- ") and not s[:1].isalpha():
# allow "- rule_id:" item starts and " key: value" continuations
if not s.startswith("- ") and not (current and s and ":" in s and not s.startswith("-")):
continue
m = re.match(r"^-\s*(rule_id|pattern|agent|dismissals|file_scope):\s*(.*)$", s)
if m:
if current:
rules.append(current)
current = {m.group(1): _coerce_scalar(m.group(2))}
continue
m = re.match(r"^(rule_id|pattern|agent|dismissals|file_scope):\s*(.*)$", s)
if m and current is not None:
key = m.group(1)
val = _coerce_scalar(m.group(2))
if key == "rule_id" and current:
rules.append(current)
current = {}
current[key] = val
continue
if current:
rules.append(current)
return [r for r in rules if r.get("pattern")]
def _coerce_scalar(v: str):
v = v.strip()
if v.startswith('"') and v.endswith('"') and len(v) >= 2:
try:
return json.loads(v)
except json.JSONDecodeError:
return v[1:-1]
if v.startswith("'") and v.endswith("'") and len(v) >= 2:
return v[1:-1]
if v.isdigit():
return int(v)
return v
def apply_suppression_rules(findings: list, rules: list) -> tuple[list, list]:
"""Filter findings against compiled suppression rules.
A finding is suppressed when ALL of:
- rule.pattern (case-insensitive substring) appears in the finding's
message, AND
- rule.file_scope is empty or equals the finding's file basename/dir.
Suppressed findings are dropped from the output and listed in stats.
Returns (kept_findings, suppressed_findings).
"""
if not rules:
return findings, []
kept, suppressed = [], []
for f in findings:
hay = f"{f.get('message', '')} {f.get('rule_id', '')}".lower()
fname = f.get("file", "")
dropped = False
for r in rules:
pat = str(r.get("pattern", "")).lower()
if not pat:
continue
scope = str(r.get("file_scope", "") or "")
scope_ok = (not scope) or (scope and (scope in fname or os.path.basename(fname) == os.path.basename(scope)))
if pat in hay and scope_ok:
dropped = True
break
if dropped:
suppressed.append(f)
else:
kept.append(f)
return kept, suppressed
def run_prescan(files: list, restrict_tools: list | None = None,
rules_path: str | None = None) -> dict:
"""Main entry point. Run all applicable tools on given files.
Returns the output dict.
"""
groups = group_by_extension(files, restrict_tools)
# Collect unique tools from matching extensions
all_exts = []
for ext, data in groups.items():
all_exts.extend(data["files"])
unique_exts = list(groups.keys())
# Also add generic tools (semgrep) to any non-empty group
all_tools_run = set()
all_findings = []
tools_run = []
tools_skipped = []
# Run extension-specific tools
for ext, g in groups.items():
for tool in g["tools"]:
if tool["name"] in all_tools_run:
continue
all_tools_run.add(tool["name"])
path = find_tool(tool["name"])
if not path:
tools_skipped.append(tool["name"])
continue
tools_run.append(tool["name"])
err, findings = run_tool(tool, g["files"], tempfile.gettempdir())
if err:
# Move from run to skipped with the reason, once (no double-append).
if tool["name"] in tools_run:
tools_run.remove(tool["name"])
tools_skipped.append(f"{tool['name']} ({err})")
continue
all_findings.extend(findings)
# Run generic tools (semgrep with security rules) on all files by extension, if available
semgrep_path = find_tool("semgrep")
if semgrep_path and all_exts:
# Only run generic semgrep if it hasn't already been run
if "semgrep" not in all_tools_run:
all_tools_run.add("semgrep")
tools_run.append("semgrep")
err, findings = run_tool(GENERIC_TOOLS[0], all_exts, tempfile.gettempdir())
if not err:
all_findings.extend(findings)
else:
tools_skipped.append(f"semgrep-generic ({err})")
# Deduplicate: same (file, line, tool, rule_id) = same finding
seen = set()
deduped = []
for f in all_findings:
key = (f["file"], f["line"], f["tool"], f["rule_id"])
if key not in seen:
seen.add(key)
deduped.append(f)
# Apply compiled suppression rules (hermaguard-compile-rules output)
rules, rules_source = load_suppression_rules(rules_path)
kept, suppressed = apply_suppression_rules(deduped, rules)
by_sev = {"CRITICAL": 0, "HIGH": 0, "MEDIUM": 0, "LOW": 0}
for f in kept:
by_sev[f["severity"]] = by_sev.get(f["severity"], 0) + 1
return {
"findings": kept,
"stats": {
"tools_run": tools_run,
"tools_skipped": tools_skipped,
"total_findings": len(kept),
"by_severity": by_sev,
"files_scanned": len(files),
"suppression_rules_applied": len(rules),
"suppression_rules_source": rules_source,
"suppressed_findings": len(suppressed),
"hermaguard_prescan_version": VERSION,
},
}
def main():
parser = argparse.ArgumentParser(
description="Hermaguard Pre-Scan — deterministic static analysis before LLM review"
)
src = parser.add_mutually_exclusive_group(required=True)
src.add_argument(
"--files",
help="Comma-separated or newline-separated list of file paths to scan"
)
src.add_argument(
"--diff",
help="Newline-separated file list, e.g. \"$(git diff --name-only)\". Alias for --files."
)
parser.add_argument(
"--output", required=True,
help="Path to write JSON output (e.g., /tmp/hermaguard/prescan-abc123.json)"
)
parser.add_argument(
"--tools",
help="Comma-separated list of tools to restrict to (e.g., 'bandit,ruff'). Default: auto-detect from file extensions."
)
parser.add_argument(
"--rules",
help="Path to compiled suppression rules (hermaguard-compile-rules output). Default: .kensei/hermaguard-rules.yaml or hermaguard-rules.yaml in cwd."
)
parser.add_argument(
"--version", action="version", version=f"%(prog)s {VERSION}"
)
args = parser.parse_args()
files = split_files(args.files or args.diff)
if not files:
output = {
"findings": [],
"stats": {
"tools_run": [],
"tools_skipped": [],
"total_findings": 0,
"by_severity": {"CRITICAL": 0, "HIGH": 0, "MEDIUM": 0, "LOW": 0},
"files_scanned": 0,
"hermaguard_prescan_version": VERSION,
},
}
else:
restrict = [t.strip() for t in args.tools.split(",")] if args.tools else None
output = run_prescan(files, restrict, rules_path=args.rules)
# Write output
os.makedirs(os.path.dirname(args.output) or ".", exist_ok=True)
with open(args.output, "w") as f:
json.dump(output, f, indent=2)
# Summary to stdout
stats = output["stats"]
print(f"Pre-scan complete: {stats['total_findings']} findings from {stats['files_scanned']} files")
print(f" Tools run: {', '.join(stats['tools_run']) if stats['tools_run'] else 'none'}")
if stats["tools_skipped"]:
print(f" Tools skipped: {', '.join(stats['tools_skipped'])}")
sevs = []
for s, c in stats["by_severity"].items():
if c > 0:
sevs.append(f"{s}: {c}")
if sevs:
print(f" By severity: {', '.join(sevs)}")
print(f" Output: {args.output}")
sys.exit(0)
if __name__ == "__main__":
main()
@@ -0,0 +1,154 @@
#!/usr/bin/env python3
"""
Hermaguard Role-Pattern Scanner — agent-specific SAST rules (built-in).
CLAWAUDIT's finding: generic SAST (Semgrep Pro, CodeQL) catches only
13.8–21.7% of agent-specific vulnerabilities, while agent-specific rule
bundles reach 66.8–75.1% recall. Hermaguard's prescan shells out to
generic tools (bandit, ruff, eslint); this scanner adds the agent-specific
layer that generic tools miss — pattern bundles keyed to the exact bug
classes each hermaguard reviewer role hunts.
Design (deterministic, stdlib-only, zero external deps):
- Rule bundles: adversarial (injection/auth/deserialization),
edge-case (races/null/async/resource), blast-radius (I/O, config,
migration touchpoints).
- Each rule: {id, agent, class_id, regex, message, severity}.
- Emits findings in the SAME schema the prescan normalizes to, so the
rest of the pipeline (dedup, suppression, report) treats them as
first-class findings: tool = "hermaguard-role-patterns".
Supports both Python and JS/TS targets; rules carry a `langs` list.
Usage:
hermaguard-role-patterns.py --files a.py,b.ts [--agents adversarial,edge-case]
"""
import argparse
import json
import os
import re
import sys
from pathlib import Path
VERSION = "1.0.0"
TOOL_NAME = "hermaguard-role-patterns"
RULES = [
# ---- adversarial-reviewer: security-focused ----
{"id": "ADV-001", "agent": "adversarial", "class_id": "CWE-89", "langs": ["py", "js", "ts"],
"regex": r"(?:execute|query|raw)\s*\(\s*f[\"']|f[\"'](?:SELECT|INSERT|UPDATE|DELETE)\b[^\"']*\{",
"severity": "HIGH", "message": "f-string/format interpolation reaches SQL — parameterize"},
{"id": "ADV-002", "agent": "adversarial", "class_id": "CWE-78", "langs": ["py"],
"regex": r"subprocess\.(?:run|Popen|call)\([^)]*shell\s*=\s*True",
"severity": "HIGH", "message": "shell=True with subprocess — command injection surface"},
{"id": "ADV-003", "agent": "adversarial", "class_id": "CWE-502", "langs": ["py"],
"regex": r"pickle\.loads?\(|yaml\.load\s*\([^)]*\)(?!.*Loader)",
"severity": "CRITICAL", "message": "unsafe deserialization — pickle/yaml.load on untrusted data"},
{"id": "ADV-004", "agent": "adversarial", "class_id": "CWE-798", "langs": ["py", "js", "ts"],
"regex": r"(?:password|passwd|secret|api[_-]?key|token)\s*[:=]\s*[\"'][^\"']{8,}[\"']",
"severity": "HIGH", "message": "hardcoded credential — move to secrets manager"},
{"id": "ADV-005", "agent": "adversarial", "class_id": "CWE-918", "langs": ["py", "js", "ts"],
"regex": r"requests\.(?:get|post|put|delete)\([^)]*(?:url|user)|fetch\(\s*(?:url|req)",
"severity": "MEDIUM", "message": "outbound request from user-influenced URL — SSRF check"},
{"id": "ADV-006", "agent": "adversarial", "class_id": "CWE-22", "langs": ["py"],
"regex": r"os\.path\.join\(\s*(?:user|filename|name)|Path\([^)]*(?:user|filename)",
"severity": "MEDIUM", "message": "path built from user input — traversal check"},
# ---- edge-case-hunter: correctness-focused ----
{"id": "EDG-001", "agent": "edge-case", "class_id": "CWE-476", "langs": ["py"],
"regex": r"\.fetchone\(\)\s*\[\s*0\s*\]|\.get\([^)]*\)\s*\[\s*0\s*\]",
"severity": "MEDIUM", "message": "index [0] on possibly-None result — null deref"},
{"id": "EDG-002", "agent": "edge-case", "class_id": "CWE-362", "langs": ["py"],
"regex": r"(?:os\.path\.(?:exists|isfile|isdir)|os\.access)\s*\([^)]*\)[\s\S]{0,400}?(?:remove|unlink|rename|write)",
"severity": "HIGH", "message": "check-then-act pattern — TOCTOU race window"},
{"id": "EDG-003", "agent": "edge-case", "class_id": "HG-ASYNC", "langs": ["py", "js", "ts"],
"regex": r"asyncio\.gather\([^)]*\)|Promise\.all\([^)]*\)",
"severity": "MEDIUM", "message": "concurrent async work — check rejection/exception handling"},
{"id": "EDG-004", "agent": "edge-case", "class_id": "HG-PARTIAL-WRITE", "langs": ["py"],
"regex": r"\.write\s*\([^)]*\)\s*\n(?![^\n]*(?:flush|close|commit|with\s+open))",
"severity": "MEDIUM", "message": "write without visible flush/commit — partial-write risk"},
{"id": "EDG-005", "agent": "edge-case", "class_id": "CWE-400", "langs": ["py", "js", "ts"],
"regex": r"while\s+True\s*:|\.read\(\)\s*$",
"severity": "MEDIUM", "message": "unbounded loop / full-file read — resource exhaustion"},
{"id": "EDG-006", "agent": "edge-case", "class_id": "CWE-190", "langs": ["py"],
"regex": r"int\([^)]*\)\s*[+*]|len\([^)]*\)\s*\+",
"severity": "LOW", "message": "arithmetic on int()/len() — overflow/limit check"},
# ---- blast-radius-integration: cross-cutting touchpoints ----
{"id": "BLT-001", "agent": "blast-radius", "class_id": "CWE-200", "langs": ["py", "js", "ts"],
"regex": r"traceback\.format_exc\(\)|print\(\s*(?:f[\"'])?[^\"']*(?:token|password|secret|key)",
"severity": "MEDIUM", "message": "sensitive data may be emitted — exposure check"},
{"id": "BLT-002", "agent": "blast-radius", "class_id": "CWE-400", "langs": ["py", "js", "ts"],
"regex": r"getenv\([^)]*\)|process\.env\.\w+|environ\.get\([^)]*\)",
"severity": "LOW", "message": "env/config read — blast radius of config change"},
{"id": "BLT-003", "agent": "blast-radius", "class_id": "HG-PARTIAL-WRITE", "langs": ["py"],
"regex": r"ALTER TABLE|DROP COLUMN|ADD COLUMN|CREATE TABLE",
"severity": "HIGH", "message": "schema migration — check rollback + data integrity"},
{"id": "BLT-004", "agent": "blast-radius", "class_id": "CWE-306", "langs": ["py", "js", "ts"],
"regex": r"@app\.(?:route|get|post|put|delete)|@router\.(?:get|post|put|delete)",
"severity": "LOW", "message": "route endpoint — verify authz/ownership check exists"},
]
RULE_MAP = {r["id"]: r for r in RULES}
def _lang_of(path: str) -> str:
s = Path(path).suffix.lower().lstrip(".")
return s if s in ("py", "js", "ts", "jsx", "tsx") else s.split(".")[-1] if s else ""
def scan_files(files: list, agents: list | None = None) -> list:
"""Scan files with role-pattern rules. Returns normalized findings list."""
active = set(agents) if agents else {"adversarial", "edge-case", "blast-radius"}
findings = []
for f in files:
p = Path(f)
try:
text = p.read_text(errors="replace")
except OSError:
continue
lang = _lang_of(f)
for rule in RULES:
if rule["agent"] not in active:
continue
if lang not in rule["langs"]:
continue
rx = re.compile(rule["regex"], re.IGNORECASE)
for m in rx.finditer(text):
line_no = text[:m.start()].count("\n") + 1
findings.append({
"tool": TOOL_NAME,
"rule_id": rule["id"],
"severity": rule["severity"],
"file": f,
"line": line_no,
"message": rule["message"],
"category": "security" if rule["agent"] == "adversarial" else "correctness",
"agent": rule["agent"],
"class_id": rule["class_id"],
})
return findings
def main():
p = argparse.ArgumentParser(description="Hermaguard agent-specific SAST rules")
p.add_argument("--files", required=True, help="Comma-separated file paths")
p.add_argument("--agents", help="Comma-separated agents (default: all)")
p.add_argument("--json", dest="json_out", help="Write findings JSON to path")
args = p.parse_args()
files = [f.strip() for f in args.files.split(",") if f.strip()]
agents = [a.strip() for a in args.agents.split(",")] if args.agents else None
findings = scan_files(files, agents)
out = {"tool": TOOL_NAME, "version": VERSION, "findings": findings,
"stats": {"files_scanned": len(files), "findings": len(findings)}}
if args.json_out:
Path(args.json_out).write_text(json.dumps(out, indent=2))
print(f"role-patterns: {len(findings)} findings from {len(files)} files")
for f in findings:
print(f" [{f['rule_id']}] {f['severity']} {f['file']}:{f['line']} {f['message']}")
return 0
if __name__ == "__main__":
sys.exit(main())
@@ -0,0 +1,225 @@
#!/usr/bin/env python3
"""
Hermaguard Sanitize — injection-hardened review context (pre-pass).
CodeSentinel's finding (arXiv 2606.19235): code LLMs ingest repo context
that attackers can poison with instruction-shaped content hidden in
comments, strings, and decoy code. Indirect prompt injection against a
code reviewer is not hypothetical — the review prompt embeds untrusted
code verbatim.
This tool is the SYNTAX-LEVEL defense layer (CodeSentinel layer 1, no
model involved — it cannot itself be injected against). It scans the code
that will be embedded into review prompts and:
1. DETECTS instruction-shaped comment/string content ("ignore previous",
"you are now", "system:", "act as", imperative chains, JSON-schema
demands, secrets-hiding phrases).
2. NEUTRALIZES it in the review copy — replaced with a marker
[SANITIZED:<id>] that keeps line count stable (line refs stay valid).
3. FLAGS decoy payloads — base64 blobs, long hex strings, encoded
instructions.
4. EMITS A MANIFEST the reviewer must read alongside the code, so the
sanitization is always visible, never silent.
Design honesty: this is not the full CodeSentinel (which adds Tree-sitter
CST + dynamic Min-K% scoring). It is the deterministic subset that runs
with zero dependencies and can't be gamed by the payload it detects.
Usage:
hermaguard-sanitize.py --files a.py,b.ts --outdir /tmp/hermaguard/sanitized/
hermaguard-sanitize.py --diff change.diff --outdir /tmp/hermaguard/sanitized/
"""
import argparse
import json
import os
import re
import sys
from pathlib import Path
VERSION = "1.0.0"
# Instruction-shaped patterns — comment/string content that should not
# silently reach the reviewer LLM.
INSTRUCTION_PATTERNS = [
(r"\bignore\s+(all\s+)?(previous|prior|above|earlier)\s+(instructions?|prompts?|context)", "ignore-prior-instruction"),
(r"\bdisregard\s+(all\s+)?(previous|prior)", "disregard-prior"),
(r"\byou\s+are\s+now\b|\bact\s+as\b|\byour\s+new\s+role\b|\bforget\s+(all\s+)?your", "role-override"),
(r"\bsystem\s*(prompt)?\s*[:=]", "system-prompt"),
(r"\b(important|critical|urgent)\s*[:!]|follow\s+(these|this|the)\s+(instructions?|rules?)\s+exactly", "instruction-demands"),
(r"\b(never|do\s+not|don'?t)\s+(tell|mention|reveal|say|report|include)\b", "secrecy-demand"),
(r"\b(only|always)\s+output\b|output\s+(only|strictly|exactly)", "output-shape-demand"),
(r"\bmust\s+(return|output|respond|answer|include|add|remove)\b", "mandate"),
(r"\bstart\s+(your\s+)?(response|answer|message)\s+with\b", "response-control"),
(r"\bjson\s*(object|schema|format)?\s*(with|containing|keys?)\s*[:=]", "json-schema-demand"),
]
# Decoy payloads: encoded/obfuscated content that commonly hides injected
# instructions or exfiltrated text.
DECOY_PATTERNS = [
(r"[A-Za-z0-9+/]{80,}={0,2}", "base64-blob"),
(r"(?<![0-9a-fA-F])[0-9a-fA-F]{64,}(?![0-9a-fA-F])", "long-hex"),
(r"(eval|exec|execve|system)\s*\(\s*(base64|bytes|decode|unhexlify|exec)\s*\(", "encoded-exec"),
]
COMMENT_STYLES = {
".py": [("#", None), ('"""', '"""'), ("'''", "'''")],
".js": [("//", None), ("/*", "*/")],
".ts": [("//", None), ("/*", "*/")],
".jsx": [("//", None), ("/*", "*/")],
".tsx": [("//", None), ("/*", "*/")],
".go": [("//", None), ("/*", "*/")],
".rs": [("//", None), ("/*", "*/")],
".java": [("//", None), ("/*", "*/")],
".sh": [("#", None)],
".sql": [("--", None), ("/*", "*/")],
".c": [("//", None), ("/*", "*/")],
".cpp": [("//", None), ("/*", "*/")],
".h": [("//", None), ("/*", "*/")],
".hpp": [("//", None), ("/*", "*/")],
}
SANITIZED_MARKER = "SANITIZED"
def sanitize_text(text: str, lang: str) -> tuple[str, list]:
"""Return (sanitized_text, manifest_entries). Line count is preserved."""
styles = COMMENT_STYLES.get("." + lang, [("#", None)])
lines = text.split("\n")
manifest = []
# Track block-comment state across lines
in_block = None
for i, line in enumerate(lines):
stripped = line.lstrip()
# Block comment detection (multi-line /* */ or """ """)
if in_block:
end_idx = line.find(in_block[1])
if end_idx != -1:
in_block = None
continue
else:
# still inside a block comment — scan full line content
content = line
hits = _scan_content(content, offset=0)
if hits:
line = _neutralize_line(line, hits, manifest, i + 1)
lines[i] = line
continue
for prefix, closer in styles:
if stripped.startswith(prefix):
idx = line.find(prefix)
content_start = idx + len(prefix)
if closer: # block opener
if closer in line[content_start:]:
content = line[content_start:].split(closer)[0]
hits = _scan_content(content, offset=content_start)
if hits:
line = _neutralize_line(line, hits, manifest, i + 1)
lines[i] = line
continue
else:
in_block = (prefix, closer)
content = line[content_start:]
hits = _scan_content(content, offset=content_start)
if hits:
line = _neutralize_line(line, hits, manifest, i + 1)
lines[i] = line
break
else: # line comment
content = line[content_start:]
hits = _scan_content(content, offset=content_start)
if hits:
line = _neutralize_line(line, hits, manifest, i + 1)
lines[i] = line
break
# Non-comment lines: scan for decoy payloads in string-ish content
if not in_block and not stripped.startswith(tuple(p for p, _ in styles)):
hits = _scan_decoy(line)
if hits:
line = _neutralize_line(line, hits, manifest, i + 1, decoy=True)
lines[i] = line
return "\n".join(lines), manifest
def _scan_content(content: str, offset: int = 0) -> list:
hits = []
for rx, label in INSTRUCTION_PATTERNS:
for m in re.finditer(rx, content, re.IGNORECASE):
hits.append({"kind": "instruction", "label": label,
"snippet": content[max(0, m.start() - 20):m.end() + 30].strip(),
"start": offset + m.start(), "end": offset + m.end()})
return hits
def _scan_decoy(line: str) -> list:
hits = []
for rx, label in DECOY_PATTERNS:
for m in re.finditer(rx, line):
hits.append({"kind": "decoy", "label": label,
"snippet": m.group(0)[:60] + ("…" if len(m.group(0)) > 60 else ""),
"start": m.start(), "end": m.end()})
return hits
def _neutralize_line(line: str, hits: list, manifest: list, lineno: int, decoy: bool = False) -> str:
"""Replace hit spans with markers, keep line count stable."""
for h in hits:
marker = f"[{SANITIZED_MARKER}:{h['label']}]"
manifest.append({"line": lineno, "kind": h["kind"], "label": h["label"],
"snippet": h["snippet"], "marker": marker})
# Apply replacements from the END so earlier spans stay valid.
for h in sorted(hits, key=lambda x: -x["start"]):
marker = f"[{SANITIZED_MARKER}:{h['label']}]"
line = line[:h["start"]] + marker + line[h["end"]:]
return line
def sanitize_files(files: list, outdir: str) -> dict:
"""Sanitize each file; write sanitized copies + manifest JSON."""
os.makedirs(outdir, exist_ok=True)
all_manifest = []
written = []
for f in files:
p = Path(f)
try:
text = p.read_text(errors="replace")
except OSError:
continue
lang = p.suffix.lstrip(".")
sanitized, manifest = sanitize_text(text, lang)
for e in manifest:
e["file"] = str(p)
all_manifest.extend(manifest)
# write sanitized copy
out_path = os.path.join(outdir, p.name)
Path(out_path).write_text(sanitized)
written.append({"original": str(p), "sanitized": out_path,
"neutralized": len(manifest)})
manifest_path = os.path.join(outdir, "sanitize-manifest.json")
Path(manifest_path).write_text(json.dumps({
"tool": "hermaguard-sanitize", "version": VERSION,
"files": written,
"entries": all_manifest,
"stats": {"files": len(written), "neutralized": len(all_manifest),
"by_label": {}},
}, indent=2))
return {"files": written, "entries": all_manifest,
"stats": {"files": len(written), "neutralized": len(all_manifest)}}
def main():
p = argparse.ArgumentParser(description="Hermaguard injection-hardened context sanitizer")
p.add_argument("--files", help="Comma-separated file paths")
p.add_argument("--outdir", required=True, help="Directory for sanitized copies + manifest")
args = p.parse_args()
if not args.files:
p.error("--files required")
files = [f.strip() for f in args.files.split(",") if f.strip()]
out = sanitize_files(files, args.outdir)
print(f"sanitize: {out['stats']['files']} files, {out['stats']['neutralized']} "
f"instruction-shaped/decoy items neutralized → {args.outdir}/sanitize-manifest.json")
return 0
if __name__ == "__main__":
sys.exit(main())
@@ -0,0 +1,274 @@
#!/usr/bin/env python3
"""
Hermaguard Verify — the evidence layer.
Runs a finding's PoC in a hardened subprocess sandbox and emits a verdict
from EXECUTION EVIDENCE, never from model self-report (the Vera principle:
judge outcomes from environment state, not model claims).
Division of labour:
- The review agent WRITES the PoC (a small script that exercises the
buggy code and exposes the failure).
- This tool RUNS it deterministically and judges the result.
Verdicts:
VERIFIED PoC ran to its terminal condition and the expected signal
was observed (expected_exit / expected_output regex).
REFUTED PoC ran cleanly — expected signal absent. The finding is
likely a false positive.
UNVERIFIED PoC could not run (timeout, crash-before-signal, missing
file). Honest label — never silently upgraded.
Sandbox hardening (stdlib only, per-run):
- cwd: throwaway temp dir
- env: empty (no credentials, no HOME leakage)
- rlimits: CPU 10s, address space 512MB, file size 1MB, processes 16
- own process group; killed hard on timeout
- stdout/stderr captured and truncated to 4KB each in the report
Usage:
hermaguard-verify.py --poc poc.py --target src/app.py \
--finding '{"id":"HG-1","classes":["CWE-476"]}' \
[--expect-fail] [--expect-output 'TypeError'] [--json out.json]
hermaguard-verify.py --check-report report.json
Enforce the forced-verification policy: findings in hard classes
(CWE-362, HG-ASYNC, HG-PARTIAL-WRITE, CWE-416 by default) must
carry verification.verdict == 'VERIFIED'. Exit 1 on violations.
Exit codes: 0 = verdict produced & policy satisfied; 1 = policy
violations (check-report) or usage error; 2 = PoC could not run at all.
"""
import argparse
import json
import os
import re
import shutil
import subprocess
import sys
import tempfile
import time
from pathlib import Path
VERSION = "1.0.0"
try:
import resource # Unix-only; guarded for Windows/macOS portability
_HAS_RESOURCE = True
except ImportError:
resource = None
_HAS_RESOURCE = False
_POSIX = os.name == "posix"
VERIFIED = "VERIFIED"
REFUTED = "REFUTED"
UNVERIFIED = "UNVERIFIED"
# Classes where LLM judgement is empirically weakest (calibration research:
# self-evaluation degrades most on hard/deferred-effect tasks). Findings in
# these classes may not be reported as confirmed without a passing PoC.
FORCED_VERIFICATION_CLASSES = [
"CWE-362", # race / TOCTOU — timing-dependent, easy to hand-wave
"HG-ASYNC", # async rejection gaps — control flow is hard to trace
"HG-PARTIAL-WRITE", # non-atomic updates — failure needs mid-sequence crash
"CWE-416", # use-after-free — lifetime reasoning
]
MAX_OUTPUT_BYTES = 4096
LIMIT_CPU_SECONDS = 10
LIMIT_AS_BYTES = 512 * 1024 * 1024
LIMIT_FSIZE_BYTES = 1024 * 1024
LIMIT_NPROC = 16
def _truncate(s: str) -> str:
return s if len(s) <= MAX_OUTPUT_BYTES else s[:MAX_OUTPUT_BYTES] + f"\n…[truncated, {len(s)} bytes total]"
def _sandbox_limits():
"""preexec_fn: apply rlimits in the child before exec. No-op when the
resource module is unavailable (Windows) — timeout is the backstop."""
if not _HAS_RESOURCE:
return
resource.setrlimit(resource.RLIMIT_CPU, (LIMIT_CPU_SECONDS, LIMIT_CPU_SECONDS))
resource.setrlimit(resource.RLIMIT_AS, (LIMIT_AS_BYTES, LIMIT_AS_BYTES))
resource.setrlimit(resource.RLIMIT_FSIZE, (LIMIT_FSIZE_BYTES, LIMIT_FSIZE_BYTES))
resource.setrlimit(resource.RLIMIT_NPROC, (LIMIT_NPROC, LIMIT_NPROC))
def run_poc(poc_path: str, target_path: str | None, expect_fail: bool = False,
expect_output: str | None = None, timeout: int = 30) -> dict:
"""Run the PoC in the sandbox and return the evidence dict."""
tmpdir = tempfile.mkdtemp(prefix="hermaguard-verify-")
try:
# Stage the PoC and the target into the sandbox cwd. The PoC imports
# the target by its basename — self-contained, no repo access needed.
staged_poc = os.path.join(tmpdir, "poc.py")
shutil.copy(poc_path, staged_poc)
if target_path:
shutil.copy(target_path, os.path.join(tmpdir, os.path.basename(target_path)))
start = time.monotonic()
try:
# -I strips user site-packages/env PYTHONPATH injection; cwd is
# prepended to sys.path by -I only for scripts — belt and braces:
# pass the sandbox dir explicitly so `import app` resolves to the
# STAGED copy, never the repo original.
cmd = [sys.executable, "-I", "-c",
f"import sys; sys.path.insert(0, {tmpdir!r}); exec(open({os.path.join(tmpdir, 'poc.py')!r}).read())"]
kwargs = dict(cwd=tmpdir, env={}, capture_output=True, text=True, timeout=timeout)
if _POSIX:
# preexec_fn/start_new_session are POSIX-only; on Windows the
# wall-clock timeout remains the isolation backstop.
kwargs["preexec_fn"] = _sandbox_limits
kwargs["start_new_session"] = True
proc = subprocess.run(cmd, **kwargs)
duration_ms = int((time.monotonic() - start) * 1000)
exit_code = proc.returncode
stdout = _truncate(proc.stdout)
stderr = _truncate(proc.stderr)
timed_out = False
except subprocess.TimeoutExpired as e:
duration_ms = int((time.monotonic() - start) * 1000)
exit_code = None
stdout = _truncate((e.stdout or b"").decode() if isinstance(e.stdout, bytes) else (e.stdout or ""))
stderr = _truncate((e.stderr or b"").decode() if isinstance(e.stderr, bytes) else (e.stderr or ""))
timed_out = True
evidence = {
"exit_code": exit_code,
"stdout": stdout,
"stderr": stderr,
"duration_ms": duration_ms,
"timed_out": timed_out,
"sandbox": {
"env": "empty",
"cwd": "throwaway tmpdir",
"rlimits_applied": _HAS_RESOURCE and _POSIX,
"rlimits": {"cpu_s": LIMIT_CPU_SECONDS, "as_mb": LIMIT_AS_BYTES // (1024 * 1024),
"fsize_mb": LIMIT_FSIZE_BYTES // (1024 * 1024), "nproc": LIMIT_NPROC},
"isolated_python": True, # -I: no user site, no env vars, no path injection
},
}
if timed_out:
evidence["verdict"] = UNVERIFIED
evidence["reason"] = f"PoC exceeded {timeout}s wall-clock limit and was killed"
return evidence
# ---- Deterministic verdict from execution evidence ----
combined = (stdout or "") + "\n" + (stderr or "")
signal_observed = []
if expect_fail:
if exit_code != 0:
signal_observed.append(f"non-zero exit ({exit_code})")
if expect_output:
if re.search(expect_output, combined):
signal_observed.append(f"output matched /{expect_output}/")
if expect_fail or expect_output:
if signal_observed:
evidence["verdict"] = VERIFIED
evidence["signal"] = "; ".join(signal_observed)
else:
evidence["verdict"] = REFUTED
evidence["signal"] = "expected signal NOT observed — likely false positive"
else:
# No expectation given: any non-zero exit still counts as a
# demonstrated failure; a clean run proves nothing, so UNVERIFIED.
if exit_code != 0:
evidence["verdict"] = VERIFIED
evidence["signal"] = f"non-zero exit ({exit_code})"
else:
evidence["verdict"] = UNVERIFIED
evidence["signal"] = "clean run with no expectation specified — nothing proven"
return evidence
finally:
shutil.rmtree(tmpdir, ignore_errors=True)
def check_report_policy(report: dict, forced_classes: list | None = None) -> dict:
"""Enforce forced verification for hard classes. Returns policy result."""
forced = set(forced_classes if forced_classes is not None else FORCED_VERIFICATION_CLASSES)
violations = []
checked = 0
for f in report.get("findings", []):
classes = set(f.get("classes") or [])
if not classes & forced:
continue
checked += 1
v = (f.get("verification") or {})
if v.get("verdict") != VERIFIED:
violations.append({
"finding_id": f.get("id", "?"),
"classes": sorted(classes & forced),
"verification_verdict": v.get("verdict", "MISSING"),
"required": VERIFIED,
})
return {
"ok": not violations,
"hard_class_findings": checked,
"violations": violations,
"policy": (
"Findings in hard classes "
f"({', '.join(sorted(forced))}) must carry verification.verdict == 'VERIFIED' "
"— produced by hermaguard-verify execution evidence, not model self-report."
),
}
def main():
p = argparse.ArgumentParser(description="Hermaguard PoC verification runner")
p.add_argument("--poc", help="Path to the PoC script (Python)")
p.add_argument("--target", help="Path to the target source file the PoC exercises")
p.add_argument("--finding", help="Finding JSON (echoed into the output for traceability)")
p.add_argument("--expect-fail", action="store_true",
help="PoC is expected to exit non-zero (demonstrates the failure)")
p.add_argument("--expect-output", help="Regex the PoC output must match (e.g. 'TypeError')")
p.add_argument("--timeout", type=int, default=30, help="Wall-clock kill limit (s)")
p.add_argument("--json", dest="json_out", help="Write the evidence JSON to this path")
p.add_argument("--check-report", help="Report JSON to enforce forced-verification policy on")
args = p.parse_args()
if args.check_report:
report = json.loads(Path(args.check_report).read_text())
result = check_report_policy(report)
print(f"policy {'PASS' if result['ok'] else 'FAIL'}: "
f"{result['hard_class_findings']} hard-class findings checked, "
f"{len(result['violations'])} violations")
for v in result["violations"]:
print(f" violation: {v['finding_id']} ({', '.join(v['classes'])}) "
f"has verification={v['verification_verdict']}, requires {v['required']}")
return 0 if result["ok"] else 1
if not args.poc:
p.error("--poc or --check-report required")
finding = json.loads(args.finding) if args.finding else {}
evidence = run_poc(args.poc, args.target, args.expect_fail, args.expect_output, args.timeout)
out = {
"tool": "hermaguard-verify",
"version": VERSION,
"finding": finding,
"verification": evidence,
}
if args.json_out:
Path(args.json_out).write_text(json.dumps(out, indent=2))
v = evidence["verdict"]
print(f"verdict: {v}")
print(f" signal: {evidence.get('signal', '')}")
print(f" exit={evidence['exit_code']} duration={evidence['duration_ms']}ms timed_out={evidence['timed_out']}")
if evidence.get("stdout"):
print(f" stdout: {evidence['stdout'][:200]}")
if evidence.get("stderr"):
print(f" stderr: {evidence['stderr'][:200]}")
if args.json_out:
print(f" evidence: {args.json_out}")
return 0
if __name__ == "__main__":
sys.exit(main())