import logging import json import pandas as pd from typing import Dict, Any, List logger = logging.getLogger(__name__) class RuleExplanationEngine: """Module 2: Rule Explanation Engine""" def explain(self, triggered_rules: str, event_data: dict) -> List[dict]: if not triggered_rules or triggered_rules == "None" or pd.isna(triggered_rules): return [] explanations = [] rules = triggered_rules.split(",") for rule in rules: if rule == "ImpossibleTravel": explanations.append({ "Rule Name": "Impossible Travel", "Rule Description": "Login originated from geographically distant location", "Evidence": f"Country: {event_data.get('country', 'Unknown')}", "Severity": "High", "Status": "Triggered" }) elif rule == "VelocityAnomaly": explanations.append({ "Rule Name": "Velocity Anomaly", "Rule Description": "Abnormal volume of authentication requests in short window", "Evidence": f"Time since last: {event_data.get('time_since_last_login', 'Unknown')}s", "Severity": "Moderate", "Status": "Triggered" }) else: explanations.append({ "Rule Name": rule, "Rule Description": f"Triggered standard rule {rule}", "Evidence": "Detected during deterministic rule evaluation", "Severity": "Elevated", "Status": "Triggered" }) return explanations class BehaviorExplanationEngine: """Module 3: Behavior Explanation Engine""" def explain(self, behavior_score: float, event_data: dict) -> List[dict]: deviations = [] if event_data.get("is_working_hour") == 0: deviations.append({ "Expected": "Login within working hours (8AM-6PM)", "Observed": "Login outside normal working hours", "Deviation": "Temporal Anomaly", "Impact": "Increased Risk" }) if event_data.get("is_weekend") == 1: deviations.append({ "Expected": "Weekday authentication", "Observed": "Weekend authentication", "Deviation": "Temporal Anomaly", "Impact": "Increased Risk" }) if behavior_score > 50: deviations.append({ "Expected": "Normal profile execution", "Observed": "Significant divergence from learned profile", "Deviation": "Behavioral Shift", "Impact": "High Risk" }) return deviations class RiskExplanationEngine: """Module 4: Risk Explanation Engine""" def explain(self, risk_record: dict) -> dict: return { "Final Risk Score": float(risk_record.get("risk_score", 0.0)), "Rule Contribution": float(risk_record.get("rule_contribution", 0.0)), "Statistical Contribution": float(risk_record.get("stat_contribution", 0.0)), "ML Contribution": float(risk_record.get("ml_contribution", 0.0)), "Behavioral Contribution": float(risk_record.get("behavioral_contribution", 0.0)), "Device Trust Contribution": float(risk_record.get("device_contribution", 0.0)), "Reasoning": "Aggregated normalized risk driven by explicit sub-engines." } class NaturalLanguageGenerator: """Module 5: Natural Language Explanation""" def generate(self, evidence_package: dict) -> dict: risk_level = evidence_package.get("Risk Level", "Unknown") attack_category = evidence_package.get("Attack Classification", "Unknown") deviations = evidence_package.get("Behavior Deviations", []) rules = evidence_package.get("Triggered Rules", []) conf = evidence_package.get("Confidence", 0.0) # 1. Executive Summary exec_summary = f"The authentication event was classified as {risk_level} Risk. " if deviations: reasons = [d['Observed'].lower() for d in deviations] exec_summary += f"This was primarily because {', '.join(reasons)}. " if rules: r_names = [r['Rule Name'] for r in rules] exec_summary += f"Additionally, the following rules triggered: {', '.join(r_names)}. " if attack_category and attack_category != "Normal Authentication": exec_summary += f"The AI classifier predicted {attack_category}." else: exec_summary += "The activity was categorized as Normal Authentication." # 2. Technical Analysis features = evidence_package.get("Feature Contributions", []) feat_str = ", ".join([f"{f['feature']} ({f['contribution']:.2f})" for f in features]) if features else "None" stat_score = evidence_package.get("Risk Breakdown", {}).get("Statistical Contribution", 0.0) model_version = evidence_package.get("Audit Metadata", {}).get("Classification_Model_Version", "Unknown") feature_values = evidence_package.get("Feature Values", {}) tech_analysis = ( f"Model Version: {model_version}\n" f"Prediction Confidence: {conf*100:.2f}%\n" f"SHAP Contributions: {feat_str}\n" f"Triggered Rules: {', '.join([r['Rule Name'] for r in rules]) if rules else 'None'}\n" f"Statistical Deviation Score: {stat_score:.2f}\n" f"Input Feature Values: {feature_values}\n" f"Recommended Action: {evidence_package.get('Recommended Action', 'None')}" ) return { "Executive_Summary": exec_summary, "Technical_Analysis": tech_analysis }