#!/usr/bin/env python3 """ Submission tracker for huntr/hackerone-style workflows. This repo already has a DB schema for this: data/findings.db.findings(status, report_path, submission_url, ...) Commands: - list-pending: show drafts that still need submission - list-next: show ranked drafts to submit next (reward-maxxing heuristic) - list-submitted: show submitted findings (and flag missing URLs) - mark: mark a finding as submitted + store submission_url Examples: python3 scripts/submission.py list-pending python3 scripts/submission.py mark --finding-id 3 --url "https://huntr.com/bounties/..." python3 scripts/submission.py mark --report reports/MLflow-FileRead-2026-02-12.md --url "https://huntr.com/bounties/..." """ from __future__ import annotations import argparse import os import sqlite3 import sys from pathlib import Path ROOT = Path(__file__).resolve().parents[1] DB_PATH = ROOT / "data" / "findings.db" def connect_db() -> sqlite3.Connection: if not DB_PATH.exists(): raise SystemExit(f"DB not found: {DB_PATH}") # sqlite3 CLI uses autocommit-ish behavior; emulate "do what I mean". conn = sqlite3.connect(str(DB_PATH)) conn.row_factory = sqlite3.Row return conn def print_table(rows: list[sqlite3.Row], cols: list[str]) -> None: if not rows: print("(none)") return def _cell(r: sqlite3.Row, c: str) -> str: try: v = r[c] except Exception: v = "" return "" if v is None else str(v) widths = {c: max(len(c), *(len(_cell(r, c)) for r in rows)) for c in cols} header = " ".join(c.ljust(widths[c]) for c in cols) sep = " ".join("-" * widths[c] for c in cols) print(header) print(sep) for r in rows: print(" ".join(_cell(r, c).ljust(widths[c]) for c in cols)) def list_pending(conn: sqlite3.Connection) -> int: rows = conn.execute( """ SELECT id, program, platform, severity, confidence, title, report_path FROM findings WHERE status='draft' ORDER BY CASE severity WHEN 'critical' THEN 1 WHEN 'high' THEN 2 WHEN 'medium' THEN 3 WHEN 'low' THEN 4 ELSE 5 END, id; """ ).fetchall() print_table(rows, ["id", "program", "platform", "severity", "confidence", "title", "report_path"]) return 0 def list_next(conn: sqlite3.Connection) -> int: # Heuristic: severity desc, confidence desc, most recently updated first. rows = conn.execute( """ SELECT id, program, repo_name, severity, confidence, title, report_path, updated_at FROM findings WHERE status='draft' ORDER BY CASE severity WHEN 'critical' THEN 4 WHEN 'high' THEN 3 WHEN 'medium' THEN 2 WHEN 'low' THEN 1 ELSE 0 END DESC, CASE confidence WHEN 'high' THEN 3 WHEN 'medium' THEN 2 WHEN 'low' THEN 1 ELSE 0 END DESC, updated_at DESC, id ASC; """ ).fetchall() print_table(rows, ["id", "program", "repo_name", "severity", "confidence", "title", "updated_at", "report_path"]) return 0 def list_submitted(conn: sqlite3.Connection) -> int: rows = conn.execute( """ SELECT id, program, platform, severity, confidence, title, submission_url, report_path FROM findings WHERE status='submitted' ORDER BY updated_at DESC, id DESC; """ ).fetchall() print_table(rows, ["id", "program", "platform", "severity", "confidence", "title", "submission_url", "report_path"]) missing = [r for r in rows if not (r["submission_url"] or "").strip()] if missing: print() print("WARNING: Some submitted findings have empty submission_url:") print_table(missing, ["id", "program", "title", "report_path"]) return 2 return 0 def _resolve_report_path_arg(report: str) -> str: p = Path(report) if not p.is_absolute(): p = (ROOT / p).resolve() return str(p) def mark_submitted(conn: sqlite3.Connection, finding_id: int | None, report: str | None, url: str) -> int: url = url.strip() if not url: raise SystemExit("--url is required") if finding_id is None and report is None: raise SystemExit("Provide either --finding-id or --report") if finding_id is not None and report is not None: raise SystemExit("Provide only one of --finding-id or --report") params: tuple[object, ...] where: str if finding_id is not None: where = "id = ?" params = (url, finding_id) else: rp_abs = _resolve_report_path_arg(report or "") # DB canonical is repo-relative, but be tolerant to callers passing absolute paths. candidates = {report or "", rp_abs} try: candidates.add(str(Path(rp_abs).resolve().relative_to(ROOT.resolve()))) except Exception: pass # Avoid ambiguous basename matching; require exact matches. c_list = [c for c in candidates if c] where = "(" + " OR ".join(["report_path = ?"] * len(c_list)) + ")" params = (url, *c_list) cur = conn.execute( f""" UPDATE findings SET status='submitted', submission_url=?, updated_at=datetime('now') WHERE {where}; """, params, ) conn.commit() if cur.rowcount != 1: print(f"ERROR: updated {cur.rowcount} rows (expected 1).", file=sys.stderr) return 1 row = conn.execute( """ SELECT id, program, platform, title, status, submission_url, report_path FROM findings WHERE submission_url = ? ORDER BY updated_at DESC LIMIT 1; """, (url,), ).fetchone() if row: print_table([row], ["id", "program", "platform", "title", "status", "submission_url", "report_path"]) return 0 def main() -> int: ap = argparse.ArgumentParser() sub = ap.add_subparsers(dest="cmd", required=True) sub.add_parser("list-pending") sub.add_parser("list-next") sub.add_parser("list-submitted") ap_mark = sub.add_parser("mark") ap_mark.add_argument("--finding-id", type=int, default=None) ap_mark.add_argument("--report", type=str, default=None) ap_mark.add_argument("--url", type=str, required=True) args = ap.parse_args() # Avoid surprises with other scripts: normalize cwd. os.chdir(str(ROOT)) conn = connect_db() try: if args.cmd == "list-pending": return list_pending(conn) if args.cmd == "list-next": return list_next(conn) if args.cmd == "list-submitted": return list_submitted(conn) if args.cmd == "mark": return mark_submitted(conn, args.finding_id, args.report, args.url) raise SystemExit(f"unknown cmd: {args.cmd}") finally: conn.close() if __name__ == "__main__": raise SystemExit(main())