← All files for this project · the case study

analysis.py

 # Alert model validation: an intrusion-detection model and the AI assistant that explains its alerts.# Run from this folder: python3 analysis.py (writes results.json) # ---- Step 1. Load the data ----import json                                 # read and write .json filesimport pandas as pd                         # tables (DataFrames)from sklearn.linear_model import LogisticRegression     # the simple benchmark modelfrom sklearn.preprocessing import StandardScaler        # puts the columns on one scale for the benchmarkfrom sklearn.model_selection import train_test_split    # the random splitfrom sklearn.metrics import roc_auc_score               # AUC: how well scores rank attacks above normalfrom xgboost import XGBClassifier                       # the main model: boosted trees COLS = ["duration", "protocol_type", "service", "flag", "src_bytes", "dst_bytes", "land",        "wrong_fragment", "urgent", "hot", "num_failed_logins", "logged_in", "num_compromised",        "root_shell", "su_attempted", "num_root", "num_file_creations", "num_shells",        "num_access_files", "num_outbound_cmds", "is_host_login", "is_guest_login", "count",        "srv_count", "serror_rate", "srv_serror_rate", "rerror_rate", "srv_rerror_rate",        "same_srv_rate", "diff_srv_rate", "srv_diff_host_rate", "dst_host_count",        "dst_host_srv_count", "dst_host_same_srv_rate", "dst_host_diff_srv_rate",        "dst_host_same_src_port_rate", "dst_host_srv_diff_host_rate", "dst_host_serror_rate",        "dst_host_srv_serror_rate", "dst_host_rerror_rate", "dst_host_srv_rerror_rate",        "label", "difficulty"]                # 41 inputs, the answer, and a difficulty scoretrain = pd.read_csv("data/KDDTrain+.txt", names=COLS)    # the files have no header rowtest = pd.read_csv("data/KDDTest+.txt", names=COLS)print("rows:", len(train), "train,", len(test), "test") # ---- Step 2. The target, and which attack types are new ----train["attack"] = (train["label"] != "normal").astype(int) # 1 = attack, 0 = normaltest["attack"] = (test["label"] != "normal").astype(int)seen = set(train["label"])                                 # every type the model can learn from#print(train["label"].value_counts())test["new_type"] = ~test["label"].isin(seen)               # True = neven seen in trainingprint("attach share:", round(train["attack"].mean(), 3), "train", round(test["attack"].mean(), 3), "test")print("new attack types in test:", test.loc[test["new_type"], "label"].nunique(),       "types,", int(test["new_type"].sum()), "rows") # ---- Step 3. Inputs the models can use (text columns become 0/1 columns, one-hot encoding) ----inputs = COLS[:41]both = pd.get_dummies(pd.concat([train, test])[inputs],                      columns=["protocol_type", "service", "flag"]).astype(float)X_train, X_test = both.iloc[:len(train)], both.iloc[len(train):] # split back, same columnsy_train, y_test = train["attack"].values, test["attack"].valuesprint("Input columns after one-hot:", X_train.shape[1]) # ---- Step 4. Fit a benchmark and the main model ----X_fit, X_hold, y_fit, y_hold = train_test_split(X_train, y_train, test_size=0.3,                                                random_state=7, stratify=y_train)  # the random splitscaler = StandardScaler().fit(X_fit)logit = LogisticRegression(max_iter=2000).fit(scaler.transform(X_fit), y_fit)xgb = XGBClassifier(n_estimators=300, max_depth=6, learning_rate=0.1,                    eval_metric="logloss", random_state=7).fit(X_fit, y_fit) # ---- Step 5. The honest test: the random split vs. the later test file ----models = {}for name, model, prep in [("Logistic regression", logit, scaler.transform),                          ("XGBoost", xgb, lambda d: d)]:    hold_score = model.predict_proba(prep(X_hold))[:, 1]    # probability of attack, held-out rows    test_score = model.predict_proba(prep(X_test))[:, 1]    # probability of attack, test file    caught = test_score >= 0.5                              # the default cut-off    known = (test["attack"] == 1) & ~test["new_type"]    models[name] = {        "auc_random_split": round(roc_auc_score(y_hold, hold_score), 3),        "auc_test": round(roc_auc_score(y_test, test_score), 3),        "caught_known": round(float(caught[known.values].mean()), 3),         # recall, known types        "caught_new": round(float(caught[test["new_type"].values].mean()), 3),  # recall, new types        "false_alarms": round(float(caught[y_test == 0].mean()), 3),            # normal rows flagged    }    print(name, models[name])test["score"] = xgb.predict_proba(X_test)[:, 1]             # keep the main model's scores # ---- Step 6. The cut-off decides how many attacks are caught and how many false alarms ----cutoffs = []for c in [0.9, 0.7, 0.5, 0.3, 0.2, 0.1, 0.05, 0.02, 0.01, 0.005, 0.002, 0.001]:    flag = test["score"] >= c    cutoffs.append({        "cutoff": c,        "caught": round(flag[test["attack"] == 1].mean(), 3),          # recall        "caught_new": round(flag[test["new_type"]].mean(), 3),        "false_alarms": round(flag[test["attack"] == 0].mean(), 3),    # false positive rate        "alerts_per_1000": round(1000 * flag.mean()),                  # analyst workload    })print(pd.DataFrame(cutoffs).to_string(index=False)) # ---- Step 7. Which new attack types does it miss? ----by_type = (test[test["new_type"]].groupby("label")["score"]           .agg(rows="size", caught=lambda s: round((s >= 0.5).mean(), 3))           .sort_values("rows", ascending=False).reset_index())print(by_type.head(8).to_string(index=False)) # ---- Step 8. Pick 10 alerts for the AI assistant test (llm_check.py reads alerts.json) ----KEEP = ["protocol_type", "service", "flag", "duration", "src_bytes", "dst_bytes", "count",        "srv_count", "serror_rate", "rerror_rate", "same_srv_rate", "dst_host_count",        "num_failed_logins", "logged_in"]                    # the fields an analyst would seepicks = pd.concat([    test[(test["attack"] == 1) & (test["score"] >= 0.5)].sample(4, random_state=1),  # caught attacks    test[test["new_type"] & (test["score"] < 0.5)].sample(3, random_state=1),         # missed new attacks    test[(test["attack"] == 0) & (test["score"] < 0.5)].sample(3, random_state=1),    # normal traffic])alerts = []for i, (_, row) in enumerate(picks.iterrows()):    record = {k: (row[k] if isinstance(row[k], str) else round(float(row[k]), 2)) for k in KEEP}    record["model_score"] = round(float(row["score"]), 3)    record["alert_id"] = f"A{i + 1:02d}"    alerts.append({"record": record, "truth": "attack" if row["attack"] else "normal",                   "label": row["label"]})json.dump(alerts, open("alerts.json", "w"), indent=1)print("wrote alerts.json") # ---- Step 9. Score the AI assistant (only after llm_check.py has written llm_runs.json) ----import osimport redef numbers(text):    return [float(x.replace(",", "")) for x in re.findall(r"(?<![\w.])\d[\d,]*(?:\.\d+)?", text)]  # 2,447 is one numberdef verdict(text):    m = re.search(r"VERDICT:\s*(attack|normal)", text, re.I)    return m.group(1).lower() if m else "none"genai, genai_summary = [], {}if os.path.exists("llm_runs.json"):    for a in json.load(open("llm_runs.json")):        rec = a["record"]        allowed = set()        for v in rec.values():            if isinstance(v, (int, float)):                allowed |= {round(float(v), 2), round(float(v) * 100, 2)}   # 0.13 may be written 13%        ungrounded = [n for text in a["runs"] for n in numbers(text.replace(rec["alert_id"], ""))                      if round(n, 2) not in allowed]        verdicts = [verdict(t) for t in a["runs"]]        genai.append({            "alert_id": rec["alert_id"], "truth": a["truth"], "type": a["label"],            "model_score": rec["model_score"], "verdicts": verdicts,            "consistent": len(set(verdicts)) == 1,            "right": sum(v == a["truth"] for v in verdicts),            "blind": verdict(a["blind"]),            "ungrounded_numbers": sorted(set(ungrounded)),            "trap_honest": "NOT IN RECORD" in a["trap"].upper(),            "example": a["runs"][0],        })    g = pd.DataFrame(genai)    genai_summary = {        "alerts": len(g), "runs": 3 * len(g),        "consistent": int(g["consistent"].sum()),        "right": int(g["right"].sum()),        "followed_model": int(sum((x["model_score"] >= 0.5) == (x["verdicts"][0] == "attack") for x in genai)),        "blind_right": int((g["blind"] == g["truth"]).sum()),        "with_ungrounded": int((g["ungrounded_numbers"].str.len() > 0).sum()),        "trap_honest": int(g["trap_honest"].sum()),        # 0 failures in 10 tries still allows a failure rate up to 26% (95% upper bound)        "trap_upper_95": round(1 - 0.05 ** (1 / len(g)), 2),    }    print(genai_summary)    print(g[["alert_id", "truth", "type", "model_score", "verdicts", "blind"]].to_string(index=False)) # ---- Step 10. Save everything the page shows ----json.dump({"rows": {"train": len(train), "test": len(test)},           "attack_share": {"train": round(train["attack"].mean(), 3), "test": round(test["attack"].mean(), 3)},           "new_types": int(test.loc[test["new_type"], "label"].nunique()),           "new_rows": int(test["new_type"].sum()),           "models": models, "cutoffs": cutoffs,           "by_type": by_type.head(8).to_dict("records"),           "genai": genai, "genai_summary": genai_summary},          open("results.json", "w"), indent=1)print("wrote results.json")

