Anomaly Detector

I Built a Tool That Catches Hackers in Real Time — Here's How It Works

Have you ever wondered how websites protect themselves from being flooded with fake traffic? Or how a server "knows" that something suspicious is happening?

I just built a tool that does exactly that — and in this post, I'll explain every part of it in plain English. No security background needed. Let's go.

What Is This Project?

Imagine you run a cloud storage website. Thousands of people visit it every day. Most of them are normal users uploading files, logging in, browsing around.

But one day, someone decides to attack your website. They write a script that sends thousands of requests per second from one IP address, trying to crash your server or sneak past your security. This is called a DDoS attack (Distributed Denial of Service).

My tool sits between the internet and the website, watches every single request coming in, and asks one question:

"Does this look normal — or is something weird happening?"

If something looks weird, it automatically blocks the attacker and sends me a message on Slack.

The Big Picture

Here's the flow in simple terms:

Someone visits the website
        ↓
Nginx (the traffic cop) lets them through and writes a note about it
        ↓
My tool reads those notes in real time
        ↓
It checks: "Is this normal traffic, or a spike?"
        ↓
If it's a spike → Block the IP + Send Slack alert
If it's normal  → Do nothing, keep watching

The "notes" Nginx writes are called access logs — one line per request, saved to a file.

Part 1: Reading the Log File (The Monitor)

Every time someone visits the website, Nginx writes a line to a log file that looks like this:

{"source_ip":"45.33.12.99","timestamp":"2025-04-25T10:32:01","method":"GET","path":"/login","status":200,"response_size":1452}

That's one visitor, at one moment in time.

My tool opens that file and reads it line by line, forever — like someone sitting next to a printer watching pages come out. In Python, this is called "tailing" a file.

with open("/var/log/nginx/hng-access.log") as f:
    f.seek(0, 2)  # Jump to the end (don't replay old history)
    while True:
        line = f.readline()
        if not line:
            time.sleep(0.05)  # No new line yet, wait a tiny bit
            continue
        process(line)  # We got a new request!

Simple, right? It just keeps checking for new lines forever.

Part 2: The Sliding Window (Counting Recent Requests)

Now that we can read each request, we need to answer: "How many requests came in the last 60 seconds?"

A naive approach would be to count all requests in the last minute. But that resets every 60 seconds and loses accuracy.

Instead, I use a sliding window — think of it like a moving spotlight that always shows you exactly the last 60 seconds.

Here's the idea using a real-world analogy:

Imagine a bouncer at a club keeping a paper roll. Every time someone enters, they write the time on the roll. Every few seconds, they tear off the old end (entries older than 60 seconds). The number of entries left = people who arrived in the last 60 seconds.

In Python, I use a deque (double-ended queue) — a list you can add to on the right and remove from the left efficiently:

from collections import deque
import time

window = deque()  # Our "paper roll"

def record_request():
    now = time.time()
    window.append(now)  # Write the time on the roll

    # Tear off old entries (older than 60 seconds)
    cutoff = now - 60
    while window and window[0] < cutoff:
        window.popleft()

    # How many requests in the last 60 seconds?
    rate = len(window) / 60
    return rate  # requests per second

I maintain one of these windows per IP address and one globally (all IPs combined).

Part 3: The Baseline (Learning What "Normal" Looks Like)

Here's the clever part. We can't just say "more than 10 requests per second is an attack" — because maybe your website normally gets 50 requests per second at peak hours, and only 2 at night.

So instead of hardcoding a number, my tool learns what normal looks like over time.

Every second, it records the current traffic rate. It keeps the last 30 minutes of these recordings. Every 60 seconds, it calculates:

Mean — the average traffic rate over the last 30 minutes
Standard deviation — how much the traffic varies

Think of it like a teacher grading on a curve. If the average test score is 70, a score of 95 is unusually high. If everyone scores between 85–95, then 95 is totally normal.

import math

samples = [2.1, 1.9, 2.3, 2.0, 1.8, ...]  # One sample per second, last 30 min

mean   = sum(samples) / len(samples)        # Average rate
stddev = math.sqrt(sum((x - mean)**2 for x in samples) / len(samples))

The tool also tracks traffic per hour separately. So if rush hour is always busier, it compares against rush hour data — not the quiet overnight average.

Part 4: Detecting the Attack (The Z-Score)

Now we have the current rate and we know what "normal" looks like. How do we decide if something is an attack?

I use a z-score — a simple math formula that answers:

"How many standard deviations above normal is this?"

z_score = (current_rate - mean) / stddev

