Collecting case data - Triage and Investigation Agent verdicts

Here’s an example script that joins TINA with Case data and outputs a CSV, perhaps may help you get started on how to bridge these two disconnected datasets, it queries the TINA (notebooks) API method, then joins this against Case data via native dashboard queries.

import csv
import json
import os
import sys
import concurrent.futures
import google.auth
import google.auth.transport.requests
import requests

CONFIG_FILE = "secops-config.json"
CSV_FILE = "tina_investigations_export.csv"

def get_credentials():
    credentials, project = google.auth.default(
        scopes=['https://www.googleapis.com/auth/cloud-platform']
    )
    auth_req = google.auth.transport.requests.Request()
    credentials.refresh(auth_req)
    return credentials

def fetch_notebooks_list(session, headers, project_id, customer_id, region, limit=50):
    url = f"https://us-chronicle.googleapis.com/v1alpha/projects/{project_id}/locations/{region}/instances/{customer_id}/notebooks"
    params = {"orderBy": "createTime desc", "pageSize": limit}
    try:
        r = session.get(url, headers=headers, params=params, timeout=15)
        if r.status_code == 200:
            return r.json().get("notebooks", [])
    except Exception as e:
        print(f"Error fetching notebooks list: {e}", file=sys.stderr)
    return []

def fetch_notebook(session, headers, name):
    url = f"https://us-chronicle.googleapis.com/v1alpha/{name}"
    try:
        r = session.get(url, headers=headers, timeout=10)
        if r.status_code == 200:
            return r.json()
    except Exception as e:
        print(f"Error fetching notebook {name}: {e}", file=sys.stderr)
    return None

def fetch_investigation(session, headers, name):
    url = f"https://us-chronicle.googleapis.com/v1alpha/{name}"
    try:
        r = session.get(url, headers=headers, timeout=10)
        if r.status_code == 200:
            return r.json()
    except Exception as e:
        print(f"Error fetching investigation {name}: {e}", file=sys.stderr)
    return None

def scan_alert_legacy(session, headers, project_id, customer_id, region, alert_id):
    url = f"https://us-chronicle.googleapis.com/v1alpha/projects/{project_id}/locations/{region}/instances/{customer_id}/legacy:legacyGetAlert?alertId={alert_id}&includeDetections=true"
    try:
        r = session.get(url, headers=headers, timeout=15)
        if r.status_code == 200:
            resp = r.json()
            alert_obj = resp.get("alert", {})
            case_name = alert_obj.get("caseName")
            if case_name:
                return alert_id, case_name
    except Exception as e:
        pass
    return alert_id, None

def resolve_legacy_case_id(session, headers, project_id, customer_id, region, case_name):
    url = f"https://us-chronicle.googleapis.com/v1alpha/projects/{project_id}/locations/{region}/instances/{customer_id}/legacy:legacyBatchGetCases?names={case_name}"
    try:
        cr = session.get(url, headers=headers, timeout=15)
        if cr.status_code == 200:
            cases_resp = cr.json()
            cases = cases_resp.get("cases", [])
            if cases:
                case_obj = cases[0]
                soar_info = case_obj.get("soarPlatformInfo", {})
                numeric_id = soar_info.get("caseId")
                priority = case_obj.get("priority", "UNKNOWN")
                return numeric_id, priority
    except Exception as e:
        pass
    return None, "UNKNOWN"

def run_mapping_query(session, headers, project_id, customer_id, region, alert_ids):
    url = f"https://us-chronicle.googleapis.com/v1alpha/projects/{project_id}/locations/{region}/instances/{customer_id}/dashboardQueries:execute"
    id_clauses = " or ".join([f'$alert_id = "{aid}"' for aid in alert_ids])
    query_string = (
        "$case_id = case.name\n"
        "$alert_id = case.alerts.metadata.id\n"
        "$priority = case.priority\n"
        f"{id_clauses}\n"
        "match:\n"
        "  $case_id, $alert_id, $priority\n"
    )
    payload = {
        "query": {
            "query": query_string,
            "input": {
                "relativeTime": {
                    "timeUnit": "DAY",
                    "startTimeVal": "365"
                }
            }
        },
        "filters": [],
        "clearCache": True
    }
    try:
        r = session.post(url, headers=headers, json=payload, timeout=60) # Increased timeout
        if r.status_code == 200:
            return r.json()
    except Exception as e:
        print(f"Error querying dashboard mapping: {e}", file=sys.stderr)
    return None

