| """GitHub PR scan service for SecBrief.""" |
|
|
| from __future__ import annotations |
|
|
| import json |
| import re |
| from typing import Any |
|
|
| import httpx |
|
|
| from llm_service import audit_code, explain_alert |
|
|
| _PLATFORM_URL = "https://adi576-secbrief.hf.space" |
|
|
|
|
| def _normalize_severity(value: Any) -> str: |
| v = str(value or "").strip().lower() |
| if v == "critical": |
| return "Critical" |
| if v == "high": |
| return "High" |
| if v == "medium": |
| return "Medium" |
| if v == "low": |
| return "Low" |
| return "Low" |
|
|
|
|
| def _severity_rank(severity: str) -> int: |
| s = _normalize_severity(severity).lower() |
| return {"low": 1, "medium": 2, "high": 3, "critical": 4}.get(s, 0) |
|
|
|
|
| def _parse_risk_score(value: Any) -> float | None: |
| if value is None: |
| return None |
| if isinstance(value, (int, float)): |
| return float(value) |
| text = str(value).strip() |
| if not text: |
| return None |
| try: |
| return float(text) |
| except Exception: |
| return None |
|
|
|
|
| def _severity_from_risk_score(score: float | None) -> str | None: |
| if score is None: |
| return None |
| if score >= 9: |
| return "Critical" |
| if score >= 7: |
| return "High" |
| if score >= 4: |
| return "Medium" |
| return "Low" |
|
|
|
|
|
|
| def _extract_owasp(value: Any) -> list[str]: |
| if not value: |
| return [] |
| text = str(value).strip() |
| if not text or text.upper() == "N/A": |
| return [] |
| match = re.search(r"(A\d{2}:2021)", text) |
| if match: |
| return [match.group(1)] |
| head = text.split("-", 1)[0].strip() |
| return [head] if head else [] |
|
|
|
|
| def _extract_cwe(value: Any) -> list[str]: |
| if not value: |
| return [] |
| text = str(value).strip() |
| if not text or text.upper() == "N/A": |
| return [] |
| match = re.search(r"(CWE-\d+)", text, re.I) |
| if match: |
| return [match.group(1).upper()] |
| return [text] |
|
|
|
|
| def _first_str(value: Any) -> str: |
| if isinstance(value, list) and value: |
| if isinstance(value[0], str): |
| return value[0] |
| if isinstance(value, str): |
| return value |
| return "" |
|
|
|
|
| def _parse_package_lock(content: str) -> list[dict[str, str]]: |
| try: |
| data = json.loads(content) |
| except Exception: |
| return [] |
|
|
| packages: list[dict[str, str]] = [] |
| pkgs = data.get("packages") |
| if isinstance(pkgs, dict): |
| for path, meta in pkgs.items(): |
| if not isinstance(meta, dict): |
| continue |
| version = meta.get("version") |
| if not version: |
| continue |
| name = str(path or "") |
| name = name.removeprefix("node_modules/") |
| if not name: |
| continue |
| packages.append({"name": name, "version": str(version)}) |
| return packages |
|
|
|
|
| def _parse_requirements_txt(content: str) -> list[dict[str, str]]: |
| packages: list[dict[str, str]] = [] |
| for line in content.splitlines(): |
| line = line.strip() |
| if not line or line.startswith("#"): |
| continue |
| match = re.match(r"^([a-zA-Z0-9_.-]+)\s*([=<>~!]+)\s*([0-9a-zA-Z.]+)", line) |
| if match: |
| packages.append({"name": match.group(1), "version": match.group(3)}) |
| return packages |
|
|
|
|
| def _parse_go_sum(content: str) -> list[dict[str, str]]: |
| packages: list[dict[str, str]] = [] |
| seen: set[str] = set() |
| for line in content.splitlines(): |
| parts = line.strip().split() |
| if len(parts) < 2: |
| continue |
| name = parts[0] |
| version = parts[1].replace("/go.mod", "") |
| key = f"{name}@{version}" |
| if key in seen: |
| continue |
| seen.add(key) |
| packages.append({"name": name, "version": version}) |
| return packages |
|
|
|
|
| def _parse_ecosystem_file(filename: str, content: str) -> list[dict[str, str]]: |
| name = filename.lower() |
| if name.endswith("package-lock.json"): |
| return _parse_package_lock(content) |
| if name.endswith("requirements.txt"): |
| return _parse_requirements_txt(content) |
| if name.endswith("go.sum"): |
| return _parse_go_sum(content) |
| return [] |
|
|
|
|
| def _query_osv(packages: list[dict[str, str]]) -> list[dict[str, Any]]: |
| if not packages: |
| return [] |
| queries = [{"package": {"name": p["name"]}, "version": p["version"]} for p in packages[:100]] |
| try: |
| r = httpx.post( |
| "https://api.osv.dev/v1/querybatch", |
| json={"queries": queries}, |
| timeout=30, |
| ) |
| r.raise_for_status() |
| data = r.json() |
| except Exception: |
| return [] |
|
|
| vulns: list[dict[str, Any]] = [] |
| for result in data.get("results", []) if isinstance(data, dict) else []: |
| for vuln in result.get("vulns", []) if isinstance(result, dict) else []: |
| if isinstance(vuln, dict): |
| vulns.append(vuln) |
| return vulns |
|
|
|
|
| def _severity_emoji(severity: str) -> str: |
| return { |
| "Critical": "🔴", |
| "High": "🟠", |
| "Medium": "🟡", |
| "Low": "🟢", |
| }.get(_normalize_severity(severity), "⚪") |
|
|
|
|
| def _format_pr_comment(overall_severity: str, findings: list[dict[str, Any]]) -> str: |
| overall = _normalize_severity(overall_severity) |
| lines: list[str] = [ |
| f"# {_severity_emoji(overall)} SecBrief Security Scan", |
| f"**Overall severity:** {_severity_emoji(overall)} {overall}", |
| f"**Total findings:** {len(findings)}", |
| "", |
| ] |
|
|
| for f in findings: |
| sev = _normalize_severity(f.get("severity")) |
| title = str(f.get("title") or "Security finding") |
| file_name = str(f.get("file") or "(dependencies)") |
| explanation = str(f.get("explanation") or "").strip() |
| fix_summary = str(f.get("fix_summary") or "").strip() |
|
|
| lines.append(f"## {_severity_emoji(sev)} {title}") |
| lines.append(f"**File:** {file_name}") |
| if explanation: |
| lines.append("") |
| lines.append(explanation) |
| tags: list[str] = [] |
| owasp = f.get("owasp") or [] |
| cwe = f.get("cwe") or [] |
| if isinstance(owasp, list) and owasp: |
| tags.append(f"**OWASP:** {', '.join(str(x) for x in owasp)}") |
| if isinstance(cwe, list) and cwe: |
| tags.append(f"**CWE:** {', '.join(str(x) for x in cwe)}") |
| if tags: |
| lines.append("") |
| lines.extend(tags) |
| if fix_summary: |
| lines.append("") |
| lines.append(f"**Fix summary:** {fix_summary}") |
| lines.append("") |
|
|
| lines.append("---") |
| lines.append(f"Powered by [SecBrief]({_PLATFORM_URL})") |
| return "\n".join(lines) |
|
|
|
|
| def run_github_scan(payload: dict[str, Any]) -> dict[str, Any]: |
| changed_files = payload.get("changed_files") or [] |
| ecosystem_files = payload.get("ecosystem_files") or [] |
|
|
| findings: list[dict[str, Any]] = [] |
|
|
| for entry in changed_files: |
| if not isinstance(entry, dict): |
| continue |
| filename = str(entry.get("filename") or "").strip() |
| content = str(entry.get("content") or "") |
| if not filename or not content.strip(): |
| continue |
|
|
| analysis, _ = audit_code(content) |
| a = analysis if isinstance(analysis, dict) else {} |
| if not a: |
| continue |
|
|
| risk_score = _parse_risk_score(a.get("risk_score")) |
| sev = _normalize_severity(a.get("severity")) |
| sev_from_risk = _severity_from_risk_score(risk_score) |
| if sev_from_risk and _severity_rank(sev_from_risk) > _severity_rank(sev): |
| sev = sev_from_risk |
|
|
| owasp = _extract_owasp(a.get("owasp_category") or a.get("owasp")) |
| cwe = _extract_cwe(a.get("cwe_id") or a.get("cwe")) |
|
|
| base_explanation = str(a.get("summary") or a.get("what_happened") or "").strip() |
| if not base_explanation: |
| base_explanation = str(a.get("formatted") or "").strip() |
|
|
| base_fix = _first_str(a.get("what_to_do")) or "" |
|
|
| vulns: Any = a.get("vulnerabilities") |
| if not isinstance(vulns, list) or not vulns: |
| for alt in ("findings", "issues"): |
| cand = a.get(alt) |
| if isinstance(cand, list) and cand: |
| vulns = cand |
| break |
|
|
| if isinstance(vulns, list) and vulns: |
| for v in vulns: |
| if not isinstance(v, dict): |
| continue |
| v_sev = _normalize_severity(v.get("severity") or sev) |
| title = str(v.get("name") or a.get("title") or "Security finding") |
| explanation = str(v.get("explanation") or "").strip() or base_explanation |
| fix_summary = str(v.get("fix") or "").strip() or base_fix or "Review and fix" |
| findings.append( |
| { |
| "file": filename, |
| "line": v.get("line_hint"), |
| "severity": v_sev, |
| "title": title, |
| "explanation": explanation, |
| "owasp": owasp, |
| "cwe": cwe, |
| "soc2": a.get("soc2_controls", []) if isinstance(a.get("soc2_controls", []), list) else [], |
| "inr_impact": str(a.get("financial_impact_inr") or ""), |
| "fix_summary": fix_summary, |
| } |
| ) |
| else: |
| should_emit = False |
| if _severity_rank(sev) > _severity_rank("Low"): |
| should_emit = True |
| if owasp or cwe: |
| should_emit = True |
| if risk_score is not None and risk_score > 3: |
| should_emit = True |
| if _severity_rank(sev) < _severity_rank("Medium"): |
| sev = "Medium" |
|
|
| if should_emit: |
| findings.append( |
| { |
| "file": filename, |
| "line": None, |
| "severity": sev, |
| "title": str(a.get("title") or "Security finding"), |
| "explanation": base_explanation, |
| "owasp": owasp, |
| "cwe": cwe, |
| "soc2": a.get("soc2_controls", []) if isinstance(a.get("soc2_controls", []), list) else [], |
| "inr_impact": str(a.get("financial_impact_inr") or ""), |
| "fix_summary": base_fix or "Review and fix", |
| } |
| ) |
|
|
| all_packages: list[dict[str, str]] = [] |
| for entry in ecosystem_files: |
| if not isinstance(entry, dict): |
| continue |
| filename = str(entry.get("filename") or "").strip() |
| content = str(entry.get("content") or "") |
| if not filename or not content.strip(): |
| continue |
| all_packages.extend(_parse_ecosystem_file(filename, content)) |
|
|
| for vuln in _query_osv(all_packages): |
| if not isinstance(vuln, dict): |
| continue |
| vuln_id = str(vuln.get("id") or "OSV") |
| summary = str(vuln.get("summary") or "Vulnerability found").strip() |
|
|
| dep_sev = "Medium" |
| for entry in vuln.get("severity", []) if isinstance(vuln.get("severity"), list) else []: |
| if not isinstance(entry, dict): |
| continue |
| if entry.get("type") != "CVSS_V3": |
| continue |
| try: |
| score = float(entry.get("score", 5.0)) |
| except Exception: |
| score = 5.0 |
| dep_sev = _normalize_severity(_severity_from_risk_score(score) or "Medium") |
| break |
|
|
| dep_analysis, _ = explain_alert(f"{vuln_id}: {summary}") |
| dep_a = dep_analysis if isinstance(dep_analysis, dict) else {} |
| owasp = _extract_owasp(dep_a.get("owasp_category") or dep_a.get("owasp")) |
| cwe = _extract_cwe(dep_a.get("cwe_id") or dep_a.get("cwe")) |
| explanation = str(dep_a.get("summary") or "").strip() or summary |
| fix_summary = _first_str(dep_a.get("what_to_do")) or "Update dependency" |
|
|
| findings.append( |
| { |
| "file": "(dependencies)", |
| "line": None, |
| "severity": dep_sev, |
| "title": f"{vuln_id}: {summary}", |
| "explanation": explanation, |
| "owasp": owasp, |
| "cwe": cwe, |
| "soc2": dep_a.get("soc2_controls", []) if isinstance(dep_a.get("soc2_controls", []), list) else [], |
| "inr_impact": str(dep_a.get("financial_impact_inr") or ""), |
| "fix_summary": fix_summary, |
| } |
| ) |
|
|
| overall = "Low" |
| for f in findings: |
| f_sev = _normalize_severity(f.get("severity")) |
| if _severity_rank(f_sev) > _severity_rank(overall): |
| overall = f_sev |
|
|
| return { |
| "overall_severity": overall, |
| "total_findings": len(findings), |
| "findings": findings, |
| "pr_comment": _format_pr_comment(overall, findings), |
| } |
|
|
