SentinelAI detects anomalous authentication behavior using behavioral profiling, hybrid AI detection, explainable machine learning, and threat intelligence to assist SOC analysts in identifying enterprise cyber threats.
", unsafe_allow_html=True) # --- Navigation Grouping --- st.sidebar.markdown("### 🧭 Navigation") page = st.sidebar.radio("Select View", [ "📊 Executive Security Overview", "🚨 Threat Hunting", "🔍 Deep Investigation", "🏛️ System Architecture" ], label_visibility="collapsed") page = page.split(" ", 1)[1] st.sidebar.markdown("---") st.sidebar.markdown("### ⚙️ Scenario Engine") scenario = st.sidebar.selectbox("Select Threat Scenario", [ "Mixed Enterprise Attack", "Normal Activity", "Brute Force Attack", "Impossible Travel", "Credential Stuffing", "Password Spray", "Insider Threat", "Suspicious Device" ]) if st.sidebar.button("🚀 Execute AI Pipeline"): run_id = str(uuid.uuid4())[:8].upper() start_time = time.time() with st.status("Executing Enterprise SOC Pipeline...", expanded=True) as status: st.write("✓ Generating Synthetic Dataset...") time.sleep(0.5) from src.runtime.orchestrator import PipelineRunner runner = PipelineRunner() config = {"scenario": scenario, "num_events": 2500} st.write("✓ Performing Feature Engineering...") st.write("✓ Building Behavioral Profiles...") runner.run_phase2(config_overrides=config) st.write("✓ Running Isolation Forest...") st.write("✓ Executing Rule Engine...") runner.run_phase3(config_overrides=config) st.write("✓ Calculating Risk Fusion...") st.write("✓ Classifying Threats (XGBoost)...") runner.run_phase4(config_overrides=config) st.write("✓ Generating SHAP Explanations...") runner.run_phase5(config_overrides=config) duration = time.time() - start_time st.write("✓ Dashboard Updated") status.update(label=f"Pipeline Execution Complete ({duration:.2f}s)!", state="complete", expanded=False) st.cache_data.clear() new_df = load_soc_telemetry() avg_risk = new_df["Risk Score"].mean() if not new_df.empty else 0 threat_cats = new_df[new_df["Attack Classification"] != "Normal Authentication"]["Attack Classification"].nunique() if not new_df.empty else 0 history_entry = { "Run ID": run_id, "Scenario": scenario, "Timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "Execution Time (s)": round(duration, 2), "Events Processed": len(new_df), "Critical Alerts": len(new_df[new_df["Risk Level"] == "Critical"]) if not new_df.empty else 0, "High Alerts": len(new_df[new_df["Risk Level"] == "High"]) if not new_df.empty else 0, "Average Risk Score": round(avg_risk, 2), "Top Threat": new_df[new_df["Attack Classification"] != "Normal Authentication"]["Attack Classification"].mode()[0] if not new_df[new_df["Attack Classification"] != "Normal Authentication"].empty else "None", "Status": "SUCCESS" } history = load_history() history.insert(0, history_entry) history = history[:5] save_history(history) st.session_state.pipeline_run = True st.session_state.run_summary = history_entry st.rerun() history = load_history() # Professional Pipeline Summary Card if st.session_state.run_summary: s = st.session_state.run_summary st.markdown(f"""Filter, prioritize, and triage high-risk alerts before deep investigation.
", unsafe_allow_html=True) if not df.empty: c1, c2, c3 = st.columns([2, 1, 1]) with c1: risk_filter = st.multiselect("Filter Risk Level", options=["Critical", "High", "Moderate", "Low"], default=["Critical", "High"]) with c3: st.write("") # Alignment hack export_csv = st.download_button( label="📥 Export Queue (CSV)", data=df.to_csv(index=False).encode('utf-8'), file_name=f"threat_queue_export_{datetime.now().strftime('%Y%m%d')}.csv", mime='text/csv' ) filtered_df = df[df["Risk Level"].isin(risk_filter)] if risk_filter else df def format_risk(x): if x == "Critical": return "🔴 Critical" if x == "High": return "🟠 High" if x == "Moderate": return "🟡 Moderate" return "🔵 Low" display_df = filtered_df[["event_id", "timestamp", "user_id", "device_id", "country", "Risk Level", "Risk Score", "Attack Classification", "authentication_method"]].copy() display_df["Severity"] = display_df["Risk Level"].apply(format_risk) display_df["Auth Type"] = display_df["authentication_method"] display_df = display_df.rename(columns={"Attack Classification": "Threat Context"}) st.dataframe( display_df[["event_id", "Severity", "Risk Score", "timestamp", "user_id", "country", "Auth Type", "Threat Context"]].head(200), use_container_width=True, hide_index=True, column_config={ "event_id": st.column_config.TextColumn("Event ID", width="small"), "Severity": st.column_config.TextColumn("Severity", width="small"), "Risk Score": st.column_config.ProgressColumn("Risk Score", format="%.2f", min_value=0, max_value=100, width="medium"), "timestamp": st.column_config.DatetimeColumn("Timestamp", format="DD-MMM-YYYY HH:mm"), "user_id": st.column_config.TextColumn("User ID"), "country": st.column_config.TextColumn("Country"), "Auth Type": st.column_config.TextColumn("Auth Type"), "Threat Context": st.column_config.TextColumn("Threat Context", width="medium") } ) else: st.warning("No alerts available. Please execute the AI pipeline.") # ----------------- DEEP INVESTIGATION ----------------- elif page == "Deep Investigation": st.markdown("{exp_dict.get("Executive_Summary", "No summary available.")}
No deterministic rules triggered for this event.
", unsafe_allow_html=True) except: st.write("Error parsing rules.") st.markdown("Understanding the multi-layered hybrid AI architecture.
", unsafe_allow_html=True) col_text, col_diagram = st.columns([1.2, 1]) with col_text: st.markdown(""" #### ⚙️ Component Responsibilities The engine processes data strictly sequentially to ensure maximum fidelity and explainability: * **Synthetic Data Generation:** Bootstraps realistic baseline enterprise logs and injects precise threat scenarios. * **Feature Engineering:** Extracts critical time-series and spatial characteristics (e.g. time since last login, distance). * **Behavior Profiling:** Builds continuous mathematical baselines for users and devices. * **Isolation Forest:** Unsupervised ML model identifying multi-dimensional spatial anomalies. * **Rule Engine:** Deterministic SOC heuristics that flag known enterprise signatures. * **Risk Fusion:** Normalizes inputs into an enterprise 0-100 risk score and fuses confidence. * **XGBoost:** Supervised learning categorizes the specific threat context (e.g. Credential Stuffing). * **SHAP (XAI):** Generates interpretable feature importance evidence for the SOC analyst. #### 🛠️ Technology Stack * **Frontend:** Streamlit, Plotly, Pandas * **AI/ML Layer:** XGBoost, Scikit-Learn (Isolation Forest), SHAP * **Data Layer:** PyArrow (Parquet), JSON * **Runtime Environment:** Python 3.10+ #### 📂 Folder Organization * `src/ai/`: Core Intelligence (Detection, Classification, XAI) * `src/data/`: Data Generation & Loaders * `src/runtime/`: Pipeline Orchestrators * `src/ui/`: Streamlit Dashboard * `data/`: Parquet storage layer """) with col_diagram: st.markdown("""