52. Project: a log-based intrusion detector

Goal: read authentication logs, detect brute-force and password-spraying patterns with sliding windows, and emit alerts as JSON lines. It is the detection-engineering loop from Security in Depth Ch 51 at small scale, and it extends the log analyzer from Ch 12.

# labs/detect.py
import json
import re
import sys
from collections import defaultdict, deque
from dataclasses import dataclass, asdict
from datetime import datetime, timedelta
from collections.abc import Iterable, Iterator

SSH_FAIL = re.compile(
    r"^(?P<ts>\w{3}\s+\d+ \d\d:\d\d:\d\d) \S+ sshd\[\d+\]: Failed password for (?:invalid user )?"
    r"(?P<user>\S+) from (?P<ip>[\d.:a-f]+)"
)
SSH_OK = re.compile(
    r"^(?P<ts>\w{3}\s+\d+ \d\d:\d\d:\d\d) \S+ sshd\[\d+\]: Accepted \S+ for (?P<user>\S+) from (?P<ip>[\d.:a-f]+)"
)

@dataclass
class Alert:
    rule: str
    ip: str
    at: str
    detail: str
    severity: str

def events(lines: Iterable[str], year: int = 2026) -> Iterator[tuple[str, datetime, str, str]]:
    for line in lines:
        for kind, rx in (("fail", SSH_FAIL), ("ok", SSH_OK)):
            if m := rx.match(line):
                ts = datetime.strptime(f"{year} {m['ts']}", "%Y %b %d %H:%M:%S")
                yield kind, ts, m["ip"], m["user"]

def detect(evts: Iterable[tuple[str, datetime, str, str]],
           window: timedelta = timedelta(minutes=5),
           brute_threshold: int = 10, spray_users: int = 5) -> Iterator[Alert]:
    fails: defaultdict[str, deque[tuple[datetime, str]]] = defaultdict(deque)
    alerted: set[tuple[str, str]] = set()
    for kind, ts, ip, user in evts:
        q = fails[ip]
        while q and ts - q[0][0] > window:                 # slide the window
            q.popleft()
        if kind == "fail":
            q.append((ts, user))
            if len(q) >= brute_threshold and ("brute", ip) not in alerted:
                alerted.add(("brute", ip))
                yield Alert("ssh-brute-force", ip, ts.isoformat(), f"{len(q)} failures in {window}", "medium")
            users = {u for _, u in q}
            if len(users) >= spray_users and ("spray", ip) not in alerted:
                alerted.add(("spray", ip))
                yield Alert("ssh-password-spray", ip, ts.isoformat(),
                            f"{len(users)} distinct users: {sorted(users)[:5]}", "high")
        elif kind == "ok" and len(q) >= 3:
            yield Alert("ssh-success-after-failures", ip, ts.isoformat(),
                        f"login as {user} after {len(q)} recent failures", "critical")

if __name__ == "__main__":
    src = open(sys.argv[1], encoding="utf-8", errors="replace") if len(sys.argv) > 1 else sys.stdin
    n = 0
    for alert in detect(events(src)):
        n += 1
        print(json.dumps(asdict(alert)))
    raise SystemExit(1 if n else 0)

# tests/test_detect.py: detections need tests too, with known-bad and known-good samples
from labs.detect import detect, events

def line(t, msg): return f"Oct  4 10:{t} web1 sshd[811]: {msg}"

def test_spray_then_success():
    log = [line(f"00:{i:02}", f"Failed password for invalid user u{i} from 198.51.100.7 port 4{i} ssh2")
           for i in range(6)]
    log.append(line("01:00", "Accepted password for deploy from 198.51.100.7 port 5000 ssh2"))
    rules = [a.rule for a in detect(events(log))]
    assert rules == ["ssh-password-spray", "ssh-success-after-failures"]

def test_normal_login_is_quiet():
    log = [line("00:01", "Failed password for ada from 10.0.0.5 port 1 ssh2"),
           line("00:05", "Accepted password for ada from 10.0.0.5 port 2 ssh2")]
    assert list(detect(events(log))) == []

What to notice:

• Detections are code, so they get tests with true-positive and true-negative samples, version control, and review, which is the core idea of detection-as-code.
• A deque per IP gives an O(1) sliding window. At scale, this logic moves into a stream processor or a SIEM query, but the windowing idea is identical.
• The most valuable rule is the last one: a success after failures is the moment a brute force worked. Part VII adds a statistical model on top of hand-written rules; rules come first because they are explainable.