import math from collections import Counter from datetime import datetime FEATURE_NAMES = [ "window_event_count", # Total events in the window "failure_ratio", # Ratio of failed events (0.0 to 1.0) "distinct_paths", # Number of unique target paths "distinct_users", # Number of unique usernames "path_entropy", # Character Shannon entropy of requested paths "inter_event_time_mean_ms", # Mean inter-event time in milliseconds "hour_of_day_norm", # Normalized hour of day (0.0 to 1.0) "sensitive_path_ratio", # Ratio of paths containing sensitive patterns ] SENSITIVE_KEYWORDS = [ ".env", ".git", "wp-config", "phpmyadmin", "actuator", "boaform", "shell", "admin", "sql", "password", "etc/passwd" ] def calculate_entropy(text: str) -> float: """Calculates Shannon entropy of string.""" if not text: return 0.0 counts = Counter(text) length = len(text) return -sum((count / length) * math.log2(count / length) for count in counts.values()) def extract_features(events_window: list) -> list: """Extracts sliding window feature vector from a sequence of events for a single IP.""" if not events_window: return [0.0] * len(FEATURE_NAMES) count = len(events_window) failures = sum(1 for e in events_window if e.get("result") in ["failed", "failed_password", "failed_publickey", "invalid_user", "not_found", "error", "blocked"]) failure_ratio = failures / count if count > 0 else 0.0 paths = set(e.get("target", "") for e in events_window if e.get("target")) distinct_paths = len(paths) users = set(e.get("user", "") for e in events_window if e.get("user")) distinct_users = len(users) all_paths_str = "".join(paths) path_entropy = calculate_entropy(all_paths_str) # Inter-event timings inter_event_ms = [] sorted_events = sorted(events_window, key=lambda e: e.get("timestamp", "")) for i in range(1, len(sorted_events)): t1 = datetime.fromisoformat(sorted_events[i - 1]["timestamp"].replace("Z", "+00:00")) t2 = datetime.fromisoformat(sorted_events[i]["timestamp"].replace("Z", "+00:00")) diff = max(0.0, (t2 - t1).total_seconds() * 1000.0) inter_event_ms.append(diff) mean_inter_event = sum(inter_event_ms) / len(inter_event_ms) if inter_event_ms else 1000.0 last_ts = datetime.fromisoformat(sorted_events[-1]["timestamp"].replace("Z", "+00:00")) hour_norm = last_ts.hour / 24.0 # Sensitive path ratio sensitive_count = sum(1 for e in events_window if any(kw in e.get("target", "").lower() for kw in SENSITIVE_KEYWORDS)) sensitive_ratio = sensitive_count / count if count > 0 else 0.0 return [ float(count), float(failure_ratio), float(distinct_paths), float(distinct_users), float(path_entropy), float(mean_inter_event), float(hour_norm), float(sensitive_ratio), ]