End-to-end example

import os
import sys
import time
import uuid
import mimetypes
import pathlib

import requests

API = "https://kalligator.com/api"
FINAL = {"accepted", "rejected", "duplicate", "insufficient_info", "withdrawn"}

session = requests.Session()
session.headers["Authorization"] = f"Bearer {os.environ['KALLIGATOR_API_KEY']}"


class KalligatorError(Exception):
    def __init__(self, status, code, detail):
        super().__init__(f"{status} {code}: {detail}")
        self.code = code


def call(method, path, **kwargs):
    """Send a request. Wait and retry when the API sends Retry-After."""
    while True:
        response = session.request(method, f"{API}{path}", timeout=150, **kwargs)
        retry_after = response.headers.get("Retry-After")
        if response.status_code in (429, 503) and retry_after:
            time.sleep(int(retry_after))
            continue
        if response.status_code >= 400:
            error = response.json()
            raise KalligatorError(response.status_code, error["code"], error["detail"])
        return response.json()


def check_account():
    me = call("GET", "/me")
    if not me["email_verified"]:
        sys.exit("Verify your email on https://kalligator.com/account first.")
    stripe = call("GET", "/stripe/status")
    if not stripe["ready"]:
        sys.exit(f"Set up payouts on https://kalligator.com/account. Stripe: {stripe['detail']}")
    if me["active_reports"] >= me["active_limit"]:
        sys.exit("All active-report slots are in use.")


def read_program(program_id):
    program = call("GET", f"/programs/{program_id}")
    if program["intake"] != "open" or program["pool_paused"]:
        sys.exit(f"{program['name']} does not accept new reports now.")
    print(f"Scope of {program['name']}:\n{program['scope']}\n")
    return program


def write_draft(report_id, program_id, description, revision):
    report = call("PUT", f"/reports/{report_id}", json={
        "program_id": program_id,
        "title": "Stored XSS in project name on the dashboard",
        "asset": "https://app.example.com/dashboard",
        "description": description,
        "cvss_vector": "CVSS:3.1/AV:N/AC:L/PR:L/UI:R/S:C/C:L/I:L/A:N",
        "revision": revision,
    })
    return report["revision"]


def upload(report_id, path, revision):
    file_id = str(uuid.uuid4())
    content_type = mimetypes.guess_type(path.name)[0] or "application/octet-stream"
    report = call(
        "PUT",
        f"/reports/{report_id}/files/{file_id}",
        params={"name": path.name, "revision": revision},
        data=path.read_bytes(),
        headers={"Content-Type": content_type},
    )
    assert report["attachments"][file_id]["state"] == "ready"
    return file_id, report["revision"]


def follow_triage(report_id):
    while True:
        report = call("GET", f"/reports/{report_id}")
        status = report["status"]
        if status in FINAL:
            decision = report["decision"] or {}
            print(f"Final status: {report['display_status']}. {decision.get('message') or ''}")
            return
        if status == "needs_info":
            messages = call("GET", f"/reports/{report_id}/messages")["messages"]
            question = [m for m in messages if m["author"] == "agent"][-1]
            print(f"Triage agent asks:\n{question['body']}\n")
            answer = input("Your answer: ")
            call("POST", f"/reports/{report_id}/messages", json={"body": answer})
        time.sleep(60)


def main(program_id, evidence):
    check_account()
    read_program(program_id)

    # Keep the report ID. Use the same ID if you run a step again.
    report_id = str(uuid.uuid4())
    print(f"Report ID: {report_id}")

    description = (
        "## Steps to reproduce\n\n"
        "1. Sign in as the program test user.\n"
        "2. Set the project name to `<img src=x onerror=alert(document.domain)>`.\n"
        "3. Open the dashboard.\n\n"
        "## Expected behavior\n\nThe dashboard shows the project name as text.\n\n"
        "## Actual behavior\n\nThe browser runs the script in the project name.\n\n"
        "## Security impact\n\n"
        "A project member can run script in the session of each user who opens the dashboard.\n"
    )
    revision = write_draft(report_id, program_id, description, 0)

    file_id, revision = upload(report_id, pathlib.Path(evidence), revision)
    description += f"\nRequest log: [{evidence}](/reports/{report_id}/files/{file_id})\n"
    revision = write_draft(report_id, program_id, description, revision)

    print(description)
    if input("Submit this report? [y/N] ").strip().lower() != "y":
        sys.exit(f"Draft {report_id} is saved. It is not submitted.")
    call("POST", f"/reports/{report_id}/submit", json={"revision": revision})
    print("Submitted. Waiting for triage.")

    follow_triage(report_id)


if __name__ == "__main__":
    main(sys.argv[1], sys.argv[2])