Neue Skills, Referenzen & OpenWiki-Doku integriert

Umfangreiche Erweiterung der Skill-Bibliothek: Neue Skills für Humanisierung (Englisch/PT-BR), Design-Validierung, AI-SEO und Coolify-Deployment inkl. Regelwerke, Presets, Pattern-Referenzen, Testfälle und Automatisierungsskripte. Zusätzliche Skills für Revenue-Centric Design, Pier Cloud, OKF, Lebenslauf- und LinkedIn-Optimierung sowie zahlreiche Referenzdateien, Checklisten und YAML/JSON/Markdown-Templates. Einführung einer vollständigen OpenWiki-Dokumentation mit Architektur-, Domain- und Workflow-Beschreibungen, zentralem Index und automatisierten Updates. Modularer Aufbau, restriktive Lizenzen und umfassende Qualitäts- und Evaluationsmechanismen für alle neuen Inhalte.
This commit is contained in:
Tim Krampitz
2026-07-26 14:00:58 +02:00
parent 070727d5cd
commit 01046b01e4
202 changed files with 31290 additions and 0 deletions

View File

@@ -0,0 +1,134 @@
#!/usr/bin/env python3
"""Seal a security scan and generate final reports."""
import argparse
import hashlib
import json
import sqlite3
from datetime import datetime, timezone
from pathlib import Path
def _connect(scan_dir: Path) -> sqlite3.Connection:
db_path = scan_dir / "scan.db"
if not db_path.exists():
raise SystemExit(f"Database not found: {db_path}")
conn = sqlite3.connect(str(db_path))
conn.row_factory = sqlite3.Row
return conn
def _get_active_scan(conn: sqlite3.Connection) -> dict:
row = conn.execute("SELECT * FROM scans WHERE status = 'active' ORDER BY started_at DESC LIMIT 1").fetchone()
if not row:
raise SystemExit("No active scan found to seal.")
return dict(row)
def _seal(conn: sqlite3.Connection, scan_id: str) -> None:
now = datetime.now(timezone.utc).isoformat()
conn.execute("UPDATE scans SET sealed_at = ?, status = 'sealed' WHERE id = ?", (now, scan_id))
conn.commit()
def _export_findings(conn: sqlite3.Connection, scan_id: str, scan_dir: Path) -> list[dict]:
rows = conn.execute("SELECT * FROM findings WHERE scan_id = ? ORDER BY severity, file_path", (scan_id,)).fetchall()
findings = [dict(r) for r in rows]
(scan_dir / "findings.json").write_text(json.dumps(findings, indent=2))
return findings
def _severity_order(sev: str) -> int:
return {"critical": 0, "high": 1, "medium": 2, "low": 3, "info": 4}.get(sev, 5)
def _generate_report(findings: list[dict], scan: dict, scan_dir: Path) -> None:
counts: dict[str, int] = {}
for f in findings:
counts[f["severity"]] = counts.get(f["severity"], 0) + 1
total = len(findings)
lines = ["# Security Scan Report\n"]
lines.append(f"**Repository:** `{scan['repo_path']}` ")
lines.append(f"**Scan ID:** `{scan['id']}` ")
lines.append(f"**Started:** {scan['started_at']} ")
lines.append(f"**Sealed:** {scan['sealed_at']}\n")
# Executive summary
lines.append("## Executive Summary\n")
if total == 0:
lines.append("No findings were recorded during this scan.\n")
else:
lines.append(f"This scan identified **{total} finding(s)** across the repository:\n")
for sev in sorted(counts, key=_severity_order):
emoji = {"critical": "🔴", "high": "🟠", "medium": "🟡", "low": "🔵", "info": ""}.get(sev, "·")
lines.append(f"- {emoji} **{sev.capitalize()}:** {counts[sev]}")
lines.append("")
if counts.get("critical", 0) > 0:
lines.append("⚠️ Critical findings require immediate attention before deployment.\n")
# Findings table
if findings:
lines.append("## Findings Overview\n")
lines.append("| # | Severity | Category | File | Status | Title |")
lines.append("|---|----------|----------|------|--------|-------|")
for i, f in enumerate(sorted(findings, key=lambda x: _severity_order(x["severity"])), 1):
loc = f"`{f['file_path']}:{f['line_number']}`" if f["file_path"] else ""
lines.append(f"| {i} | {f['severity']} | {f['category']} | {loc} | {f['status']} | {f['title']} |")
lines.append("")
# Detailed findings
if findings:
lines.append("## Detailed Findings\n")
for i, f in enumerate(sorted(findings, key=lambda x: _severity_order(x["severity"])), 1):
lines.append(f"### {i}. {f['title']}\n")
lines.append(f"- **Severity:** {f['severity']}")
lines.append(f"- **Category:** {f['category']}")
lines.append(f"- **Status:** {f['status']}")
if f["file_path"]:
lines.append(f"- **Location:** `{f['file_path']}:{f['line_number']}`")
if f["tracking_url"]:
lines.append(f"- **Tracking:** {f['tracking_url']}")
lines.append(f"\n{f['description']}\n")
if f["evidence"]:
lines.append("**Evidence:**\n")
lines.append(f"```\n{f['evidence']}\n```\n")
(scan_dir / "report.md").write_text("\n".join(lines))
def _write_integrity(scan_dir: Path) -> str:
content = (scan_dir / "findings.json").read_bytes()
digest = hashlib.sha256(content).hexdigest()
(scan_dir / "integrity.sha256").write_text(f"{digest} findings.json\n")
return digest
def main() -> None:
parser = argparse.ArgumentParser(description="Seal scan and generate reports")
parser.add_argument("--scan-dir", required=True, help="Path to .security/ directory")
args = parser.parse_args()
scan_dir = Path(args.scan_dir).resolve()
conn = _connect(scan_dir)
scan = _get_active_scan(conn)
_seal(conn, scan["id"])
scan["sealed_at"] = datetime.now(timezone.utc).isoformat()
findings = _export_findings(conn, scan["id"], scan_dir)
_generate_report(findings, scan, scan_dir)
digest = _write_integrity(scan_dir)
conn.close()
total = len(findings)
counts = {}
for f in findings:
counts[f["severity"]] = counts.get(f["severity"], 0) + 1
print(f"Scan sealed: {scan['id']}")
print(f"Findings: {total} total — " + ", ".join(f"{k}: {v}" for k, v in sorted(counts.items(), key=lambda x: _severity_order(x[0]))))
print(f"Reports: {scan_dir / 'report.md'}, {scan_dir / 'findings.json'}")
print(f"Integrity: {digest}")
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,478 @@
#!/usr/bin/env python3
"""Penetration testing automation helpers.
Wraps system tools, pip-installed packages, and stdlib fallbacks.
Priority: system binary > pip package > stdlib.
"""
from __future__ import annotations
import argparse
import json
import shutil
import socket
import subprocess
import sys
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
from urllib.error import URLError
from urllib.request import Request, urlopen
def _run(cmd: list[str], timeout: int = 120) -> tuple[int, str, str]:
try:
r = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
return r.returncode, r.stdout, r.stderr
except FileNotFoundError:
return -1, "", f"Tool not found: {cmd[0]}"
except subprocess.TimeoutExpired:
return -2, "", f"Timeout after {timeout}s"
def _has_bin(name: str) -> bool:
return shutil.which(name) is not None
def _has_pkg(name: str) -> bool:
try:
__import__(name)
return True
except ImportError:
return False
def _now() -> str:
return datetime.now(timezone.utc).isoformat()
def _warn(msg: str) -> None:
print(f"[!] {msg}", file=sys.stderr)
def _info(msg: str) -> None:
print(f"[*] {msg}", file=sys.stderr)
# ─── DNS & WHOIS ────────────────────────────────────────────────────────────
def _dns_lookup(domain: str) -> dict[str, list[str]]:
"""DNS enumeration: dnspython > dig > socket."""
records: dict[str, list[str]] = {}
if _has_pkg("dns"):
import dns.resolver
_info("Using dnspython for DNS")
for rtype in ["A", "AAAA", "MX", "NS", "TXT", "CNAME", "SOA"]:
try:
answers = dns.resolver.resolve(domain, rtype)
records[rtype] = [str(r) for r in answers]
except Exception:
records[rtype] = []
elif _has_bin("dig"):
_info("Using dig for DNS")
for rtype in ["A", "AAAA", "MX", "NS", "TXT", "CNAME"]:
rc, out, _ = _run(["dig", "+short", domain, rtype])
records[rtype] = out.strip().splitlines() if rc == 0 and out.strip() else []
else:
_info("Stdlib fallback for DNS (limited to A records)")
try:
addrs = socket.getaddrinfo(domain, None)
records["A"] = list({a[4][0] for a in addrs if a[0] == socket.AF_INET})
records["AAAA"] = list({a[4][0] for a in addrs if a[0] == socket.AF_INET6})
except socket.gaierror:
records["A"] = []
return records
def _whois_lookup(domain: str) -> str | None:
"""WHOIS: python-whois > system whois > None."""
if _has_pkg("whois"):
import whois
_info("Using python-whois")
try:
w = whois.whois(domain)
return str(w)
except Exception:
return None
elif _has_bin("whois"):
_info("Using system whois")
rc, out, _ = _run(["whois", domain])
return out if rc == 0 else None
else:
_warn("No whois tool available (pip install python-whois)")
return None
def _subdomain_enum(domain: str) -> list[str]:
"""Subdomains: subfinder > bbot > crt.sh."""
if _has_bin("subfinder"):
_info("Using subfinder")
rc, out, _ = _run(["subfinder", "-d", domain, "-silent"], timeout=60)
return sorted(set(out.strip().splitlines())) if rc == 0 else []
if _has_pkg("bbot"):
_info("Using bbot (pip)")
rc, out, _ = _run([sys.executable, "-m", "bbot", "-t", domain, "-f", "subdomain-enum", "--silent"], timeout=120)
return sorted(set(out.strip().splitlines())) if rc == 0 else []
# crt.sh fallback
_info("Using crt.sh CT logs for subdomains")
try:
url = f"https://crt.sh/?q=%.{domain}&output=json"
req = Request(url, headers={"User-Agent": "security-specialist/1.0"})
with urlopen(req, timeout=15) as resp:
certs = json.loads(resp.read())
return sorted({e["name_value"].strip() for e in certs if "name_value" in e})[:100]
except Exception as e:
_warn(f"crt.sh failed: {e}")
return []
# ─── PORT SCANNING ──────────────────────────────────────────────────────────
def _port_scan(target: str, ports: str) -> dict[str, Any]:
"""Port scan: nmap > python3-nmap > socket scan."""
if _has_bin("nmap"):
_info("Using nmap")
port_arg = "-p-" if ports == "all" else "--top-ports 1000"
cmd = ["nmap", "-sC", "-sV", "--open", "-oN", "-"] + port_arg.split() + [target]
rc, out, _ = _run(cmd, timeout=300)
return {"tool": "nmap", "raw": out} if rc == 0 else {"tool": "nmap", "error": "scan failed"}
if _has_pkg("nmap3"):
_info("Using python3-nmap (pip)")
import nmap3
nm = nmap3.NmapScanTechniques()
try:
result = nm.nmap_tcp_scan(target, args="--top-ports 1000" if ports != "all" else "-p-")
return {"tool": "python3-nmap", "results": result}
except Exception as e:
return {"tool": "python3-nmap", "error": str(e)}
if _has_pkg("nmap"):
_info("Using python-nmap (pip)")
import nmap
nm = nmap.PortScanner()
try:
port_range = "1-65535" if ports == "all" else "1-1024"
nm.scan(target, port_range, arguments="-sV")
results = []
for host in nm.all_hosts():
for proto in nm[host].all_protocols():
for port in nm[host][proto]:
info = nm[host][proto][port]
if info["state"] == "open":
results.append({"port": port, "service": info.get("name", ""), "version": info.get("version", "")})
return {"tool": "python-nmap", "open_ports": results}
except Exception as e:
return {"tool": "python-nmap", "error": str(e)}
# Stdlib fallback
_info("Stdlib socket scan (slow, no service detection)")
return {"tool": "socket", "open_ports": _socket_scan(target, ports)}
def _socket_scan(target: str, ports: str) -> list[dict]:
port_list = list(range(1, 65536)) if ports == "all" else list(range(1, 1025))
open_ports = []
def check(port: int) -> dict | None:
try:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.settimeout(0.5)
if s.connect_ex((target, port)) == 0:
return {"port": port, "state": "open"}
except (socket.timeout, OSError):
pass
return None
with ThreadPoolExecutor(max_workers=200) as pool:
futures = {pool.submit(check, p): p for p in port_list}
for f in as_completed(futures):
r = f.result()
if r:
open_ports.append(r)
return sorted(open_ports, key=lambda x: x["port"])
# ─── WEB ENUMERATION ────────────────────────────────────────────────────────
def _dir_brute(url: str) -> list[str]:
"""Directory brute: gobuster > feroxbuster > dirsearch > urllib."""
if _has_bin("gobuster"):
wl = _find_wordlist()
if wl:
_info("Using gobuster")
rc, out, _ = _run(["gobuster", "dir", "-u", url, "-w", wl, "-q", "--no-error"], timeout=180)
return out.strip().splitlines() if rc == 0 else []
if _has_bin("feroxbuster"):
_info("Using feroxbuster")
rc, out, _ = _run(["feroxbuster", "-u", url, "-q", "--no-state"], timeout=180)
return out.strip().splitlines() if rc == 0 else []
if _has_bin("dirsearch"):
_info("Using dirsearch")
rc, out, _ = _run(["dirsearch", "-u", url, "--format=plain", "-q"], timeout=180)
return out.strip().splitlines() if rc == 0 else []
# Python fallback
_info("Stdlib URL brute (limited wordlist)")
return _python_dir_brute(url)
def _find_wordlist() -> str | None:
for path in [
"/usr/share/wordlists/dirb/common.txt",
"/usr/share/seclists/Discovery/Web-Content/common.txt",
"/usr/share/dirbuster/wordlists/directory-list-2.3-small.txt",
"/opt/wordlists/common.txt",
]:
if Path(path).exists():
return path
return None
def _python_dir_brute(url: str) -> list[str]:
common = [
"admin", "login", "api", "wp-admin", "wp-login.php", ".git", ".git/HEAD",
".env", ".env.local", "config", "backup", "phpmyadmin", "console", "debug",
"server-status", "actuator", "actuator/health", "swagger", "swagger-ui.html",
"graphql", "graphiql", ".well-known/security.txt", "robots.txt", "sitemap.xml",
"wp-json", "xmlrpc.php", "solr", "jenkins", "manager/html", "_debug_toolbar",
"elmah.axd", "trace.axd", "info.php", "phpinfo.php", ".DS_Store", ".htaccess",
"web.config", "crossdomain.xml", "clientaccesspolicy.xml",
]
found = []
for path in common:
try:
req = Request(f"{url.rstrip('/')}/{path}", method="HEAD",
headers={"User-Agent": "security-specialist/1.0"})
with urlopen(req, timeout=5) as resp:
if resp.status < 400:
found.append(f"/{path} [{resp.status}]")
except Exception:
pass
return found
def _tech_detect(url: str) -> dict | None:
"""Tech detection: whatweb > webtech > header analysis."""
if _has_bin("whatweb"):
_info("Using whatweb")
rc, out, _ = _run(["whatweb", "--color=never", "-a", "3", url])
return {"tool": "whatweb", "raw": out.strip()} if rc == 0 else None
if _has_pkg("webtech"):
_info("Using webtech (pip)")
try:
from webtech import WebTech
wt = WebTech(options={"json": True})
result = wt.start_from_url(url)
return {"tool": "webtech", "technologies": result}
except Exception:
pass
# Header-based fallback
_info("Header-based tech detection")
try:
req = Request(url, headers={"User-Agent": "security-specialist/1.0"})
with urlopen(req, timeout=10) as resp:
headers = dict(resp.headers)
tech = {}
if "X-Powered-By" in headers:
tech["powered_by"] = headers["X-Powered-By"]
if "Server" in headers:
tech["server"] = headers["Server"]
if "X-Generator" in headers:
tech["generator"] = headers["X-Generator"]
for h in ["X-AspNet-Version", "X-AspNetMvc-Version"]:
if h in headers:
tech["aspnet"] = headers[h]
return {"tool": "headers", "detected": tech}
except Exception:
return None
# ─── VULNERABILITY SCANNING ─────────────────────────────────────────────────
def _vuln_scan_web(target: str) -> dict[str, Any]:
"""Web vuln scan: nikto > wapiti3 > nuclei > basic checks."""
results: dict[str, Any] = {}
if _has_bin("nikto"):
_info("Using nikto")
rc, out, _ = _run(["nikto", "-h", target, "-Format", "txt"], timeout=300)
if rc == 0:
results["nikto"] = out
if _has_bin("wapiti"):
_info("Using wapiti3")
rc, out, _ = _run(["wapiti", "-u", target, "--flush-session", "-f", "txt", "--no-bugreport"], timeout=300)
if rc == 0:
results["wapiti"] = out
elif _has_pkg("wapitiCore"):
_info("wapiti3 available via pip — run: wapiti -u <target>")
results["wapiti_note"] = "wapiti3 installed but requires CLI invocation"
if _has_bin("nuclei"):
_info("Using nuclei")
rc, out, _ = _run(["nuclei", "-u", target, "-silent", "-nc"], timeout=300)
if rc == 0:
results["nuclei"] = out.strip().splitlines()
if not results:
_info("Running basic security header checks")
results["header_checks"] = _check_security_headers(target)
return results
def _check_security_headers(url: str) -> dict[str, str]:
"""Check common security headers as minimal vuln scan fallback."""
expected = [
"Strict-Transport-Security",
"Content-Security-Policy",
"X-Content-Type-Options",
"X-Frame-Options",
"X-XSS-Protection",
"Referrer-Policy",
"Permissions-Policy",
]
try:
req = Request(url, headers={"User-Agent": "security-specialist/1.0"})
with urlopen(req, timeout=10) as resp:
headers = dict(resp.headers)
results = {}
for h in expected:
if h in headers:
results[h] = f"{headers[h]}"
else:
results[h] = "✗ MISSING"
return results
except Exception as e:
return {"error": str(e)}
# ─── CLI COMMANDS ───────────────────────────────────────────────────────────
def recon_passive(args: argparse.Namespace) -> None:
target = args.target
results = {
"target": target,
"timestamp": _now(),
"dns": _dns_lookup(target),
"whois": _whois_lookup(target),
"subdomains": _subdomain_enum(target),
}
_output(results, args)
def recon_active(args: argparse.Namespace) -> None:
results = {
"target": args.target,
"timestamp": _now(),
"scan_type": "active",
"port_scan": _port_scan(args.target, args.ports or "top1000"),
}
_output(results, args)
def enumerate_web(args: argparse.Namespace) -> None:
url = args.url.rstrip("/")
results = {
"target": url,
"timestamp": _now(),
"directories": _dir_brute(url),
"technologies": _tech_detect(url),
}
_output(results, args)
def vuln_scan(args: argparse.Namespace) -> None:
results = {
"target": args.target,
"timestamp": _now(),
"type": args.type,
"findings": _vuln_scan_web(args.target) if args.type == "web" else _port_scan(args.target, "top1000"),
}
_output(results, args)
def check_tools(args: argparse.Namespace) -> None:
"""Show which tools are available on this system."""
tools = {
"Port scan": [("nmap", "bin"), ("python3-nmap (nmap3)", "pkg:nmap3"), ("python-nmap", "pkg:nmap"), ("socket", "stdlib")],
"DNS": [("dig", "bin"), ("dnspython", "pkg:dns"), ("socket", "stdlib")],
"WHOIS": [("whois", "bin"), ("python-whois", "pkg:whois")],
"Subdomains": [("subfinder", "bin"), ("amass", "bin"), ("bbot", "pkg:bbot"), ("crt.sh", "stdlib")],
"Dir brute": [("gobuster", "bin"), ("feroxbuster", "bin"), ("dirsearch", "bin"), ("urllib", "stdlib")],
"Tech detect": [("whatweb", "bin"), ("webtech", "pkg:webtech"), ("headers", "stdlib")],
"Vuln scan": [("nikto", "bin"), ("wapiti", "bin"), ("nuclei", "bin"), ("header check", "stdlib")],
"SQLi": [("sqlmap", "bin")],
}
print("Tool availability:\n")
for category, items in tools.items():
print(f" {category}:")
for name, check in items:
if check == "stdlib":
status = "✓ (always available)"
elif check.startswith("pkg:"):
pkg = check.split(":")[1]
status = "" if _has_pkg(pkg) else f"✗ (pip install {name.split('(')[0].strip().replace(' ', '-')})"
else:
status = "" if _has_bin(name) else "✗ (not in PATH)"
print(f" {name:30s} {status}")
print()
# ─── OUTPUT & MAIN ──────────────────────────────────────────────────────────
def _output(data: dict, args: argparse.Namespace) -> None:
out = json.dumps(data, indent=2, default=str)
if hasattr(args, "out") and args.out:
Path(args.out).write_text(out)
print(f"Results written to {args.out}")
else:
print(out)
def main() -> None:
parser = argparse.ArgumentParser(description="Penetration testing automation")
sub = parser.add_subparsers(dest="command", required=True)
p = sub.add_parser("recon-passive", help="Passive recon (DNS, WHOIS, subdomains)")
p.add_argument("--target", required=True)
p.add_argument("--out")
p.set_defaults(func=recon_passive)
p = sub.add_parser("recon-active", help="Active recon (port scanning)")
p.add_argument("--target", required=True)
p.add_argument("--ports", choices=["top1000", "all"], default="top1000")
p.add_argument("--out")
p.set_defaults(func=recon_active)
p = sub.add_parser("enumerate-web", help="Web enumeration (dirs, tech)")
p.add_argument("--url", required=True)
p.add_argument("--out")
p.set_defaults(func=enumerate_web)
p = sub.add_parser("vuln-scan", help="Vulnerability scanning")
p.add_argument("--target", required=True)
p.add_argument("--type", choices=["web", "infra"], default="web")
p.add_argument("--out")
p.set_defaults(func=vuln_scan)
p = sub.add_parser("check-tools", help="Show available tools on this system")
p.set_defaults(func=check_tools)
args = parser.parse_args()
args.func(args)
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,95 @@
#!/usr/bin/env python3
"""Rank repository files by security relevance for analysis prioritization."""
import argparse
import json
import subprocess
from pathlib import Path
SKIP_DIRS = {"node_modules", "vendor", ".git", "__pycache__", "dist", "build", ".next", "coverage", ".venv", "venv"}
SKIP_PATTERNS = {"test", "tests", "spec", "specs", "__tests__", "fixtures", "mocks", "generated"}
HIGH_KEYWORDS = ("auth", "login", "session", "token", "password", "secret", "crypto", "permission")
MEDIUM_HIGH_KEYWORDS = ("api", "handler", "controller", "route", "endpoint", "middleware")
MEDIUM_KEYWORDS = ("config", "env", "settings", "database", "migration")
CODE_EXTENSIONS = {".py", ".js", ".ts", ".tsx", ".jsx", ".go", ".rs", ".java", ".rb", ".php", ".c", ".cpp", ".h", ".cs", ".yml", ".yaml", ".toml", ".json", ".env"}
def _should_skip(path: Path) -> bool:
parts = set(path.parts)
if parts & SKIP_DIRS:
return True
return bool(parts & SKIP_PATTERNS)
def _score(path: str) -> tuple[int, str]:
low = path.lower()
for kw in HIGH_KEYWORDS:
if kw in low:
return 5, f"contains '{kw}' — security-sensitive"
for kw in MEDIUM_HIGH_KEYWORDS:
if kw in low:
return 4, f"contains '{kw}' — attack surface"
for kw in MEDIUM_KEYWORDS:
if kw in low:
return 3, f"contains '{kw}' — configuration"
return 1, "general code"
def cmd_from_repo(args: argparse.Namespace) -> None:
"""Walk repo and score all code files."""
repo = Path(args.repo).resolve()
results = []
for f in repo.rglob("*"):
if not f.is_file() or f.suffix not in CODE_EXTENSIONS:
continue
rel = f.relative_to(repo)
if _should_skip(rel):
continue
priority, reason = _score(str(rel))
results.append({"path": str(rel), "priority": priority, "reason": reason})
results.sort(key=lambda x: -x["priority"])
Path(args.out).write_text(json.dumps(results, indent=2))
print(f"Ranked {len(results)} files → {args.out}")
def cmd_from_diff(args: argparse.Namespace) -> None:
"""Rank files changed between two git refs."""
repo = Path(args.repo).resolve()
result = subprocess.run(
["git", "diff", "--name-only", args.base, args.head],
capture_output=True, text=True, cwd=str(repo), check=True,
)
results = []
for line in result.stdout.strip().splitlines():
rel = Path(line)
if _should_skip(rel) or rel.suffix not in CODE_EXTENSIONS:
continue
priority, reason = _score(line)
results.append({"path": line, "priority": priority, "reason": reason})
results.sort(key=lambda x: -x["priority"])
Path(args.out).write_text(json.dumps(results, indent=2))
print(f"Ranked {len(results)} changed files → {args.out}")
def main() -> None:
parser = argparse.ArgumentParser(description="Rank files by security relevance")
sub = parser.add_subparsers(dest="command", required=True)
p_repo = sub.add_parser("from-repo")
p_repo.add_argument("--repo", required=True)
p_repo.add_argument("--out", required=True)
p_diff = sub.add_parser("from-diff")
p_diff.add_argument("--repo", required=True)
p_diff.add_argument("--base", required=True)
p_diff.add_argument("--head", required=True)
p_diff.add_argument("--out", required=True)
args = parser.parse_args()
{"from-repo": cmd_from_repo, "from-diff": cmd_from_diff}[args.command](args)
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,175 @@
#!/usr/bin/env python3
"""SQLite-based security scan database manager."""
import argparse
import json
import sqlite3
import uuid
from datetime import datetime, timezone
from pathlib import Path
SCHEMA = """
CREATE TABLE IF NOT EXISTS scans (
id TEXT PRIMARY KEY,
repo_path TEXT NOT NULL,
started_at TEXT NOT NULL,
sealed_at TEXT,
status TEXT NOT NULL CHECK(status IN ('active', 'sealed'))
);
CREATE TABLE IF NOT EXISTS findings (
id TEXT PRIMARY KEY,
scan_id TEXT NOT NULL REFERENCES scans(id),
title TEXT NOT NULL,
severity TEXT NOT NULL CHECK(severity IN ('critical', 'high', 'medium', 'low', 'info')),
category TEXT NOT NULL,
status TEXT NOT NULL DEFAULT 'open'
CHECK(status IN ('open', 'fixed', 'false-positive', 'accepted-risk', 'tracked')),
file_path TEXT,
line_number INTEGER,
description TEXT,
evidence TEXT,
created_at TEXT NOT NULL,
tracking_url TEXT,
notes TEXT
);
"""
def _connect(repo: str) -> sqlite3.Connection:
db_path = Path(repo) / ".security" / "scan.db"
if not db_path.exists():
raise SystemExit(f"Database not found: {db_path}")
conn = sqlite3.connect(str(db_path))
conn.row_factory = sqlite3.Row
return conn
def _now() -> str:
return datetime.now(timezone.utc).isoformat()
def cmd_init(args: argparse.Namespace) -> None:
"""Initialize .security/scan.db and create a new scan record."""
sec_dir = Path(args.repo) / ".security"
sec_dir.mkdir(parents=True, exist_ok=True)
db_path = sec_dir / "scan.db"
conn = sqlite3.connect(str(db_path))
conn.executescript(SCHEMA)
scan_id = str(uuid.uuid4())
conn.execute(
"INSERT INTO scans (id, repo_path, started_at, status) VALUES (?, ?, ?, ?)",
(scan_id, str(Path(args.repo).resolve()), _now(), "active"),
)
conn.commit()
conn.close()
print(json.dumps({"scan_id": scan_id, "db": str(db_path)}))
def cmd_add_finding(args: argparse.Namespace) -> None:
"""Insert a finding into the database."""
conn = _connect(args.repo)
finding_id = str(uuid.uuid4())
conn.execute(
"""INSERT INTO findings
(id, scan_id, title, severity, category, status, file_path, line_number,
description, evidence, created_at)
VALUES (?, ?, ?, ?, ?, 'open', ?, ?, ?, ?, ?)""",
(finding_id, args.scan_id, args.title, args.severity, args.category,
args.file, args.line, args.description, args.evidence, _now()),
)
conn.commit()
conn.close()
print(json.dumps({"finding_id": finding_id}))
def cmd_list_findings(args: argparse.Namespace) -> None:
"""List findings as JSON, with optional filters."""
conn = _connect(args.repo)
query = "SELECT * FROM findings WHERE scan_id = ?"
params: list = [args.scan_id]
if args.severity:
query += " AND severity = ?"
params.append(args.severity)
if args.status:
query += " AND status = ?"
params.append(args.status)
rows = conn.execute(query, params).fetchall()
conn.close()
print(json.dumps([dict(r) for r in rows], indent=2))
def cmd_update_status(args: argparse.Namespace) -> None:
"""Update a finding's status and optional tracking metadata."""
conn = _connect(args.repo)
parts = ["status = ?"]
params: list = [args.status]
if args.tracking_url:
parts.append("tracking_url = ?")
params.append(args.tracking_url)
if args.note:
parts.append("notes = ?")
params.append(args.note)
params.append(args.finding_id)
conn.execute(f"UPDATE findings SET {', '.join(parts)} WHERE id = ?", params)
conn.commit()
conn.close()
print(json.dumps({"updated": args.finding_id}))
def cmd_stats(args: argparse.Namespace) -> None:
"""Print severity/category/status counts for a scan."""
conn = _connect(args.repo)
result: dict = {"by_severity": {}, "by_category": {}, "by_status": {}}
for col, key in [("severity", "by_severity"), ("category", "by_category"), ("status", "by_status")]:
rows = conn.execute(
f"SELECT {col}, COUNT(*) as cnt FROM findings WHERE scan_id = ? GROUP BY {col}",
(args.scan_id,),
).fetchall()
result[key] = {r[col]: r["cnt"] for r in rows}
conn.close()
print(json.dumps(result, indent=2))
def main() -> None:
parser = argparse.ArgumentParser(description="Security scan database manager")
sub = parser.add_subparsers(dest="command", required=True)
p_init = sub.add_parser("init")
p_init.add_argument("--repo", required=True)
p_add = sub.add_parser("add-finding")
p_add.add_argument("--repo", required=True)
p_add.add_argument("--scan-id", required=True)
p_add.add_argument("--title", required=True)
p_add.add_argument("--severity", required=True, choices=["critical", "high", "medium", "low", "info"])
p_add.add_argument("--category", required=True)
p_add.add_argument("--file", required=True)
p_add.add_argument("--line", type=int, required=True)
p_add.add_argument("--description", required=True)
p_add.add_argument("--evidence", required=True)
p_list = sub.add_parser("list-findings")
p_list.add_argument("--repo", required=True)
p_list.add_argument("--scan-id", required=True)
p_list.add_argument("--severity", choices=["critical", "high", "medium", "low", "info"])
p_list.add_argument("--status", choices=["open", "fixed", "false-positive", "accepted-risk", "tracked"])
p_upd = sub.add_parser("update-status")
p_upd.add_argument("--repo", required=True)
p_upd.add_argument("--finding-id", required=True)
p_upd.add_argument("--status", required=True, choices=["open", "fixed", "false-positive", "accepted-risk", "tracked"])
p_upd.add_argument("--tracking-url")
p_upd.add_argument("--note")
p_stats = sub.add_parser("stats")
p_stats.add_argument("--repo", required=True)
p_stats.add_argument("--scan-id", required=True)
args = parser.parse_args()
{"init": cmd_init, "add-finding": cmd_add_finding, "list-findings": cmd_list_findings,
"update-status": cmd_update_status, "stats": cmd_stats}[args.command](args)
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,150 @@
#!/usr/bin/env node
/**
* Valida findings.json contra report-schema.json.
* Usage: node validate-findings.cjs <findings.json>
*
* Zero dependências. Exit 0 = success, exit 1 = falha.
*/
const fs = require("fs");
const path = require("path");
const file = process.argv[2];
if (!file) {
console.error("Usage: node validate-findings.cjs <findings.json>");
process.exit(1);
}
const schemaPath = path.join(__dirname, "..", "references", "report-schema.json");
let itemSchema;
try {
const doc = JSON.parse(fs.readFileSync(schemaPath, "utf8"));
itemSchema = doc.output_schema;
if (!itemSchema) throw new Error('report-schema.json missing "output_schema"');
} catch (e) {
console.error(`Failed to load schema from ${schemaPath}:`, e.message);
process.exit(1);
}
let findings;
try {
findings = JSON.parse(fs.readFileSync(file, "utf8"));
} catch (e) {
console.error("Failed to parse JSON:", e.message);
process.exit(1);
}
if (!Array.isArray(findings)) {
console.error("findings.json must be an array");
process.exit(1);
}
function typeOf(v) {
if (Array.isArray(v)) return "array";
if (v === null) return "null";
return typeof v;
}
function findDiscriminator(schema) {
if (!schema.properties) return null;
for (const [key, sub] of Object.entries(schema.properties)) {
if (sub && Object.prototype.hasOwnProperty.call(sub, "const")) {
return { key, value: sub.const };
}
}
return null;
}
function validate(value, schema, p, errors) {
if (schema.oneOf) {
for (const branch of schema.oneOf) {
const disc = findDiscriminator(branch);
if (disc && value && typeof value === "object" && value[disc.key] === disc.value) {
validate(value, branch, p, errors);
return;
}
}
const discs = schema.oneOf.map(findDiscriminator).filter(Boolean);
if (discs.length === schema.oneOf.length && value && typeof value === "object") {
const key = discs[0].key;
const allowed = discs.map((d) => JSON.stringify(d.value)).join(", ");
errors.push(`${p}: "${key}" must be one of ${allowed}, got ${JSON.stringify(value[key])}`);
return;
}
errors.push(`${p}: does not match exactly one of the allowed schemas`);
return;
}
if (Object.prototype.hasOwnProperty.call(schema, "const") && value !== schema.const) {
errors.push(`${p}: must equal ${JSON.stringify(schema.const)}, got ${JSON.stringify(value)}`);
}
if (schema.enum && !schema.enum.includes(value)) {
errors.push(`${p}: invalid value ${JSON.stringify(value)} (expected one of ${schema.enum.join(", ")})`);
}
switch (schema.type) {
case "object": {
if (typeOf(value) !== "object") { errors.push(`${p}: expected object, got ${typeOf(value)}`); return; }
for (const req of schema.required || []) {
if (!(req in value)) errors.push(`${p}: missing required field "${req}"`);
}
for (const key of Object.keys(value)) {
if (schema.properties && key in schema.properties) {
validate(value[key], schema.properties[key], `${p}.${key}`, errors);
} else if (schema.additionalProperties === false) {
errors.push(`${p}: unexpected field "${key}"`);
}
}
break;
}
case "array": {
if (typeOf(value) !== "array") { errors.push(`${p}: expected array, got ${typeOf(value)}`); return; }
if (typeof schema.minItems === "number" && value.length < schema.minItems) {
errors.push(`${p}: must have at least ${schema.minItems} item(s), got ${value.length}`);
}
if (schema.items) {
value.forEach((el, i) => validate(el, schema.items, `${p}[${i}]`, errors));
}
break;
}
case "integer": {
if (typeOf(value) !== "number" || !Number.isInteger(value)) {
errors.push(`${p}: expected integer, got ${typeOf(value)}`);
}
break;
}
case "string": {
if (typeOf(value) !== "string") errors.push(`${p}: expected string, got ${typeOf(value)}`);
break;
}
}
}
let errorCount = 0;
findings.forEach((f, i) => {
const label = `[${i}] ${(f && (f.title || f.reason)) || "(untitled)"}`;
console.log(`Checking ${label}`);
const errs = [];
validate(f, itemSchema, `[${i}]`, errs);
// Semantic: confirmed trace must start at entrypoint and end at sink
if (f && f.verdict === "confirmed" && Array.isArray(f.trace) && f.trace.length > 0) {
if (f.trace[0] && f.trace[0].kind !== "entrypoint") {
errs.push(`[${i}].trace[0].kind must be "entrypoint", got ${JSON.stringify(f.trace[0].kind)}`);
}
const last = f.trace.length - 1;
if (f.trace[last] && f.trace[last].kind !== "sink") {
errs.push(`[${i}].trace[${last}].kind must be "sink", got ${JSON.stringify(f.trace[last].kind)}`);
}
}
for (const msg of errs) console.error(" ERROR:", msg);
errorCount += errs.length;
});
console.log();
if (errorCount === 0) {
console.log(`PASS: ${findings.length} findings valid`);
} else {
console.error(`FAIL: ${errorCount} error(s) across ${findings.length} findings`);
process.exit(1);
}