Real-Time Nginx Log Parser and Discord Alerting with Python

Admin

October 5, 2026

10 min read

Repeated authentication failures can be a useful signal during troubleshooting, but a single HTTP 401 or 403 is not enough to identify an attack. This guide follows a Linux Nginx access log, counts those responses by client IP over a rolling window, and reports when a configurable threshold is reached. It can send an optional Discord webhook alert; it does not block addresses or prove malicious activity.

The watcher starts at the end of the current log, handles file rotation and truncation, and keeps a bounded set of client counters. The sample regex expects Nginx’s common/combined access-log format.

Step 1: Build a Rotating Log Follower and Burst Counter

Count Authentication Failures by Client IP

Python Script

Save this as log_watcher.py. Set THRESHOLD and WINDOW_SECONDS to fit your service. The default raises a console alert when at least ten 401/403 responses from the same address arrive within 60 seconds. Each client’s event queue is capped at the threshold to keep memory use bounded.

#!/usr/bin/env python3
import ipaddress
import os
import re
import time
from collections import OrderedDict, deque
from pathlib import Path

LOG_PATH = "/var/log/nginx/access.log"
WINDOW_SECONDS = 60
THRESHOLD = 10
ALERT_COOLDOWN_SECONDS = 300
MAX_TRACKED_CLIENTS = 5000

ACCESS_PATTERN = re.compile(
    r'^(?P<client>\S+) \S+ \S+ \[[^\]]+\] "(?:[^"]*)" (?P<status>401|403)\b'
)
recent_by_ip = OrderedDict()
last_alert_at = {}


def follow_log(filepath):
    """Yield new lines and reopen the path after rotation or truncation."""
    path = Path(filepath)
    start_at_end = True

    while True:
        try:
            log_file = path.open("r", encoding="utf-8", errors="replace")
        except FileNotFoundError:
            time.sleep(0.5)
            continue

        opened_stat = os.fstat(log_file.fileno())
        if start_at_end:
            log_file.seek(0, os.SEEK_END)
            start_at_end = False

        while True:
            line = log_file.readline()
            if line:
                yield line
                continue

            try:
                current_stat = path.stat()
            except FileNotFoundError:
                current_stat = None

            rotated = (
                current_stat is None
                or (current_stat.st_dev, current_stat.st_ino)
                != (opened_stat.st_dev, opened_stat.st_ino)
            )
            truncated = (
                current_stat is not None
                and current_stat.st_size < log_file.tell()
            )
            if rotated:
                log_file.close()
                break
            if truncated:
                log_file.seek(0)
            time.sleep(0.5)


def dispatch_alert(client_ip, count):
    """Console-only dispatcher; Step 2 replaces this with Discord delivery."""
    print(
        f"[ALERT] {client_ip} reached at least {count} HTTP 401/403 responses "
        f"within {WINDOW_SECONDS} seconds."
    )


def process_line(line):
    match = ACCESS_PATTERN.search(line)
    if not match:
        return

    try:
        client_ip = str(ipaddress.ip_address(match.group("client")))
    except ValueError:
        return

    now = time.monotonic()
    recent = recent_by_ip.get(client_ip)
    if recent is None:
        if len(recent_by_ip) >= MAX_TRACKED_CLIENTS:
            oldest_ip, _ = recent_by_ip.popitem(last=False)
            last_alert_at.pop(oldest_ip, None)
        recent = deque(maxlen=THRESHOLD)
        recent_by_ip[client_ip] = recent
    else:
        recent_by_ip.move_to_end(client_ip)

    recent.append(now)
    cutoff = now - WINDOW_SECONDS
    while recent and recent[0] < cutoff:
        recent.popleft()

    if (
        len(recent) >= THRESHOLD
        and now - last_alert_at.get(client_ip, float("-inf"))
        >= ALERT_COOLDOWN_SECONDS
    ):
        last_alert_at[client_ip] = now
        dispatch_alert(client_ip, len(recent))


def main():
    print(f"Watching {LOG_PATH}; threshold={THRESHOLD} failures/{WINDOW_SECONDS}s.")
    for line in follow_log(LOG_PATH):
        process_line(line)


if __name__ == "__main__":
    main()

Step 2: Replace the Console Alert with a Discord Webhook

Send One Privacy-Conscious Alert per Cooldown

Notification

Add these imports and the environment variable near the top of the script, then replace dispatch_alert with this version. It sends the client address and count only; it does not include request paths or query strings. Enter the URL without echoing it when testing interactively.

import json
import urllib.request
from urllib.error import URLError

DISCORD_WEBHOOK_URL = os.environ.get("DISCORD_WEBHOOK_URL", "")


def dispatch_alert(client_ip, count):
    if not DISCORD_WEBHOOK_URL:
        print(
            f"[ALERT] {client_ip}: at least {count} HTTP 401/403 responses "
            f"in {WINDOW_SECONDS}s; webhook is not configured."
        )
        return

    payload = json.dumps({
        "content": (
            "**Nginx authentication-failure threshold reached**\n"
            f"Client: `{client_ip}`\n"
            f"Responses: at least {count} in {WINDOW_SECONDS} seconds"
        )
    }).encode("utf-8")
    request = urllib.request.Request(
        DISCORD_WEBHOOK_URL,
        data=payload,
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    try:
        with urllib.request.urlopen(request, timeout=10) as response:
            response.read()
        print(f"Discord alert sent for {client_ip}.")
    except (OSError, URLError) as error:
        print(f"Discord webhook request failed: {error}")

read -rsp "Discord webhook URL: " DISCORD_WEBHOOK_URL
export DISCORD_WEBHOOK_URL
printf '\n'

Step 3: Run with a Restricted systemd Service

Grant Read-Only Log Access and Install the Service

Deployment

These commands assume the Nginx log is readable by the adm group, as on common Debian-family installs. Verify access to the current and rotated log files on your distribution; grant the service account read access only to those logs. The systemd manager reads the root-only environment file before dropping privileges to logwatcher.

sudo useradd --system --user-group --no-create-home --groups adm --shell /usr/sbin/nologin logwatcher
sudo -u logwatcher test -r /var/log/nginx/access.log
sudo install -o root -g root -m 0755 log_watcher.py /opt/log_watcher.py
sudo install -o root -g root -m 0600 /dev/null /etc/logwatcher.env
sudoedit /etc/logwatcher.env

DISCORD_WEBHOOK_URL=https://discord.com/api/webhooks/replace-with-your-secret

sudo tee /etc/systemd/system/logwatcher.service >/dev/null <<'EOF'
[Unit]
Description=Python Nginx authentication-failure monitor
Wants=network-online.target
After=network-online.target

[Service]
Type=simple
User=logwatcher
Group=logwatcher
EnvironmentFile=/etc/logwatcher.env
ExecStart=/usr/bin/python3 /opt/log_watcher.py
Restart=on-failure
RestartSec=5

[Install]
WantedBy=multi-user.target
EOF

sudo systemctl daemon-reload
sudo systemctl enable --now logwatcher.service
sudo systemctl status logwatcher.service --no-pager

❯ View Expected Console Output

● logwatcher.service - Python Nginx authentication-failure monitor
     Loaded: loaded (/etc/systemd/system/logwatcher.service; enabled)
     Active: active (running)