From 052deca6a26f2f58a645f611ee08916d3f339e12 Mon Sep 17 00:00:00 2001 From: Chocapikk Date: Thu, 16 Apr 2026 20:36:48 +0200 Subject: [PATCH] Refactor: Extract shared code, eliminate duplication across modules - Extract github_get() into src/github_api.py (was duplicated in deep_scan.py and silent_scan.py) - Extract score_with_fingerprints() into fingerprint.py (scoring logic duplicated in both scanners) - Extract build_analysis_result() into analyze.py (result building duplicated in backfill_local.py) - Extract _enrich_advisory() in render.py (cleaning logic repeated in 3 render functions) - Fix undefined suspects_count crash in deep_scan.py - Ruff cleanup: unused imports, f-strings without placeholders across 11 files --- src/analyze.py | 29 +++++++++++------ src/backfill_local.py | 42 ++++-------------------- src/db.py | 1 - src/deep_scan.py | 65 +++++--------------------------------- src/fetch.py | 2 +- src/fingerprint.py | 25 +++++++++++++-- src/fingerprint_builder.py | 3 +- src/github_api.py | 44 ++++++++++++++++++++++++++ src/main.py | 9 +++--- src/render.py | 45 +++++++++++--------------- src/render_cli.py | 6 ++-- src/silent_scan.py | 51 ++++-------------------------- 12 files changed, 135 insertions(+), 187 deletions(-) create mode 100644 src/github_api.py diff --git a/src/analyze.py b/src/analyze.py index 2c1698210d..bc66b1bda9 100644 --- a/src/analyze.py +++ b/src/analyze.py @@ -6,7 +6,6 @@ GEMINI_API_KEY, GEMINI_API_URL, LLM_SYSTEM_PROMPT, LLM_USER_PROMPT_TEMPLATE, TAXONOMY_PATH, RETRY_ATTEMPTS, RETRY_BACKOFF, - RATE_LIMIT_DELAY, ) @@ -42,7 +41,7 @@ def analyze_advisory(advisory: dict, filtered_diff: str) -> Optional[dict]: raw_response = _call_gemini(user_prompt) if not raw_response: - print(f" DEBUG: _call_gemini returned None") + print(" DEBUG: _call_gemini returned None") return None parsed = _parse_llm_response(raw_response) @@ -53,6 +52,16 @@ def analyze_advisory(advisory: dict, filtered_diff: str) -> Optional[dict]: if parsed.get("pattern_id") not in taxonomy_ids: parsed["pattern_id"] = "UNCLASSIFIED" + return build_analysis_result(advisory, parsed) + + +def build_analysis_result(advisory: dict, parsed: dict) -> dict: + """Build a standardized analysis result from advisory metadata and LLM output.""" + def _str(val): + if isinstance(val, (dict, list)): + return json.dumps(val) + return str(val) if val else "" + return { "ghsa_id": advisory["ghsa_id"], "date": advisory["published_at"][:10], @@ -63,12 +72,12 @@ def analyze_advisory(advisory: dict, filtered_diff: str) -> Optional[dict]: "cvss_score": advisory["cvss_score"], "package_name": advisory["package_name"], "pattern_id": parsed["pattern_id"], - "vuln_type": parsed.get("vuln_type", ""), - "root_cause": parsed.get("root_cause", ""), - "impact": parsed.get("impact", ""), - "fix_summary": parsed.get("fix_summary", ""), - "key_diff": parsed.get("key_diff", ""), - "confidence": parsed.get("confidence", "LOW"), + "vuln_type": _str(parsed.get("vuln_type", "")), + "root_cause": _str(parsed.get("root_cause", "")), + "impact": _str(parsed.get("impact", "")), + "fix_summary": _str(parsed.get("fix_summary", "")), + "key_diff": _str(parsed.get("key_diff", "")), + "confidence": _str(parsed.get("confidence", "LOW")), "commit_url": advisory["commit_url"], } @@ -107,13 +116,13 @@ def _call_gemini(user_prompt: str) -> Optional[str]: data = response.json() candidates = data.get("candidates", []) if not candidates: - print(f" Gemini: no candidates in response") + print(" Gemini: no candidates in response") return None content = candidates[0].get("content", {}) parts = content.get("parts", []) if not parts: - print(f" Gemini: no parts in response") + print(" Gemini: no parts in response") return None return parts[0].get("text", "") diff --git a/src/backfill_local.py b/src/backfill_local.py index 7d9b82a93d..e58dcc0d7c 100644 --- a/src/backfill_local.py +++ b/src/backfill_local.py @@ -1,16 +1,12 @@ -import json import sys -import time import requests from datetime import datetime, timedelta, timezone from src.config import ( - RATE_LIMIT_DELAY, MAX_DIFF_LINES, LLM_SYSTEM_PROMPT, LLM_USER_PROMPT_TEMPLATE, - TAXONOMY_PATH, STATE_PATH, ) from src.fetch import fetch_advisories, fetch_commit_diff from src.diff_filter import filter_diff -from src.analyze import load_taxonomy, _parse_llm_response, _ecosystem_to_language +from src.analyze import load_taxonomy, _parse_llm_response, build_analysis_result from src.db import ( rebuild_from_jsonl, advisory_exists, insert_analysis, export_to_jsonl, get_stats, @@ -64,7 +60,7 @@ def analyze_with_ollama(advisory: dict, filtered_diff: str) -> dict | None: raw_response = call_ollama(user_prompt) if not raw_response: - print(f" Ollama returned nothing") + print(" Ollama returned nothing") return None parsed = _parse_llm_response(raw_response) @@ -75,31 +71,7 @@ def analyze_with_ollama(advisory: dict, filtered_diff: str) -> dict | None: if parsed.get("pattern_id") not in taxonomy_ids: parsed["pattern_id"] = "UNCLASSIFIED" - def _str(val): - if isinstance(val, dict): - return json.dumps(val) - if isinstance(val, list): - return json.dumps(val) - return str(val) if val else "" - - return { - "ghsa_id": advisory["ghsa_id"], - "date": advisory["published_at"][:10], - "cve_id": "", - "repo": advisory["repo"], - "language": _ecosystem_to_language(advisory["ecosystem"]), - "severity": advisory["severity"], - "cvss_score": advisory["cvss_score"], - "package_name": advisory["package_name"], - "pattern_id": parsed["pattern_id"], - "vuln_type": _str(parsed.get("vuln_type", "")), - "root_cause": _str(parsed.get("root_cause", "")), - "impact": _str(parsed.get("impact", "")), - "fix_summary": _str(parsed.get("fix_summary", "")), - "key_diff": _str(parsed.get("key_diff", "")), - "confidence": _str(parsed.get("confidence", "LOW")), - "commit_url": advisory["commit_url"], - } + return build_analysis_result(advisory, parsed) def backfill_local(days: int): @@ -154,13 +126,13 @@ def backfill_local(days: int): raw_diff = fetch_commit_diff(advisory["commit_url"]) if not raw_diff: - print(f" SKIP: no diff") + print(" SKIP: no diff") errors += 1 continue filtered = filter_diff(raw_diff) if not filtered: - print(f" SKIP: no relevant files") + print(" SKIP: no relevant files") continue result = analyze_with_ollama(advisory, filtered) @@ -195,12 +167,12 @@ def backfill_local(days: int): render_html_index() stats = get_stats() - print(f"\n=== Summary ===") + print("\n=== Summary ===") print(f"Processed: {processed}") print(f"New patterns: {new_patterns}") print(f"Errors: {errors}") print(f"Total DB: {stats['total_advisories']} advisories, {stats['total_patterns']} patterns") - print(f"\nCommit and push:") + print("\nCommit and push:") print(f" git add -A && git commit -m 'feat: local backfill — {processed} advisories' && git push") diff --git a/src/db.py b/src/db.py index 4c390b8d74..1533c9c20f 100644 --- a/src/db.py +++ b/src/db.py @@ -1,6 +1,5 @@ import json import sqlite3 -from pathlib import Path from typing import Optional from src.config import DB_PATH, JSONL_PATH diff --git a/src/deep_scan.py b/src/deep_scan.py index 211606d5e0..a996a8aac9 100644 --- a/src/deep_scan.py +++ b/src/deep_scan.py @@ -1,58 +1,17 @@ import json import sys import time -import os import requests from datetime import datetime, timezone -from pathlib import Path -from src.config import DATA_DIR +from src.config import DATA_DIR, GITHUB_TOKEN from src.heuristics import score_commit -from src.fingerprint import match_fingerprints +from src.fingerprint import score_with_fingerprints +from src.github_api import github_get, get_link_next, GITHUB_API, REQUEST_DELAY -GITHUB_API = "https://api.github.com" -GITHUB_TOKEN = os.environ.get("GITHUB_TOKEN", "") -REQUEST_DELAY = 0.8 RESULTS_DIR = DATA_DIR / "deep_scans" -def github_get(url: str, params: dict = None) -> dict | list | None: - headers = {"Accept": "application/vnd.github.v3+json"} - if GITHUB_TOKEN: - headers["Authorization"] = f"Bearer {GITHUB_TOKEN}" - - full_url = f"{GITHUB_API}{url}" if url.startswith("/") else url - - try: - response = requests.get(full_url, headers=headers, params=params, timeout=30) - if response.status_code == 403: - remaining = response.headers.get("X-RateLimit-Remaining", "?") - reset = response.headers.get("X-RateLimit-Reset", "?") - print(f" Rate limited (remaining: {remaining}, reset: {reset})") - if remaining == "0": - wait = max(int(reset) - int(time.time()), 10) - print(f" Waiting {wait}s for rate limit reset...") - time.sleep(wait) - return github_get(url, params) - return None - if response.status_code == 404: - return None - response.raise_for_status() - return response.json() - except requests.RequestException as exc: - print(f" API error: {exc}") - return None - - -def get_link_next(response_headers: dict) -> str | None: - link = response_headers.get("Link", "") - for part in link.split(","): - if 'rel="next"' in part: - url = part.split(";")[0].strip().strip("<>") - return url - return None - - def fetch_all_commits(repo: str, since: str = None, until: str = None, per_page: int = 100) -> list[dict]: all_commits = [] params = {"per_page": per_page} @@ -146,7 +105,7 @@ def deep_scan(repo: str, since: str = None, until: str = None, max_commits: int message = commit.get("commit", {}).get("message", "").split("\n")[0] if (i + 1) % 50 == 0 or (i + 1) == len(new_commits): - print(f" [{i+1}/{len(new_commits)}] {suspects_count} suspects so far...", flush=True) if 'suspects_count' in dir() else None + print(f" [{i+1}/{len(new_commits)}] {len(suspects)} suspects so far...", flush=True) detail = github_get(f"/repos/{repo}/commits/{sha}") if not detail: @@ -166,16 +125,9 @@ def deep_scan(repo: str, since: str = None, until: str = None, max_commits: int time.sleep(REQUEST_DELAY) continue - combined_patch = "\n".join(f.get("patch", "") for f in files if f.get("patch")) - fingerprint_matches = match_fingerprints(combined_patch) - - best_fp = fingerprint_matches[0] if fingerprint_matches else None - fp_score = best_fp["score"] if best_fp else 0.0 - - raw_combined = heuristic_result["score"] + (fp_score * 20) - normalized = heuristic_result["normalized_score"] - if best_fp: - normalized = min(normalized + (fp_score * 30), 100) + normalized, best_fp, fp_score = score_with_fingerprints( + heuristic_result, files + ) top_file = heuristic_result["files"][0] if heuristic_result["files"] else {} author = detail.get("commit", {}).get("author", {}) @@ -206,7 +158,6 @@ def deep_scan(repo: str, since: str = None, until: str = None, max_commits: int f.write(json.dumps(result, ensure_ascii=False) + "\n") suspects.append(result) - suspects_count = len(suspects) severity = "HIGH" if normalized >= 60 else "MEDIUM" if normalized >= 30 else "LOW" print(f" [{i+1}/{len(new_commits)}] {severity} score={normalized} {sha[:8]} {message[:60]}") @@ -221,7 +172,7 @@ def deep_scan(repo: str, since: str = None, until: str = None, max_commits: int if suspects: print(f"\nResults saved to: {results_path}") - print(f"\nTop suspects:") + print("\nTop suspects:") top = sorted(suspects, key=lambda s: s["normalized_score"], reverse=True)[:10] for s in top: print(f" score={s['normalized_score']:5.1f} {s['commit_sha'][:8]} {s['message'][:60]}") diff --git a/src/fetch.py b/src/fetch.py index 68a52a9666..f9d2bc258c 100644 --- a/src/fetch.py +++ b/src/fetch.py @@ -28,7 +28,7 @@ def graphql_request(query: str, variables: dict) -> dict: if "errors" in data: raise RuntimeError(f"GraphQL errors: {data['errors']}") return data - except (requests.RequestException, RuntimeError) as exc: + except (requests.RequestException, RuntimeError): if attempt == RETRY_ATTEMPTS - 1: raise time.sleep(RETRY_BACKOFF[attempt]) diff --git a/src/fingerprint.py b/src/fingerprint.py index 3c26ae64be..f1433dd584 100644 --- a/src/fingerprint.py +++ b/src/fingerprint.py @@ -1,6 +1,5 @@ import json import re -from pathlib import Path from src.config import DATA_DIR @@ -115,4 +114,26 @@ def get_best_match(patch_text: str) -> dict | None: matches = match_fingerprints(patch_text) if matches and matches[0]["score"] >= 0.1: return matches[0] - return None \ No newline at end of file + return None + + +def score_with_fingerprints( + heuristic_result: dict, files: list[dict] +) -> tuple[float, dict | None, float]: + """Combine heuristic score with fingerprint matching. + + Returns (normalized_score, best_fingerprint, fp_score). + """ + combined_patch = "\n".join( + f.get("patch", "") for f in files if f.get("patch") + ) + fingerprint_matches = match_fingerprints(combined_patch) + + best_fp = fingerprint_matches[0] if fingerprint_matches else None + fp_score = best_fp["score"] if best_fp else 0.0 + + normalized = heuristic_result["normalized_score"] + if best_fp: + normalized = min(normalized + (fp_score * 30), 100) + + return normalized, best_fp, fp_score \ No newline at end of file diff --git a/src/fingerprint_builder.py b/src/fingerprint_builder.py index ee3ab30732..ba8ba9cb30 100644 --- a/src/fingerprint_builder.py +++ b/src/fingerprint_builder.py @@ -2,7 +2,6 @@ import re import sys from pathlib import Path -from collections import defaultdict EXPERT_FINGERPRINTS = { @@ -246,7 +245,7 @@ def build_fingerprints(output_path: str = "data/fingerprints.json"): print(f" Expert CWE: {len(EXPERT_FINGERPRINTS)}") print(f" OSDC live: {len([k for k in fingerprints if k.startswith('OSDC:')])}") - print(f"\nTop patterns by token count:") + print("\nTop patterns by token count:") top = sorted(fingerprints.items(), key=lambda x: len(x[1]["add_tokens"]) + len(x[1]["del_tokens"]), reverse=True)[:10] for pid, data in top: total = len(data["add_tokens"]) + len(data["del_tokens"]) diff --git a/src/github_api.py b/src/github_api.py new file mode 100644 index 0000000000..d053e305e1 --- /dev/null +++ b/src/github_api.py @@ -0,0 +1,44 @@ +import time +import requests +from src.config import GITHUB_TOKEN + + +GITHUB_API = "https://api.github.com" +REQUEST_DELAY = 0.8 + + +def github_get(endpoint: str, params: dict = None) -> dict | list | None: + headers = {"Accept": "application/vnd.github.v3+json"} + if GITHUB_TOKEN: + headers["Authorization"] = f"Bearer {GITHUB_TOKEN}" + + url = f"{GITHUB_API}{endpoint}" if endpoint.startswith("/") else endpoint + + try: + response = requests.get(url, headers=headers, params=params, timeout=30) + if response.status_code == 403: + remaining = response.headers.get("X-RateLimit-Remaining", "?") + reset = response.headers.get("X-RateLimit-Reset", "0") + if remaining == "0": + wait = max(int(reset) - int(time.time()), 10) + print(f" Rate limited, waiting {wait}s...") + time.sleep(wait) + return github_get(endpoint, params) + print(f" Rate limited (remaining: {remaining})") + return None + if response.status_code == 404: + return None + response.raise_for_status() + return response.json() + except requests.RequestException as exc: + print(f" API error: {exc}") + return None + + +def get_link_next(response_headers: dict) -> str | None: + link = response_headers.get("Link", "") + for part in link.split(","): + if 'rel="next"' in part: + url = part.split(";")[0].strip().strip("<>") + return url + return None diff --git a/src/main.py b/src/main.py index 2b50f54619..6ddb33566a 100644 --- a/src/main.py +++ b/src/main.py @@ -1,7 +1,6 @@ import json import time import signal -import sys from datetime import datetime, timezone from src.config import ( STATE_PATH, MAX_DAILY_CALLS, RATE_LIMIT_DELAY, @@ -98,18 +97,18 @@ def run(): raw_diff = fetch_commit_diff(advisory["commit_url"]) if not raw_diff: - print(f" SKIP: no diff available") + print(" SKIP: no diff available") errors += 1 continue filtered = filter_diff(raw_diff) if not filtered: - print(f" SKIP: no relevant files in diff") + print(" SKIP: no relevant files in diff") continue result = analyze_advisory(advisory, filtered) if not result: - print(f" ERROR: LLM analysis failed") + print(" ERROR: LLM analysis failed") errors += 1 continue @@ -174,7 +173,7 @@ def run(): encoding="utf-8", ) - print(f"\n=== Summary ===") + print("\n=== Summary ===") print(f"Analyzed: {len(analyzed)}") print(f"New patterns: {new_patterns}") print(f"Errors: {errors}") diff --git a/src/render.py b/src/render.py index 8aba29087c..8bef3ab545 100644 --- a/src/render.py +++ b/src/render.py @@ -1,15 +1,14 @@ import json import re from datetime import date -from pathlib import Path from jinja2 import Environment, FileSystemLoader from src.config import ( TEMPLATES_DIR, PATCHES_DIR, DOCS_DIR, ROOT_DIR, - README_DAYS_SHOWN, DATA_DIR, + DATA_DIR, ) from src.db import ( - get_advisories_for_date, get_pattern_info, - get_recent_dates, get_all_advisories, get_stats, + get_pattern_info, + get_all_advisories, get_stats, ) @@ -52,22 +51,25 @@ def _clean_text(raw: str) -> str: return str(raw).strip() +def _enrich_advisory(adv: dict) -> dict: + """Clean text fields and attach pattern info.""" + pattern_info = get_pattern_info(adv["pattern_id"]) + adv["pattern_info"] = pattern_info + adv["occurrences"] = pattern_info["occurrences"] if pattern_info else 1 + adv["key_diff"] = _clean_diff(adv.get("key_diff", "")) + adv["root_cause"] = _clean_text(adv.get("root_cause", "")) + adv["impact"] = _clean_text(adv.get("impact", "")) + adv["fix_summary"] = _clean_text(adv.get("fix_summary", "")) + adv["commit_url"] = adv.get("commit_url", "") + return adv + + def render_daily_patch(target_date: str, advisories: list[dict]): PATCHES_DIR.mkdir(parents=True, exist_ok=True) env = init_renderer() template = env.get_template("patch.md.j2") - enriched = [] - for adv in advisories: - pattern_info = get_pattern_info(adv["pattern_id"]) - enriched.append({ - **adv, - "pattern_info": pattern_info, - "key_diff": _clean_diff(adv.get("key_diff", "")), - "root_cause": _clean_text(adv.get("root_cause", "")), - "impact": _clean_text(adv.get("impact", "")), - "fix_summary": _clean_text(adv.get("fix_summary", "")), - }) + enriched = [_enrich_advisory({**adv}) for adv in advisories] content = template.render( date=target_date, @@ -90,13 +92,7 @@ def render_readme(): )[:50] for adv in top_advisories: - pattern_info = get_pattern_info(adv["pattern_id"]) - adv["occurrences"] = pattern_info["occurrences"] if pattern_info else 1 - adv["root_cause"] = _clean_text(adv.get("root_cause", "")) - adv["impact"] = _clean_text(adv.get("impact", "")) - adv["fix_summary"] = _clean_text(adv.get("fix_summary", "")) - adv["key_diff"] = _clean_diff(adv.get("key_diff", "")) - adv["commit_url"] = adv.get("commit_url", "") + _enrich_advisory(adv) stats = get_stats() @@ -118,10 +114,7 @@ def render_html_index(): all_advisories = get_all_advisories() for adv in all_advisories: - adv["key_diff"] = _clean_diff(adv.get("key_diff", "")) - adv["root_cause"] = _clean_text(adv.get("root_cause", "")) - adv["impact"] = _clean_text(adv.get("impact", "")) - adv["fix_summary"] = _clean_text(adv.get("fix_summary", "")) + _enrich_advisory(adv) stats = get_stats() diff --git a/src/render_cli.py b/src/render_cli.py index 15a9578d7b..d98d04e7d1 100644 --- a/src/render_cli.py +++ b/src/render_cli.py @@ -1,9 +1,7 @@ import json -from datetime import date -from pathlib import Path from src.config import DOCS_DIR from src.db import ( - rebuild_from_jsonl, get_all_advisories, get_recent_dates, + rebuild_from_jsonl, get_recent_dates, get_advisories_for_date, get_stats, ) from src.render import render_readme, render_html_index, render_daily_patch, render_silent_page @@ -32,7 +30,7 @@ def generate_badge(stats: dict): (DOCS_DIR / "badge-patterns.json").write_text( json.dumps(badge_patterns), encoding="utf-8" ) - print(f" Generated badge endpoints") + print(" Generated badge endpoints") def run(): diff --git a/src/silent_scan.py b/src/silent_scan.py index acaafb4b70..01d206e98a 100644 --- a/src/silent_scan.py +++ b/src/silent_scan.py @@ -1,20 +1,16 @@ import json import time import sys -import os -import requests from datetime import datetime, timedelta, timezone -from pathlib import Path -from src.config import DATA_DIR, GITHUB_TOKEN +from src.config import DATA_DIR from src.heuristics import score_commit -from src.fingerprint import match_fingerprints +from src.fingerprint import score_with_fingerprints +from src.github_api import github_get, REQUEST_DELAY SILENT_STATE_PATH = DATA_DIR / "silent_state.json" SILENT_RESULTS_PATH = DATA_DIR / "silent_results.jsonl" WATCHLIST_PATH = DATA_DIR / "watchlist.json" -GITHUB_API = "https://api.github.com" -REQUEST_DELAY = 0.8 def load_watchlist() -> list[str]: @@ -57,33 +53,6 @@ def append_result(result: dict): f.write(json.dumps(result, ensure_ascii=False) + "\n") -def github_get(endpoint: str, params: dict = None) -> dict | list | None: - headers = {"Accept": "application/vnd.github.v3+json"} - if GITHUB_TOKEN: - headers["Authorization"] = f"Bearer {GITHUB_TOKEN}" - - url = f"{GITHUB_API}{endpoint}" if endpoint.startswith("/") else endpoint - - try: - response = requests.get(url, headers=headers, params=params, timeout=30) - if response.status_code == 403: - remaining = response.headers.get("X-RateLimit-Remaining", "?") - reset = response.headers.get("X-RateLimit-Reset", "0") - if remaining == "0": - wait = max(int(reset) - int(time.time()), 10) - print(f" Rate limited, waiting {wait}s...") - time.sleep(wait) - return github_get(endpoint, params) - return None - if response.status_code == 404: - return None - response.raise_for_status() - return response.json() - except requests.RequestException as exc: - print(f" API error: {exc}") - return None - - def run(hours: int = 24): now = datetime.now(timezone.utc) since = (now - timedelta(hours=hours)).isoformat() @@ -141,15 +110,9 @@ def run(hours: int = 24): layer1_pass += 1 - combined_patch = "\n".join(f.get("patch", "") for f in files if f.get("patch")) - fingerprint_matches = match_fingerprints(combined_patch) - - best_fp = fingerprint_matches[0] if fingerprint_matches else None - fp_score = best_fp["score"] if best_fp else 0.0 - - normalized = heuristic_result["normalized_score"] - if best_fp: - normalized = min(normalized + (fp_score * 30), 100) + normalized, best_fp, fp_score = score_with_fingerprints( + heuristic_result, files + ) if normalized < 20: continue @@ -202,7 +165,7 @@ def run(hours: int = 24): } save_silent_state(state) - print(f"\n=== Summary ===") + print("\n=== Summary ===") print(f"Repos scanned: {len(watchlist) - skipped_repos}/{len(watchlist)}") print(f"Commits analyzed: {total_commits}") print(f"Layer 1 pass (heuristics >= 8): {layer1_pass}")