File size: 2,999 Bytes
a7d517c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 | 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),
]
|