Benjamin Admin
9e9d780902
Liest den Lebenszyklus jedes Befunds (status + tracker_issue_url) aus dem Scanner zurück und rollt ihn zu einem Management-Bild auf: % erledigt, 4-Phasen (offen/in Arbeit/erledigt/ausgeschlossen), offenes Restrisiko nach Schweregrad, Fortschritt je CRA-Anforderung und eine Aufgaben-/Ticket-Tabelle mit Jira-Link. Neuer Endpoint GET/POST /api/v1/cra/progress (dünn → Service cra_progress, rein deterministisch, kein /assess-Schema-Drift). Frontend: ProgressView in Ebene 1 (CRACyberView), live je Scanner-Repo, sonst Demo-Status. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
40 lines
1.5 KiB
Python
40 lines
1.5 KiB
Python
"""CRA project-progress endpoint (management view).
|
|
|
|
Reads each finding's lifecycle back from the scanner (status + tracker ticket)
|
|
and rolls it up into a completion picture: % done, what's left by risk, and
|
|
per-CRA-requirement coverage. Pull-flow (GET ?repo_id=) reads live from the
|
|
scanner's MCP; POST takes findings in the body (demo / direct).
|
|
"""
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
from fastapi import APIRouter
|
|
from pydantic import BaseModel
|
|
|
|
from compliance.services.cra_progress import build_progress
|
|
from compliance.services.scanner_mcp_client import fetch_findings
|
|
|
|
router = APIRouter(prefix="/v1/cra", tags=["cra"])
|
|
|
|
|
|
class ProgressRequest(BaseModel):
|
|
# Raw finding dicts (scanner shape: status, tracker_issue_url, cwe, severity …).
|
|
findings: List[Dict[str, Any]] = []
|
|
|
|
|
|
@router.get("/progress")
|
|
async def progress(repo_id: Optional[str] = None, severity: Optional[str] = None):
|
|
"""Pull-flow: fetch the repo's findings from the scanner and roll up progress.
|
|
Returns an empty rollup if no scanner is configured."""
|
|
findings = await fetch_findings(repo_id=repo_id, severity=severity, limit=500)
|
|
result = build_progress(findings)
|
|
result["source"] = {"scanner": True, "pulled": len(findings), "repo_id": repo_id}
|
|
return result
|
|
|
|
|
|
@router.post("/progress")
|
|
async def progress_from_body(body: ProgressRequest):
|
|
"""Roll up progress for findings supplied directly (demo / offline)."""
|
|
result = build_progress(body.findings)
|
|
result["source"] = {"scanner": False, "pulled": len(body.findings)}
|
|
return result
|