Let findings be dismissed as false positives or accepted risk

AGENTS.md

@@ -716,6 +716,19 @@ Get specific analysis by ID.
 ### DELETE `/api/analyses/{id}`
 Delete analysis by ID.
 
+### POST `/api/analyses/{id}/dismissals`
+Dismiss a finding (`{"fingerprint": "...", "reason": "..."}`) for the analysis's repository or action. Returns the analysis with dismissals applied.
+
+### DELETE `/api/analyses/{id}/dismissals/{fingerprint}`
+Restore a dismissed finding. Returns the analysis with dismissals applied.
+
+### GET `/api/dismissals?target=owner/repo`
+List dismissals for a repository or action.
+
+## Dismissed Findings
+
+`backend/dismissals.py` gives every issue a `fingerprint`: a hash of its node id, type, identifying fields and message, leaving out what shifts between runs (`line_number`, `latest_version`, `days_old`, `commit_date`, digits in the message). Dismissals are stored per audit target in `data/dismissals.json` and applied when an audit is built and whenever a stored analysis is read, so one dismissal covers every run of that target. A dismissed issue carries `"dismissed": {"reason", "dismissed_at"}` and is left out of `issue_count`, node `severity` and the statistics (`dismissed_issues` counts them). If you add a volatile field to an issue, add it to `_VOLATILE_FIELDS` or dismissals of that finding won't survive a re-run.
+
 ## Configuration
 
 ### Trusted Action Publishers

backend/dismissals.py