If z_score > 3.0, that means the current rate is so far above normal that there's only a 0.1% chance it's a coincidence. That's our signal to act.

I also have a backup rule: if the rate is more than 5 times the average, it's flagged regardless of the z-score. Whichever rule fires first wins.

Real example:

Normal traffic: mean = 5 req/s, stddev = 1.5
Attacker sending: 50 req/s
Z-score = (50 - 5) / 1.5 = 30 → Way above 3.0 → 🚨 ATTACK DETECTED

There's also a bonus feature: if an IP is sending lots of error responses (like hitting bad login attempts over and over), we tighten the thresholds even further for that IP specifically.

Part 5: Blocking the Attacker (iptables)

Once we've decided an IP is an attacker, we need to stop them immediately.

iptables is a built-in Linux tool that acts like a firewall. You can tell it: "Drop all traffic from this IP address — don't even respond to them."

My tool runs this command in Python:

import subprocess

def ban_ip(ip):
    subprocess.run(["iptables", "-I", "INPUT", "-s", ip, "-j", "DROP"])
    print(f"Blocked: {ip}")

That's it. One command. The attacker's packets are now silently dropped by the operating system — they never even reach the web server.

You can verify it worked with:

sudo iptables -L INPUT -n

You'll see a line like:

DROP  all  --  45.33.12.99  0.0.0.0/0

Part 6: Auto-Unban (Giving Second Chances)

What if we accidentally blocked a real user? Or the attacker stopped and wants to try again?

My tool uses a backoff schedule — bans get longer each time the same IP misbehaves:

Offense	Ban Duration
1st time	10 minutes
2nd time	30 minutes
3rd time	2 hours
4th time	Permanent

Every 30 seconds, the tool checks: "Has anyone's ban expired?" If yes, it removes the iptables rule and sends a Slack notification.

Part 7: Slack Notifications

Every time something happens — a ban, an unban, or a global traffic surge — my tool sends a message to a Slack channel so I know about it immediately.

It looks like this:

🚨 IP BANNED
• IP: 45.33.12.99
• Condition: z-score=18.4
• Current rate: 120.00 req/s
• Baseline mean: 5.20 req/s
• Ban duration: 10 min
• Time: 2025-04-25T14:22:01

This uses Slack's "Incoming Webhooks" feature — you post a JSON message to a special URL, and it appears in your Slack channel.

Part 8: The Live Dashboard

Finally, the whole system has a web dashboard you can open in your browser. It shows:

Current global requests per second
The baseline mean and standard deviation
CPU and memory usage of the server
List of currently banned IPs
Top 10 most active IP addresses

It refreshes automatically every 3 seconds so you always see what's happening live.

Putting It All Together

Here's the whole system in one picture:

                        ┌─────────────────────────┐
Internet requests  ───► │  Nginx (reverse proxy)  │
                        │  writes JSON log lines  │
                        └──────────┬──────────────┘
                                   │ log file
                        ┌──────────▼──────────────┐
                        │     Monitor Thread       │
                        │  tails log line by line  │
                        └──────────┬──────────────┘
                                   │ parsed request
                        ┌──────────▼──────────────┐
                        │    Detector Thread       │
                        │  z-score & rate check    │
                        └──────┬──────────┬────────┘
                               │          │
               ┌───────────────▼─┐    ┌───▼──────────────┐
               │  Blocker        │    │  Notifier         │
               │  iptables DROP  │    │  Slack alert      │
               └────────┬────────┘    └───────────────────┘
                        │
               ┌────────▼────────┐
               │  Unbanner       │
               │  removes rule   │
               │  after timeout  │
               └─────────────────┘

Every component runs in its own thread — meaning they all run simultaneously, like workers in a factory each doing their own job at the same time.

What I Learned

Building this taught me a few things I hadn't truly understood before:

Security is about patterns, not rules. The most powerful part of this system isn't the blocking — it's the baseline that learns what's normal. A hardcoded rule would break on the first unusual-but-legitimate traffic spike.
Simple data structures go a long way. A deque + a timestamp is enough to build a production-grade sliding window. You don't need a database.
Linux tools are powerful. iptables has been in Linux for decades. One shell command blocks an IP at the kernel level — nothing the attacker sends can get through.

Want to Build It Yourself?

The full code is available at: https://github.com/ucheenyi/hng-detector

The README has step-by-step setup instructions starting from a fresh Linux server.

Thanks for reading! If you found this helpful, drop a like or comment — I'd love to hear what you're building.

推荐订阅源

DEV Community