From 363fce0f8799c9d963a1cfbccd5c9be6d4fb992a Mon Sep 17 00:00:00 2001 From: saidai-bhuvanesh Date: Mon, 6 Jul 2026 18:56:19 +0000 Subject: [PATCH 1/4] feat: Add SARIF 2.1.0 export for external platform compliance (Issue #107) - Added SARIF 2.1.0 compliant export module in backend/app/reports/sarif.py - Added /api/scans/{job_id}/report/sarif endpoint for single scan reports - Added /api/scans/org/{org_job_id}/report/sarif for org-level reports - Added comprehensive tests with 14 test cases - SARIF output compatible with GitHub Advanced Security, SonarQube, DefectDojo Closes #107 --- backend/app/main.py | 138 +++++++++++++++++++++++ backend/app/reports/sarif.py | 209 +++++++++++++++++++++++++++++++++++ backend/tests/test_sarif.py | 133 ++++++++++++++++++++++ 3 files changed, 480 insertions(+) create mode 100644 backend/app/reports/sarif.py create mode 100644 backend/tests/test_sarif.py diff --git a/backend/app/main.py b/backend/app/main.py index 66168de..19435ff 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -68,6 +68,7 @@ from .remediation.engine import propose_fixes from .reports.evidence_pack import build_evidence_pack from .reports.pdf_builder import generate_audit_pdf, generate_org_audit_pdf +from .reports.sarif import generate_sarif_report from .sandbox.verify import verify_repo from .scanners.entropy import run_entropy from .scanners.gitleaks import run_gitleaks @@ -1105,6 +1106,61 @@ async def download_audit_pdf(job_id: str): ) +@app.get("/api/scans/{job_id}/report/sarif", tags=["Reports"]) +async def download_audit_sarif(job_id: str): + """Generate and download SARIF report for a scan job. + + SARIF (Static Analysis Results Interchange Format) is compatible with: + - GitHub Advanced Security + - SonarQube + - DefectDojo + """ + db = await get_db() + try: + job_row = await get_job(db, job_id) + if job_row is None: + raise HTTPException( + status_code=404, detail=f"No job found with id '{job_id}'" + ) + + project_name = job_row["project_name"] + rows = await get_findings_by_job_id(db, job_id) + finally: + await db.close() + + findings_list = [] + for row in rows: + loc = None + if row["file_path"]: + loc = Location(path=row["file_path"], start_line=row["line_number"]) + + findings_list.append( + Finding( + id=row["id"], + title=row.get("title") or row.get("rule_id") or "Unknown", + severity=row.get("severity") or "INFO", + category=row.get("category") or "Unknown", + location=loc, + description=row.get("message") or "", + metadata=row.get("metadata") or {}, + ) + ) + + sarif_report = generate_sarif_report( + findings=findings_list, + project_name=project_name, + run_id=job_id, + ) + + return Response( + content=json.dumps(sarif_report, indent=2), + media_type="application/json", + headers={ + "Content-Disposition": f"attachment; filename=PatchPilot-Audit-{job_id}.sarif" + }, + ) + + @app.get("/jobs/{job_id}/findings") async def get_findings(job_id: str): db = await get_db() @@ -1924,6 +1980,88 @@ async def download_org_audit_pdf(org_job_id: str): ) +@app.get("/api/scans/org/{org_job_id}/report/sarif", tags=["Reports"]) +async def download_org_audit_sarif(org_job_id: str): + """Generate and download SARIF report for an organization scan job. + + SARIF (Static Analysis Results Interchange Format) is compatible with: + - GitHub Advanced Security + - SonarQube + - DefectDojo + """ + db = await get_db() + try: + db.row_factory = aiosqlite.Row + + cur = await db.execute( + "SELECT org_name FROM org_jobs WHERE id = ?", (org_job_id,) + ) + org_row = await cur.fetchone() + if not org_row: + raise HTTPException(status_code=404, detail="Organization job not found") + org_name = org_row["org_name"] + + cur = await db.execute( + """ + SELECT + f.id, + j.project_name as repo_name, + f.rule_id as title, + f.message as description, + f.severity, + f.file_path, + f.line_number, + f.category + FROM findings f + JOIN jobs j ON f.job_id = j.job_id + WHERE j.org_job_id = ? + ORDER BY + CASE f.severity + WHEN 'CRITICAL' THEN 1 + WHEN 'HIGH' THEN 2 + WHEN 'MEDIUM' THEN 3 + WHEN 'LOW' THEN 4 + ELSE 5 + END + """, + (org_job_id,), + ) + findings_rows = await cur.fetchall() + + findings_list = [] + for row in findings_rows: + loc = None + if row["file_path"]: + loc = Location(path=row["file_path"], start_line=row["line_number"]) + + findings_list.append( + Finding( + id=row["id"], + title=row["title"] or "Unknown", + severity=row["severity"] or "INFO", + category=row.get("category") or "Unknown", + location=loc, + description=row.get("description") or "", + ) + ) + + sarif_report = generate_sarif_report( + findings=findings_list, + project_name=org_name, + run_id=org_job_id, + ) + + return Response( + content=json.dumps(sarif_report, indent=2), + media_type="application/json", + headers={ + "Content-Disposition": f"attachment; filename=PatchPilot-Org-{org_name}-{org_job_id}.sarif" + }, + ) + finally: + await db.close() + + @app.get("/api/scans/org/{org_job_id}/blast-radius", tags=["Organization"]) async def get_blast_radius(org_job_id: str): db = await get_db() diff --git a/backend/app/reports/sarif.py b/backend/app/reports/sarif.py new file mode 100644 index 0000000..4695d68 --- /dev/null +++ b/backend/app/reports/sarif.py @@ -0,0 +1,209 @@ +"""SARIF (Static Analysis Results Interchange Format) export module. + +Generates SARIF 2.1.0 compliant output for integration with: +- GitHub Advanced Security +- SonarQube +- DefectDojo +""" + +from __future__ import annotations + +import json +import uuid +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Dict, List, Optional + +from ..models import Finding + + +SARIF_VERSION = "2.1.0" +SARIF_SCHEMA = "https://json.schemastore.org/sarif-2.1.0.json" + + +def _severity_to_sarif_level(severity: str) -> str: + """Map PatchPilot severity to SARIF level. + + SARIF levels: "error", "warning", "note", "none" + """ + mapping = { + "CRITICAL": "error", + "HIGH": "error", + "MEDIUM": "warning", + "LOW": "warning", + "INFO": "note", + } + return mapping.get(severity.upper(), "warning") + + +def _tool_to_sarif_rule(tool_name: str) -> Dict[str, Any]: + """Map scanner tool name to SARIF rule properties.""" + tool_info = { + "semgrep": { + "id": "SEMGREP", + "name": "Semgrep", + "short_description": "Static analysis for security patterns", + "full_description": "Semgrep findings from pattern matching", + "default_level": "error", + }, + "osv": { + "id": "OSV", + "name": "OSV", + "short_description": "Open Source Vulnerabilities database", + "full_description": "Dependency vulnerability findings from OSV", + "default_level": "error", + }, + "gitleaks": { + "id": "GITLEAKS", + "name": "Gitleaks", + "short_description": "Secret scanning in repositories", + "full_description": "Secret detection findings from Gitleaks", + "default_level": "error", + }, + } + return tool_info.get(tool_name.lower(), { + "id": tool_name.upper(), + "name": tool_name, + "short_description": f"{tool_name} findings", + "full_description": f"Findings from {tool_name}", + "default_level": "warning", + }) + + +def finding_to_sarif_result(finding: Finding, tool_driver: str = "PatchPilot") -> Dict[str, Any]: + """Convert a PatchPilot Finding to a SARIF Result object.""" + result: Dict[str, Any] = { + "id": finding.id, + "rule_id": finding.metadata.get("engine", "unknown").upper(), + "level": _severity_to_sarif_level(finding.severity), + "message": { + "text": finding.title, + }, + "properties": { + "category": finding.category, + "ml_score": finding.ml_score, + }, + } + + # Add location if available + if finding.location and finding.location.path: + result["locations"] = [{ + "physical_location": { + "artifact_location": { + "uri": finding.location.path, + "uri_base_id": "%SRCROOT%", + }, + } + }] + + # Add line information + if finding.location.start_line: + result["locations"][0]["physical_location"]["region"] = { + "start_line": finding.location.start_line, + } + if finding.location.end_line and finding.location.end_line != finding.location.start_line: + result["locations"][0]["physical_location"]["region"]["end_line"] = finding.location.end_line + + # Add description as markdown with code context + if finding.description: + result["message"] = { + "text": finding.title, + "markdown": f"**{finding.title}**\n\n{finding.description}", + } + + return result + + +def generate_sarif_report( + findings: List[Finding], + project_name: str = "PatchPilot Scan", + run_id: Optional[str] = None, +) -> Dict[str, Any]: + """Generate a complete SARIF 2.1.0 report from PatchPilot findings. + + Args: + findings: List of Finding objects to export + project_name: Name of the scanned project + run_id: Optional run identifier + + Returns: + SARIF 2.1.0 compliant JSON object + """ + if run_id is None: + run_id = str(uuid.uuid4()) + + # Collect unique tools from findings + tools_seen: set = set() + rules: List[Dict[str, Any]] = [] + + for finding in findings: + tool = finding.metadata.get("engine", "unknown") + if tool.lower() not in tools_seen: + tools_seen.add(tool.lower()) + rule_info = _tool_to_sarif_rule(tool) + rules.append({ + "id": rule_info["id"], + "name": rule_info["name"], + "short_description": { + "text": rule_info["short_description"], + }, + "full_description": { + "text": rule_info["full_description"], + }, + "properties": { + "tags": ["security", "vulnerability"], + }, + }) + + # Convert findings to SARIF results + sarif_results = [finding_to_sarif_result(f) for f in findings] + + # Build the complete SARIF document + sarif_report = { + "$schema": SARIF_SCHEMA, + "version": SARIF_VERSION, + "runs": [{ + "tool": { + "driver": { + "name": "PatchPilot", + "full_name": "PatchPilot Security Engine", + "version": "1.0.0", + "information_uri": "https://github.com/ionfwsrijan/PatchPilot", + "rules": rules, + } + }, + "invocations": [{ + "execution_successful": True, + "end_time_utc": datetime.now(timezone.utc).isoformat(), + }], + "results": sarif_results, + "properties": { + "project_name": project_name, + "run_id": run_id, + "exported_at": datetime.now(timezone.utc).isoformat(), + "total_findings": len(findings), + "metrics": { + "critical": len([f for f in findings if f.severity.upper() == "CRITICAL"]), + "high": len([f for f in findings if f.severity.upper() == "HIGH"]), + "medium": len([f for f in findings if f.severity.upper() == "MEDIUM"]), + "low": len([f for f in findings if f.severity.upper() == "LOW"]), + "info": len([f for f in findings if f.severity.upper() == "INFO"]), + }, + }, + }], + } + + return sarif_report + + +def save_sarif_report(sarif_report: Dict[str, Any], output_path: str) -> None: + """Save SARIF report to a file. + + Args: + sarif_report: The SARIF report dictionary + output_path: Path to save the .sarif file + """ + Path(output_path).write_text( + json.dumps(sarif_report, indent=2), + encoding="utf-8" + ) diff --git a/backend/tests/test_sarif.py b/backend/tests/test_sarif.py new file mode 100644 index 0000000..f86d129 --- /dev/null +++ b/backend/tests/test_sarif.py @@ -0,0 +1,133 @@ +"""Tests for SARIF export module.""" + +import pytest +from app.models import Finding, Location +from app.reports.sarif import ( + generate_sarif_report, + finding_to_sarif_result, + _severity_to_sarif_level, + _tool_to_sarif_rule, + SARIF_VERSION, +) + + +class TestSeverityMapping: + def test_critical_maps_to_error(self): + assert _severity_to_sarif_level("CRITICAL") == "error" + assert _severity_to_sarif_level("critical") == "error" + + def test_high_maps_to_error(self): + assert _severity_to_sarif_level("HIGH") == "error" + + def test_medium_maps_to_warning(self): + assert _severity_to_sarif_level("MEDIUM") == "warning" + + def test_low_maps_to_warning(self): + assert _severity_to_sarif_level("LOW") == "warning" + + def test_info_maps_to_note(self): + assert _severity_to_sarif_level("INFO") == "note" + + +class TestToolMapping: + def test_semgrep_rule(self): + rule = _tool_to_sarif_rule("semgrep") + assert rule["id"] == "SEMGREP" + assert rule["name"] == "Semgrep" + + def test_osv_rule(self): + rule = _tool_to_sarif_rule("osv") + assert rule["id"] == "OSV" + assert rule["name"] == "OSV" + + def test_gitleaks_rule(self): + rule = _tool_to_sarif_rule("gitleaks") + assert rule["id"] == "GITLEAKS" + assert rule["name"] == "Gitleaks" + + def test_unknown_tool(self): + rule = _tool_to_sarif_rule("custom-tool") + assert rule["id"] == "CUSTOM-TOOL" + assert rule["name"] == "custom-tool" + + +class TestFindingConversion: + def test_finding_with_location(self): + finding = Finding( + id="test-1", + title="SQL Injection vulnerability", + severity="HIGH", + category="Security", + location=Location(path="src/app.py", start_line=42, end_line=45), + description="Potential SQL injection in user input", + ) + result = finding_to_sarif_result(finding) + assert result["id"] == "test-1" + assert result["level"] == "error" + assert result["message"]["text"] == "SQL Injection vulnerability" + assert len(result["locations"]) == 1 + assert result["locations"][0]["physical_location"]["artifact_location"]["uri"] == "src/app.py" + + def test_finding_without_location(self): + finding = Finding( + id="test-2", + title="Missing dependency", + severity="MEDIUM", + category="Dependency", + ) + result = finding_to_sarif_result(finding) + assert result["id"] == "test-2" + assert result["level"] == "warning" + assert "locations" not in result + + +class TestSarifReportGeneration: + def test_empty_report(self): + report = generate_sarif_report([]) + assert report["version"] == SARIF_VERSION + assert len(report["runs"]) == 1 + assert report["runs"][0]["tool"]["driver"]["name"] == "PatchPilot" + assert len(report["runs"][0]["results"]) == 0 + + def test_report_with_findings(self): + findings = [ + Finding( + id="finding-1", + title="XSS Vulnerability", + severity="CRITICAL", + category="Security", + location=Location(path="index.html", start_line=10), + metadata={"engine": "semgrep"}, + ), + Finding( + id="finding-2", + title="Outdated dependency", + severity="MEDIUM", + category="Dependency", + metadata={"engine": "osv"}, + ), + ] + report = generate_sarif_report(findings, project_name="Test Project") + + assert report["version"] == SARIF_VERSION + run = report["runs"][0] + assert run["tool"]["driver"]["name"] == "PatchPilot" + assert len(run["results"]) == 2 + assert run["properties"]["project_name"] == "Test Project" + assert run["properties"]["metrics"]["critical"] == 1 + assert run["properties"]["metrics"]["medium"] == 1 + + def test_report_includes_rules(self): + findings = [ + Finding( + id="finding-1", + title="Test", + severity="HIGH", + category="Test", + metadata={"engine": "semgrep"}, + ), + ] + report = generate_sarif_report(findings) + rules = report["runs"][0]["tool"]["driver"]["rules"] + assert len(rules) == 1 + assert rules[0]["id"] == "SEMGREP" From 85261a262be26e99b137018c271534884b55b66d Mon Sep 17 00:00:00 2001 From: saidai-bhuvanesh Date: Mon, 6 Jul 2026 19:25:08 +0000 Subject: [PATCH 2/4] Trigger CI re-run From ee1fdfb30a90982a5fa3931f2571a9cef61b18e4 Mon Sep 17 00:00:00 2001 From: saidai-bhuvanesh Date: Mon, 6 Jul 2026 19:32:02 +0000 Subject: [PATCH 3/4] Trigger CI re-run with updated PR template From 11b3a49822a12a6ea09bb047b21634e26dd4cc8d Mon Sep 17 00:00:00 2001 From: saidai-bhuvanesh Date: Mon, 6 Jul 2026 19:36:21 +0000 Subject: [PATCH 4/4] Re-trigger CI