features.py

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),
    ]