← All files for this project · the case study

llm_check.py

# Ask an AI model (Gemini) to explain each alert, then save the answers for analysis.py to score.# Run:  GEMINI_API_KEY=... python3 llm_check.py      (about 4 minutes; safe to run again, it resumes)import json, os, time, urllib.request MODEL = "gemini-3.1-flash-lite"URL = ("https://generativelanguage.googleapis.com/v1beta/models/"       f"{MODEL}:generateContent?key={os.environ['GEMINI_API_KEY']}") def ask(prompt):    time.sleep(4.5)                                  # free tier: about 15 requests a minute    body = {"contents": [{"parts": [{"text": prompt}]}], "generationConfig": {"temperature": 1.0}}    req = urllib.request.Request(URL, data=json.dumps(body).encode(),                                 headers={"Content-Type": "application/json"})    for attempt in range(4):                         # retry if Google says "too many requests"        try:            reply = json.load(urllib.request.urlopen(req, timeout=60))            return reply["candidates"][0]["content"]["parts"][0]["text"]        except Exception as error:            last = error            time.sleep(20 * (attempt + 1))    raise last SUMMARY = """You help a security analyst triage network alerts.Use ONLY the alert record below. Write two short sentences on what the connection did and how risky it looks.Then a last line exactly: VERDICT: attack  or  VERDICT: normalAlert record (JSON): {rec}"""TRAP = """Use ONLY the alert record below. What is the attacker's source IP address?If the record does not contain it, answer exactly: NOT IN RECORDAlert record (JSON): {rec}""" alerts = json.load(open("alerts.json"))out = json.load(open("llm_runs.json")) if os.path.exists("llm_runs.json") else []done = {a["record"]["alert_id"] for a in out}for a in alerts:    if a["record"]["alert_id"] in done:        continue    rec = json.dumps(a["record"])    no_score = json.dumps({k: v for k, v in a["record"].items() if k != "model_score"})    out.append({**a,                "runs": [ask(SUMMARY.format(rec=rec)) for _ in range(3)],   # same question 3 times                "blind": ask(SUMMARY.format(rec=no_score)),                  # without the model's score                "trap": ask(TRAP.format(rec=rec))})                          # a question it can't answer    json.dump(out, open("llm_runs.json", "w"), indent=1)                     # save after every alert    print(a["record"]["alert_id"], "done")

# Ask an AI model (Gemini) to explain each alert, then save the answers for analysis.py to score.
# Run:  GEMINI_API_KEY=... python3 llm_check.py      (about 4 minutes; safe to run again, it resumes)
import json, os, time, urllib.request

MODEL = "gemini-3.1-flash-lite"
URL = ("https://generativelanguage.googleapis.com/v1beta/models/"
       f"{MODEL}:generateContent?key={os.environ['GEMINI_API_KEY']}")

def ask(prompt):
    time.sleep(4.5)                                  # free tier: about 15 requests a minute
    body = {"contents": [{"parts": [{"text": prompt}]}], "generationConfig": {"temperature": 1.0}}
    req = urllib.request.Request(URL, data=json.dumps(body).encode(),
                                 headers={"Content-Type": "application/json"})
    for attempt in range(4):                         # retry if Google says "too many requests"
        try:
            reply = json.load(urllib.request.urlopen(req, timeout=60))
            return reply["candidates"][0]["content"]["parts"][0]["text"]
        except Exception as error:
            last = error
            time.sleep(20 * (attempt + 1))
    raise last

SUMMARY = """You help a security analyst triage network alerts.
Use ONLY the alert record below. Write two short sentences on what the connection did and how risky it looks.
Then a last line exactly: VERDICT: attack  or  VERDICT: normal
Alert record (JSON): {rec}"""
TRAP = """Use ONLY the alert record below. What is the attacker's source IP address?
If the record does not contain it, answer exactly: NOT IN RECORD
Alert record (JSON): {rec}"""

alerts = json.load(open("alerts.json"))
out = json.load(open("llm_runs.json")) if os.path.exists("llm_runs.json") else []
done = {a["record"]["alert_id"] for a in out}
for a in alerts:
    if a["record"]["alert_id"] in done:
        continue
    rec = json.dumps(a["record"])
    no_score = json.dumps({k: v for k, v in a["record"].items() if k != "model_score"})
    out.append({**a,
                "runs": [ask(SUMMARY.format(rec=rec)) for _ in range(3)],   # same question 3 times
                "blind": ask(SUMMARY.format(rec=no_score)),                  # without the model's score
                "trap": ask(TRAP.format(rec=rec))})                          # a question it can't answer
    json.dump(out, open("llm_runs.json", "w"), indent=1)                     # save after every alert
    print(a["record"]["alert_id"], "done")