Function set_finding_triage

plugins/codex-security/scripts/workbench_db.py:1998–2053

(connection: sqlite3.Connection, args: argparse.Namespace)

def set_finding_triage(connection: sqlite3.Connection, args: argparse.Namespace) -> dict[str, Any]:
    close_reason = args.close_reason
    if args.status == "open" and close_reason is not None:
        raise SystemExit("An open finding cannot keep a close reason.")
    if args.status == "closed" and close_reason is None:
        raise SystemExit("Choose why this finding is being closed.")
    note = optional_text(args.note, maximum=2400)
    if close_reason == "wont_fix" and note is None:
        raise SystemExit("Explain why this finding will not be fixed.")
    connection.execute("BEGIN IMMEDIATE")
    try:
        timestamp = now()
        occurrence = require_occurrence(connection, args.occurrence_id)
        if args.status == "closed":
            remediation = connection.execute(
                """
                SELECT *
                FROM finding_remediation_attempts
                WHERE occurrence_id = ?
                ORDER BY created_at DESC, rowid DESC
                LIMIT 1
                """,
                (occurrence["id"],),
            ).fetchone()
            if remediation is not None and remediation["pending_action"] is not None:
                raise SystemExit(
                    "Wait for the pending remediation operation to finish before closing this finding."
                )
            if (
                close_reason == "already_fixed"
                and remediation is not None
                and remediation["state"] == "verified"
            ):
                scan = require_scan(connection, occurrence["scan_id"])
                require_remediation_checkout_unchanged(
                    scan,
                    remediation,
                    require_applied_content=True,
                )
        connection.execute(
            """
            INSERT INTO finding_triage (occurrence_id, status, close_reason, note, updated_at)
            VALUES (?, ?, ?, ?, ?)
            ON CONFLICT(occurrence_id) DO UPDATE SET
                status = excluded.status,
                close_reason = excluded.close_reason,
                note = excluded.note,
                updated_at = excluded.updated_at
            """,
            (occurrence["id"], args.status, close_reason, note, timestamp),
        )
        connection.commit()
    except BaseException:
        connection.rollback()
        raise
    return scan_context(connection, occurrence["scan_id"])