update
Browse files
app.py
CHANGED
|
@@ -1,20 +1,8 @@
|
|
| 1 |
-
"""
|
| 2 |
-
TextScout β red-team threat-intel chatbot (Streamlit, for Hugging Face Spaces).
|
| 3 |
-
|
| 4 |
-
Per turn: user asks in natural language -> RAG retrieves the matching record
|
| 5 |
-
(bundled MITRE ATT&CK / KEV snapshot, or LIVE NVD + CISA KEV for any CVE not in
|
| 6 |
-
the snapshot) -> builds the EXACT training-format prompt -> the fine-tuned model
|
| 7 |
-
(Llama-3.2-3B + LoRA Maximuz23/Text-OSINT) analyzes it. Empty lookup -> the model
|
| 8 |
-
refuses instead of fabricating.
|
| 9 |
-
|
| 10 |
-
Nothing is hardcoded: the answer is always model.generate(); only the retrieved
|
| 11 |
-
record (real data) and a one-line help message are non-model text. Record
|
| 12 |
-
templates mirror scripts/build_grounded_records.py; SYSTEM_PROMPT mirrors
|
| 13 |
-
ai-test.ipynb c03 β do not drift either.
|
| 14 |
-
"""
|
| 15 |
import os
|
| 16 |
import re
|
| 17 |
import json
|
|
|
|
|
|
|
| 18 |
import pathlib
|
| 19 |
import urllib.request
|
| 20 |
import urllib.parse
|
|
@@ -317,7 +305,6 @@ def _vt(ioc, ioc_type):
|
|
| 317 |
elif ioc_type == "file":
|
| 318 |
path = "files/" + ioc
|
| 319 |
elif ioc_type == "url":
|
| 320 |
-
import base64
|
| 321 |
path = "urls/" + base64.urlsafe_b64encode(ioc.encode()).decode().strip("=")
|
| 322 |
else:
|
| 323 |
return None
|
|
@@ -398,7 +385,6 @@ def recent_threatfox_ioc():
|
|
| 398 |
rows = d.get("data") if d.get("query_status") == "ok" else None
|
| 399 |
if not isinstance(rows, list) or not rows:
|
| 400 |
return None
|
| 401 |
-
import random
|
| 402 |
pool = [r for r in rows if r.get("ioc_type") == "ip:port"] or rows
|
| 403 |
return random.choice(pool).get("ioc", "").split(":")[0]
|
| 404 |
|
|
@@ -457,46 +443,56 @@ HELP = (
|
|
| 457 |
)
|
| 458 |
|
| 459 |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 460 |
def route(q):
|
| 461 |
-
"""
|
| 462 |
actor_idx, cve_idx, tech_idx, soft_idx, actor_re, soft_re = load_corpus()
|
| 463 |
|
|
|
|
| 464 |
m = re.search(r"CVE-\d{4}-\d{4,}", q, re.I)
|
| 465 |
if m:
|
| 466 |
cid = m.group(0).upper()
|
| 467 |
if cid in cve_idx:
|
| 468 |
-
rec = cve_idx[cid]
|
| 469 |
-
return
|
| 470 |
try:
|
| 471 |
rec = live_cve(cid)
|
| 472 |
except Exception:
|
| 473 |
return None, None, None, "β οΈ Live NVD/CISA KEV lookup failed β please try again in a moment."
|
| 474 |
if rec:
|
| 475 |
-
|
| 476 |
-
|
| 477 |
|
|
|
|
| 478 |
m = re.search(r"\bT\d{4}(?:\.\d{3})?\b", q)
|
| 479 |
if m and m.group(0).upper() in tech_idx:
|
| 480 |
-
|
| 481 |
|
|
|
|
| 482 |
if IOC_RE.search(q):
|
| 483 |
ioc, ioc_type = detect_ioc(q)
|
| 484 |
if ioc and _have_ioc_keys():
|
| 485 |
-
|
| 486 |
-
if
|
| 487 |
-
return
|
| 488 |
return q, None, "the report you provided", None
|
| 489 |
|
|
|
|
| 490 |
a = _longest(actor_re, q)
|
| 491 |
s = _longest(soft_re, q)
|
| 492 |
if a or s:
|
| 493 |
if len(s or "") > len(a or ""):
|
| 494 |
-
|
| 495 |
-
rec = actor_idx[a.lower()]
|
| 496 |
-
return
|
| 497 |
|
|
|
|
| 498 |
if ACTOR_KW.search(q):
|
| 499 |
-
|
| 500 |
if REPORT_KW.search(q):
|
| 501 |
return q, None, "the report you provided", None
|
| 502 |
return None, None, None, HELP
|
|
@@ -523,7 +519,7 @@ SUGGESTIONS = {
|
|
| 523 |
"π οΈ Who uses Cobalt Strike?": "Which threat actors use the Cobalt Strike tool?",
|
| 524 |
"π Explain technique T1059.001": "Explain MITRE ATT&CK technique T1059.001",
|
| 525 |
"π Assess CVE-2025-0282 (live)": "Assess CVE-2025-0282 for offensive relevance",
|
| 526 |
-
"π Look up a fresh live IOC": None,
|
| 527 |
}
|
| 528 |
|
| 529 |
if "messages" not in st.session_state:
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 1 |
import os
|
| 2 |
import re
|
| 3 |
import json
|
| 4 |
+
import base64
|
| 5 |
+
import random
|
| 6 |
import pathlib
|
| 7 |
import urllib.request
|
| 8 |
import urllib.parse
|
|
|
|
| 305 |
elif ioc_type == "file":
|
| 306 |
path = "files/" + ioc
|
| 307 |
elif ioc_type == "url":
|
|
|
|
| 308 |
path = "urls/" + base64.urlsafe_b64encode(ioc.encode()).decode().strip("=")
|
| 309 |
else:
|
| 310 |
return None
|
|
|
|
| 385 |
rows = d.get("data") if d.get("query_status") == "ok" else None
|
| 386 |
if not isinstance(rows, list) or not rows:
|
| 387 |
return None
|
|
|
|
| 388 |
pool = [r for r in rows if r.get("ioc_type") == "ip:port"] or rows
|
| 389 |
return random.choice(pool).get("ioc", "").split(":")[0]
|
| 390 |
|
|
|
|
| 443 |
)
|
| 444 |
|
| 445 |
|
| 446 |
+
def _hit(prompt, source):
|
| 447 |
+
"""A successful retrieval: the same text is both fed to the model and shown in the RAG expander."""
|
| 448 |
+
return prompt, prompt, source, None
|
| 449 |
+
|
| 450 |
+
|
| 451 |
def route(q):
|
| 452 |
+
"""Map a user query to (user_prompt, retrieved_record_or_None, source_label, note_or_None)."""
|
| 453 |
actor_idx, cve_idx, tech_idx, soft_idx, actor_re, soft_re = load_corpus()
|
| 454 |
|
| 455 |
+
# CVE id -> bundled record, else live NVD + CISA KEV
|
| 456 |
m = re.search(r"CVE-\d{4}-\d{4,}", q, re.I)
|
| 457 |
if m:
|
| 458 |
cid = m.group(0).upper()
|
| 459 |
if cid in cve_idx:
|
| 460 |
+
rec = cve_idx[cid]
|
| 461 |
+
return _hit(rec.get("_prompt") or t_cve(rec), "bundled MITRE/KEV record")
|
| 462 |
try:
|
| 463 |
rec = live_cve(cid)
|
| 464 |
except Exception:
|
| 465 |
return None, None, None, "β οΈ Live NVD/CISA KEV lookup failed β please try again in a moment."
|
| 466 |
if rec:
|
| 467 |
+
return _hit(t_cve(rec), "live NVD + CISA KEV")
|
| 468 |
+
return _hit(t_cve_empty(cid), "live NVD + CISA KEV (no record)")
|
| 469 |
|
| 470 |
+
# ATT&CK technique id
|
| 471 |
m = re.search(r"\bT\d{4}(?:\.\d{3})?\b", q)
|
| 472 |
if m and m.group(0).upper() in tech_idx:
|
| 473 |
+
return _hit(t_tech(tech_idx[m.group(0).upper()]), "MITRE ATT&CK")
|
| 474 |
|
| 475 |
+
# IOC -> live enrichment; if no keys, pass the raw report straight to the model
|
| 476 |
if IOC_RE.search(q):
|
| 477 |
ioc, ioc_type = detect_ioc(q)
|
| 478 |
if ioc and _have_ioc_keys():
|
| 479 |
+
prompt = _safe(enrich_ioc, ioc, ioc_type)
|
| 480 |
+
if prompt:
|
| 481 |
+
return _hit(prompt, "live ThreatFox + OTX + VirusTotal")
|
| 482 |
return q, None, "the report you provided", None
|
| 483 |
|
| 484 |
+
# actor or software name β longest match wins (so "Cobalt Strike" beats "Cobalt")
|
| 485 |
a = _longest(actor_re, q)
|
| 486 |
s = _longest(soft_re, q)
|
| 487 |
if a or s:
|
| 488 |
if len(s or "") > len(a or ""):
|
| 489 |
+
return _hit(t_soft(soft_idx[s.lower()]), "MITRE ATT&CK Software")
|
| 490 |
+
rec = actor_idx[a.lower()]
|
| 491 |
+
return _hit(rec.get("_prompt") or t_actor(rec), "MITRE ATT&CK")
|
| 492 |
|
| 493 |
+
# actor-intent but no match -> honest empty lookup; else a report keyword; else help
|
| 494 |
if ACTOR_KW.search(q):
|
| 495 |
+
return _hit(t_actor_empty(_guess_name(q)), "MITRE ATT&CK (no record)")
|
| 496 |
if REPORT_KW.search(q):
|
| 497 |
return q, None, "the report you provided", None
|
| 498 |
return None, None, None, HELP
|
|
|
|
| 519 |
"π οΈ Who uses Cobalt Strike?": "Which threat actors use the Cobalt Strike tool?",
|
| 520 |
"π Explain technique T1059.001": "Explain MITRE ATT&CK technique T1059.001",
|
| 521 |
"π Assess CVE-2025-0282 (live)": "Assess CVE-2025-0282 for offensive relevance",
|
| 522 |
+
"π Look up a fresh live IOC": None,
|
| 523 |
}
|
| 524 |
|
| 525 |
if "messages" not in st.session_state:
|