From 0d2b01df10ca5277b7e4e67d275d6d8d972d4546 Mon Sep 17 00:00:00 2001 From: Ankit Bari Date: Sun, 11 Oct 2026 19:44:03 +0530 Subject: [PATCH 1/2] Create export.py --- openhack/export.py | 398 +++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 398 insertions(+) create mode 100644 openhack/export.py diff --git a/openhack/export.py b/openhack/export.py new file mode 100644 index 0000000..9890169 --- /dev/null +++ b/openhack/export.py @@ -0,0 +1,398 @@ +"""Render and write findings as SARIF, Markdown, or JSON. + +Pure module: no TUI or prompt_toolkit imports, so it can be used from the +interactive agent and from tests alike. +""" + +from __future__ import annotations + +import hashlib +import json +import os +import re +import tempfile +from datetime import datetime, timezone +from pathlib import Path, PurePosixPath +from typing import Iterable, Optional, Union + +from openhack import __version__ as OPENHACK_VERSION +from openhack.agents.eventlog import redact +from openhack.agents.session import Finding +from openhack.categories import CATEGORY_SEVERITY, normalize_category + +FORMATS = ("sarif", "md", "json") + +_ALIASES = {"sarif": "sarif", "md": "md", "markdown": "md", "json": "json"} +_EXTENSIONS = {".sarif": "sarif", ".json": "json", ".md": "md", ".markdown": "md"} + +_LEVEL = {"critical": "error", "high": "error", "medium": "warning", "low": "note", "info": "note"} +_SCORE = {"critical": "9.5", "high": "8.0", "medium": "5.5", "low": "3.0", "info": "1.0"} +_SEVERITY_ORDER = ("critical", "high", "medium", "low", "info") + +CWE_BY_CATEGORY: dict[str, int] = { + "SQL Injection": 89, + "Command Injection": 78, + "RCE": 94, + "XSS": 79, + "SSRF": 918, + "Open Redirect": 601, + "Path Traversal": 22, + "IDOR": 639, + "Authentication Bypass": 287, + "Authorization Bypass": 285, + "CSRF": 352, + "Data Exposure": 200, + "Information Disclosure": 200, + "Hardcoded Secret": 798, + "Security Misconfiguration": 16, + "Missing RLS": 284, + "RPC Function Abuse": 284, + "Storage Misconfiguration": 732, + "Mass Assignment": 915, + "Business Logic Flaw": 840, + "Denial of Service": 400, +} + +_LOCATION_RE = re.compile(r"^(?P.+?):(?P\d+)(?::(?P\d+))?$") +_SARIF_SCHEMA = "https://json.schemastore.org/sarif-2.1.0.json" +_INFORMATION_URI = "https://openhack.com" +_SRCROOT = "%SRCROOT%" +_FINGERPRINT_KEY = "openhackFingerprint/v1" + + +class ExportError(ValueError): + """Raised when an export format, destination, or write is invalid.""" + + +def normalize_format(raw: str) -> str: + key = (raw or "").strip().lower() + if key not in _ALIASES: + raise ExportError( + f"unknown export format {raw!r} (use sarif, md, or json)" + ) + return _ALIASES[key] + + +def infer_format_from_path(path: Union[str, Path]) -> Optional[str]: + return _EXTENSIONS.get(Path(path).suffix.lower()) + + +def default_filename(fmt: str) -> str: + return f"openhack-findings.{normalize_format(fmt)}" + + +def findings_to_sarif( + findings: Iterable[Finding], + *, + target: Union[str, Path, None] = None, +) -> dict: + target_path = _target_path(target) + rules: list[dict] = [] + rule_index: dict[str, int] = {} + results: list[dict] = [] + seen_fingerprints: set[str] = set() + + for finding in findings: + category = normalize_category(finding.category) + if category not in rule_index: + rule_index[category] = len(rules) + rules.append(_sarif_rule(category)) + severity = _severity(finding.severity) + uri, relative, line, column = _location(finding, target_path) + + region: dict = {"startLine": line or 1} + if column: + region["startColumn"] = column + if finding.code_snippet: + region["snippet"] = {"text": _scrub(finding.code_snippet)} + + artifact: dict = {"uri": uri} + if relative: + artifact["uriBaseId"] = _SRCROOT + + fingerprint = _unique_fingerprint(finding, seen_fingerprints) + properties: dict = { + "openhackSeverity": severity, + "security-severity": _security_severity(finding, severity), + "confidence": finding.confidence, + "validated": finding.validated, + } + if finding.cvss_score is not None: + properties["cvssScore"] = finding.cvss_score + + results.append({ + "ruleId": rules[rule_index[category]]["id"], + "ruleIndex": rule_index[category], + "level": _LEVEL[severity], + "message": {"text": _message(finding, category)}, + "locations": [{ + "physicalLocation": { + "artifactLocation": artifact, + "region": region, + }, + }], + "partialFingerprints": {_FINGERPRINT_KEY: fingerprint}, + "properties": properties, + }) + + run: dict = { + "tool": { + "driver": { + "name": "OpenHack", + "version": OPENHACK_VERSION, + "informationUri": _INFORMATION_URI, + "rules": rules, + }, + }, + "columnKind": "utf16CodeUnits", + "results": results, + } + if target_path is not None: + run["originalUriBaseIds"] = { + _SRCROOT: {"uri": target_path.as_uri() + "/"}, + } + return {"$schema": _SARIF_SCHEMA, "version": "2.1.0", "runs": [run]} + + +def findings_to_markdown( + findings: Iterable[Finding], + *, + target: Union[str, Path, None] = None, + generated_at: Optional[str] = None, +) -> str: + items = list(findings) + target_path = _target_path(target) + stamp = generated_at or _now() + + lines = [ + "# OpenHack security findings", + "", + f"- **Target:** `{target_path or 'unknown'}`", + f"- **Generated:** {stamp}", + f"- **Findings:** {len(items)}", + "", + "## Summary", + "", + "| Severity | Count |", + "| --- | ---: |", + ] + counts = {name: 0 for name in _SEVERITY_ORDER} + for finding in items: + counts[_severity(finding.severity)] += 1 + lines += [f"| {name} | {counts[name]} |" for name in _SEVERITY_ORDER] + lines += ["", "## Findings", ""] + + if not items: + lines += ["No findings.", ""] + + for index, finding in enumerate(items, 1): + category = normalize_category(finding.category) + severity = _severity(finding.severity) + cwe = CWE_BY_CATEGORY.get(category) + uri, _relative, line, _column = _location(finding, target_path) + location = f"{uri}:{line}" if line else uri + + lines += [f"### {index}. {finding.title or category}", ""] + lines.append(f"- **Severity:** {severity}") + category_line = f"- **Category:** {category}" + if cwe is not None: + category_line += f" (CWE-{cwe})" + lines.append(category_line) + lines.append(f"- **Location:** `{location}`") + if finding.cvss_score is not None: + lines.append(f"- **CVSS:** {finding.cvss_score:g}") + lines.append(f"- **Confidence:** {finding.confidence}") + lines.append(f"- **Validated:** {'yes' if finding.validated else 'no'}") + lines.append("") + + if finding.description: + lines += [_scrub(finding.description), ""] + if finding.code_snippet: + snippet = _scrub(finding.code_snippet) + lines += ["**Vulnerable code**", "", _fence(snippet), snippet, _fence(snippet), ""] + if finding.fix: + lines += ["**Recommended fix**", "", _scrub(finding.fix), ""] + + return "\n".join(lines).rstrip() + "\n" + + +def findings_to_json( + findings: Iterable[Finding], + *, + target: Union[str, Path, None] = None, + generated_at: Optional[str] = None, +) -> str: + items = list(findings) + target_path = _target_path(target) + payload = { + "tool": "openhack", + "version": OPENHACK_VERSION, + "generatedAt": generated_at or _now(), + "target": str(target_path) if target_path is not None else None, + "count": len(items), + "findings": [redact(finding.to_dict()) for finding in items], + } + return json.dumps(payload, indent=2, ensure_ascii=False) + "\n" + + +def render( + findings: Iterable[Finding], + fmt: str, + *, + target: Union[str, Path, None] = None, +) -> str: + fmt = normalize_format(fmt) + items = list(findings) + if fmt == "sarif": + return json.dumps(findings_to_sarif(items, target=target), indent=2, ensure_ascii=False) + "\n" + if fmt == "md": + return findings_to_markdown(items, target=target) + return findings_to_json(items, target=target) + + +def write_export( + findings: Iterable[Finding], + fmt: str, + dest: Union[str, Path], + *, + target: Union[str, Path, None] = None, +) -> Path: + text = render(findings, fmt, target=target) + dest_path = Path(dest) + if dest_path.is_dir(): + raise ExportError(f"{dest_path} is a directory") + parent = dest_path.parent + if not parent.is_dir(): + raise ExportError(f"directory does not exist: {parent}") + + fd, tmp_name = tempfile.mkstemp(dir=parent, prefix=f".{dest_path.name}.", suffix=".tmp") + tmp_path = Path(tmp_name) + try: + with os.fdopen(fd, "w", encoding="utf-8", newline="\n") as handle: + handle.write(text) + _restrict_permissions(tmp_path) + os.replace(tmp_path, dest_path) + except OSError as exc: + _unlink_quietly(tmp_path) + raise ExportError(f"could not write {dest_path}: {exc}") from exc + except BaseException: + _unlink_quietly(tmp_path) + raise + return dest_path + + +def _now() -> str: + return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def _target_path(target: Union[str, Path, None]) -> Optional[Path]: + if target is None or not str(target).strip(): + return None + return Path(target).expanduser().absolute() + + +def _severity(raw: Optional[str]) -> str: + key = (raw or "").strip().lower() + return key if key in _LEVEL else "medium" + + +def _scrub(text: str) -> str: + return str(redact(text)) + + +def _security_severity(finding: Finding, severity: str) -> str: + if finding.cvss_score is not None: + return f"{float(finding.cvss_score):.1f}" + return _SCORE[severity] + + +def _message(finding: Finding, category: str) -> str: + title = finding.title or category + text = f"{title}: {finding.description}" if finding.description else title + return _scrub(text) + + +def _sarif_rule(category: str) -> dict: + severity = _severity(CATEGORY_SEVERITY.get(category, "medium")) + slug = "-".join(category.lower().split()) + tags = ["security"] + cwe = CWE_BY_CATEGORY.get(category) + if cwe is not None: + tags.append(f"external/cwe/cwe-{cwe}") + return { + "id": f"openhack/{slug}", + "name": _rule_name(category), + "shortDescription": {"text": category}, + "defaultConfiguration": {"level": _LEVEL[severity]}, + "properties": { + "tags": tags, + "security-severity": _SCORE[severity], + }, + } + + +def _rule_name(category: str) -> str: + return "".join( + word if word.isupper() else word[:1].upper() + word[1:] + for word in category.split() + ) + + +def _split_location(file_path: str) -> tuple[str, Optional[int], Optional[int]]: + match = _LOCATION_RE.match(file_path.strip()) + if match is None: + return file_path.strip(), None, None + line = int(match.group("line")) + column = int(match.group("col")) if match.group("col") else None + return match.group("path"), line, column + + +def _artifact_uri(path_text: str, target: Optional[Path]) -> tuple[str, bool]: + if not path_text: + return "unknown", True + path = Path(path_text) + if not path.is_absolute(): + return PurePosixPath(*path.parts).as_posix(), True + if target is not None: + try: + relative = path.relative_to(target) + except ValueError: + pass + else: + return PurePosixPath(*relative.parts).as_posix() or ".", True + return path.as_uri(), False + + +def _location(finding: Finding, target: Optional[Path]) -> tuple[str, bool, Optional[int], Optional[int]]: + path_text, parsed_line, parsed_column = _split_location(finding.file_path or "") + line = finding.line_number or parsed_line + uri, relative = _artifact_uri(path_text, target) + return uri, relative, line, parsed_column + + +def _unique_fingerprint(finding: Finding, seen: set[str]) -> str: + base = hashlib.sha256(f"{finding.fingerprint()}::{finding.title}".encode()).hexdigest()[:32] + fingerprint = base + if fingerprint in seen: + fingerprint = hashlib.sha256(f"{base}::{finding.id}".encode()).hexdigest()[:32] + seen.add(fingerprint) + return fingerprint + + +def _fence(text: str) -> str: + longest = max((len(run) for run in re.findall(r"`+", text)), default=0) + return "`" * max(3, longest + 1) + + +def _restrict_permissions(path: Path) -> None: + try: + os.chmod(path, 0o600) + except OSError: + pass + + +def _unlink_quietly(path: Path) -> None: + try: + path.unlink() + except OSError: + pass From 9b5ba43158870a860daed9a18b77a4fd726a9551 Mon Sep 17 00:00:00 2001 From: Ankit Bari Date: Mon, 12 Oct 2026 21:43:13 +0530 Subject: [PATCH 2/2] Update tui.py --- openhack/tui.py | 48 ++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 48 insertions(+) diff --git a/openhack/tui.py b/openhack/tui.py index 3fe924b..8e629e8 100644 --- a/openhack/tui.py +++ b/openhack/tui.py @@ -78,6 +78,7 @@ ) from openhack.providers import CURATED_PROVIDER_IDS from openhack.setup import run_provider_connect, run_setup_command +from openhack import export as export_mod from openhack.shells import ShellManager from openhack.tools.registry import ToolRegistry from openhack.prompts.project_context import build_project_context @@ -353,6 +354,7 @@ def _sev_label(severity: str) -> str: ("/findings", "Show findings from the current session"), ("/verify", "Verify loaded findings (`/verify sandbox` or `/verify browser`)"), ("/copy", "Copy the selected finding as an AI-fix prompt to the clipboard"), + ("/export", "Export findings to SARIF / Markdown / JSON — /export [sarif|md|json] [path]"), ("/pause", "Pause the running scan/agent (Ctrl+C also pauses)"), ("/resume", "Resume a paused run"), ("/cancel", "Stop the running scan/agent"), @@ -398,6 +400,7 @@ def _sev_label(severity: str) -> str: "Use /verify sandbox to reproduce findings safely.", "Use /verify browser to validate browser-exploitable findings.", "Use /copy to copy a finding as a remediation prompt.", + "Use /export to save findings as SARIF for GitHub code scanning.", "Use /sessions to reopen an earlier investigation.", "Use /cost to inspect token usage and inference cost.", "Use /cd to change the active project.", @@ -4614,6 +4617,8 @@ async def _handle_input(self, text: str) -> None: self._open_shells_view() elif cmd == "/copy": self._cmd_copy_fix() + elif cmd == "/export": + self._cmd_export(arg) elif cmd == "/verify": self._cmd_verify(arg) elif cmd == "/mouse": @@ -5129,6 +5134,49 @@ def _cmd_copy_fix(self) -> None: "couldn't find a clipboard tool (pbcopy/xclip/wl-copy/clip)" ) + def _cmd_export(self, arg: str) -> None: + """/export [sarif|md|json] [path] — write all loaded findings to a file.""" + findings = self._current_findings() + if not findings: + self.last_status_line = ( + "no findings to export — run /scan or load a session from /sessions first" + ) + return + + tokens = arg.split() + fmt: Optional[str] = None + if tokens: + fmt = export_mod.normalize_format(tokens[0]) + if fmt is not None: + tokens = tokens[1:] + raw_path = " ".join(tokens) + + if fmt is None and raw_path: + fmt = export_mod.infer_format_from_path(raw_path) + if fmt is None: + self.last_status_line = ( + f"usage: /export [{'|'.join(export_mod.FORMATS)}] [path] — " + f"can't infer a format from '{raw_path}'" + ) + return + fmt = fmt or "sarif" + + dest = _resolve_path_argument(raw_path, default=os.getcwd()) + target = ( + self.viewing_target + or (self.scan.target if self.scan is not None and self.scan.target else "") + or None + ) + try: + written = export_mod.write_export(findings, fmt, dest, target=target) + except export_mod.ExportError as exc: + self.last_status_line = f"error: {exc}" + return + self.last_status_line = ( + f"exported {len(findings)} finding(s) as {fmt.upper()} → " + f"{_abbrev_home(str(written))}" + ) + # ── Sessions overlay ────────────────────────────────────────── @staticmethod