Spaces:
Sleeping
Sleeping
| import gradio as gr | |
| import pandas as pd | |
| import numpy as np | |
| import joblib | |
| import datetime | |
| import json | |
| import os | |
| import matplotlib | |
| matplotlib.use("Agg") | |
| import matplotlib.pyplot as plt | |
| import matplotlib.patches as mpatches | |
| from matplotlib.gridspec import GridSpec | |
| # ββ Load artifacts βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| dt_model = joblib.load("models/decision_tree_model.pkl") | |
| lr_model = joblib.load("models/logistic_regression_model.pkl") | |
| svm_model = joblib.load("models/svm_model.pkl") | |
| scaler = joblib.load("models/scaler.pkl") | |
| features = joblib.load("models/features.pkl") | |
| MODELS = { | |
| "π³ Decision Tree": (dt_model, False), | |
| "π Logistic Regression": (lr_model, True), | |
| "β‘ SVM (RBF Kernel)": (svm_model, True), | |
| } | |
| # Load pre-computed training metrics if available | |
| TRAIN_METRICS = {} | |
| _mp = "models/metrics_summary.json" | |
| if os.path.exists(_mp): | |
| with open(_mp) as f: | |
| TRAIN_METRICS = json.load(f) | |
| # ββ Session state ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| session_log = [] | |
| total_scanned = 0 | |
| total_attacks = 0 | |
| attack_types = {"DoS": 0, "Probe": 0, "R2L": 0, "U2R": 0, "Normal": 0} | |
| # ββ Feature metadata βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| FEATURE_INFO = { | |
| "serror_rate": "SYN error rate β high = DoS/SYN-flood", | |
| "srv_serror_rate": "SYN error rate for same service", | |
| "dst_host_serror_rate": "SYN error rate at destination host", | |
| "dst_host_srv_serror_rate": "SYN error rate for dest-host service", | |
| "same_srv_rate": "% connections to same service", | |
| "diff_srv_rate": "% connections to diff services (scan indicator)", | |
| "dst_host_same_srv_rate": "Rate of same-service connections at dest host", | |
| "dst_host_srv_count": "# connections to same service on dest host", | |
| "count": "# connections to same host (last 2 s)", | |
| "srv_count": "# connections to same service (last 2 s)", | |
| "dst_host_count": "# connections to dest host", | |
| "logged_in": "1 = login successful, 0 = not logged in", | |
| "flag_sf": "SF = normal successful connection", | |
| "flag_s0": "S0 = incomplete connection (suspicious)", | |
| "service_http": "1 = HTTP/web service traffic", | |
| "src_bytes": "Bytes sent from source to destination", | |
| "dst_bytes": "Bytes sent from destination to source", | |
| "duration": "Connection duration in seconds", | |
| } | |
| ATTACK_TIPS = { | |
| "CRITICAL": "β‘ Immediate action β block source IP and alert SOC team.", | |
| "HIGH": "π΄ High-risk β investigate source, log for forensic review.", | |
| "MEDIUM": "π‘ Suspicious pattern β monitor closely, review connection logs.", | |
| "LOW": "π’ Low-confidence β continue passive monitoring.", | |
| } | |
| # ββ Colors βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| DARK_BG = "#0a0e1a" | |
| PANEL_BG = "#0d1526" | |
| CARD_BG = "#111d35" | |
| CYAN = "#00d4ff" | |
| RED = "#ff3c6e" | |
| GREEN = "#39ff14" | |
| YELLOW = "#f5a623" | |
| PURPLE = "#c084fc" | |
| TEXT = "#c8e6ff" | |
| GRID_COL = "#1e3a5a" | |
| # ββ Attack type inference ββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| def infer_attack_type(data: dict) -> tuple: | |
| serror = float(data.get("serror_rate", 0)) | |
| srv_serr = float(data.get("srv_serror_rate", 0)) | |
| dh_serr = float(data.get("dst_host_serror_rate", 0)) | |
| diff_srv = float(data.get("diff_srv_rate", 0)) | |
| cnt = float(data.get("count", 0)) | |
| srv_cnt = float(data.get("srv_count", 0)) | |
| logged = float(data.get("logged_in", 0)) | |
| src_b = float(data.get("src_bytes", 0)) | |
| dst_b = float(data.get("dst_bytes", 0)) | |
| flag_s0 = float(data.get("flag_s0", 0)) | |
| if (serror > 0.5 or srv_serr > 0.5 or dh_serr > 0.5 or flag_s0 == 1) and cnt > 50: | |
| return ("DoS", | |
| "High SYN/connection error rate with large connection count β " | |
| "classic Denial-of-Service pattern (neptune, smurf, pod).") | |
| if diff_srv > 0.5 and cnt > 30 and serror < 0.3: | |
| return ("Probe", | |
| "High proportion of connections to different services β " | |
| "network scanning / probing detected (portsweep, nmap).") | |
| if logged == 1 and src_b > 0 and dst_b < src_b * 0.1 and cnt < 10: | |
| return ("R2L", | |
| "Authenticated session with unusual byte asymmetry β " | |
| "possible remote-to-local exploit (ftp_write, guess_passwd).") | |
| if logged == 1 and cnt < 5 and srv_cnt < 5 and src_b < 500: | |
| return ("U2R", | |
| "Very low traffic volume with successful login β " | |
| "possible privilege escalation (buffer_overflow).") | |
| return ("Unknown Attack", | |
| "Does not clearly match DoS, Probe, R2L, or U2R β " | |
| "could be a novel or combined attack vector.") | |
| # ββ Chart helpers ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| def _dark(fig, axes): | |
| fig.patch.set_facecolor(DARK_BG) | |
| for ax in axes: | |
| ax.set_facecolor(PANEL_BG) | |
| ax.tick_params(colors=TEXT, labelsize=8) | |
| ax.xaxis.label.set_color(TEXT) | |
| ax.yaxis.label.set_color(TEXT) | |
| ax.title.set_color(CYAN) | |
| for sp in ax.spines.values(): | |
| sp.set_edgecolor(GRID_COL) | |
| ax.grid(color=GRID_COL, linewidth=0.5, alpha=0.6) | |
| def radar_chart(values, feat_names, title): | |
| N = len(values) | |
| angles = np.linspace(0, 2 * np.pi, N, endpoint=False).tolist() | |
| vals = values + [values[0]] | |
| angles += [angles[0]] | |
| fig, ax = plt.subplots(figsize=(4.5, 4.5), subplot_kw=dict(polar=True)) | |
| fig.patch.set_facecolor(DARK_BG) | |
| ax.set_facecolor(PANEL_BG) | |
| ax.plot(angles, vals, color=CYAN, linewidth=2) | |
| ax.fill(angles, vals, color=CYAN, alpha=0.18) | |
| ax.set_xticks(angles[:-1]) | |
| short = [f.replace("dst_host_", "dh_").replace("serror", "serr") | |
| .replace("_rate", "_r") for f in feat_names] | |
| ax.set_xticklabels(short, color=TEXT, size=7) | |
| ax.set_yticklabels([], color=TEXT) | |
| ax.tick_params(colors=TEXT) | |
| ax.spines["polar"].set_color(GRID_COL) | |
| ax.grid(color=GRID_COL, linewidth=0.5) | |
| ax.set_title(title, color=CYAN, pad=14, fontsize=10, fontweight="bold") | |
| plt.tight_layout() | |
| return fig | |
| def confidence_chart(vote_results): | |
| names = list(vote_results.keys()) | |
| confs = [vote_results[n]["confidence"] for n in names] | |
| colors = [RED if vote_results[n]["is_attack"] else GREEN for n in names] | |
| fig, ax = plt.subplots(figsize=(5, 2.6)) | |
| bars = ax.barh(names, confs, color=colors, height=0.45, edgecolor=GRID_COL) | |
| ax.set_xlim(0, 110) | |
| ax.set_xlabel("Confidence (%)") | |
| ax.set_title("Model Confidence Comparison", fontsize=10, fontweight="bold") | |
| for bar, val in zip(bars, confs): | |
| ax.text(val + 1, bar.get_y() + bar.get_height() / 2, | |
| f"{val:.1f}%", va="center", color=TEXT, fontsize=9) | |
| _dark(fig, [ax]) | |
| plt.tight_layout() | |
| return fig | |
| def session_chart(): | |
| fig = plt.figure(figsize=(10, 4)) | |
| gs = GridSpec(1, 2, figure=fig, wspace=0.38) | |
| # Donut: attack type breakdown | |
| ax1 = fig.add_subplot(gs[0]) | |
| keys = [k for k, v in attack_types.items() if v > 0] | |
| vals = [attack_types[k] for k in keys] | |
| if vals: | |
| pal = [RED, YELLOW, PURPLE, CYAN, GREEN] | |
| colors = pal[:len(keys)] | |
| wedges, texts, autotexts = ax1.pie( | |
| vals, labels=keys, autopct="%1.0f%%", colors=colors, | |
| startangle=90, | |
| wedgeprops=dict(width=0.55, edgecolor=DARK_BG, linewidth=1.5), | |
| textprops=dict(color=TEXT, fontsize=8)) | |
| for at in autotexts: | |
| at.set_color(DARK_BG); at.set_fontsize(7) | |
| else: | |
| ax1.text(0.5, 0.5, "No scans yet", ha="center", va="center", | |
| color=TEXT, fontsize=9, transform=ax1.transAxes) | |
| ax1.set_title("Traffic Classification", color=CYAN, fontsize=10, fontweight="bold") | |
| ax1.set_facecolor(PANEL_BG) | |
| fig.patch.set_facecolor(DARK_BG) | |
| # Bar: scan history | |
| ax2 = fig.add_subplot(gs[1]) | |
| recent = session_log[-15:] | |
| if recent: | |
| idxs = list(range(1, len(recent) + 1)) | |
| clrs = [RED if e["result"] == "ATTACK" else GREEN for e in recent] | |
| confs = [float(e["confidence"].rstrip("%")) for e in recent] | |
| ax2.bar(idxs, confs, color=clrs, edgecolor=DARK_BG, linewidth=0.8) | |
| ax2.set_ylim(0, 108) | |
| ax2.set_xlabel("Scan #"); ax2.set_ylabel("Confidence %") | |
| ax2.set_title("Scan History (last 15)", fontsize=10, fontweight="bold") | |
| ax2.legend(handles=[ | |
| mpatches.Patch(color=RED, label="Attack"), | |
| mpatches.Patch(color=GREEN, label="Normal")], | |
| fontsize=7, facecolor=CARD_BG, edgecolor=GRID_COL, labelcolor=TEXT) | |
| else: | |
| ax2.text(0.5, 0.5, "No scans yet", ha="center", va="center", | |
| color=TEXT, fontsize=9, transform=ax2.transAxes) | |
| ax2.set_title("Scan History (last 15)", fontsize=10, fontweight="bold") | |
| _dark(fig, [ax2]) | |
| ax2.set_facecolor(PANEL_BG) | |
| plt.tight_layout() | |
| return fig | |
| def metrics_chart(): | |
| if not TRAIN_METRICS: | |
| fig, ax = plt.subplots(figsize=(7, 3)) | |
| ax.text(0.5, 0.5, "Run train_models.py first to generate metrics_summary.json", | |
| ha="center", va="center", color=TEXT, fontsize=9, | |
| transform=ax.transAxes, wrap=True) | |
| _dark(fig, [ax]) | |
| return fig | |
| keys = ["accuracy", "precision", "recall", "f1", "roc_auc"] | |
| labels = ["Accuracy", "Precision", "Recall", "F1", "ROC-AUC"] | |
| mnames = list(TRAIN_METRICS.keys()) | |
| palette = [CYAN, YELLOW, RED] | |
| x = np.arange(len(labels)); w = 0.22 | |
| fig, ax = plt.subplots(figsize=(8.5, 4)) | |
| for i, (mname, color) in enumerate(zip(mnames, palette)): | |
| vals = [TRAIN_METRICS[mname].get(k, 0) for k in keys] | |
| bars = ax.bar(x + i * w, vals, w, label=mname, | |
| color=color, edgecolor=DARK_BG, linewidth=0.8, alpha=0.88) | |
| for bar, val in zip(bars, vals): | |
| ax.text(bar.get_x() + bar.get_width() / 2, | |
| bar.get_height() + 0.008, f"{val:.3f}", | |
| ha="center", va="bottom", color=TEXT, fontsize=6.5) | |
| ax.set_xticks(x + w); ax.set_xticklabels(labels) | |
| ax.set_ylim(0, 1.14); ax.set_ylabel("Score") | |
| ax.set_title("Model Performance Comparison (Training Evaluation)", | |
| fontsize=11, fontweight="bold") | |
| ax.legend(facecolor=CARD_BG, edgecolor=GRID_COL, labelcolor=TEXT, fontsize=8) | |
| _dark(fig, [ax]) | |
| plt.tight_layout() | |
| return fig | |
| # ββ Core prediction ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| def predict(selected_model_name, *args): | |
| global total_scanned, total_attacks | |
| data = dict(zip(features, args)) | |
| df_in = pd.DataFrame([data])[features] | |
| model, needs_scale = MODELS[selected_model_name] | |
| X = scaler.transform(df_in) if needs_scale else df_in.values | |
| pred = model.predict(X)[0] | |
| proba = model.predict_proba(X)[0] | |
| prob_attack = proba[1] | |
| prob_normal = proba[0] | |
| is_attack = pred == 1 | |
| confidence = prob_attack * 100 if is_attack else prob_normal * 100 | |
| severity = ("NONE" if not is_attack else | |
| "CRITICAL" if prob_attack >= 0.90 else | |
| "HIGH" if prob_attack >= 0.70 else | |
| "MEDIUM" if prob_attack >= 0.50 else "LOW") | |
| attack_type, attack_explanation = ( | |
| infer_attack_type(data) if is_attack else | |
| ("Normal", "Traffic behaves within expected norms.") | |
| ) | |
| # Feature importance / weight | |
| if hasattr(model, "feature_importances_"): | |
| imps = model.feature_importances_ | |
| elif hasattr(model, "coef_"): | |
| imps = np.abs(model.coef_[0]) | |
| else: | |
| imps = np.ones(len(features)) | |
| contribs = sorted(zip(features, imps, list(args)), | |
| key=lambda x: abs(x[1]), reverse=True)[:3] | |
| top3_text = "\n".join( | |
| f" β’ {f:<36} val={v:.3f} wt={w:.4f}" | |
| for f, w, v in contribs) | |
| # All-model vote | |
| vote_results = {} | |
| for mname, (m, scaled) in MODELS.items(): | |
| Xv = scaler.transform(df_in) if scaled else df_in.values | |
| p = m.predict(Xv)[0] | |
| pr = m.predict_proba(Xv)[0] | |
| atk = p == 1 | |
| cf = pr[1] * 100 if atk else pr[0] * 100 | |
| short = mname.split(" ", 1)[1].split("(")[0].strip() | |
| vote_results[short] = {"is_attack": atk, "confidence": cf} | |
| # Session update | |
| total_scanned += 1 | |
| if is_attack: | |
| total_attacks += 1 | |
| attack_types[attack_type] = attack_types.get(attack_type, 0) + 1 | |
| else: | |
| attack_types["Normal"] += 1 | |
| ts = datetime.datetime.now().strftime("%H:%M:%S") | |
| session_log.append({ | |
| "time": ts, | |
| "result": "ATTACK" if is_attack else "NORMAL", | |
| "severity": severity, | |
| "confidence": f"{confidence:.1f}%", | |
| "type": attack_type, | |
| }) | |
| # Format result | |
| border = "β" * 54 if is_attack else "β" * 54 | |
| cb = "β" * int(confidence / 5) + "β" * (20 - int(confidence / 5)) | |
| tip = ATTACK_TIPS.get(severity, "") | |
| status = f"π¨ ATTACK DETECTED Β· {severity}" if is_attack else "β NORMAL TRAFFIC" | |
| vote_lines = "\n".join( | |
| f" {'βοΈ' if v['is_attack'] else 'β '} {n:<28} {v['confidence']:.1f}%" | |
| for n, v in vote_results.items()) | |
| result_text = ( | |
| f"{border}\n {status}\n{border}\n\n" | |
| f" Confidence : [{cb}] {confidence:.2f}%\n" | |
| f" Model Used : {selected_model_name}\n" | |
| f" Timestamp : {ts}\n\n" | |
| ) | |
| if is_attack: | |
| result_text += ( | |
| f" Attack Type : {attack_type}\n" | |
| f" Explanation : {attack_explanation}\n\n" | |
| f" Severity : {severity}\n" | |
| f" Advice : {tip}\n\n" | |
| ) | |
| result_text += ( | |
| f" Top Contributing Features:\n{top3_text}\n\n" | |
| f" ββ All-Model Consensus βββββββββββββββββββββββββββββ\n" | |
| f"{vote_lines}\n{border}" | |
| ) | |
| # Stats | |
| rate = (total_attacks / total_scanned * 100) if total_scanned else 0 | |
| stats_text = ( | |
| f"π SESSION STATISTICS\n{'β'*32}\n" | |
| f" Total Scanned : {total_scanned}\n" | |
| f" Attacks Found : {total_attacks}\n" | |
| f" Normal Traffic : {total_scanned - total_attacks}\n" | |
| f" Attack Rate : {rate:.1f}%\n\n" | |
| f" Attack Types Seen:\n" + | |
| "".join(f" {k:<18} {v}\n" for k, v in attack_types.items() if v > 0) | |
| ) | |
| # History | |
| recent = session_log[-8:][::-1] | |
| hist = ["π RECENT PREDICTIONS\n" + "β" * 46] + [ | |
| f" {'π΄' if e['result']=='ATTACK' else 'π’'} {e['time']} " | |
| f"{e['result']:<7} {e.get('type','β'):<18} {e['confidence']}" | |
| for e in recent] | |
| history_text = "\n".join(hist) | |
| # Normalise feature values β [0,1] for radar | |
| norm = [] | |
| for feat, val in zip(features, args): | |
| fv = float(val) | |
| if "rate" in feat or feat.startswith("flag_") or feat in ["logged_in","service_http"]: | |
| norm.append(min(fv, 1.0)) | |
| elif "count" in feat: | |
| norm.append(min(fv / 255.0, 1.0)) | |
| else: | |
| norm.append(min(fv / max(fv, 10000.0), 1.0)) | |
| return (result_text, stats_text, history_text, | |
| radar_chart(norm, features, "Input Feature Profile"), | |
| confidence_chart(vote_results), | |
| session_chart()) | |
| def reset_session(): | |
| global session_log, total_scanned, total_attacks, attack_types | |
| session_log = [] | |
| total_scanned = 0 | |
| total_attacks = 0 | |
| attack_types = {"DoS": 0, "Probe": 0, "R2L": 0, "U2R": 0, "Normal": 0} | |
| return ( | |
| "β"*54 + "\n Session cleared. Ready for new scan.\n" + "β"*54, | |
| "π SESSION STATISTICS\n" + "β"*32 + "\n No data yet.", | |
| "π RECENT PREDICTIONS\n" + "β"*46 + "\n No predictions yet.", | |
| None, None, session_chart() | |
| ) | |
| # ββ Build input widgets ββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| inputs = [] | |
| for feature in features: | |
| info = FEATURE_INFO.get(feature, "Network traffic feature") | |
| if "rate" in feature: | |
| inputs.append(gr.Slider(0, 1, value=0, step=0.01, label=feature, info=info)) | |
| elif feature.startswith("flag_") or feature in ["logged_in", "service_http"]: | |
| inputs.append(gr.Radio([0, 1], value=0, label=feature, info=info)) | |
| elif "count" in feature: | |
| inputs.append(gr.Slider(0, 255, value=0, step=1, label=feature, info=info)) | |
| else: | |
| inputs.append(gr.Number(value=0, label=feature, info=info)) | |
| flag_inputs, rate_inputs, count_inputs, other_inputs = [], [], [], [] | |
| for i, feature in enumerate(features): | |
| if feature.startswith("flag_") or feature in ["logged_in", "service_http"]: | |
| flag_inputs.append((i, inputs[i])) | |
| elif "rate" in feature: | |
| rate_inputs.append((i, inputs[i])) | |
| elif "count" in feature: | |
| count_inputs.append((i, inputs[i])) | |
| else: | |
| other_inputs.append((i, inputs[i])) | |
| all_inputs = [inp for _, inp in flag_inputs + rate_inputs + count_inputs + other_inputs] | |
| # ββ CSS ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| CSS = """ | |
| @import url('https://fonts.googleapis.com/css2?family=Share+Tech+Mono&family=Rajdhani:wght@400;600;700&display=swap'); | |
| :root{--bg:#0a0e1a;--panel:#0d1526;--card:#111d35;--cyan:#00d4ff;--red:#ff3c6e; | |
| --green:#39ff14;--yellow:#f5a623;--text:#c8e6ff;--muted:#5a8aaa;--border:#1e3a5a; | |
| --glow:0 0 14px rgba(0,212,255,0.45);} | |
| body,.gradio-container{background:var(--bg)!important;font-family:'Rajdhani',sans-serif!important;color:var(--text)!important;} | |
| .gradio-container h1{font-family:'Rajdhani',sans-serif!important;font-weight:700!important;font-size:2rem!important; | |
| color:var(--cyan)!important;text-shadow:var(--glow)!important;letter-spacing:2px!important;} | |
| .gr-block,.gr-box,.gradio-group,.gr-form,div[data-testid="block"]{ | |
| background:var(--panel)!important;border:1px solid var(--border)!important;border-radius:8px!important;} | |
| label span,.gr-label,.label-wrap span{font-family:'Share Tech Mono',monospace!important; | |
| font-size:0.73rem!important;color:var(--cyan)!important;letter-spacing:1px!important;text-transform:uppercase!important;} | |
| .gr-info{color:var(--muted)!important;font-size:0.68rem!important;} | |
| input[type=range]{accent-color:var(--cyan)!important;} | |
| input[type=number]{background:var(--card)!important;border:1px solid var(--border)!important; | |
| color:var(--cyan)!important;font-family:'Share Tech Mono',monospace!important;border-radius:4px!important;} | |
| textarea{background:var(--card)!important;border:1px solid var(--border)!important; | |
| color:var(--green)!important;font-family:'Share Tech Mono',monospace!important; | |
| font-size:0.8rem!important;line-height:1.65!important;border-radius:6px!important;} | |
| button.primary{background:linear-gradient(135deg,#003c6e,#006aaa)!important; | |
| border:1px solid var(--cyan)!important;color:var(--cyan)!important; | |
| font-family:'Rajdhani',sans-serif!important;font-weight:700!important; | |
| font-size:1.05rem!important;letter-spacing:3px!important;text-transform:uppercase!important; | |
| border-radius:6px!important;box-shadow:var(--glow)!important;} | |
| button.primary:hover{background:linear-gradient(135deg,#005090,#0088cc)!important; | |
| box-shadow:0 0 22px rgba(0,212,255,0.7)!important;} | |
| button.secondary{background:#1a0a14!important;border:1px solid var(--red)!important; | |
| color:var(--red)!important;font-family:'Rajdhani',sans-serif!important; | |
| font-weight:600!important;letter-spacing:2px!important;border-radius:6px!important;} | |
| .tab-nav button{font-family:'Rajdhani',sans-serif!important;font-weight:600!important; | |
| color:var(--muted)!important;background:var(--panel)!important; | |
| border:1px solid var(--border)!important;letter-spacing:1px!important;} | |
| .tab-nav button.selected{color:var(--cyan)!important; | |
| border-bottom:2px solid var(--cyan)!important;box-shadow:var(--glow)!important;} | |
| select,select *{background:var(--card)!important;border:1px solid var(--border)!important; | |
| color:var(--cyan)!important;font-family:'Share Tech Mono',monospace!important;} | |
| .gr-accordion summary{color:var(--cyan)!important;font-family:'Rajdhani',sans-serif!important; | |
| font-weight:600!important;letter-spacing:1px!important;} | |
| ::-webkit-scrollbar{width:5px;}::-webkit-scrollbar-track{background:var(--bg);} | |
| ::-webkit-scrollbar-thumb{background:var(--border);border-radius:3px;} | |
| """ | |
| # ββ Interface ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| with gr.Blocks(css=CSS, title="π‘οΈ IDS", | |
| theme=gr.themes.Base(primary_hue="cyan", secondary_hue="pink", | |
| neutral_hue="slate")) as app: | |
| gr.HTML(""" | |
| <div style="text-align:center;padding:18px 0 6px;"> | |
| <div style="font-size:2.8rem;line-height:1;">π‘οΈ</div> | |
| <h1 style="font-family:'Rajdhani',sans-serif;font-size:2rem;color:#00d4ff; | |
| letter-spacing:3px;margin:8px 0 4px; | |
| text-shadow:0 0 16px rgba(0,212,255,0.6);"> | |
| INTRUSION DETECTION SYSTEM | |
| </h1> | |
| <p style="font-family:'Share Tech Mono',monospace;color:#5a8aaa; | |
| font-size:0.74rem;letter-spacing:2px;margin:0;"> | |
| DECISION TREE Β· LOGISTIC REGRESSION Β· SVM Β· NSL-KDD Β· CHI-SQUARE FEATURES | |
| </p> | |
| <div style="height:2px;background:linear-gradient(90deg,transparent,#00d4ff,transparent); | |
| margin:12px auto;width:55%;"></div> | |
| </div>""") | |
| with gr.Tabs(): | |
| # ββ Tab 1: Live Scanner βββββββββββββββββββββββββββββββββββββββββββββββ | |
| with gr.Tab("π LIVE SCANNER"): | |
| model_selector = gr.Dropdown( | |
| choices=list(MODELS.keys()), value=list(MODELS.keys())[0], | |
| label="SELECT MODEL", | |
| info="Choose which trained model performs the classification") | |
| with gr.Row(): | |
| with gr.Column(scale=3): | |
| gr.HTML('<p style="font-family:\'Share Tech Mono\',monospace;color:#5a8aaa;' | |
| 'font-size:0.7rem;letter-spacing:1px;margin-bottom:6px;">' | |
| 'βΈ CONFIGURE NETWORK TRAFFIC PARAMETERS</p>') | |
| with gr.Accordion("β FLAG & BINARY FEATURES", open=True): | |
| for _, inp in flag_inputs: inp.render() | |
| with gr.Accordion("π RATE FEATURES", open=True): | |
| for _, inp in rate_inputs: inp.render() | |
| with gr.Accordion("π’ COUNT FEATURES", open=False): | |
| for _, inp in count_inputs: inp.render() | |
| if other_inputs: | |
| with gr.Accordion("π§ OTHER FEATURES", open=False): | |
| for _, inp in other_inputs: inp.render() | |
| with gr.Column(scale=2): | |
| gr.HTML('<p style="font-family:\'Share Tech Mono\',monospace;color:#5a8aaa;' | |
| 'font-size:0.7rem;letter-spacing:1px;margin-bottom:6px;">' | |
| 'βΈ ANALYSIS OUTPUT</p>') | |
| result_out = gr.Textbox(label="π DETECTION RESULT", lines=18, interactive=False) | |
| stats_out = gr.Textbox(label="π SESSION STATS", lines=9, interactive=False) | |
| history_out = gr.Textbox(label="π SCAN HISTORY", lines=10, interactive=False) | |
| with gr.Row(): | |
| scan_btn = gr.Button("β‘ SCAN TRAFFIC", variant="primary") | |
| reset_btn = gr.Button("π RESET SESSION", variant="secondary") | |
| gr.HTML('<div style="height:1px;background:#1e3a5a;margin:18px 0 10px;"></div>' | |
| '<p style="font-family:\'Share Tech Mono\',monospace;color:#5a8aaa;' | |
| 'font-size:0.7rem;letter-spacing:1px;margin-bottom:6px;">βΈ VISUAL ANALYSIS</p>') | |
| with gr.Row(): | |
| radar_plot = gr.Plot(label="Feature Profile (Radar)") | |
| conf_plot = gr.Plot(label="Model Confidence Comparison") | |
| session_plot = gr.Plot(label="Session Dashboard") | |
| scan_btn.click(fn=predict, | |
| inputs=[model_selector] + all_inputs, | |
| outputs=[result_out, stats_out, history_out, | |
| radar_plot, conf_plot, session_plot]) | |
| reset_btn.click(fn=reset_session, inputs=[], | |
| outputs=[result_out, stats_out, history_out, | |
| radar_plot, conf_plot, session_plot]) | |
| # ββ Tab 2: Model Comparison ββββββββββββββββββββββββββββββββββββββββββββ | |
| with gr.Tab("π MODEL COMPARISON"): | |
| gr.HTML('<div style="font-family:\'Share Tech Mono\',monospace;color:#5a8aaa;' | |
| 'font-size:0.72rem;letter-spacing:1px;padding:8px 0 14px;">' | |
| 'βΈ TRAINING PERFORMANCE METRICS ACROSS ALL THREE MODELS</div>') | |
| metrics_plot = gr.Plot(label="Model Metrics") | |
| gr.Button("π LOAD METRICS CHART", variant="primary").click( | |
| fn=metrics_chart, inputs=[], outputs=[metrics_plot]) | |
| if TRAIN_METRICS: | |
| rows = [{"Model": m, | |
| "Accuracy": f"{v['accuracy']:.4f}", | |
| "Precision": f"{v['precision']:.4f}", | |
| "Recall": f"{v['recall']:.4f}", | |
| "F1": f"{v['f1']:.4f}", | |
| "ROC-AUC": f"{v['roc_auc']:.4f}"} | |
| for m, v in TRAIN_METRICS.items()] | |
| gr.Dataframe(pd.DataFrame(rows), label="Metrics Table", interactive=False) | |
| # ββ Tab 3: Preset Scenarios ββββββββββββββββββββββββββββββββββββββββββββ | |
| with gr.Tab("π― PRESET SCENARIOS"): | |
| gr.HTML('<div style="font-family:\'Share Tech Mono\',monospace;color:#5a8aaa;' | |
| 'font-size:0.72rem;letter-spacing:1px;padding:8px 0 14px;">' | |
| 'βΈ LOAD A KNOWN SCENARIO β SEE EXPECTED VALUES, THEN TEST IN SCANNER</div>') | |
| scenario_out = gr.Textbox(label="Scenario Description", lines=18, interactive=False) | |
| def make_scenario(name, expected, rules): | |
| vals = [] | |
| for f in features: | |
| matched = any(k in f and (vals.append(v) or True) | |
| for k, v in rules.items()) | |
| if not matched: | |
| vals.append( | |
| 1 if f in ["logged_in","flag_sf","service_http"] else 0) | |
| lines = [f"SCENARIO : {name}", f"EXPECTED : {expected}", "β"*44] | |
| lines += [f" {f:<40} = {v}" for f, v in zip(features, vals)] | |
| lines += ["β"*44, "β€ Set values in LIVE SCANNER tab and click SCAN."] | |
| return "\n".join(lines) | |
| with gr.Row(): | |
| gr.Button("π₯ DoS Attack").click( | |
| fn=lambda: make_scenario("Denial-of-Service (DoS)", | |
| "π¨ ATTACK β CRITICAL | Type: DoS", | |
| {"serror":0.95,"count":200,"srv_count":200, | |
| "flag_s0":1,"flag_sf":0,"same_srv":0.95, | |
| "diff_srv":0.05,"logged_in":0}), | |
| outputs=scenario_out) | |
| gr.Button("β Normal Session").click( | |
| fn=lambda: make_scenario("Normal HTTP Web Session", | |
| "β NORMAL TRAFFIC", | |
| {"serror":0.0,"count":5,"srv_count":5,"flag_sf":1, | |
| "flag_s0":0,"logged_in":1,"same_srv":0.95, | |
| "diff_srv":0.0,"service_http":1, | |
| "src_bytes":2000,"dst_bytes":8000}), | |
| outputs=scenario_out) | |
| gr.Button("π Port Scan").click( | |
| fn=lambda: make_scenario("Network Port Scan (Probe)", | |
| "π¨ ATTACK β MEDIUM/HIGH | Type: Probe", | |
| {"diff_srv":0.85,"same_srv":0.10,"count":120, | |
| "serror":0.1,"logged_in":0,"flag_sf":0}), | |
| outputs=scenario_out) | |
| gr.Button("π R2L / Brute-Force").click( | |
| fn=lambda: make_scenario("Remote-to-Local (R2L) Attempt", | |
| "π¨ ATTACK β HIGH | Type: R2L", | |
| {"logged_in":1,"src_bytes":500,"dst_bytes":20, | |
| "count":3,"serror":0.0,"flag_sf":1,"diff_srv":0.1}), | |
| outputs=scenario_out) | |
| # ββ Tab 4: Feature Reference βββββββββββββββββββββββββββββββββββββββββββ | |
| with gr.Tab("π FEATURE REFERENCE"): | |
| rows = [{"Feature": f, | |
| "Type": ("Binary 0/1" if f.startswith("flag_") or f in ["logged_in","service_http"] | |
| else "Rate [0β1]" if "rate" in f | |
| else "Count [0β255]" if "count" in f | |
| else "Numeric"), | |
| "Description": FEATURE_INFO.get(f, "Network traffic feature")} | |
| for f in features] | |
| gr.Dataframe(pd.DataFrame(rows), label="Selected Features", | |
| interactive=False, wrap=True) | |
| gr.HTML('<div style="font-family:\'Share Tech Mono\',monospace;color:#5a8aaa;' | |
| 'font-size:0.7rem;letter-spacing:1px;margin-top:14px;padding:10px 14px;' | |
| 'border:1px solid #1e3a5a;border-radius:6px;">' | |
| 'Pipeline: Pearson Correlation (top-25) β Chi-Square SelectKBest (final 12).<br>' | |
| 'Trained on NSL-KDD 20,000 rows Β· 80/20 split Β· class_weight=balanced</div>') | |
| # ββ Tab 5: About βββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| with gr.Tab("βΉοΈ ABOUT"): | |
| gr.HTML(""" | |
| <div style="font-family:'Rajdhani',sans-serif;max-width:720px; | |
| margin:0 auto;padding:20px 0;line-height:1.8;"> | |
| <h2 style="color:#00d4ff;letter-spacing:2px;border-bottom:1px solid #1e3a5a; | |
| padding-bottom:8px;">ML-Based Intrusion Detection System</h2> | |
| <p style="color:#c8e6ff;font-size:0.95rem;"> | |
| Classifies network connections as <b style="color:#39ff14;">Normal</b> or | |
| <b style="color:#ff3c6e;">Attack</b> using three ML models. Attack type is | |
| inferred via rule-based heuristics on top of binary classification. | |
| </p> | |
| <h3 style="color:#00d4ff;margin-top:18px;">Three Models</h3> | |
| <ul style="color:#c8e6ff;font-size:0.92rem;"> | |
| <li><b style="color:#00d4ff;">π³ Decision Tree</b> β Interpretable tree splits. | |
| max_depth=10, balanced weights.</li> | |
| <li><b style="color:#f5a623;">π Logistic Regression</b> β Linear probabilistic. | |
| Scaled input, lbfgs, max_iter=1000.</li> | |
| <li><b style="color:#ff3c6e;">β‘ SVM (RBF)</b> β Non-linear kernel SVM. | |
| C=1.0, gamma=scale, probability=True.</li> | |
| </ul> | |
| <h3 style="color:#00d4ff;margin-top:18px;">Attack Categories</h3> | |
| <ul style="color:#c8e6ff;font-size:0.92rem;"> | |
| <li><b style="color:#ff3c6e;">DoS</b> β High error rate + large count (neptune, smurf)</li> | |
| <li><b style="color:#f5a623;">Probe</b> β Many services scanned (portsweep, nmap)</li> | |
| <li><b style="color:#c084fc;">R2L</b> β Asymmetric bytes after login (ftp_write)</li> | |
| <li><b style="color:#00d4ff;">U2R</b> β Low-volume logged-in session (buffer_overflow)</li> | |
| </ul> | |
| <div style="margin-top:20px;padding:10px 14px;background:#0d1526; | |
| border:1px solid #1e3a5a;border-radius:6px; | |
| font-family:'Share Tech Mono',monospace;font-size:0.72rem; | |
| color:#5a8aaa;letter-spacing:1px;"> | |
| Dataset: Mireu-Lab/NSL-KDD (Hugging Face) | | |
| Models: sklearn | UI: Gradio | |
| </div> | |
| </div>""") | |
| app.launch() |