Function upsert_finding

core/db.py:186–217

Insert or update a finding. Returns the finding row id.

(run_id: str, finding: dict)

# ── Finding CRUD ──────────────────────────────────────────────────────────────

def upsert_finding(run_id: str, finding: dict) -> int:
    """Insert or update a finding. Returns the finding row id."""
    sql = """
        INSERT INTO findings
            (run_id, ip, port, service, version, cve_id, cvss_score,
             exploit_status, attack_tags, already_exploited, hunt_evidence, created_at)
        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
        ON CONFLICT(run_id, ip, port) DO UPDATE SET
            cve_id=excluded.cve_id,
            cvss_score=excluded.cvss_score,
            exploit_status=excluded.exploit_status,
            attack_tags=excluded.attack_tags,
            already_exploited=excluded.already_exploited,
            hunt_evidence=excluded.hunt_evidence
        RETURNING id
    """
    with get_db() as conn:
        row = conn.execute(sql, [
            run_id,
            finding.get("ip", ""),
            finding.get("port", 0),
            finding.get("service", ""),
            finding.get("version", ""),
            finding.get("cve_id", ""),
            finding.get("cvss_score", 0.0),
            finding.get("exploit_status", "NOT_CHECKED"),
            json.dumps(finding.get("attack_tags", [])),
            int(finding.get("already_exploited", False)),
            json.dumps(finding.get("hunt_evidence", [])),
            datetime.now(timezone.utc).isoformat(),
        ]).fetchone()
        return row[0] if row else -1


def list_findings(run_id: str) -> list[dict]: