How to Build a Log Parser That Detects Brute Force Attacks

Brute force attacks against SSH, web logins, and APIs are constant background noise on any public-facing server. The problem is not knowing they happen — your logs already record every failed attempt. The problem is catching the pattern fast enough to block the source before it succeeds, and doing it without generating so many alerts that your team starts ignoring them.

This post walks through building a practical log parser in Python that identifies brute force patterns using a sliding window, with configurable thresholds and structured JSON output you can pipe into any alerting or blocking pipeline.

What counts as brute force?

A working definition matters before writing a single line of code. In log analysis, a brute force attack typically exhibits:

• Multiple failed authentication attempts from the same source IP
• Within a short time window (commonly 30–120 seconds)
• Targeting the same service — SSH, a login endpoint, or an API

The tunable knobs are threshold (how many failures trigger an alert) and window (over what duration). Set the threshold too low and you flag a user who mistyped their password twice. Set the window too large and you fire an alert hours after the attacker has moved on.

We will start with sshd logs (/var/log/auth.log on Debian/Ubuntu, /var/log/secure on RHEL). The same pattern extends to nginx or any structured log format.

Parsing sshd auth logs

A typical failed SSH attempt looks like:

Aug  7 04:23:11 prod-01 sshd[1842]: Failed password for invalid user admin from 203.0.113.5 port 52341 ssh2
Aug  7 04:23:12 prod-01 sshd[1842]: Failed password for root from 203.0.113.5 port 52342 ssh2

The following parser extracts the timestamp, source IP, and attempted username from every such line:

import re
from datetime import datetime
from typing import Iterator, NamedTuple

class AuthEvent(NamedTuple):
    timestamp: datetime
    source_ip: str
    username: str
    raw_line: str

# Handles both "Failed password for root" and "Failed password for invalid user admin"
SSH_PATTERN = re.compile(
    r"(\w{3}\s+\d{1,2}\s+\d{2}:\d{2}:\d{2})"           # "Aug  7 04:23:11"
    r".+?Failed password for (?:invalid user )?(\S+)"    # username
    r" from (\d{1,3}(?:\.\d{1,3}){3})"                  # IPv4 source
)

def parse_auth_log(filepath: str, year: int = 2026) -> Iterator[AuthEvent]:
    with open(filepath, encoding="utf-8", errors="replace") as fh:
        for line in fh:
            m = SSH_PATTERN.search(line)
            if not m:
                continue
            raw_ts, username, ip = m.groups()
            # auth.log omits the year — inject it
            ts = datetime.strptime(f"{year} {raw_ts.strip()}", "%Y %b %d %H:%M:%S")
            yield AuthEvent(ts, ip, username, line.rstrip())

One common gotcha: auth.log does not include the year. You inject it manually, but watch for year-boundary wrap-around if you parse rotated logs from late December.

Detecting brute force with a sliding window

The detection logic maintains a per-IP bucket of recent events and evicts anything outside the current window before each check:

from collections import defaultdict
from datetime import timedelta

def detect_brute_force(
    events: Iterator[AuthEvent],
    threshold: int = 5,
    window_seconds: int = 60,
) -> list[dict]:
    buckets: dict[str, list[AuthEvent]] = defaultdict(list)
    incidents: list[dict] = []
    fired: set[tuple] = set()  # dedup key: (ip, window_start)

    for event in events:
        ip = event.source_ip
        cutoff = event.timestamp - timedelta(seconds=window_seconds)

        # Evict stale events first, then append the new one
        buckets[ip] = [e for e in buckets[ip] if e.timestamp >= cutoff]
        buckets[ip].append(event)

        if len(buckets[ip]) >= threshold:
            window_start = buckets[ip][0].timestamp
            key = (ip, window_start)
            if key not in fired:
                fired.add(key)
                incidents.append({
                    "ip": ip,
                    "first_seen": window_start.isoformat(),
                    "last_seen": event.timestamp.isoformat(),
                    "attempt_count": len(buckets[ip]),
                    "usernames_tried": sorted({e.username for e in buckets[ip]}),
                    "window_seconds": window_seconds,
                })

    return incidents

A few design decisions worth noting:

Evict before counting. If you count then evict, you get off-by-one errors at the window boundary. Trim first, always.

Dedup by (ip, window_start). Without this, a 20-attempt burst fires 16 separate alerts — one for every event after the threshold. This approach fires exactly once per distinct burst, regardless of how large it grows.

Collect usernames. An attacker trying root, admin, ubuntu, pi, and deploy in 30 seconds tells you whether this is automated credential stuffing or something more targeted. That context matters when you are writing the incident report.

Wiring it into a CLI

import json
import sys

def main() -> None:
    log_path    = sys.argv[1] if len(sys.argv) > 1 else "/var/log/auth.log"
    threshold   = int(sys.argv[2]) if len(sys.argv) > 2 else 5
    window      = int(sys.argv[3]) if len(sys.argv) > 3 else 60

    events    = parse_auth_log(log_path)
    incidents = detect_brute_force(events, threshold=threshold, window_seconds=window)

    if not incidents:
        print("No brute force activity detected.")
        return

    print(json.dumps(incidents, indent=2))
    sys.stderr.write(f"[!] {len(incidents)} incident(s) detected\n")

if __name__ == "__main__":
    main()

Running against a real log from an exposed VPS produces output like:

[
  {
    "ip": "203.0.113.5",
    "first_seen": "2026-08-07T04:23:11",
    "last_seen": "2026-08-07T04:23:42",
    "attempt_count": 23,
    "usernames_tried": ["admin", "deploy", "pi", "root", "ubuntu"],
    "window_seconds": 60
  }
]

Connecting to blocking and alerting pipelines

The JSON output is intentionally pipeline-friendly.

Auto-block with iptables:

python bf_parser.py /var/log/auth.log | \
  jq -r '.[].ip' | \
  xargs -I{} sudo iptables -A INPUT -s {} -j DROP

Forward to a Slack or webhook endpoint:

import requests

def send_alert(incident: dict, webhook_url: str) -> None:
    text = (
        f":rotating_light: Brute force from {incident['ip']} — "
        f"{incident['attempt_count']} attempts in {incident['window_seconds']}s. "
        f"Users tried: {', '.join(incident['usernames_tried'])}"
    )
    requests.post(webhook_url, json={"text": text}, timeout=5)

If you run this across multiple servers, the in-memory buckets dict is the scaling constraint: it is per-process and per-host. Moving it to Redis with sorted sets and ZRANGEBYSCORE for window queries lets you aggregate across a fleet and survive process restarts without losing event history.

Brute force detection covers one layer. Hardening the service itself — SSH key-only auth, fail2ban tuning, firewall ingress rules, and sudoers auditing — covers the rest. The security hardening checklists we publish walk through the full SSH and Linux server surface step by step, including the specific sshd_config options that eliminate most of this log noise in the first place.

The takeaway

The core logic is around 60 lines of Python. What makes it production-useful rather than a toy:

1. Regex tailored to your log format — generic parsers miss edge cases like the invalid user prefix sshd adds for non-existent accounts
2. Sliding window with explicit eviction — naive counting over fixed time buckets misses cross-boundary bursts
3. Deduplication on burst identity — one burst, one alert, regardless of how many events it contains
4. Structured JSON output — the next tool in the chain does not need to parse your output

Tune threshold and window against real data. A fresh VPS sees its first scan within minutes of getting a public IP — run this parser on a week of logs to find natural thresholds that separate legitimate failures from attack traffic before you start auto-blocking.

---

I run AYI NEDJIMI Consultants, a cybersecurity consulting firm. We publish free security hardening checklists — PDF and Excel.