From 821fa8a5eb09e3a6cf8a2b019d9dfe9362fe0853 Mon Sep 17 00:00:00 2001 From: prasiddhi-105 Date: Mon, 6 Jul 2026 09:53:58 +0530 Subject: [PATCH] fix: resolve type hint mismatch in deduplicator and broaden embedder exceptions (#259) --- backend/app/main.py | 1069 ++++++++++++++++++++++++-------- backend/app/ml/deduplicator.py | 75 +++ backend/app/ml/embedder.py | 41 ++ 3 files changed, 932 insertions(+), 253 deletions(-) create mode 100644 backend/app/ml/deduplicator.py create mode 100644 backend/app/ml/embedder.py diff --git a/backend/app/main.py b/backend/app/main.py index 81e8359..166b06f 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -1,6 +1,7 @@ from __future__ import annotations import asyncio +import functools import json import logging import os @@ -8,6 +9,7 @@ import re import shutil import tempfile +import threading import uuid from datetime import datetime, timezone from pathlib import Path @@ -25,23 +27,35 @@ Request, UploadFile, ) +from fastapi.concurrency import run_in_threadpool from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import FileResponse, Response, StreamingResponse -from pydantic import BaseModel - -from app.ml.ranker import load_ranker, scoring_function +from pydantic import BaseModel, Field from .db import ( + create_findings, + create_job, + delete_job, get_cwe_distribution, get_db, get_dependency_diff, + get_finding, + get_findings_by_job_id, + get_job, get_leaderboard_stats, get_trend_data, init_db, + update_finding_status, + update_job_status, upsert_contributor_stat, ) +from .ml.deduplicator import SENTENCE_TRANSFORMERS_AVAILABLE, deduplicate +from .ml.fp_predictor import predictor +from .ml.ranker import load_ranker, scoring_function from .models import ( Finding, + FindingStatusUpdate, + Fix, FixRequest, FixResponse, Location, @@ -59,7 +73,7 @@ from .scanners.gitleaks import run_gitleaks from .scanners.osv import run_osv_scanner from .scanners.semgrep import run_semgrep -from .utils.fs import ensure_dir, safe_rmtree, unzip_to_dir +from .utils.fs import ensure_dir, safe_job_dir, safe_rmtree, unzip_to_dir _MAX_UPLOAD_MB_RAW = os.environ.get("MAX_UPLOAD_MB") RANKER = load_ranker() @@ -75,9 +89,27 @@ logger = logging.getLogger(__name__) app = FastAPI(title="PatchPilot API", version="0.1.0") +ALLOWED_ORIGINS = [ + "http://localhost:5173", + "http://127.0.0.1:5173", + "http://localhost:3000", + "http://127.0.0.1:3000", +] + +env_origins = os.environ.get("ALLOWED_ORIGINS") or os.environ.get( + "VITE_API_BASE_URL", "" +) +if env_origins: + for origin in env_origins.split(","): + cleaned_origin = origin.strip() + if cleaned_origin: + if cleaned_origin.endswith("/"): + cleaned_origin = cleaned_origin.rstrip("/") + ALLOWED_ORIGINS.append(cleaned_origin) + app.add_middleware( CORSMiddleware, - allow_origins=["*"], + allow_origins=ALLOWED_ORIGINS, allow_credentials=True, allow_methods=["*"], allow_headers=["*"], @@ -111,6 +143,42 @@ def health(): } +@app.get("/api/health/ollama", tags=["Health"]) +async def ollama_health(): + """ + Ping the local Ollama service to verify availability and list installed models. + Fails gracefully to prevent 500 errors if the service is down. + """ + base_url = os.getenv("OLLAMA_BASE_URL", "http://localhost:11434").rstrip("/") + timeout = httpx.Timeout(3.0) + + try: + async with httpx.AsyncClient(timeout=timeout) as client: + response = await client.get(f"{base_url}/api/tags") + + if response.status_code == 200: + data = response.json() + models = [model.get("name") for model in data.get("models", [])] + return { + "available": True, + "models": models, + "base_url": base_url, + } + else: + return { + "available": False, + "models": [], + "base_url": base_url, + } + except (httpx.RequestError, ValueError, AttributeError, TypeError) as e: + logger.warning(f"Ollama health check failed: {e}") + return { + "available": False, + "models": [], + "base_url": base_url, + } + + def _prioritize_findings(findings: List[Finding]) -> List[Finding]: def score(f: Finding) -> int: sev = {"CRITICAL": 100, "HIGH": 80, "MEDIUM": 50, "LOW": 20, "INFO": 5}.get( @@ -163,10 +231,55 @@ def _extract_dependencies(repo_dir: Path) -> List[tuple[str, str]]: return deps -def _scan_repo_dir(repo_dir: Path): - semgrep = run_semgrep(repo_dir) - osv = run_osv_scanner(repo_dir) - gitleaks = run_gitleaks(repo_dir) +ACTIVE_SCANS = {} +ORG_CANCEL_EVENTS = {} + + +def _scan_repo_dir( + repo_dir: Path, + progress_cb=None, + job_dir: Path = None, + cancel_event: threading.Event = None, + raw_dir_name: str = "raw", +): + if cancel_event and cancel_event.is_set(): + raise asyncio.CancelledError() + if progress_cb: + progress_cb("sast", "in_progress") + + semgrep_raw_out = (job_dir / raw_dir_name / "semgrep.json") if job_dir else None + semgrep = run_semgrep(repo_dir, raw_out=semgrep_raw_out) + + if cancel_event and cancel_event.is_set(): + raise asyncio.CancelledError() + + if progress_cb: + progress_cb("sast", "completed") + + if progress_cb: + progress_cb("dependency", "in_progress") + + osv_raw_out = (job_dir / raw_dir_name / "osv.json") if job_dir else None + osv = run_osv_scanner(repo_dir, raw_out=osv_raw_out) + + if cancel_event and cancel_event.is_set(): + raise asyncio.CancelledError() + + if progress_cb: + progress_cb("dependency", "completed") + + if progress_cb: + progress_cb("secrets", "in_progress") + + gitleaks_raw_out = (job_dir / raw_dir_name / "gitleaks.json") if job_dir else None + gitleaks = run_gitleaks(repo_dir, raw_out=gitleaks_raw_out) + + if cancel_event and cancel_event.is_set(): + raise asyncio.CancelledError() + + if progress_cb: + progress_cb("secrets", "completed") + entropy = run_entropy(repo_dir) findings: List[Finding] = [] @@ -189,6 +302,10 @@ def _scan_repo_dir(repo_dir: Path): def finding_key(f: Finding): + """Generate a stable identifier for a finding. + Uses rule identifier, file path, and start line so that findings from the + same rule on different lines are treated as distinct findings. + """ metadata = f.metadata or {} rule_id = ( @@ -230,10 +347,16 @@ def github_zip_url(repo_url: str, ref: str = "main") -> str: MAX_REDIRECTS = 5 -async def download_to_path(url: str, dest_path: Path, max_retries: int = 5) -> None: +async def download_to_path( + url: str, + dest_path: Path, + max_retries: int = 5, + cancel_event: threading.Event = None, +) -> None: """ Download *url* to *dest_path*, following redirects only to hosts in ALLOWED_REDIRECT_HOSTS. Implements exponential backoff for GitHub rate limits. + Now securely streams the download to prevent RAM/Disk exhaustion. """ dest_path.parent.mkdir(parents=True, exist_ok=True) timeout = httpx.Timeout(120.0, connect=30.0) @@ -242,6 +365,8 @@ async def download_to_path(url: str, dest_path: Path, max_retries: int = 5) -> N async with httpx.AsyncClient(timeout=timeout, follow_redirects=False) as client: for attempt in range(max_retries): current_url = url + status_code_for_retry = None + try: for _ in range(MAX_REDIRECTS): parsed = httpx.URL(current_url) @@ -250,32 +375,49 @@ async def download_to_path(url: str, dest_path: Path, max_retries: int = 5) -> N status_code=400, detail=f"Redirect to disallowed host '{parsed.host}' was blocked.", ) - - r = await client.get(current_url) - - if r.status_code in (301, 302, 303, 307, 308): - location = r.headers.get("location") - if not location: + async with client.stream("GET", current_url) as r: + if r.status_code in (301, 302, 303, 307, 308): + location = r.headers.get("location") + if not location: + raise HTTPException( + status_code=400, + detail="Redirect missing Location header.", + ) + current_url = str(location) + continue + + if r.status_code in (403, 429): + status_code_for_retry = r.status_code + break + + if r.status_code != 200: raise HTTPException( status_code=400, - detail="Redirect missing Location header.", + detail=f"Failed to download repo ZIP ({r.status_code}).", ) - current_url = str(location) - continue - - if r.status_code in (403, 429): - break - - if r.status_code != 200: - raise HTTPException( - status_code=400, - detail=f"Failed to download repo ZIP ({r.status_code}).", - ) - - dest_path.write_bytes(r.content) - return - - if r.status_code in (403, 429): + bytes_received = 0 + chunk_size = 1024 * 1024 + + try: + with open(dest_path, "wb") as f: + async for chunk in r.aiter_bytes(chunk_size=chunk_size): + bytes_received += len(chunk) + if bytes_received > MAX_UPLOAD_SIZE: + raise HTTPException( + status_code=413, + detail=f"Remote repository exceeds the maximum limit of {MAX_UPLOAD_MB}MB.", + ) + f.write(chunk) + except Exception: + try: + if dest_path.exists(): + dest_path.unlink() + except Exception: + pass + raise + return + + if status_code_for_retry in (403, 429): if attempt == max_retries - 1: raise HTTPException( status_code=429, @@ -285,7 +427,7 @@ async def download_to_path(url: str, dest_path: Path, max_retries: int = 5) -> N jitter = random.uniform(0.5, 1.5) sleep_time = (base_delay * (2**attempt)) + jitter logger.warning( - f"Rate limited (status {r.status_code}) on {url}. Retrying in {sleep_time:.2f}s..." + f"Rate limited (status {status_code_for_retry}) on {url}. Retrying in {sleep_time:.2f}s..." ) await asyncio.sleep(sleep_time) continue @@ -303,6 +445,30 @@ async def download_to_path(url: str, dest_path: Path, max_retries: int = 5) -> N await asyncio.sleep((base_delay * (2**attempt)) + jitter) +async def _apply_fp_predictor(findings: List[Finding]) -> None: + ml_input = [] + for f in findings: + rule_id = ( + (f.metadata or {}).get("check_id") + or (f.metadata or {}).get("rule") + or (f.metadata or {}).get("osv_id") + or f.title + ) + ml_input.append( + { + "rule_id": rule_id, + "message": f.description or f.title, + "file_path": f.location.path if f.location else "", + "ml_score": getattr(f, "ml_score", 1.0), + } + ) + + adjusted_scores = await run_in_threadpool(predictor.adjust_scores, ml_input) + + for f, new_score in zip(findings, adjusted_scores): + f.ml_score = new_score + + def _maybe_use_single_top_folder(repo_dir: Path) -> Path: """ If the extracted folder contains exactly one top-level directory (typical GitHub ZIP), @@ -320,52 +486,51 @@ def _maybe_use_single_top_folder(repo_dir: Path) -> Path: return repo_dir -@app.post("/scan", response_model=ScanResponse) -async def scan( - request: Request, - project: UploadFile = File(...), - project_name: str = Form("project"), +async def _run_single_scan_task( + job_id: str, project_name: str, scan_method: str, scan_root: Path ): - content_length = request.headers.get("content-length") + def update_progress(phase, status): + if job_id in ACTIVE_SCANS: + ACTIVE_SCANS[job_id][phase] = status + + ACTIVE_SCANS[job_id] = { + "sast": "pending", + "dependency": "pending", + "secrets": "pending", + "status": "running", + } try: - content_length = int(content_length) if content_length else None - except ValueError: - content_length = None + db = await get_db() + try: + await create_job(db, job_id, project_name, scan_method) + finally: + await db.close() - if content_length and content_length > MAX_UPLOAD_SIZE: - raise HTTPException( - status_code=413, - detail=f"File too large. Maximum upload size is {MAX_UPLOAD_MB}MB.", + job_dir = safe_job_dir(WORK_ROOT, job_id) + semgrep, osv, gitleaks, entropy, findings = await run_in_threadpool( + functools.partial( + _scan_repo_dir, scan_root, update_progress, job_dir=job_dir + ) ) - job_id = next(tempfile._get_candidate_names()) - job_dir = WORK_ROOT / job_id - ensure_dir(job_dir) - - archive_path = job_dir / project.filename - content = await project.read() - archive_path.write_bytes(content) + raw_finding_count = len(findings) - repo_dir = job_dir / "repo" - ensure_dir(repo_dir) + disable_dedup = os.environ.get("DISABLE_DEDUP", "").lower() == "true" + try: + epsilon = float(os.environ.get("DEDUP_EPSILON", 0.15)) + except ValueError: + epsilon = 0.15 - try: - unzip_to_dir(archive_path, repo_dir) - except Exception as e: - safe_rmtree(job_dir) - raise HTTPException(status_code=400, detail=f"Invalid zip upload: {e}") + if not disable_dedup and SENTENCE_TRANSFORMERS_AVAILABLE: + findings = deduplicate(findings, epsilon) - scan_root = _maybe_use_single_top_folder(repo_dir) + await _apply_fp_predictor(findings) - semgrep, osv, gitleaks, entropy, findings = _scan_repo_dir(scan_root) + finding_count = len(findings) - try: - async with await get_db() as db: - await db.execute( - "INSERT INTO jobs (job_id, project_name, scan_method) VALUES (?, ?, ?)", - (job_id, project_name, "zip"), - ) + db = await get_db() + try: rows = [] for f in findings: engine = (f.metadata or {}).get("engine") @@ -379,16 +544,15 @@ async def scan( file_path = f.location.path if f.location else None line_number = f.location.start_line if f.location else None message = f.description or f.title - pkg_info = (f.metadata or {}).get("package") or {} pkg_name = pkg_info.get("name") pkg_version = pkg_info.get("version") - rows.append( ( str(uuid.uuid4()), job_id, rule_id, + f.title, f.severity, f.category, file_path, @@ -398,43 +562,168 @@ async def scan( message, pkg_name, pkg_version, + f.ml_score, + json.dumps(f.features) if f.features else None, ) ) - await db.executemany( - "INSERT INTO findings (id, job_id, rule_id, severity, category, file_path, line_number, cwe, scanner, message, package_name, package_version) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", - rows, + if rows: + await create_findings(db, rows) + await update_job_status( + db, job_id, "completed", raw_finding_count, finding_count ) - await db.commit() + finally: + await db.close() + + if job_id in ACTIVE_SCANS: + ACTIVE_SCANS[job_id]["status"] = "completed" + ACTIVE_SCANS[job_id]["findings_count"] = finding_count except Exception: - logger.exception("DB write failed for job %s", job_id) - return ScanResponse( - job_id=job_id, - project_name=project_name, - repo_path=str(scan_root), - findings=findings, - scanners={ - "semgrep": {"ok": True, "count": len(semgrep)}, - "osv": {"ok": True, "count": len(osv)}, - "gitleaks": {"ok": True, "count": len(gitleaks)}, - "entropy": {"ok": True, "count": len(entropy)}, - }, - ) + logger.exception("Failed scan task for %s", job_id) + if job_id in ACTIVE_SCANS: + ACTIVE_SCANS[job_id]["status"] = "failed" + try: + db = await get_db() + try: + await update_job_status(db, job_id, "failed") + finally: + await db.close() + except Exception: + logger.exception("Failed to write failed status for job %s", job_id) + + +@app.post( + "/scan", + summary="Scan an uploaded project archive", + description=( + "Uploads a ZIP archive containing a project, extracts it, " + "and starts an asynchronous security scan. " + "The endpoint immediately returns a job ID that can be used " + "to monitor scan progress and retrieve results." + ), + response_model=dict, + responses={ + 400: {"description": "Invalid ZIP archive or upload error."}, + 413: {"description": "Uploaded file exceeds the maximum allowed size."}, + 500: {"description": "Internal server error."}, + }, +) +async def scan( + request: Request, + background_tasks: BackgroundTasks, + project: UploadFile = File( + ..., + description="ZIP archive containing the project source code to scan.", + ), + project_name: str = Form( + "project", + description="Display name assigned to the uploaded project.", + ), +): + """ + Upload a project archive and start an asynchronous security scan. + + The uploaded ZIP file is extracted into a temporary workspace, + validated, and queued for background scanning. The endpoint + returns immediately with a unique job ID that can be used to + track scan progress and retrieve scan results. + """ + content_length = request.headers.get("content-length") + try: + content_length = int(content_length) if content_length else None + except ValueError: + content_length = None + + if content_length and content_length > MAX_UPLOAD_SIZE: + raise HTTPException( + status_code=413, + detail=f"Header indicates file is too large. Maximum upload size is {MAX_UPLOAD_MB}MB.", + ) + + job_id = next(tempfile._get_candidate_names()) + job_dir = WORK_ROOT / job_id + ensure_dir(job_dir) + archive_path = job_dir / project.filename + bytes_received = 0 + chunk_size = 1024 * 1024 + try: + with open(archive_path, "wb") as f: + while chunk := await project.read(chunk_size): + bytes_received += len(chunk) + if bytes_received > MAX_UPLOAD_SIZE: + raise HTTPException( + status_code=413, + detail=f"Actual file size exceeds the maximum limit of {MAX_UPLOAD_MB}MB.", + ) + f.write(chunk) + except HTTPException: + safe_rmtree(job_dir) + raise + except Exception as e: + safe_rmtree(job_dir) + raise HTTPException(status_code=400, detail=f"Error saving upload: {e}") + + repo_dir = job_dir / "repo" + ensure_dir(repo_dir) -@app.post("/scan-url", response_model=ScanResponse) + try: + unzip_to_dir(archive_path, repo_dir) + except Exception as e: + safe_rmtree(job_dir) + raise HTTPException(status_code=400, detail=f"Invalid zip upload: {e}") + + scan_root = _maybe_use_single_top_folder(repo_dir) + background_tasks.add_task( + _run_single_scan_task, job_id, project_name, "zip", scan_root + ) + return {"job_id": job_id, "project_name": project_name, "status": "running"} + + +@app.post( + "/scan-url", + summary="Scan a GitHub repository", + description=( + "Downloads a GitHub repository as a ZIP archive using the " + "specified repository URL and branch, then starts an " + "asynchronous security scan. Returns a job ID for tracking " + "the scan progress and results." + ), + responses={ + 400: {"description": "Invalid repository URL or repository download failed."}, + 404: {"description": "Repository or branch not found."}, + 429: {"description": "GitHub API rate limit exceeded."}, + 500: {"description": "Internal server error."}, + }, +) async def scan_url( - repo_url: str = Form(...), - ref: str = Form("main"), - project_name: str = Form("project"), + background_tasks: BackgroundTasks, + repo_url: str = Form( + ..., + description="Public GitHub repository URL to scan.", + ), + ref: str = Form( + "main", + description="Git branch to download and scan.", + ), + project_name: str = Form( + "project", + description="Display name assigned to this scan.", + ), ): + """ + Download a GitHub repository and start an asynchronous security scan. + + The repository is downloaded as a ZIP archive, extracted into a + temporary workspace, and queued for background scanning. A job ID + is returned immediately for tracking scan progress and retrieving + results. + """ job_id = next(tempfile._get_candidate_names()) job_dir = WORK_ROOT / job_id ensure_dir(job_dir) - archive_path = job_dir / "repo.zip" repo_dir = job_dir / "repo" ensure_dir(repo_dir) - zip_url = github_zip_url(repo_url, ref=ref) try: @@ -448,78 +737,112 @@ async def scan_url( raise HTTPException(status_code=400, detail=f"Import from URL failed: {e}") scan_root = _maybe_use_single_top_folder(repo_dir) + background_tasks.add_task( + _run_single_scan_task, job_id, project_name, "url", scan_root + ) + return {"job_id": job_id, "project_name": project_name, "status": "running"} + - semgrep, osv, gitleaks, entropy, findings = _scan_repo_dir(scan_root) +@app.get("/api/scans/{job_id}/stream") +async def stream_single_scan_status(job_id: str): + async def event_generator(): + while True: + if job_id not in ACTIVE_SCANS: + yield f"data: {json.dumps({'error': 'Job not found'})}\n\n" + break + state = ACTIVE_SCANS[job_id] + yield f"data: {json.dumps(state)}\n\n" + if state["status"] in ["completed", "failed"]: + break + await asyncio.sleep(0.5) + + return StreamingResponse(event_generator(), media_type="text/event-stream") + +async def _record_fixes_to_db(job_id: str, fixes: List[Fix]): try: db = await get_db() try: - await db.execute( - "INSERT INTO jobs (job_id, project_name, scan_method) VALUES (?, ?, ?)", - (job_id, project_name, "url"), - ) rows = [] - for f in findings: - engine = (f.metadata or {}).get("engine") - scanner = {"osv-scanner": "osv"}.get(engine, engine) - rule_id = ( - (f.metadata or {}).get("check_id") - or (f.metadata or {}).get("rule") - or (f.metadata or {}).get("osv_id") - or f.title - ) - file_path = f.location.path if f.location else None - line_number = f.location.start_line if f.location else None - message = f.description or f.title - - pkg_info = (f.metadata or {}).get("package") or {} - pkg_name = pkg_info.get("name") - pkg_version = pkg_info.get("version") + for f in fixes: + if not f.diff: + continue + diff_str = f.diff + adds = 0 + dels = 0 + for line in diff_str.split("\n"): + if line.startswith("+") and not line.startswith("+++"): + adds += 1 + elif line.startswith("-") and not line.startswith("---"): + dels += 1 + + diff_line_count = adds + dels + files = f.files_changed + diff_file_count = len(files) if files else 0 + + if adds == 0 and dels == 0: + fix_type = "none" + elif adds > 0 and dels == 0: + fix_type = "insert" + elif dels > 0 and adds == 0: + fix_type = "delete" + else: + fix_type = "mixed" rows.append( ( str(uuid.uuid4()), job_id, - rule_id, - f.severity, - f.category, - file_path, - line_number, - None, - scanner, - message, - pkg_name, - pkg_version, + f.finding_id, + diff_line_count, + diff_file_count, + fix_type, ) ) if rows: await db.executemany( - "INSERT INTO findings (id, job_id, rule_id, severity, category, file_path, line_number, cwe, scanner, message, package_name, package_version) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + "INSERT INTO fixes " + "(id, job_id, finding_id, diff_line_count, " + "diff_file_count, fix_type) " + "VALUES (?, ?, ?, ?, ?, ?)", rows, ) - await db.commit() + await db.commit() finally: await db.close() except Exception: - logger.exception("DB write failed for job %s", job_id) - - return ScanResponse( - job_id=job_id, - project_name=project_name, - repo_path=str(scan_root), - findings=findings, - scanners={ - "semgrep": {"ok": True, "count": len(semgrep)}, - "osv": {"ok": True, "count": len(osv)}, - "gitleaks": {"ok": True, "count": len(gitleaks)}, - "entropy": {"ok": True, "count": len(entropy)}, - }, - ) + logger.exception("Failed to record fixes to DB for job %s", job_id) + + +@app.post( + "/fix", + response_model=FixResponse, + summary="Generate remediation suggestions", + description=( + "Generates automated fix suggestions for one or more selected " + "security findings within a previously scanned project. " + "The generated fixes are returned to the client and recorded " + "asynchronously for reporting purposes." + ), + responses={ + 404: {"description": "Specified scan job was not found."}, + 422: {"description": "Invalid request payload."}, + 500: {"description": "Internal server error."}, + }, +) +def fix(req: FixRequest, background_tasks: BackgroundTasks): + """ + Generate remediation suggestions for selected findings. + Uses the specified job ID and finding IDs to produce proposed + fixes for detected security issues. Generated fixes are returned + immediately and stored asynchronously for later reporting. + """ + try: + job_dir = safe_job_dir(WORK_ROOT, req.job_id) + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) -@app.post("/fix", response_model=FixResponse) -def fix(req: FixRequest): - job_dir = WORK_ROOT / req.job_id repo_dir = job_dir / "repo" if not repo_dir.exists(): raise HTTPException(status_code=404, detail="Unknown job_id") @@ -527,6 +850,8 @@ def fix(req: FixRequest): repo_dir = _maybe_use_single_top_folder(repo_dir) fixes = propose_fixes(repo_dir, req.finding_ids) + background_tasks.add_task(_record_fixes_to_db, req.job_id, fixes) + return FixResponse(job_id=req.job_id, fixes=fixes) @@ -535,7 +860,7 @@ async def get_baseline_findings(job_id: str): try: cursor = await db.execute( """ - SELECT rule_id, file_path, line_number + SELECT rule_id, file_path FROM findings WHERE job_id = ? """, @@ -543,21 +868,52 @@ async def get_baseline_findings(job_id: str): ) rows = await cursor.fetchall() - return { - ( - row[0], - row[1], - row[2], - ) - for row in rows - } + return {(row[0], row[1]) for row in rows} finally: await db.close() -@app.post("/verify", response_model=VerifyResponse) -async def verify(job_id: str = Form(...)): - job_dir = WORK_ROOT / job_id +@app.post( + "/verify", + response_model=VerifyResponse, + summary="Verify repository fixes", + description=( + "Verifies whether the applied fixes successfully resolve the " + "detected security issues. The repository is rescanned and " + "compared against the baseline scan to identify any newly " + "introduced issues." + ), + responses={ + 404: {"description": "Specified scan job was not found."}, + 422: {"description": "Invalid request parameters."}, + 500: {"description": "Internal server error."}, + }, +) +async def verify( + job_id: str = Form( + ..., + description="Identifier of the scan job to verify.", + ), + baseline_job_id: str | None = Form( + None, + description=( + "Optional baseline scan job ID used for comparison. " + "If omitted, the current job is used as the baseline." + ), + ), +): + """ + Verify that applied fixes resolve previously detected security issues. + + The repository is rescanned and compared with a baseline scan to + determine whether fixes were successful and whether any new + vulnerabilities were introduced. + """ + try: + job_dir = safe_job_dir(WORK_ROOT, job_id) + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) + repo_dir = job_dir / "repo" if not repo_dir.exists(): raise HTTPException(status_code=404, detail="Unknown job_id") @@ -566,9 +922,18 @@ async def verify(job_id: str = Form(...)): result = verify_repo(repo_dir) - baseline_findings = await get_baseline_findings(job_id) + if baseline_job_id is not None: + try: + safe_job_dir(WORK_ROOT, baseline_job_id) + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) + + baseline_job_id = baseline_job_id or job_id + baseline_findings = await get_baseline_findings(baseline_job_id) - _, _, _, _, findings = _scan_repo_dir(repo_dir) + _, _, _, _, findings = _scan_repo_dir( + repo_dir, job_dir=job_dir, raw_dir_name="raw_verify" + ) current_findings = {finding_key(f) for f in findings} @@ -582,6 +947,19 @@ async def verify(job_id: str = Form(...)): ) passed = 1 if result.ok and new_issues_introduced == 0 else 0 + + verify_dir = job_dir / "raw_verify" + verify_dir.mkdir(parents=True, exist_ok=True) + verify_report = { + "verified_at": datetime.now(timezone.utc).isoformat(), + "passed": bool(passed), + "new_issues_introduced": new_issues_introduced, + "baseline_job_id": baseline_job_id, + } + (verify_dir / "verification-report.json").write_text( + json.dumps(verify_report, indent=2), encoding="utf-8" + ) + try: db = await get_db() try: @@ -606,20 +984,70 @@ async def verify(job_id: str = Form(...)): return result -@app.post("/evidence-pack") -def evidence_pack(job_id: str = Form(...), project_name: str = Form("project")): - job_dir = WORK_ROOT / job_id +@app.post( + "/evidence-pack", + summary="Generate an evidence package", + description=( + "Generates a downloadable evidence package containing scan " + "artifacts, reports, and supporting files for a completed " + "security scan. Optionally refreshes the raw scan outputs " + "before creating the evidence package." + ), + responses={ + 200: {"description": "Evidence package generated successfully."}, + 404: {"description": "Specified scan job was not found."}, + 422: {"description": "Invalid request parameters."}, + 500: {"description": "Internal server error."}, + }, +) +def evidence_pack( + job_id: str = Form( + ..., + description="Identifier of the scan job for which the evidence package should be generated.", + ), + project_name: str = Form( + "project", + description="Display name of the project included in the generated reports.", + ), + update_raw: bool = Form( + False, + description="If true, refreshes the raw scan results before generating the evidence package.", + ), +): + """ + Generate a downloadable evidence package for a completed scan. + + The evidence package includes reports, raw scanner outputs, and + supporting artifacts that can be used for auditing, compliance, + or sharing scan results. + """ + try: + job_dir = safe_job_dir(WORK_ROOT, job_id) + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) + repo_dir = job_dir / "repo" if not repo_dir.exists(): raise HTTPException(status_code=404, detail="Unknown job_id") repo_dir = _maybe_use_single_top_folder(repo_dir) + if update_raw: + verify_dir = job_dir / "raw_verify" + if verify_dir.exists(): + _scan_repo_dir(repo_dir, job_dir=job_dir, raw_dir_name="raw_verify") + else: + _scan_repo_dir(repo_dir, job_dir=job_dir, raw_dir_name="raw") + out_dir = job_dir / "out" ensure_dir(out_dir) pack_path = build_evidence_pack( - repo_dir=repo_dir, out_dir=out_dir, project_name=project_name, job_id=job_id + repo_dir=repo_dir, + out_dir=out_dir, + project_name=project_name, + job_id=job_id, + job_dir=job_dir, ) return FileResponse( path=str(pack_path), filename=pack_path.name, media_type="application/zip" @@ -630,49 +1058,31 @@ def evidence_pack(job_id: str = Form(...), project_name: str = Form("project")): async def download_audit_pdf(job_id: str): db = await get_db() try: - cur = await db.execute( - "SELECT project_name FROM jobs WHERE job_id = ?", (job_id,) - ) - job_row = await cur.fetchone() - + 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[0] - - cur = await db.execute( - """ - SELECT id, rule_id, severity, category, file_path, line_number, message - FROM findings - WHERE job_id = ? - """, - (job_id,), - ) - columns = [col[0] for col in cur.description] - rows = await cur.fetchall() + 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: - row_dict = dict(zip(columns, row)) - loc = None - if row_dict["file_path"]: - loc = Location( - path=row_dict["file_path"], start_line=row_dict["line_number"] - ) + if row["file_path"]: + loc = Location(path=row["file_path"], start_line=row["line_number"]) findings_list.append( Finding( - id=row_dict["id"], - title=row_dict["rule_id"] or "Unknown", - severity=row_dict["severity"] or "INFO", - category=row_dict["category"] or "Unknown", + 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_dict["message"] or "", + description=row.get("message") or "", ) ) @@ -699,40 +1109,102 @@ async def download_audit_pdf(job_id: str): async def get_findings(job_id: str): db = await get_db() try: - cur = await db.execute("SELECT job_id FROM jobs WHERE job_id = ?", (job_id,)) - job_row = await cur.fetchone() - + 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}'" ) - cur = await db.execute( + raw_finding_count = job_row.get("raw_finding_count") + finding_count = job_row.get("finding_count") + + findings = await get_findings_by_job_id(db, job_id) + finally: + await db.close() + + if raw_finding_count is None: + raw_finding_count = len(findings) + if finding_count is None: + finding_count = len(findings) + + return { + "job_id": job_id, + "raw_finding_count": raw_finding_count, + "finding_count": finding_count, + "findings": findings, + } + + +class LabelFindingRequest(BaseModel): + false_positive: bool + expected_version: int + + +@app.post("/findings/{finding_id}/label") +async def label_finding(finding_id: str, payload: LabelFindingRequest): + fp_int = 1 if payload.false_positive else 0 + + db = await get_db() + try: + cursor = await db.execute( """ - SELECT id, rule_id, severity, category, file_path, - line_number, cwe, scanner, message, package_name, package_version, created_at - FROM findings - WHERE job_id = ? - ORDER BY created_at + UPDATE findings + SET false_positive = ?, labeled_at = datetime('now'), version = version + 1 + WHERE id = ? AND version = ? """, - (job_id,), + (fp_int, finding_id, payload.expected_version), ) - columns = [col[0] for col in cur.description] - rows = await cur.fetchall() + + if cursor.rowcount == 0: + # Distinguish between a missing finding (404) and a stale version (409) + cur2 = await db.execute( + "SELECT id FROM findings WHERE id = ?", (finding_id,) + ) + if not await cur2.fetchone(): + raise HTTPException(status_code=404, detail="Finding not found") + raise HTTPException( + status_code=409, + detail="Finding has been modified by another user.", + ) + + await db.commit() + finally: + await db.close() + + return { + "status": "success", + "finding_id": finding_id, + "false_positive": payload.false_positive, + } + + +@app.patch("/findings/{finding_id}/status") +async def update_finding_status_endpoint(finding_id: str, payload: FindingStatusUpdate): + if payload.status not in ("open", "accepted", "ignored"): + raise HTTPException( + status_code=400, + detail="Invalid status value. Must be 'open', 'accepted', or 'ignored'.", + ) + + db = await get_db() + try: + finding = await get_finding(db, finding_id) + if not finding: + raise HTTPException( + status_code=404, detail=f"Finding '{finding_id}' not found." + ) + await update_finding_status(db, finding_id, payload.status) finally: await db.close() - findings = [dict(zip(columns, row)) for row in rows] - return {"job_id": job_id, "finding_count": len(findings), "findings": findings} + return {"id": finding_id, "status": payload.status} @app.get("/jobs/{job_id}/verify") async def get_verify(job_id: str): db = await get_db() try: - cur = await db.execute("SELECT job_id FROM jobs WHERE job_id = ?", (job_id,)) - job_row = await cur.fetchone() - + 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}'" @@ -762,10 +1234,23 @@ async def get_verify(job_id: str): @app.delete("/jobs/{job_id}") -def delete_job(job_id: str): - job_dir = WORK_ROOT / job_id - if job_dir.exists(): - safe_rmtree(job_dir) +async def delete_job_endpoint(job_id: str): + try: + job_dir = safe_job_dir(WORK_ROOT, job_id) + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) + + if not job_dir.exists(): + raise HTTPException(status_code=404, detail="Job not found") + + safe_rmtree(job_dir) + + db = await get_db() + try: + await delete_job(db, job_id) + finally: + await db.close() + return {"deleted": True} @@ -790,7 +1275,12 @@ async def dependency_diff_endpoint(): class LeaderboardUpdateRequest(BaseModel): - github_username: str + github_username: str = Field( + ..., + max_length=39, + pattern=r"^[a-zA-Z0-9](?:-?[a-zA-Z0-9])*$", + description="GitHub username must be max 39 chars, alphanumeric or single hyphens.", + ) pr_description: str = "" fixes_passed: int = 0 is_pr_merged: bool = False @@ -834,13 +1324,19 @@ async def fetch_org_repos(org_name: str) -> List[dict]: token = os.environ.get("GITHUB_PAT") if token: headers["Authorization"] = f"token {token}" + timeout = httpx.Timeout(30.0, connect=10.0) - async with httpx.AsyncClient() as client: - resp = await client.get(url, headers=headers, follow_redirects=True) - if resp.status_code != 200: - raise HTTPException(status_code=400, detail="Failed to fetch org repos") - repos = resp.json() - return [r for r in repos if not r.get("archived")][:20] + try: + async with httpx.AsyncClient(timeout=timeout) as client: + resp = await client.get(url, headers=headers, follow_redirects=True) + if resp.status_code != 200: + raise HTTPException(status_code=400, detail="Failed to fetch org repos") + repos = resp.json() + return [r for r in repos if not r.get("archived")][:20] + except httpx.RequestError as e: + raise HTTPException( + status_code=504, detail=f"GitHub API request failed or timed out: {e}" + ) async def _run_repo_scan_task( @@ -850,9 +1346,12 @@ async def _run_repo_scan_task( ref: str, project_name: str, org_job_id: str, + cancel_event: threading.Event = None, ): async with sem: try: + if cancel_event and cancel_event.is_set(): + raise asyncio.CancelledError() db = await get_db() try: db.row_factory = aiosqlite.Row @@ -862,31 +1361,57 @@ async def _run_repo_scan_task( org_row = await cur.fetchone() if org_row and org_row["status"] == "aborted": - await db.execute( - "UPDATE jobs SET status = 'aborted' WHERE job_id = ?", (job_id,) - ) - await db.commit() + await update_job_status(db, job_id, "aborted") return - await db.execute( - "UPDATE jobs SET status = 'scanning' WHERE job_id = ?", (job_id,) - ) - await db.commit() + await update_job_status(db, job_id, "scanning") finally: await db.close() - job_dir = WORK_ROOT / job_id + job_dir = safe_job_dir(WORK_ROOT, job_id) ensure_dir(job_dir) archive_path = job_dir / "repo.zip" repo_dir = job_dir / "repo" ensure_dir(repo_dir) zip_url = github_zip_url(repo_url, ref=ref) - await download_to_path(zip_url, archive_path) + await download_to_path(zip_url, archive_path, cancel_event=cancel_event) + if cancel_event and cancel_event.is_set(): + raise asyncio.CancelledError() + unzip_to_dir(archive_path, repo_dir) + if cancel_event and cancel_event.is_set(): + raise asyncio.CancelledError() scan_root = _maybe_use_single_top_folder(repo_dir) - semgrep, osv, gitleaks, entropy, findings = _scan_repo_dir(scan_root) + semgrep, osv, gitleaks, entropy, findings = await run_in_threadpool( + functools.partial( + _scan_repo_dir, + scan_root, + job_dir=job_dir, + cancel_event=cancel_event, + ) + ) + + raw_finding_count = len(findings) + + disable_dedup = os.environ.get("DISABLE_DEDUP", "").lower() == "true" + try: + epsilon = float(os.environ.get("DEDUP_EPSILON", 0.15)) + except ValueError: + epsilon = 0.15 + + # CORRECT WAY: + if not disable_dedup and SENTENCE_TRANSFORMERS_AVAILABLE: + # The assertion must be safely inside the block body + assert all(isinstance(f, Finding) for f in findings), \ + f"Expected Finding objects, got {set(type(f).__name__ for f in findings)}" + + findings = deduplicate(findings, epsilon) + + await _apply_fp_predictor(findings) + + finding_count = len(findings) deps = _extract_dependencies(scan_root) @@ -914,6 +1439,7 @@ async def _run_repo_scan_task( str(uuid.uuid4()), job_id, rule_id, + f.title, f.severity, f.category, file_path, @@ -923,14 +1449,13 @@ async def _run_repo_scan_task( message, pkg_name, pkg_version, + f.ml_score, + json.dumps(f.features) if f.features else None, ) ) if rows: - await db.executemany( - "INSERT INTO findings (id, job_id, rule_id, severity, category, file_path, line_number, cwe, scanner, message, package_name, package_version) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", - rows, - ) + await create_findings(db, rows) dep_rows = [] for pkg_n, pkg_v in deps: @@ -944,27 +1469,37 @@ async def _run_repo_scan_task( dep_rows, ) - await db.execute( - "UPDATE jobs SET status = 'completed' WHERE job_id = ?", (job_id,) + await update_job_status( + db, job_id, "completed", raw_finding_count, finding_count ) - await db.commit() finally: await db.close() + except asyncio.CancelledError: + logger.info("Scan task %s was aborted", job_id) + # No database write here except Exception: logger.exception("Failed repo scan task for job %s", job_id) db = await get_db() try: - await db.execute( - "UPDATE jobs SET status = 'failed' WHERE job_id = ?", (job_id,) - ) - await db.commit() + await update_job_status(db, job_id, "failed") finally: await db.close() async def _run_org_batch(org_job_id: str, repos: List[dict]): + cancel_event = threading.Event() + ORG_CANCEL_EVENTS[org_job_id] = cancel_event + db = await get_db() try: + cur = await db.execute( + "SELECT status FROM org_jobs WHERE id = ?", (org_job_id,) + ) + row = await cur.fetchone() + if row and row[0] == "aborted": + ORG_CANCEL_EVENTS.pop(org_job_id, None) + return + await db.execute( "UPDATE org_jobs SET status = 'scanning' WHERE id = ?", (org_job_id,) ) @@ -983,19 +1518,48 @@ async def _run_org_batch(org_job_id: str, repos: List[dict]): db = await get_db() try: - await db.execute( - "INSERT INTO jobs (job_id, project_name, scan_method, org_job_id, status) VALUES (?, ?, ?, ?, ?)", - (job_id, project_name, "org_batch", org_job_id, "pending"), + await create_job( + db, job_id, project_name, "org_batch", org_job_id, "pending" ) - await db.commit() finally: await db.close() tasks.append( - _run_repo_scan_task(sem, job_id, repo_url, ref, project_name, org_job_id) + asyncio.create_task( + _run_repo_scan_task( + sem, job_id, repo_url, ref, project_name, org_job_id, cancel_event + ) + ) ) - await asyncio.gather(*tasks, return_exceptions=True) + cancel_task = asyncio.create_task(asyncio.to_thread(cancel_event.wait)) + wait_tasks = asyncio.create_task(asyncio.gather(*tasks, return_exceptions=True)) + + try: + done, pending = await asyncio.wait( + [wait_tasks, cancel_task], return_when=asyncio.FIRST_COMPLETED + ) + + if cancel_task in done: + for t in tasks: + if not t.done(): + t.cancel() + await asyncio.gather(*tasks, return_exceptions=True) + + db = await get_db() + try: + await db.execute( + "UPDATE jobs SET status = 'aborted' WHERE org_job_id = ? AND status NOT IN ('completed', 'failed')", + (org_job_id,), + ) + await db.commit() + finally: + await db.close() + else: + cancel_task.cancel() + finally: + cancel_task.cancel() + ORG_CANCEL_EVENTS.pop(org_job_id, None) db = await get_db() try: @@ -1079,6 +1643,9 @@ async def get_org_status(org_job_id: str): @app.post("/api/scans/org/{org_job_id}/abort") async def abort_org_scan(org_job_id: str, mode: str = Query("pending")): + if org_job_id in ORG_CANCEL_EVENTS: + ORG_CANCEL_EVENTS[org_job_id].set() + for attempt in range(5): try: db = await get_db() @@ -1150,14 +1717,9 @@ async def event_generator(): } yield f"data: {json.dumps(payload)}\n\n" - if org_row["status"] in ["completed", "failed"]: + if org_row["status"] in ["completed", "failed", "aborted"]: break - if org_row["status"] == "aborted": - is_scanning = any(r["status"] == "scanning" for r in job_rows) - if not is_scanning: - break - await asyncio.sleep(1.5) return StreamingResponse(event_generator(), media_type="text/event-stream") @@ -1237,7 +1799,8 @@ async def get_org_findings(org_job_id: str): f.severity, f.file_path, f.line_number, - f.cwe + f.cwe, + f.ml_score FROM findings f JOIN jobs j ON f.job_id = j.job_id WHERE j.org_job_id = ? diff --git a/backend/app/ml/deduplicator.py b/backend/app/ml/deduplicator.py new file mode 100644 index 0000000..eed1d63 --- /dev/null +++ b/backend/app/ml/deduplicator.py @@ -0,0 +1,75 @@ +from collections import defaultdict +from typing import Union, List, Dict, Any +from sklearn.cluster import DBSCAN +from app.models import Finding + +def get_model(): + return None + + +# Import embedder in a way that keeps the module importable during tests. +# Tests may patch SENTENCE_TRANSFORMERS_AVAILABLE and/or embed functions. +try: + from app.ml.embedder import embed_findings as _embed_findings +except Exception: # pragma: no cover + _embed_findings = None + +SENTENCE_TRANSFORMERS_AVAILABLE = _embed_findings is not None + + +def embed_findings(findings): + """Wrapper so tests can patch deduplicator.embed_findings if needed.""" + if _embed_findings is None: + raise RuntimeError( + "sentence-transformers is not installed. " + "Install it using: pip install sentence-transformers" + ) + return _embed_findings(findings) + + +def deduplicate( + findings: List[Union[Finding, Dict[str, Any]]], + epsilon: float = 0.15, +) -> List[Union[Finding, Dict[str, Any]]]: + """ + Group similar findings using DBSCAN and return + representative findings with duplicate metadata. + """ + if not findings: + return [] + + if not SENTENCE_TRANSFORMERS_AVAILABLE: + return findings + + embeddings = embed_findings(findings) + + clustering = DBSCAN( + eps=epsilon, + min_samples=2, + metric="cosine", + ) + + labels = clustering.fit_predict(embeddings) + + results = [] + clusters = defaultdict(list) + + for idx, label in enumerate(labels): + clusters[label].append(findings[idx]) + + for label, cluster_findings in clusters.items(): + if label == -1: + for finding in cluster_findings: + # finding["duplicate_count"] = 0 + # finding["related_files"] = [] + results.append(finding) + continue + + representative = cluster_findings[0] + + # representative["duplicate_count"] = len(cluster_findings) + # representative["related_files"] = related_files + + results.append(representative) + + return results \ No newline at end of file diff --git a/backend/app/ml/embedder.py b/backend/app/ml/embedder.py new file mode 100644 index 0000000..70a071e --- /dev/null +++ b/backend/app/ml/embedder.py @@ -0,0 +1,41 @@ +import logging +import numpy as np + +try: + from sentence_transformers import SentenceTransformer + + MODEL = SentenceTransformer("all-MiniLM-L6-v2") +except Exception: + MODEL = None + logging.getLogger(__name__).warning( + "Failed to load sentence-transformers model: all-MiniLM-L6-v2", + exc_info=True + ) + + +def _extract_text(finding) -> str: + """Safely extracts title and description from either a Pydantic Finding object or a raw dict.""" + if isinstance(finding, dict): + return f"{finding.get('title', '')} {finding.get('description', '')}".strip() + return f"{getattr(finding, 'title', '')} {getattr(finding, 'description', '')}".strip() + + +def embed_findings(findings: list) -> np.ndarray: + """ + Convert findings into embeddings. + + Each finding is converted to: + "{title} {description}" + + Returns: + np.ndarray of shape (n, 384) + """ + if MODEL is None: + raise RuntimeError( + "sentence-transformers is not installed or failed to initialize. " + "Install it using: pip install sentence-transformers" + ) + + texts = [_extract_text(finding) for finding in findings] + + return MODEL.encode(texts, convert_to_numpy=True)