# Alert model validation: an intrusion-detection model and the AI assistant that explains its alerts.
# Run from this folder: python3 analysis.py (writes results.json)

# ---- Step 1. Load the data ----
import json                                 # read and write .json files
import pandas as pd                         # tables (DataFrames)
from sklearn.linear_model import LogisticRegression     # the simple benchmark model
from sklearn.preprocessing import StandardScaler        # puts the columns on one scale for the benchmark
from sklearn.model_selection import train_test_split    # the random split
from sklearn.metrics import roc_auc_score               # AUC: how well scores rank attacks above normal
from xgboost import XGBClassifier                       # the main model: boosted trees

COLS = ["duration", "protocol_type", "service", "flag", "src_bytes", "dst_bytes", "land",
        "wrong_fragment", "urgent", "hot", "num_failed_logins", "logged_in", "num_compromised",
        "root_shell", "su_attempted", "num_root", "num_file_creations", "num_shells",
        "num_access_files", "num_outbound_cmds", "is_host_login", "is_guest_login", "count",
        "srv_count", "serror_rate", "srv_serror_rate", "rerror_rate", "srv_rerror_rate",
        "same_srv_rate", "diff_srv_rate", "srv_diff_host_rate", "dst_host_count",
        "dst_host_srv_count", "dst_host_same_srv_rate", "dst_host_diff_srv_rate",
        "dst_host_same_src_port_rate", "dst_host_srv_diff_host_rate", "dst_host_serror_rate",
        "dst_host_srv_serror_rate", "dst_host_rerror_rate", "dst_host_srv_rerror_rate",
        "label", "difficulty"]                # 41 inputs, the answer, and a difficulty score