@@ -0,0 +1,144 @@
+"""Dismissed findings: stable issue fingerprints and a per-target store.
+
+A dismissal is recorded against an audit target (``owner/repo`` or an action
+reference) and an issue fingerprint, so it carries over to later audits of the
+same target. The fingerprint identifies a finding by the node it sits on, its
+type, its identifying fields and its message, and leaves out what shifts
+between runs without changing the finding: line numbers, the latest available
+version, commit ages, and numbers in the message.
+"""
+import datetime
+import hashlib
+import json
+import re
+import threading
+from pathlib import Path
+from typing import Any, Dict, Iterable, List, Optional
+
+# Fields that change between runs (or carry presentation only) and so must not
+# change a finding's identity.
+_VOLATILE_FIELDS = {
+    "severity", "message", "evidence", "recommendation", "line_number",
+    "latest_version", "days_old", "commit_date",
+    "versions", "version_count", "workflows", "workflow_count",
+    # Set by this module.
+    "fingerprint", "dismissed",
+}
+
+# Node types a finding is mirrored onto from the node it was raised on, so
+# the graph is navigable (see main._add_package_dependency_nodes). The last two
+# are the names analyses stored by older versions use for image nodes.
+MIRROR_NODE_TYPES = {"package", "image", "container_image", "docker_image"}
+
+_MAX_REASON_LENGTH = 500
+
+
+def _normalized_message(issue: Dict[str, Any]) -> str:
+    message = str(issue.get("message", ""))
+    for field in ("latest_version", "days_old", "commit_date", "line_number"):
+        value = issue.get(field)
+        if value not in (None, ""):
+            message = message.replace(str(value), f"<{field}>")
+    return re.sub(r"\d+", "#", message)
+
+
+def _content_key(issue: Dict[str, Any]) -> str:
+    identity = {k: v for k, v in issue.items() if k not in _VOLATILE_FIELDS}
+    return json.dumps([identity, _normalized_message(issue)], sort_keys=True, default=str)
+
+
+def fingerprint(node_id: str, issue: Dict[str, Any]) -> str:
+    """Stable identifier for a finding on a node."""
+    digest = hashlib.sha256(f"{node_id}\n{_content_key(issue)}".encode()).hexdigest()
+    return digest[:20]
+
+
+def apply_dismissals(nodes: Iterable[Dict[str, Any]], dismissed: Dict[str, Dict[str, Any]]) -> None:
+    """Tag every issue with its fingerprint and, when dismissed, the dismissal.
+
+    Mutates the issues in place. A finding mirrored onto a package or image
+    node shares the fingerprint of the finding it mirrors, so dismissing one
+    dismisses both.
+    """
+    nodes = list(nodes)
+    source_fingerprints: Dict[str, str] = {}
+    ordered = sorted(nodes, key=lambda n: n.get("type") in MIRROR_NODE_TYPES)
+    for node in ordered:
+        mirror = node.get("type") in MIRROR_NODE_TYPES
+        for issue in node.get("issues", []):
+            key = _content_key(issue)
+            fp = source_fingerprints.get(key) if mirror else None
+            if fp is None:
+                fp = fingerprint(node["id"], issue)
+            if not mirror:
+                source_fingerprints.setdefault(key, fp)
+            issue["fingerprint"] = fp
+            record = dismissed.get(fp)
+            if record:
+                issue["dismissed"] = {k: record.get(k) for k in ("reason", "dismissed_at")}
+            else:
+                issue.pop("dismissed", None)
+
+
+def is_dismissed(issue: Dict[str, Any]) -> bool:
+    return bool(issue.get("dismissed"))
+
+
+class DismissalStore:
+    """Dismissals keyed by audit target, persisted as one JSON file."""
+
+    def __init__(self, path: Optional[str] = None):
+        self.path = Path(path) if path else Path(__file__).parent.parent / "data" / "dismissals.json"
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        self._lock = threading.Lock()
+
+    def _read(self) -> Dict[str, Dict[str, Dict[str, Any]]]:
+        if not self.path.exists():
+            return {}
+        with open(self.path, "r") as f:
+            return json.load(f)
+
+    def _write(self, data: Dict[str, Dict[str, Dict[str, Any]]]) -> None:
+        tmp_path = self.path.with_suffix(".tmp")
+        try:
+            with open(tmp_path, "w") as f:
+                json.dump(data, f, indent=2)
+            tmp_path.replace(self.path)
+        except Exception:
+            tmp_path.unlink(missing_ok=True)
+            raise
+
+    def get(self, target: Optional[str]) -> Dict[str, Dict[str, Any]]:
+        """Dismissals for a target, keyed by fingerprint."""
+        if not target:
+            return {}
+        with self._lock:
+            return self._read().get(target, {})
+
+    def list(self, target: str) -> List[Dict[str, Any]]:
+        return [{"fingerprint": fp, **record} for fp, record in self.get(target).items()]
+
+    def add(self, target: str, fp: str, issue: Dict[str, Any], node_id: str, reason: str = "") -> Dict[str, Any]:
+        record = {
+            "reason": reason.strip()[:_MAX_REASON_LENGTH],
+            "dismissed_at": datetime.datetime.now(datetime.UTC).isoformat(),
+            "type": issue.get("type"),
+            "node": node_id,
+            "message": issue.get("message"),
+        }
+        with self._lock:
+            data = self._read()
+            data.setdefault(target, {})[fp] = record
+            self._write(data)
+        return record
+
+    def remove(self, target: str, fp: str) -> bool:
+        with self._lock:
+            data = self._read()
+            if fp not in data.get(target, {}):
+                return False
+            del data[target][fp]
+            if not data[target]:
+                del data[target]
+            self._write(data)
+        return True

backend/graph_builder.py

@@ -12,6 +12,19 @@ def __init__(self):
         self.issues: Dict[str, List[Dict[str, Any]]] = defaultdict(list)
         self._edge_keys: Set[Tuple[str, str]] = set()
 
+    @classmethod
+    def from_graph_data(cls, graph_data: Dict[str, Any]) -> "GraphBuilder":
+        """Rebuild a builder from get_graph_data() output, e.g. a stored analysis."""
+        graph = cls()
+        for node in graph_data.get("nodes", []):
+            graph.nodes[node["id"]] = node
+            node.setdefault("issues", [])
+            if node["issues"]:
+                graph.issues[node["id"]] = node["issues"]
+        for edge in graph_data.get("edges", []):
+            graph.add_edge(edge["source"], edge["target"], edge.get("type", "uses"))
+        return graph
+
     def add_node(self, node_id: str, label: str, node_type: str = "action", metadata: Optional[Dict] = None):
         """Add a node to the graph."""
         if node_id not in self.nodes:
