I secure systems and automate the boring parts.

# --- excerpt from siem_log_monitor.py ---
# Full source: https://github.com/olokunde/siem-log-monitor
import re
from collections import defaultdict
from datetime import datetime, timedelta

FAILED_LOGIN = re.compile(
    r"(?P<ts>\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}).*"
    r"Failed password for .* from (?P<ip>\d+\.\d+\.\d+\.\d+)"
)
THRESHOLD = 5
WINDOW = timedelta(minutes=10)


def scan_log(path: str) -> dict[str, list[datetime]]:
    """Group failed-login timestamps by source IP."""
    attempts: dict[str, list[datetime]] = defaultdict(list)
    with open(path, "r", encoding="utf-8") as handle:
        for line in handle:
            match = FAILED_LOGIN.search(line)
            if not match:
                continue
            ts = datetime.strptime(match["ts"], "%Y-%m-%d %H:%M:%S")
            attempts[match["ip"]].append(ts)
    return attempts


def flag_brute_force(attempts: dict[str, list[datetime]]) -> list[dict]:
    """Raise an alert when an IP exceeds THRESHOLD attempts inside WINDOW."""
    alerts = []
    for ip, timestamps in attempts.items():
        timestamps.sort()
        for i in range(len(timestamps) - THRESHOLD + 1):
            window = timestamps[i : i + THRESHOLD]
            if window[-1] - window[0] <= WINDOW:
                alerts.append({
                    "ip": ip,
                    "attempts": len(window),
                    "first_seen": window[0].isoformat(),
                    "last_seen": window[-1].isoformat(),
                })
                break
    return alerts