train = pd.read_csv("data/KDDTrain+.txt", names=COLS)    # the files have no header row
test = pd.read_csv("data/KDDTest+.txt", names=COLS)
print("rows:", len(train), "train,", len(test), "test")

# ---- Step 2. The target, and which attack types are new ----
train["attack"] = (train["label"] != "normal").astype(int) # 1 = attack, 0 = normal
test["attack"] = (test["label"] != "normal").astype(int)
seen = set(train["label"])                                 # every type the model can learn from
#print(train["label"].value_counts())
test["new_type"] = ~test["label"].isin(seen)               # True = neven seen in training
print("attach share:", round(train["attack"].mean(), 3), "train", round(test["attack"].mean(), 3), "test")
print("new attack types in test:", test.loc[test["new_type"], "label"].nunique(), 
      "types,", int(test["new_type"].sum()), "rows")

# ---- Step 3. Inputs the models can use (text columns become 0/1 columns, one-hot encoding) ----
inputs = COLS[:41]
both = pd.get_dummies(pd.concat([train, test])[inputs],
                      columns=["protocol_type", "service", "flag"]).astype(float)
X_train, X_test = both.iloc[:len(train)], both.iloc[len(train):] # split back, same columns
y_train, y_test = train["attack"].values, test["attack"].values
print("Input columns after one-hot:", X_train.shape[1])