@@ -116,7 +129,7 @@ def get_graph_data(self) -> Dict[str, Any]:
         """Get graph data in format suitable for visualization."""
         depths = self._compute_depths()
         for node_id, node in self.nodes.items():
-            issues = node.get("issues", [])
+            issues = [i for i in node.get("issues", []) if not i.get("dismissed")]
             node["issue_count"] = len(issues)
             node["depth"] = depths.get(node_id, 0)
             severities = {issue.get("severity", "low") for issue in issues}
@@ -129,24 +142,28 @@ def get_graph_data(self) -> Dict[str, Any]:
         }
 
     def _unique_issues(self) -> List[Dict[str, Any]]:
-        """All issues, counting an issue object once even if attached to several nodes.
+        """All issues, counting an issue once even if attached to several nodes.
 
         Findings are deliberately mirrored onto related nodes (a package-install
         finding onto the package node, an unpinned-image finding onto the image
-        node) so the graph is navigable; they are still one finding.
+        node) so the graph is navigable; they are still one finding. Mirrors
+        share one object in a fresh audit and one fingerprint once dismissals
+        are applied (a stored analysis has lost the shared objects).
         """
-        seen: Set[int] = set()
+        seen: Set[Any] = set()
         unique = []
         for issues in self.issues.values():
             for issue in issues:
-                if id(issue) not in seen:
-                    seen.add(id(issue))
+                key = issue.get("fingerprint") or id(issue)
+                if key not in seen:
+                    seen.add(key)
                     unique.append(issue)
         return unique
 
     def get_statistics(self) -> Dict[str, Any]:
         """Get statistics about the graph."""
-        unique_issues = self._unique_issues()
+        all_issues = self._unique_issues()
+        unique_issues = [i for i in all_issues if not i.get("dismissed")]
         severity_counts = defaultdict(int)
         for issue in unique_issues:
             severity_counts[issue.get("severity", "low")] += 1
@@ -157,6 +174,9 @@ def get_statistics(self) -> Dict[str, Any]:
             "total_edges": len(self.edges),
             "total_issues": len(unique_issues),
             "severity_counts": dict(severity_counts),
-            "nodes_with_issues": sum(1 for issues in self.issues.values() if issues),
+            "dismissed_issues": len(all_issues) - len(unique_issues),
+            "nodes_with_issues": sum(
+                1 for issues in self.issues.values() if any(not i.get("dismissed") for i in issues)
+            ),
             "max_depth": max(depths.values(), default=0),
         }

backend/main.py

@@ -20,6 +20,7 @@
 from graph_builder import GraphBuilder
 from repo_cloner import RepoCloner, CloneError
 from analysis_storage import AnalysisStorage
+from dismissals import DismissalStore, apply_dismissals
 import pin_client
 
 app = FastAPI(
@@ -48,6 +49,7 @@
 auditor = SecurityAuditor()
 cloner = RepoCloner()
 storage = AnalysisStorage()
+dismissals = DismissalStore()
 logger = logging.getLogger(__name__)
 
 # Serve frontend static files if they exist (for production builds)
@@ -72,6 +74,11 @@ class AuditRequest(BaseModel):
     use_clone: bool = False  # New option to use clone instead of API
 
 
+class DismissRequest(BaseModel):
+    fingerprint: str
+    reason: str = ""
+
+
 class AuditYAMLRequest(BaseModel):
     yaml_content: str
     github_token: Optional[str] = None
@@ -605,6 +612,7 @@ async def _run_audit(request: AuditRequest, log_fn: Optional[Callable[[str], Non
 
         if log_fn:
             log_fn("Building final graph...")
+        apply_dismissals(graph.nodes.values(), dismissals.get(repository or action))
         graph_data = graph.get_graph_data()
         statistics = graph.get_statistics()
 
@@ -615,7 +623,13 @@ async def _run_audit(request: AuditRequest, log_fn: Optional[Callable[[str], Non
             statistics=statistics,
             method="clone" if request.use_clone else "api"
         )
-        return {"id": analysis_id, "graph": graph_data, "statistics": statistics}
+        return {
+            "id": analysis_id,
+            "repository": repository,
+            "action": action,
+            "graph": graph_data,
+            "statistics": statistics,
+        }
     finally:
         await _close_client(client)
 
@@ -1312,6 +1326,9 @@ async def _audit_inline_workflow(request: AuditYAMLRequest, workflow: Dict[str,
 
     _add_workflow_container_image_nodes(graph, workflow, workflow_node_id, workflow_issues)
 
+    # An inline workflow has no target to remember dismissals against; this
+    # only gives its findings fingerprints.
+    apply_dismissals(graph.nodes.values(), {})
     graph_data = graph.get_graph_data()
     statistics = graph.get_statistics()
 
@@ -1359,15 +1376,63 @@ async def list_analyses(repository: Optional[str] = None, limit: int = 50):
     return storage.list_analyses(limit=limit, repository=repository)
 
 
-@app.get("/api/analyses/{analysis_id}")
-async def get_analysis(analysis_id: str):
-    """Get a specific analysis by ID."""
+def _analysis_target(analysis: Dict[str, Any]) -> Optional[str]:
+    return analysis.get("repository") or analysis.get("action")
+
+
+def _load_analysis(analysis_id: str) -> Dict[str, Any]:
+    """A stored analysis with the target's current dismissals applied.
+
+    Dismissals live apart from analyses, so one made on any run of a target
+    shows on every run of it, including ones stored before it was made.
+    """
     analysis = storage.get_analysis(analysis_id)
     if not analysis:
         raise HTTPException(status_code=404, detail="Analysis not found")
+    graph = GraphBuilder.from_graph_data(analysis.get("graph", {}))
+    apply_dismissals(graph.nodes.values(), dismissals.get(_analysis_target(analysis)))
+    analysis["graph"] = graph.get_graph_data()
+    analysis["statistics"] = graph.get_statistics()
     return analysis
 
 
+@app.get("/api/analyses/{analysis_id}")
+async def get_analysis(analysis_id: str):
+    """Get a specific analysis by ID."""
+    return _load_analysis(analysis_id)
+
+
+@app.post("/api/analyses/{analysis_id}/dismissals")
+async def dismiss_issue(analysis_id: str, request: DismissRequest):
+    """Dismiss a finding for this analysis's target; returns the updated analysis."""
+    analysis = _load_analysis(analysis_id)
+    target = _analysis_target(analysis)
+    if not target:
+        raise HTTPException(status_code=400, detail="Only repository and action audits can dismiss findings")
+    for node in analysis["graph"]["nodes"]:
+        issue = next((i for i in node.get("issues", []) if i.get("fingerprint") == request.fingerprint), None)
+        if issue:
+            dismissals.add(target, request.fingerprint, issue, node["id"], request.reason)
+            return _load_analysis(analysis_id)
+    raise HTTPException(status_code=404, detail="Finding not found in this analysis")
+
+
+@app.delete("/api/analyses/{analysis_id}/dismissals/{fingerprint}")
+async def restore_issue(analysis_id: str, fingerprint: str):
+    """Undo a dismissal; returns the updated analysis."""
+    analysis = _load_analysis(analysis_id)
+    target = _analysis_target(analysis)
+    if not target or not dismissals.remove(target, fingerprint):
+        raise HTTPException(status_code=404, detail="Dismissal not found")
+    return _load_analysis(analysis_id)
+
+
+@app.get("/api/dismissals")
+async def list_dismissals(target: str):
+    """Dismissed findings for a repository or action."""
+    return dismissals.list(target)
+
+
 @app.delete("/api/analyses/{analysis_id}")
 async def delete_analysis(analysis_id: str):
     """Delete an analysis by ID."""