def main():
    print("==========================================================")
    print("Standalone TINA to CSV Exporter")
    print("==========================================================")

    if not os.path.exists(CONFIG_FILE):
        print(f"Error: {CONFIG_FILE} not found.", file=sys.stderr)
        sys.exit(1)

    with open(CONFIG_FILE, "r") as f:
        config = json.load(f)
    project_id = config.get("project_id")
    customer_id = config.get("customer_id")
    region = config.get("region", "us")

    try:
        creds = get_credentials()
        headers = {
            "Authorization": f"Bearer {creds.token}",
            "Accept": "application/json",
            "Content-Type": "application/json"
        }
    except Exception as e:
        print(f"Authentication failed: {e}", file=sys.stderr)
        sys.exit(1)

    session = requests.Session()

    print("1. Fetching latest notebooks...")
    notebooks = fetch_notebooks_list(session, headers, project_id, customer_id, region, limit=50)
    print(f"Retrieved {len(notebooks)} notebooks.")

    print("2. Fetching detailed notebooks to extract investigation IDs...")
    detailed_notebooks = []
    investigation_names = set()
    with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
        futures = {executor.submit(fetch_notebook, session, headers, nb["name"]): nb for nb in notebooks}
        for future in concurrent.futures.as_completed(futures):
            res = future.result()
            if res:
                detailed_notebooks.append(res)
                for inv_name in res.get("investigations", []):
                    investigation_names.add(inv_name)

    print(f"Found {len(investigation_names)} investigations.")

    print("3. Fetching investigation details to get Alert IDs...")
    investigations = []
    all_alert_ids = set()
    with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
        futures = {executor.submit(fetch_investigation, session, headers, name): name for name in investigation_names}
        for future in concurrent.futures.as_completed(futures):
            res = future.result()
            if res:
                investigations.append(res)
                for aid in res.get("alerts", {}).get("ids", []):
                    all_alert_ids.add(aid)

    print(f"Found {len(all_alert_ids)} unique alert IDs.")

    print("4. Mapping Alert IDs to Case IDs via dashboardQueries...")
    alert_to_case_id = {}
    alert_to_priority = {}

    alert_list = sorted(list(all_alert_ids))
    batch_size = 30
    unmapped_alerts = set(alert_list)

    for start_idx in range(0, len(alert_list), batch_size):
        batch_alerts = alert_list[start_idx:start_idx + batch_size]
        res_json = run_mapping_query(session, headers, project_id, customer_id, region, batch_alerts)
        if res_json:
            results = res_json.get("results", [])
            columns = {col["column"]: col.get("values", []) for col in results}
            if columns and "alert_id" in columns and "case_id" in columns:
                num_rows = len(columns["alert_id"])
                for i in range(num_rows):
                    aid = columns["alert_id"][i].get("value", {}).get("stringVal")
                    case_guid = columns["case_id"][i].get("value", {}).get("stringVal")
                    priority = columns.get("priority", [])[i].get("value", {}).get("stringVal", "UNKNOWN") if "priority" in columns else "UNKNOWN"

                    # Extract numeric SOAR Case ID
                    links = columns["case_id"][i].get("value", {}).get("metadata", {}).get("links", [])
                    numeric_case_id = "UNKNOWN"
                    if links:
                        url_val = links[0].get("url", "")
                        if "/cases/" in url_val:
                            numeric_case_id = url_val.split("/cases/")[-1]

                    if aid and numeric_case_id != "UNKNOWN":
                        alert_to_case_id[aid] = numeric_case_id
                        alert_to_priority[aid] = priority
                        if aid in unmapped_alerts:
                            unmapped_alerts.remove(aid)

    print(f"5. Fallback mapping for {len(unmapped_alerts)} alerts via legacy APIs...")
    legacy_mappings = []
    with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
        futures = {executor.submit(scan_alert_legacy, session, headers, project_id, customer_id, region, aid): aid for aid in unmapped_alerts}
        for future in concurrent.futures.as_completed(futures):
            aid, case_name = future.result()
            if case_name:
                legacy_mappings.append((aid, case_name))

    if legacy_mappings:
        for aid, case_name in legacy_mappings:
            numeric_id, priority = resolve_legacy_case_id(session, headers, project_id, customer_id, region, case_name)
            if numeric_id:
                alert_to_case_id[aid] = numeric_id
                alert_to_priority[aid] = priority
                if aid in unmapped_alerts:
                    unmapped_alerts.remove(aid)

    print(f"Remaining unmapped alerts: {len(unmapped_alerts)}")

    print("6. Writing data to CSV...")
    with open(CSV_FILE, mode='w', newline='', encoding='utf-8') as file:
        writer = csv.writer(file)
        writer.writerow(["Notebook Name", "Investigation ID", "Verdict", "Alert ID", "Case ID (SOAR)", "Priority"])

        # Map Notebook to Investigations
        inv_to_notebook = {}
        for nb in detailed_notebooks:
            for inv_name in nb.get("investigations", []):
                inv_to_notebook[inv_name] = nb.get("name", "UNKNOWN")

        row_count = 0
        for inv in investigations:
            inv_name = inv.get("name", "UNKNOWN")
            inv_id = inv_name.split("/")[-1]
            verdict = inv.get("verdict", "UNKNOWN")
            notebook_name = inv_to_notebook.get(inv_name, "UNKNOWN").split("/")[-1]

            alerts = inv.get("alerts", {}).get("ids", [])
            if not alerts:
                # Write a row even if no alerts
                writer.writerow([notebook_name, inv_id, verdict, "NONE", "NONE", "UNKNOWN"])
                row_count += 1
            else:
                for aid in alerts:
                    case_id = alert_to_case_id.get(aid, "NOT_FOUND")
                    priority = alert_to_priority.get(aid, "UNKNOWN")
                    writer.writerow([notebook_name, inv_id, verdict, aid, case_id, priority])
                    row_count += 1

    print(f"Done! Successfully wrote {row_count} rows to {CSV_FILE}")

if __name__ == "__main__":
    main()