Spaces:
Runtime error
Runtime error
| """ | |
| CTI-lite — Gradio CTI benchmark dashboard backed directly by the HF Dataset. | |
| Three tabs: | |
| 1. CTI Leaderboard — prequential score per task; click a row to drill into Model Detail | |
| 2. CTI Model Detail — per-task scores + per-item results for one model | |
| 3. Task Guide — explainer for each of the six benchmark tasks | |
| No database, no API server. | |
| Set HF_CTI_DATASET_REPO (and optionally HF_TOKEN) before running. | |
| """ | |
| import os | |
| import gradio as gr | |
| import pandas as pd | |
| import data | |
| _TASK_ORDER = ["rcm", "vsp", "ate", "taa", "forecast", "syn"] | |
| _TASK_LABELS = {"rcm": "RCM", "vsp": "VSP", "ate": "ATE", "taa": "TAA", "forecast": "Forecast", "syn": "SYN"} | |
| _LB_COLS = ["Model", "Provider"] + [_TASK_LABELS[t] for t in _TASK_ORDER] + ["Overall"] | |
| _MODEL_COLS = ["Task", "Score", "Items", "Scored", "Completed", "AUC"] | |
| _RESULT_COLS = ["Item", "Score", "Correct", "Pre-Cutoff", "Breakdown"] | |
| # --------------------------------------------------------------------------- | |
| # Task explainer content | |
| # --------------------------------------------------------------------------- | |
| _TASK_EXPLAINERS = { | |
| "rcm": { | |
| "title": "RCM — Root Cause Mapping", | |
| "description": ( | |
| "Given a CVE description, the model must identify the underlying weakness class(es) " | |
| "using the CWE (Common Weakness Enumeration) taxonomy. " | |
| "This tests whether a model understands *why* a vulnerability exists, not just what it does." | |
| ), | |
| "prompt_note": ( | |
| "Prompt: *You are a vulnerability analyst. Given the CVE description below, identify the most " | |
| "appropriate CWE identifier(s). Respond with only the CWE id(s) in the form CWE-XXX.*" | |
| ), | |
| "example_input": ( | |
| "CVE-2024-23897: Jenkins 2.441 and earlier, LTS 2.426.2 and earlier does not disable a feature " | |
| "of its CLI command parser that replaces an '@' character followed by a file path in an argument " | |
| "with the file's contents, allowing unauthenticated attackers to read arbitrary files on the " | |
| "Jenkins controller file system." | |
| ), | |
| "example_label": "CWE-22, CWE-88", | |
| "example_response_good": "CWE-22", | |
| "example_response_bad": "CWE-200", | |
| "scoring": ( | |
| "**Set F1** — precision and recall over the predicted vs. true CWE set. " | |
| "Partial credit when a model gets one CWE right out of two. " | |
| "Score is prequentially aggregated with a 0.99 fading factor so recent items count more." | |
| ), | |
| }, | |
| "vsp": { | |
| "title": "VSP — CVSS Score Prediction", | |
| "description": ( | |
| "Given a CVE description, the model must predict the full CVSS v3.1 base vector string. " | |
| "This tests calibrated severity reasoning across eight independent metrics: " | |
| "attack vector, complexity, privileges required, user interaction, scope, " | |
| "and confidentiality/integrity/availability impact." | |
| ), | |
| "prompt_note": ( | |
| "Prompt: *You are a vulnerability analyst. Given the CVE description below, predict the CVSS " | |
| "v3.1 base vector string. Respond with only the CVSS v3.1 vector, e.g. " | |
| "CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H.*" | |
| ), | |
| "example_input": ( | |
| "CVE-2024-6387: A race condition in OpenSSH's server (sshd) in glibc-based Linux systems " | |
| "allows unauthenticated remote code execution as root due to a signal handler that calls " | |
| "async-signal-unsafe functions." | |
| ), | |
| "example_label": "CVSS:3.1/AV:N/AC:H/PR:N/UI:N/S:U/C:H/I:H/A:H", | |
| "example_response_good": "CVSS:3.1/AV:N/AC:H/PR:N/UI:N/S:U/C:H/I:H/A:H", | |
| "example_response_bad": "CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H", | |
| "scoring": ( | |
| "**Normalised MAD** — mean absolute deviation per metric, scaled so that a perfect prediction " | |
| "scores 1.0 and maximum error scores 0.0. Severity band agreement and component-level " | |
| "partial credit are also applied. Models that get AC:L vs AC:H wrong (a common mistake on " | |
| "regreSSHion-class bugs) take a meaningful hit." | |
| ), | |
| }, | |
| "ate": { | |
| "title": "ATE — ATT&CK Technique Extraction", | |
| "description": ( | |
| "Given an excerpt from a threat intelligence report (CISA advisory, NCSC alert, DFIR write-up, etc.), " | |
| "the model must identify the MITRE ATT&CK techniques described. " | |
| "This tests structured threat-behaviour recognition against a living, versioned taxonomy." | |
| ), | |
| "prompt_note": ( | |
| "Prompt: *You are a threat-intelligence analyst. Identify the MITRE ATT&CK techniques described " | |
| "in the report excerpt below. Respond with only the ATT&CK technique ids (e.g. T1059), " | |
| "comma-separated.*" | |
| ), | |
| "example_input": ( | |
| "The threat actor gained initial access via spear-phishing emails containing a malicious LNK file. " | |
| "After execution, a PowerShell script downloaded a Cobalt Strike beacon from a compromised web server. " | |
| "Credentials were harvested using a memory-scraping tool, and lateral movement was achieved " | |
| "via pass-the-hash. Data was staged in a ZIP archive before exfiltration over HTTPS." | |
| ), | |
| "example_label": "T1566, T1059, T1105, T1003, T1550, T1560, T1041", | |
| "example_response_good": "T1566, T1059, T1003, T1550, T1560, T1041", | |
| "example_response_bad": "T1566, T1059", | |
| "scoring": ( | |
| "**Set F1** over predicted vs. true technique IDs. Revoked techniques are resolved to their " | |
| "successor before scoring, so a model citing a deprecated ID is not penalised if the concept " | |
| "is correct. Sub-technique IDs (e.g. T1059.001) are normalised to the parent." | |
| ), | |
| }, | |
| "taa": { | |
| "title": "TAA — Threat Actor Attribution", | |
| "description": ( | |
| "Given an intrusion narrative from a public advisory or report, the model must name the most " | |
| "likely responsible threat actor. This tests attribution reasoning — matching TTPs, targeting, " | |
| "and infrastructure clues to a known group — and handles the challenge that the same group " | |
| "is tracked under dozens of different names across vendors." | |
| ), | |
| "prompt_note": ( | |
| "Prompt: *You are a threat-intelligence analyst. Based on the intrusion narrative below, name " | |
| "the most likely threat actor responsible. Respond with only the threat-actor name.*" | |
| ), | |
| "example_input": ( | |
| "The intrusion targeted South Korean cryptocurrency exchanges and defence contractors. " | |
| "The attackers used trojanised cryptocurrency trading software delivered via LinkedIn recruiter messages. " | |
| "Malware included a custom backdoor communicating over TLS with rotating C2 infrastructure, " | |
| "and the operation involved significant cryptocurrency theft routed through mixing services." | |
| ), | |
| "example_label": "Lazarus Group", | |
| "example_response_good": "Lazarus Group", | |
| "example_response_bad": "APT28", | |
| "scoring": ( | |
| "**Synonym-graph credit**: canonical name = 1.0, known alias (e.g. *Hidden Cobra*, *ZINC*) = 0.5, " | |
| "related group = 0.0. The alias graph is built from MISP galaxy threat-actor data, " | |
| "so vendor-specific names are correctly mapped. A model that says 'Hidden Cobra' when " | |
| "the label is 'Lazarus Group' scores 0.5, not 0." | |
| ), | |
| }, | |
| "forecast": { | |
| "title": "Forecast — Exploitation Probability", | |
| "description": ( | |
| "Given a CVE description at publication time, the model must estimate the probability " | |
| "that the vulnerability will be actively exploited in the wild. " | |
| "Labels are resolved 14+ days later against the CISA Known Exploited Vulnerabilities (KEV) catalogue, " | |
| "making this a genuine prospective prediction task that cannot be gamed by memorising KEV entries." | |
| ), | |
| "prompt_note": ( | |
| "Prompt: *You are a vulnerability prioritisation analyst. Given the CVE description below, " | |
| "estimate the probability that this vulnerability will be exploited in the wild. " | |
| "Respond with only a probability between 0 and 1 (e.g. 0.75).*" | |
| ), | |
| "example_input": ( | |
| "CVE-2024-21762: A out-of-bounds write vulnerability in Fortinet FortiOS SSL VPN " | |
| "may allow a remote unauthenticated attacker to execute arbitrary code or command " | |
| "via specially crafted HTTP requests." | |
| ), | |
| "example_label": "exploited=True (later confirmed in CISA KEV)", | |
| "example_response_good": "0.90", | |
| "example_response_bad": "0.10", | |
| "scoring": ( | |
| "**Brier score** (lower is better, inverted to [0,1]) measures calibration of the probability " | |
| "estimate against the binary KEV outcome. **AUC** measures discrimination across all Forecast " | |
| "items in a run. Items are withheld for 14 days after creation to prevent KEV leakage " | |
| "into the label, and the newest 14-day slice is always held out from scoring." | |
| ), | |
| }, | |
| "syn": { | |
| "title": "SYN — Threat Assessment Synthesis", | |
| "description": ( | |
| "Given a bundle of raw CTI inputs (CVE details, observed TTPs, actor indicators, affected sectors), " | |
| "the model must produce a coherent, calibrated threat assessment. " | |
| "This is the most open-ended task: it tests whether a model can integrate heterogeneous " | |
| "evidence into actionable intelligence without hallucinating facts or over-claiming certainty." | |
| ), | |
| "prompt_note": ( | |
| "Prompt: *You are a CTI analyst. From the raw inputs below, produce a concise threat assessment: " | |
| "name the most likely threat actor, relevant CVEs, MITRE ATT&CK techniques, targeted sectors, " | |
| "indicators of compromise, and recommended mitigations. Hedge explicitly where the evidence is weak.*" | |
| ), | |
| "example_input": ( | |
| "CVE-2024-3400 (CVSS 10.0) exploited in Palo Alto Networks GlobalProtect. " | |
| "Observed TTPs: T1190 (Exploit Public-Facing Application), T1059.006 (Python), T1105 (Ingress Tool Transfer). " | |
| "Targeted: US government, critical infrastructure. " | |
| "IOCs: 144.172.79[.]92, update.pkgs-cloud[.]com. " | |
| "CISA emergency directive issued." | |
| ), | |
| "example_label": "Claims: [actor=UTA0218, cve=CVE-2024-3400, techniques={T1190,T1059,T1105}, sectors={government,critical_infrastructure}, mitigation=patch_immediately]", | |
| "example_response_good": ( | |
| "Based on the targeting profile and TTPs, this activity is attributed with moderate confidence to " | |
| "UTA0218, a likely nation-state actor. CVE-2024-3400 enables unauthenticated RCE and is being " | |
| "actively exploited. Recommended actions: apply PAN-OS hotfix immediately, block listed IOCs, " | |
| "audit GlobalProtect logs for T1190 indicators. Note: actor attribution remains preliminary " | |
| "pending further forensic analysis." | |
| ), | |
| "example_response_bad": ( | |
| "This is a critical vulnerability exploited by Lazarus Group targeting all sectors globally. " | |
| "The attack uses T1059, T1190, T1003, T1078, T1071, T1486. Patch everything immediately. " | |
| "All systems are at risk." | |
| ), | |
| "scoring": ( | |
| "**Claim-set scoring** via LLM judge: the reference assessment is decomposed into atomic claims, " | |
| "then the model response is evaluated for (1) **recall** — what fraction of true claims are covered, " | |
| "(2) **faithfulness** — no hallucinated facts, (3) **calibration** — appropriate hedging where " | |
| "evidence is weak. Final score = recall × faithfulness × calibration." | |
| ), | |
| }, | |
| } | |
| def _task_explainer_md(task: str) -> str: | |
| e = _TASK_EXPLAINERS[task] | |
| return f""" | |
| {e['description']} | |
| _{e['prompt_note']}_ | |
| --- | |
| **Example input** | |
| ``` | |
| {e['example_input']} | |
| ``` | |
| **Ground-truth label** | |
| ``` | |
| {e['example_label']} | |
| ``` | |
| **Strong model response** | |
| ``` | |
| {e['example_response_good']} | |
| ``` | |
| **Weak model response** | |
| ``` | |
| {e['example_response_bad']} | |
| ``` | |
| --- | |
| **Scoring** — {e['scoring']} | |
| """ | |
| # --------------------------------------------------------------------------- | |
| # DataFrame builders | |
| # --------------------------------------------------------------------------- | |
| def build_leaderboard_df() -> pd.DataFrame: | |
| empty = pd.DataFrame(columns=_LB_COLS) | |
| rows = data.get_leaderboard() | |
| if not rows: | |
| return empty | |
| result = [] | |
| for m in rows: | |
| row: dict = {"Model": m["model_name"], "Provider": m["provider"]} | |
| for task in _TASK_ORDER: | |
| label = _TASK_LABELS[task] | |
| score = m["task_scores"].get(task) | |
| row[label] = f"{score:.1%}" if score is not None else "—" | |
| overall = m.get("overall") | |
| row["Overall"] = f"{overall:.1%}" if overall is not None else "—" | |
| result.append(row) | |
| return pd.DataFrame(result, columns=_LB_COLS) | |
| def build_model_detail(model_id: str) -> tuple[pd.DataFrame, list[str], pd.DataFrame]: | |
| empty_tasks = pd.DataFrame(columns=_MODEL_COLS) | |
| empty_results = pd.DataFrame(columns=_RESULT_COLS) | |
| if not model_id: | |
| return empty_tasks, [], empty_results | |
| runs = data.get_model_runs(model_id) | |
| if not runs: | |
| return empty_tasks, [], empty_results | |
| rows = [] | |
| run_ids: list[str] = [] | |
| for r in runs: | |
| score = r.get("prequential_score") | |
| auc = r.get("auc") | |
| rows.append({ | |
| "Task": _TASK_LABELS.get(r["task"], r["task"]), | |
| "Score": f"{score:.1%}" if score is not None else "—", | |
| "Items": r.get("item_count", 0), | |
| "Scored": r.get("scored_count", 0), | |
| "Completed": r.get("completed_at") or "—", | |
| "AUC": f"{auc:.3f}" if auc is not None else "—", | |
| }) | |
| run_ids.append(r["run_id"]) | |
| return pd.DataFrame(rows, columns=_MODEL_COLS), run_ids, empty_results | |
| def build_results_df(run_id: str) -> pd.DataFrame: | |
| empty = pd.DataFrame(columns=_RESULT_COLS) | |
| if not run_id: | |
| return empty | |
| results = data.get_run_results(run_id) | |
| if not results: | |
| return empty | |
| rows = [] | |
| for r in results: | |
| score = r.get("score") | |
| correct = r.get("correct") | |
| pre_cutoff = r.get("pre_cutoff") | |
| rows.append({ | |
| "Item": str(r["item_id"])[:8], | |
| "Score": f"{score:.3f}" if score is not None else "—", | |
| "Correct": "✓" if correct is True else ("✗" if correct is False else "—"), | |
| "Pre-Cutoff": "yes" if pre_cutoff is True else ("no" if pre_cutoff is False else "—"), | |
| "Breakdown": r.get("score_summary", ""), | |
| }) | |
| return pd.DataFrame(rows, columns=_RESULT_COLS) | |
| # --------------------------------------------------------------------------- | |
| # Gradio UI | |
| # --------------------------------------------------------------------------- | |
| def build_app() -> gr.Blocks: | |
| with gr.Blocks(title="Glokta CTI Leaderboard", theme=gr.themes.Soft()) as demo: | |
| gr.Markdown( | |
| f""" | |
| # Glokta — CTI Benchmark Leaderboard | |
| Cyber-threat-intelligence living benchmark · data from `{data.HF_CTI_DATASET_REPO or "HF_CTI_DATASET_REPO not set"}` | |
| """ | |
| ) | |
| model_name_to_id = gr.State(value={}) | |
| with gr.Tabs() as tabs: | |
| # ---------------------------------------------------------------- | |
| # Tab 1: CTI Leaderboard | |
| # ---------------------------------------------------------------- | |
| with gr.Tab("CTI Leaderboard", id="cti_leaderboard"): | |
| gr.Markdown( | |
| "Prequential (fading-factor weighted) accuracy per task. " | |
| "**Click a row to drill into CTI Model Detail.**" | |
| ) | |
| with gr.Row(): | |
| refresh_btn = gr.Button("Refresh Data from HF", variant="primary") | |
| leaderboard_table = gr.Dataframe( | |
| label="CTI Leaderboard", | |
| interactive=False, | |
| wrap=True, | |
| ) | |
| # ---------------------------------------------------------------- | |
| # Tab 2: CTI Model Detail | |
| # ---------------------------------------------------------------- | |
| with gr.Tab("CTI Model Detail", id="cti_model_detail") as model_detail_tab: | |
| gr.Markdown( | |
| "Select a model to see its latest prequential score per task. " | |
| "**Click a task row to inspect individual item results.**" | |
| ) | |
| model_dropdown = gr.Dropdown( | |
| label="Model", | |
| choices=[], | |
| value=None, | |
| interactive=True, | |
| ) | |
| model_table = gr.Dataframe( | |
| label="Task Scores (click a row to see results)", | |
| interactive=False, | |
| wrap=True, | |
| ) | |
| gr.Markdown("### Item Results") | |
| result_table = gr.Dataframe( | |
| label="Per-Item Results", | |
| interactive=False, | |
| wrap=True, | |
| ) | |
| # ---------------------------------------------------------------- | |
| # Tab 3: Task Guide | |
| # ---------------------------------------------------------------- | |
| with gr.Tab("Task Guide", id="task_guide"): | |
| gr.Markdown(""" | |
| ## What is the CTI Benchmark? | |
| The Glokta CTI benchmark is a **living, temporally-anchored** evaluation suite for | |
| cyber-threat-intelligence reasoning. Items are derived from public sources (NVD, CISA advisories, | |
| MITRE ATT&CK, MISP galaxy, DFIR reports) and anchored to a `first_available_date` — the date | |
| the information was publicly known. Models are only tested on items published **after** their | |
| training cutoff, preventing label memorisation and making scores a genuine measure of reasoning | |
| rather than recall. | |
| Scores are aggregated using **prequential (fading-factor) accuracy** with a 0.99 decay weight, | |
| so recent items count slightly more than older ones. This keeps the leaderboard sensitive to | |
| model updates and knowledge cutoff improvements. | |
| The benchmark comprises six tasks spanning the full CTI workflow: | |
| """) | |
| for task in _TASK_ORDER: | |
| e = _TASK_EXPLAINERS[task] | |
| with gr.Accordion(e["title"], open=False): | |
| gr.Markdown(_task_explainer_md(task)) | |
| gr.Markdown(""" | |
| --- | |
| ## Further Reading | |
| - [Glokta on GitHub](https://github.com/JakeBx/Glokta) — self-hosting instructions, architecture overview, and contribution guide | |
| """) | |
| # State: run_ids parallel to model_table rows | |
| task_run_ids = gr.State(value=[]) | |
| # -------------------------------------------------------------------- | |
| # Event handlers | |
| # -------------------------------------------------------------------- | |
| def on_load(): | |
| try: | |
| data.load_data() | |
| except Exception as exc: | |
| print(f"[app] Data load failed: {exc}") | |
| lb_df = build_leaderboard_df() | |
| models = [(m["name"], m["id"]) for m in data.get_models()] | |
| name_to_id = {m["name"]: m["id"] for m in data.get_models()} | |
| return ( | |
| lb_df, | |
| name_to_id, | |
| gr.update(choices=models, value=None), | |
| ) | |
| def on_refresh(): | |
| try: | |
| data.load_data() | |
| except Exception as exc: | |
| print(f"[app] Reload failed: {exc}") | |
| lb_df = build_leaderboard_df() | |
| models = [(m["name"], m["id"]) for m in data.get_models()] | |
| name_to_id = {m["name"]: m["id"] for m in data.get_models()} | |
| return ( | |
| lb_df, | |
| name_to_id, | |
| gr.update(choices=models, value=None), | |
| ) | |
| def on_lb_row_click(evt: gr.SelectData, lb_df: pd.DataFrame, name_to_id: dict): | |
| try: | |
| model_name = str(lb_df.iloc[evt.index[0]]["Model"]) | |
| except Exception: | |
| return gr.update(), gr.update() | |
| model_id = name_to_id.get(model_name, "") | |
| return ( | |
| gr.update(value=model_id), | |
| gr.update(selected="cti_model_detail"), | |
| ) | |
| def on_model_change(model_id: str | None): | |
| task_df, run_ids, result_df = build_model_detail(model_id or "") | |
| return task_df, run_ids, result_df | |
| def on_task_select(evt: gr.SelectData, run_ids: list[str]): | |
| try: | |
| run_id = run_ids[evt.index[0]] | |
| except (IndexError, TypeError): | |
| return pd.DataFrame(columns=_RESULT_COLS) | |
| return build_results_df(run_id) | |
| # -------------------------------------------------------------------- | |
| # Wire events | |
| # -------------------------------------------------------------------- | |
| demo.load( | |
| fn=on_load, | |
| inputs=None, | |
| outputs=[leaderboard_table, model_name_to_id, model_dropdown], | |
| ) | |
| refresh_btn.click( | |
| fn=on_refresh, | |
| inputs=None, | |
| outputs=[leaderboard_table, model_name_to_id, model_dropdown], | |
| ) | |
| leaderboard_table.select( | |
| fn=on_lb_row_click, | |
| inputs=[leaderboard_table, model_name_to_id], | |
| outputs=[model_dropdown, tabs], | |
| ) | |
| model_dropdown.change( | |
| fn=on_model_change, | |
| inputs=[model_dropdown], | |
| outputs=[model_table, task_run_ids, result_table], | |
| ) | |
| model_detail_tab.select( | |
| fn=on_model_change, | |
| inputs=[model_dropdown], | |
| outputs=[model_table, task_run_ids, result_table], | |
| ) | |
| model_table.select( | |
| fn=on_task_select, | |
| inputs=[task_run_ids], | |
| outputs=[result_table], | |
| ) | |
| return demo | |
| if __name__ == "__main__": | |
| port = int(os.environ.get("GRADIO_SERVER_PORT", 7860)) | |
| app = build_app() | |
| app.launch(server_name="0.0.0.0", server_port=port, show_api=False) | |