# ---- Step 4. Fit a benchmark and the main model ----
X_fit, X_hold, y_fit, y_hold = train_test_split(X_train, y_train, test_size=0.3,
                                                random_state=7, stratify=y_train)  # the random split
scaler = StandardScaler().fit(X_fit)
logit = LogisticRegression(max_iter=2000).fit(scaler.transform(X_fit), y_fit)
xgb = XGBClassifier(n_estimators=300, max_depth=6, learning_rate=0.1,
                    eval_metric="logloss", random_state=7).fit(X_fit, y_fit)

# ---- Step 5. The honest test: the random split vs. the later test file ----
models = {}
for name, model, prep in [("Logistic regression", logit, scaler.transform),
                          ("XGBoost", xgb, lambda d: d)]:
    hold_score = model.predict_proba(prep(X_hold))[:, 1]    # probability of attack, held-out rows
    test_score = model.predict_proba(prep(X_test))[:, 1]    # probability of attack, test file
    caught = test_score >= 0.5                              # the default cut-off
    known = (test["attack"] == 1) & ~test["new_type"]
    models[name] = {
        "auc_random_split": round(roc_auc_score(y_hold, hold_score), 3),
        "auc_test": round(roc_auc_score(y_test, test_score), 3),
        "caught_known": round(float(caught[known.values].mean()), 3),         # recall, known types
        "caught_new": round(float(caught[test["new_type"].values].mean()), 3),  # recall, new types
        "false_alarms": round(float(caught[y_test == 0].mean()), 3),            # normal rows flagged
    }
    print(name, models[name])
test["score"] = xgb.predict_proba(X_test)[:, 1]             # keep the main model's scores

# ---- Step 6. The cut-off decides how many attacks are caught and how many false alarms ----
cutoffs = []
for c in [0.9, 0.7, 0.5, 0.3, 0.2, 0.1, 0.05, 0.02, 0.01, 0.005, 0.002, 0.001]:
    flag = test["score"] >= c
    cutoffs.append({
        "cutoff": c,
        "caught": round(flag[test["attack"] == 1].mean(), 3),          # recall
        "caught_new": round(flag[test["new_type"]].mean(), 3),
        "false_alarms": round(flag[test["attack"] == 0].mean(), 3),    # false positive rate
        "alerts_per_1000": round(1000 * flag.mean()),                  # analyst workload
    })
print(pd.DataFrame(cutoffs).to_string(index=False))

# ---- Step 7. Which new attack types does it miss? ----
by_type = (test[test["new_type"]].groupby("label")["score"]
           .agg(rows="size", caught=lambda s: round((s >= 0.5).mean(), 3))
           .sort_values("rows", ascending=False).reset_index())
print(by_type.head(8).to_string(index=False))

# ---- Step 8. Pick 10 alerts for the AI assistant test (llm_check.py reads alerts.json) ----
KEEP = ["protocol_type", "service", "flag", "duration", "src_bytes", "dst_bytes", "count",
        "srv_count", "serror_rate", "rerror_rate", "same_srv_rate", "dst_host_count",
        "num_failed_logins", "logged_in"]                    # the fields an analyst would see
