|
| 1 | +"""Load reviewed false-positive feedback for a security scan.""" |
| 2 | + |
| 3 | +from __future__ import annotations |
| 4 | + |
| 5 | +import argparse |
| 6 | +import sqlite3 |
| 7 | +import sys |
| 8 | +from pathlib import Path |
| 9 | +from typing import Any |
| 10 | + |
| 11 | +sys.path.insert(0, str(Path(__file__).resolve().parent)) |
| 12 | +from workbench_constants import ( |
| 13 | + FINDING_LOCATION_PATH_BYTES, |
| 14 | + FINDING_SUMMARY_BYTES, |
| 15 | + FINDING_TITLE_BYTES, |
| 16 | +) |
| 17 | +from workbench_validation import bounded_output_text |
| 18 | + |
| 19 | + |
| 20 | +def get_scan_feedback(connection: sqlite3.Connection, scan: sqlite3.Row) -> dict[str, Any]: |
| 21 | + rows = connection.execute( |
| 22 | + """ |
| 23 | + WITH ranked_decisions AS ( |
| 24 | + SELECT findings.id AS finding_id, findings.fingerprint, findings.rule_id, |
| 25 | + findings.identity_anchor, findings.identity_instance, occurrences.title, |
| 26 | + occurrences.summary, COALESCE(triage.status, 'open') AS triage_status, |
| 27 | + triage.close_reason, triage.note, |
| 28 | + COALESCE(triage.updated_at, source_scans.completed_at) AS updated_at, |
| 29 | + source_scans.id AS source_scan_id, |
| 30 | + source_scans.completed_at AS source_completed_at, |
| 31 | + locations.relative_path, locations.start_line, locations.end_line, locations.role, |
| 32 | + ROW_NUMBER() OVER ( |
| 33 | + PARTITION BY findings.id |
| 34 | + ORDER BY COALESCE(triage.updated_at, source_scans.completed_at) DESC, |
| 35 | + source_scans.completed_at DESC, |
| 36 | + source_scans.id DESC, occurrences.id DESC |
| 37 | + ) AS decision_rank |
| 38 | + FROM finding_occurrences AS occurrences |
| 39 | + JOIN findings ON findings.id = occurrences.finding_id |
| 40 | + JOIN scans AS source_scans ON source_scans.id = occurrences.scan_id |
| 41 | + LEFT JOIN finding_triage AS triage ON triage.occurrence_id = occurrences.id |
| 42 | + JOIN finding_locations AS locations ON locations.id = ( |
| 43 | + SELECT candidate.id |
| 44 | + FROM finding_locations AS candidate |
| 45 | + WHERE candidate.occurrence_id = occurrences.id |
| 46 | + ORDER BY CASE WHEN candidate.role = 'root_control' THEN 0 ELSE 1 END, |
| 47 | + candidate.sort_order |
| 48 | + LIMIT 1 |
| 49 | + ) |
| 50 | + WHERE source_scans.target_id = ? |
| 51 | + AND source_scans.id != ? |
| 52 | + AND source_scans.status = 'complete' |
| 53 | + ) |
| 54 | + SELECT * |
| 55 | + FROM ranked_decisions |
| 56 | + WHERE decision_rank = 1 |
| 57 | + AND triage_status = 'closed' |
| 58 | + AND close_reason = 'false_positive' |
| 59 | + AND note IS NOT NULL |
| 60 | + AND trim(note) != '' |
| 61 | + ORDER BY updated_at DESC, source_completed_at DESC, source_scan_id DESC, finding_id DESC |
| 62 | + LIMIT 50 |
| 63 | + """, |
| 64 | + (scan["target_id"], scan["id"]), |
| 65 | + ) |
| 66 | + false_positives = [] |
| 67 | + for row in rows: |
| 68 | + identity = {"anchor": row["identity_anchor"]} |
| 69 | + if row["identity_instance"] is not None: |
| 70 | + identity["instance"] = row["identity_instance"] |
| 71 | + location = { |
| 72 | + "path": bounded_output_text(row["relative_path"], FINDING_LOCATION_PATH_BYTES), |
| 73 | + "startLine": row["start_line"], |
| 74 | + "endLine": row["end_line"], |
| 75 | + } |
| 76 | + if row["role"] is not None: |
| 77 | + location["role"] = row["role"] |
| 78 | + false_positives.append( |
| 79 | + { |
| 80 | + "findingId": row["finding_id"], |
| 81 | + "fingerprint": row["fingerprint"], |
| 82 | + "ruleId": row["rule_id"], |
| 83 | + "identity": identity, |
| 84 | + "title": bounded_output_text(row["title"], FINDING_TITLE_BYTES), |
| 85 | + "summary": bounded_output_text(row["summary"], FINDING_SUMMARY_BYTES), |
| 86 | + "reason": row["note"], |
| 87 | + "locations": [location], |
| 88 | + "sourceScanId": row["source_scan_id"], |
| 89 | + "updatedAt": row["updated_at"], |
| 90 | + } |
| 91 | + ) |
| 92 | + return {"scanId": scan["id"], "targetId": scan["target_id"], "falsePositives": false_positives} |
| 93 | + |
| 94 | + |
| 95 | +if __name__ == "__main__": |
| 96 | + argparse.ArgumentParser(description=__doc__).parse_args() |
0 commit comments