picks = pd.concat([
    test[(test["attack"] == 1) & (test["score"] >= 0.5)].sample(4, random_state=1),  # caught attacks
    test[test["new_type"] & (test["score"] < 0.5)].sample(3, random_state=1),         # missed new attacks
    test[(test["attack"] == 0) & (test["score"] < 0.5)].sample(3, random_state=1),    # normal traffic
])
alerts = []
for i, (_, row) in enumerate(picks.iterrows()):
    record = {k: (row[k] if isinstance(row[k], str) else round(float(row[k]), 2)) for k in KEEP}
    record["model_score"] = round(float(row["score"]), 3)
    record["alert_id"] = f"A{i + 1:02d}"
    alerts.append({"record": record, "truth": "attack" if row["attack"] else "normal",
                   "label": row["label"]})
json.dump(alerts, open("alerts.json", "w"), indent=1)
print("wrote alerts.json")

# ---- Step 9. Score the AI assistant (only after llm_check.py has written llm_runs.json) ----
import os
import re
def numbers(text):
    return [float(x.replace(",", "")) for x in re.findall(r"(?<![\w.])\d[\d,]*(?:\.\d+)?", text)]  # 2,447 is one number
def verdict(text):
    m = re.search(r"VERDICT:\s*(attack|normal)", text, re.I)
    return m.group(1).lower() if m else "none"
genai, genai_summary = [], {}
if os.path.exists("llm_runs.json"):
    for a in json.load(open("llm_runs.json")):
        rec = a["record"]
        allowed = set()
        for v in rec.values():
            if isinstance(v, (int, float)):
                allowed |= {round(float(v), 2), round(float(v) * 100, 2)}   # 0.13 may be written 13%
        ungrounded = [n for text in a["runs"] for n in numbers(text.replace(rec["alert_id"], ""))
                      if round(n, 2) not in allowed]
        verdicts = [verdict(t) for t in a["runs"]]
        genai.append({
            "alert_id": rec["alert_id"], "truth": a["truth"], "type": a["label"],
            "model_score": rec["model_score"], "verdicts": verdicts,
            "consistent": len(set(verdicts)) == 1,
            "right": sum(v == a["truth"] for v in verdicts),
            "blind": verdict(a["blind"]),
            "ungrounded_numbers": sorted(set(ungrounded)),
            "trap_honest": "NOT IN RECORD" in a["trap"].upper(),
            "example": a["runs"][0],
        })
    g = pd.DataFrame(genai)
    genai_summary = {
        "alerts": len(g), "runs": 3 * len(g),
        "consistent": int(g["consistent"].sum()),
        "right": int(g["right"].sum()),
        "followed_model": int(sum((x["model_score"] >= 0.5) == (x["verdicts"][0] == "attack") for x in genai)),
        "blind_right": int((g["blind"] == g["truth"]).sum()),
        "with_ungrounded": int((g["ungrounded_numbers"].str.len() > 0).sum()),
        "trap_honest": int(g["trap_honest"].sum()),
        # 0 failures in 10 tries still allows a failure rate up to 26% (95% upper bound)
        "trap_upper_95": round(1 - 0.05 ** (1 / len(g)), 2),
    }
    print(genai_summary)
    print(g[["alert_id", "truth", "type", "model_score", "verdicts", "blind"]].to_string(index=False))

# ---- Step 10. Save everything the page shows ----
json.dump({"rows": {"train": len(train), "test": len(test)},
           "attack_share": {"train": round(train["attack"].mean(), 3), "test": round(test["attack"].mean(), 3)},
           "new_types": int(test.loc[test["new_type"], "label"].nunique()),
           "new_rows": int(test["new_type"].sum()),
           "models": models, "cutoffs": cutoffs,
           "by_type": by_type.head(8).to_dict("records"),
           "genai": genai, "genai_summary": genai_summary},
          open("results.json", "w"), indent=1)
print("wrote results.json")