Build a Log-Based DDoS Detection Engine

Introduction

30 Second Summary

Every server connected to the internet faces unwanted traffic. Automated bots, port scanners, and DDoS attacks can overwhelm a system in seconds if nothing is watching the logs.

In this project, you will build a DDoS detection engine in Python that monitors log files and automatically bans suspicious IPs. You will also build a live web dashboard that displays traffic rates, banned IPs, and system health in real time.

What You'll Build

A DDoS detection engine that monitors log files, flags anomalous IPs with a sliding window algorithm, auto-bans them via nftables, and displays live metrics on a dark-themed dashboard built with FastAPI and HTMX.

By the end of this project, you'll have:

• A real-time log tailer and sliding window detector that monitors server logs and flags IPs exceeding configurable request thresholds.
• An nftables auto-banning system that blocks offending IPs at the Linux kernel level using dynamic firewall sets.
• A live web dashboard built with FastAPI and HTMX that auto-refreshes every 2 seconds to display traffic rates, banned IPs, top talkers, and system resources.
• Secret Mission: add auto-unban with configurable TTL and a manual unban button on the dashboard.

Are there any prerequisites?

This project assumes basic Python knowledge, including functions, dictionaries, and file I/O. You should also be comfortable with the Linux command line.

Before We Start

Before We Start...

Set Up the Project with UV

Before writing any detection logic, you need a properly structured Python project with the right dependencies in place. A clean foundation now means fewer headaches later.

In this step, you'll use uv to initialize a new project, install the packages your engine and dashboard need, and create every file you'll work with throughout the project.

In this step, get ready to:

• Initialize a UV project and verify its structure.
• Install FastAPI, Jinja2, and psutil as project dependencies.
• Create the file and directory structure for all modules and templates.

Initialize a new UV project

uv is a fast Python package manager that creates a project scaffold for you. It generates a pyproject.toml file (your project's configuration), a virtual environment, and a few starter files.

• Move to your Desktop by running this command:

cd ~/Desktop

• Create a new UV project called ddos-detector and move into it by running these commands:

uv init ddos-detector
cd ddos-detector

• Confirm the project was created by running:

ls

You should see pyproject.toml, main.py, README.md, and .python-version in the output.

What did uv init create?

pyproject.toml is the configuration file for your Python project. It tracks your project's name, version, and all its dependencies.

main.py is a starter script. You won't use it directly in this project, but it's harmless to leave in place. .python-version pins the Python version so the project behaves consistently.

Seeing 'command not found' for uv?

Make sure uv is installed and on your system PATH. Run uv --version to check.

If you already ran uv init but forgot to cd, double check you're inside the ddos-detector folder before continuing.

Help me troubleshoot my uv init setup

Add project dependencies

Your detection engine needs three packages. FastAPI serves the live dashboard. Jinja2 renders HTML templates on the server. psutil reads system resources like CPU and memory usage.

• Install all three dependencies by running:

uv add "fastapi[standard]" jinja2 psutil

What are these packages?

• fastapi[standard] installs FastAPI along with extras like uvicorn (the web server), Jinja2, and other production essentials.
• jinja2 is already included with fastapi[standard]. Listing it explicitly makes the dependency visible in your pyproject.toml.
• psutil provides cross-platform CPU and memory monitoring. Your dashboard uses it to display live system resource stats.

You should see output showing packages being resolved and installed. A uv.lock file now appears in your project folder, pinning the exact versions of every package.

Installation failed?

Check your internet connection. uv add downloads packages from PyPI, so it needs network access.

Make sure you're inside the ddos-detector folder. Running uv add outside a UV project gives an error because there's no pyproject.toml to update.

Help me fix my uv add error

Create the project file structure

Your project is split into six Python modules and two HTML templates. Creating the empty files now gives you a clear map of the codebase before you start writing logic.

• Create all six Python module files by running this command:

touch tailer.py detector.py banner.py engine.py simulator.py dashboard.py

• Create the template directories and HTML files by running these commands:

mkdir -p templates/partials
touch templates/index.html templates/partials/metrics.html

• Confirm everything was created by running:

ls

You should see banner.py, dashboard.py, detector.py, engine.py, main.py, simulator.py, tailer.py, and a templates directory alongside your pyproject.toml and uv.lock.

What is each file for?

• tailer.py follows a log file in real time, reading new lines as they appear.
• detector.py tracks per-IP request rates using a sliding window algorithm.
• banner.py bans and unbans IPs by managing nftables firewall rules.
• engine.py wires the tailer, detector, and banner together into a monitoring loop.
• simulator.py generates fake normal and attack traffic for testing.
• dashboard.py serves the live web dashboard using FastAPI.
• templates/index.html is the main dashboard page layout.
• templates/partials/metrics.html is the HTML fragment that HTMX polls every 2 seconds to refresh the dashboard.

Missing files after running touch?

Check that you're inside the ddos-detector folder by running pwd. The path should end with ddos-detector.

If the templates/partials directory is missing, make sure you ran mkdir -p templates/partials before the touch command. The -p flag creates both the parent and child directory at once.

Help me verify my project file structure

Your project skeleton is ready. Next up, you'll bring tailer.py to life by building a real-time log file tailer that watches for new lines as they appear.

Build the Log File Tailer

Your project structure is set up and all the dependencies are installed. Time to write real code.

Every intrusion detection system needs a way to watch log files in real time. The first module you'll build is a log tailer that follows a file as it grows, grabbing new lines the moment they appear. This is the same core pattern that tools like fail2ban use under the hood.

In this step, get ready to:

• Implement a LogTailer class that follows a file using Python's seek() and tell() file I/O primitives.
• Test the tailer by tailing a log file and appending lines from another terminal.

Implement the LogTailer class

The LogTailer class tracks its position in the file using a byte offset. Each time you call it, it opens the file, jumps to where it left off, reads any new content, and updates its position.

• Open tailer.py in your editor and add the imports, class definition, and start() method by copying this code:

import os
import time

class LogTailer:
    """Tails a log file in real time, yielding new lines as they appear."""

    def __init__(self, filepath: str):
        self.filepath = filepath
        self._position = 0

    def start(self):
        """Seek to the end of the file so we only read new lines."""
        if os.path.exists(self.filepath):
            with open(self.filepath, "r") as f:
                f.seek(0, os.SEEK_END)
                self._position = f.tell()

What does this code do?

• self._position stores the byte offset of where you last read. This is how the tailer remembers its place in the file between calls.
• start() uses f.seek(0, os.SEEK_END) to jump to the very end of the file. This means you'll only see lines added after the tailer starts, not the entire file history.
• f.tell() returns the current byte position. Storing it in self._position gives the tailer a bookmark to jump back to later.

You won't see a change yet. The next chunk adds the method that actually reads new lines.

• Still in tailer.py, add the read_new_lines() method below the start() method, inside the LogTailer class:

    def read_new_lines(self) -> list[str]:
        """Read any new lines appended since last check."""
        lines = []
        try:
            with open(self.filepath, "r") as f:
                f.seek(self._position)
                new_content = f.read()
                if new_content:
                    lines = new_content.splitlines()
                self._position = f.tell()
        except FileNotFoundError:
            pass
        return lines

What does this method do?

• f.seek(self._position) jumps back to where the tailer left off on the last read.
• f.read() grabs everything new from that position to the current end of the file.
• splitlines() breaks the raw text into individual lines so each log entry is its own string.
• The FileNotFoundError catch prevents the tailer from crashing if the log file hasn't been created yet.

• Save tailer.py.

Seeing a syntax error?

• Make sure read_new_lines() is indented inside the LogTailer class, at the same level as start().
• Check that the try: block uses consistent indentation. Mixing tabs and spaces will cause Python to throw an IndentationError.
• Help me fix a syntax error in my LogTailer class

✔️ Awesome, I've got everything!

Great. Double check you've saved tailer.py before moving on.

ⓧ I'd like to double check the full code

Here is the complete tailer.py file. Make sure yours matches this exactly.

import os
import time


class LogTailer:
    """Tails a log file in real time, yielding new lines as they appear."""

    def __init__(self, filepath: str):
        self.filepath = filepath
        self._position = 0

    def start(self):
        """Seek to the end of the file so we only read new lines."""
        if os.path.exists(self.filepath):
            with open(self.filepath, "r") as f:
                f.seek(0, os.SEEK_END)
                self._position = f.tell()

    def read_new_lines(self) -> list[str]:
        """Read any new lines appended since last check."""
        lines = []
        try:
            with open(self.filepath, "r") as f:
                f.seek(self._position)
                new_content = f.read()
                if new_content:
                    lines = new_content.splitlines()
                self._position = f.tell()
        except FileNotFoundError:
            pass
        return lines

Write a test loop that tails the log file

Before wiring the tailer into the detection engine, you should prove it works on its own. You'll create a test log file, then run a small Python loop that watches it for new lines.

• Create an empty test log file by running this command:

touch test.log

• Confirm the file exists by running:

ls test.log

You should see test.log printed in the output.

• Start a test loop that watches test.log for new lines by running this command:

uv run python -c "
from tailer import LogTailer
import time

t = LogTailer('test.log')
t.start()
print('Tailing test.log... (press Ctrl+C to stop)')

while True:
    lines = t.read_new_lines()
    for line in lines:
        print('NEW:', line)
    time.sleep(1)
"

You should see Tailing test.log... (press Ctrl+C to stop) printed in the terminal. The script is now watching the file. Leave this running.

What does this test loop do?

It creates a LogTailer pointed at test.log, seeks to the end of the file, then enters an infinite loop. Every second it checks for new lines and prints any it finds.

Seeing a ModuleNotFoundError?

• Make sure you're running the command from the project root folder where tailer.py lives.
• Check that you saved tailer.py before running the command. Unsaved changes won't be picked up.
• Help me debug a ModuleNotFoundError when importing my LogTailer

Verify the tailer from another terminal

The test loop is running and waiting for new content. Now you'll open a second terminal and write lines to test.log to confirm the tailer picks them up.

• Open a new terminal window. Keep the first one running.
• Navigate to your project folder in the new terminal by running:

cd ~/Desktop/ddos-detector

• Append a test line to test.log by running this command:

echo "2025-01-01 10:00:00 server sshd[12345]: Connection from 10.0.0.1 port 22" >> test.log

• Switch back to the first terminal where the test loop is running.

You should see the line printed with a NEW: prefix, like this:

NEW: 2025-01-01 10:00:00 server sshd[12345]: Connection from 10.0.0.1 port 22

• Try appending a few more lines from the second terminal to confirm the tailer catches every one:

echo "2025-01-01 10:00:01 server sshd[12345]: Connection from 192.168.1.50 port 443" >> test.log
echo "2025-01-01 10:00:02 server sshd[12345]: Connection from 10.0.0.2 port 80" >> test.log

Both lines should appear in the first terminal within a second.

• Once you've confirmed it works, stop the test loop in the first terminal by pressing Ctrl+C.

Not seeing any output?

• Make sure you use >> (append) instead of > (overwrite). A single > rewrites the file from scratch, which can confuse the tailer's position tracking.
• Confirm that both terminals are in the same project folder. The echo command must write to the same test.log file the tailer is watching.
• Help me debug my LogTailer that isn't picking up new lines

Your log tailer is working. It can follow a file in real time and grab every new line as it appears. Next up, you'll build the anomaly detector that takes those lines and decides which IPs are sending too many requests.

Build the Sliding Window Anomaly Detector

Your log tailer can now follow a file in real time. The next challenge is figuring out which IPs are hitting your server too often.

A DDoS attack floods a server with requests from one or more IPs. In this step, you'll build a detector that tracks how many requests each IP sends within a rolling time window and flags anything above a configurable threshold.

In this step, get ready to:

• Implement the SlidingWindowDetector class with per-IP request tracking.
• Add threshold-based detection methods that flag offending IPs.
• Test the detector with simulated IP hits.

Implement the SlidingWindowDetector class

DDoS detection relies on counting how frequently each IP address hits your server. A sliding window algorithm does this by storing the timestamp of every request.

When you check for anomalies, the algorithm discards timestamps older than the window and counts what remains. If the count exceeds a threshold, that IP is flagged.

Why store every timestamp instead of just a counter?

A plain counter would tell you how many total requests an IP made, but not when they happened. By storing timestamps, you can prune old ones and focus only on the recent burst. This is what makes the window "slide" forward in time.

This is the same approach that tools like fail2ban use under the hood for rate-based detection.

• Open detector.py and add the class with its core tracking methods by pasting this code:

import time
from collections import defaultdict


class SlidingWindowDetector:
    """Tracks per-IP request rates using a sliding window counter."""

    def __init__(self, window_seconds: int = 60, max_requests: int = 100):
        self.window_seconds = window_seconds
        self.max_requests = max_requests
        self._requests: dict[str, list[float]] = defaultdict(list)

    def record(self, ip: str):
        """Record a request from the given IP."""
        now = time.time()
        self._requests[ip].append(now)

    def _prune(self, ip: str):
        """Remove timestamps outside the current window."""
        cutoff = time.time() - self.window_seconds
        self._requests[ip] = [
            t for t in self._requests[ip] if t > cutoff
        ]

What does this code do?

• _requests is a defaultdict(list) that maps each IP address to a list of timestamps. Any IP you've never seen before automatically gets an empty list.
• record() appends the current time to that IP's list every time a request comes in.
• _prune() is the heart of the sliding window. It calculates a cutoff time and throws away any timestamp older than that. Only recent activity survives.

• Save detector.py.

You won't see a change yet. The next chunk adds the detection methods that query this data structure.

Seeing a syntax error?

• Make sure the import line reads from collections import defaultdict. A missing s in collections is a common typo.
• Check that all methods are indented under the class SlidingWindowDetector: line. Python uses indentation to determine what belongs to a class.
• Why is my SlidingWindowDetector class not working?

Add detection and query methods

The core data structure is in place. Now you need methods to ask questions about it: which IPs are over the threshold, how many requests has a specific IP made, and what is the overall traffic rate.

• In detector.py, add these four methods inside the SlidingWindowDetector class, below the _prune method:

    def get_offenders(self) -> list[str]:
        """Return list of IPs that exceed the max request threshold."""
        offenders = []
        for ip in list(self._requests.keys()):
            self._prune(ip)
            if len(self._requests[ip]) > self.max_requests:
                offenders.append(ip)
        return offenders

    def get_request_count(self, ip: str) -> int:
        """Return current request count for an IP within the window."""
        self._prune(ip)
        return len(self._requests[ip])

    def get_all_ips(self) -> list[str]:
        """Return all IPs currently being tracked."""
        return list(self._requests.keys())

    def get_total_requests_per_second(self) -> float:
        """Return the global request rate across all IPs."""
        now = time.time()
        cutoff = now - self.window_seconds
        total = 0
        for ip in list(self._requests.keys()):
            self._prune(ip)
            total += len(self._requests[ip])
        if self.window_seconds == 0:
            return 0.0
        return total / self.window_seconds

What does this code do?

• get_offenders() loops through every tracked IP, prunes old timestamps, and collects any IP whose remaining count exceeds max_requests. This is the method that actually flags attackers.
• get_request_count() returns how many recent requests a single IP has made. The dashboard will use this to display per-IP stats.
• get_all_ips() returns every IP the detector is currently tracking.
• get_total_requests_per_second() calculates the global traffic rate by summing all requests across all IPs and dividing by the window size. The zero-check prevents a division error.

• Save detector.py.

You won't see a visible change yet. The next substep tests these methods with simulated traffic.

Getting an IndentationError?

• All four methods must be indented at the same level as record() and _prune() above. They are class methods, so they need exactly one level of indentation inside the class body.
• Make sure there is no blank line between the class header and the first method. Stray whitespace or mixed tabs/spaces can cause Python indentation errors.
• Help me fix an IndentationError in my detector.py file.

✔️ Awesome, I've got everything!

Double check that you've saved detector.py before moving on to the test.

ⓧ I'd like to double check the full code

Here is the complete detector.py file. Make sure yours matches exactly.

import time
from collections import defaultdict


class SlidingWindowDetector:
    """Tracks per-IP request rates using a sliding window counter."""

    def __init__(self, window_seconds: int = 60, max_requests: int = 100):
        self.window_seconds = window_seconds
        self.max_requests = max_requests
        self._requests: dict[str, list[float]] = defaultdict(list)

    def record(self, ip: str):
        """Record a request from the given IP."""
        now = time.time()
        self._requests[ip].append(now)

    def _prune(self, ip: str):
        """Remove timestamps outside the current window."""
        cutoff = time.time() - self.window_seconds
        self._requests[ip] = [
            t for t in self._requests[ip] if t > cutoff
        ]

    def get_offenders(self) -> list[str]:
        """Return list of IPs that exceed the max request threshold."""
        offenders = []
        for ip in list(self._requests.keys()):
            self._prune(ip)
            if len(self._requests[ip]) > self.max_requests:
                offenders.append(ip)
        return offenders

    def get_request_count(self, ip: str) -> int:
        """Return current request count for an IP within the window."""
        self._prune(ip)
        return len(self._requests[ip])

    def get_all_ips(self) -> list[str]:
        """Return all IPs currently being tracked."""
        return list(self._requests.keys())

    def get_total_requests_per_second(self) -> float:
        """Return the global request rate across all IPs."""
        now = time.time()
        cutoff = now - self.window_seconds
        total = 0
        for ip in list(self._requests.keys()):
            self._prune(ip)
            total += len(self._requests[ip])
        if self.window_seconds == 0:
            return 0.0
        return total / self.window_seconds

Test the detector with simulated hits

Before wiring the detector into the full engine, you should verify it works on its own. You'll feed it a mix of normal and attack traffic and confirm it only flags the right IP.

• Create a temporary test file called test_detector.py in your project root and paste in this code:

from detector import SlidingWindowDetector

# Create a detector with a small window and low threshold for testing
detector = SlidingWindowDetector(window_seconds=10, max_requests=5)

# Simulate normal traffic: 3 requests from a regular IP
for i in range(3):
    detector.record("10.0.0.1")

# Simulate attack traffic: 10 rapid requests from a suspicious IP
for i in range(10):
    detector.record("192.168.1.100")

# Check results
print("Offenders:", detector.get_offenders())
print("10.0.0.1 count:", detector.get_request_count("10.0.0.1"))
print("192.168.1.100 count:", detector.get_request_count("192.168.1.100"))
print("All IPs:", detector.get_all_ips())
print("Global rate:", detector.get_total_requests_per_second(), "req/s")

What does this test do?

The test creates a detector with a 10-second window and a threshold of 5 requests. It records 3 requests from 10.0.0.1 (normal) and 10 requests from 192.168.1.100 (attacker).

Only 192.168.1.100 exceeds the threshold. If the detector works correctly, it should appear in the offenders list while 10.0.0.1 does not.

• Save test_detector.py.
• Run the test by executing this command:

uv run python test_detector.py

You should see output similar to this:

Offenders: ['192.168.1.100']
10.0.0.1 count: 3
192.168.1.100 count: 10
All IPs: ['10.0.0.1', '192.168.1.100']
Global rate: 1.3 req/s

What should I see?

• Offenders should only contain 192.168.1.100. The normal IP stays below the threshold.
• The request counts should be 3 and 10 respectively.
• The exact Global rate number varies depending on how fast your machine runs the script. Any positive number means it is working.

Not seeing the expected output?

• If you see ModuleNotFoundError: No module named 'detector', make sure test_detector.py is in the same folder as detector.py.
• If the offenders list is empty, check that you set max_requests=5 in the test file. The default is 100, which is higher than the 10 simulated hits.
• If both IPs appear in offenders, check that the normal IP loop runs only 3 times, not 10.
• Why is my detector test not producing the expected output?

• Once the test passes, delete the test file by running:

rm test_detector.py

Your detector can now spot suspicious IPs in real time. Next up, you'll build the firewall module that actually bans those IPs at the kernel level using nftables.

Build the nftables IP Banner

Your detector can now spot IPs that exceed your traffic threshold. The next question is what you actually do about those offending IPs.

In this step, you'll build a Firewall class that uses nftables to block malicious IPs at the Linux kernel level. This is the same approach that tools like fail2ban use under the hood.

In this step, get ready to:

• Create a Firewall class with nftables setup and teardown methods.
• Add ban_ip() and unban_ip() methods to control the blackhole set.
• Verify the firewall works by banning and unbanning a test IP from the terminal.

Implement the Firewall class and nftables setup

nftables is the modern Linux firewall framework. It lets you create named sets of IP addresses and write a single rule that drops all traffic from any IP in the set.

Why nftables instead of iptables?

nftables is the successor to iptables, included in Linux kernels 3.13 and above. It offers simpler syntax, atomic rule changes, and a generic set infrastructure that is perfect for dynamic ban lists.

Your Firewall class will call the nft command via Python's subprocess module. This is more portable than the python3-nftables package, which requires a system-level install and is not pip-installable.

The class starts with imports, three module-level constants for the nftables table, chain, and set names, and an __init__ that tracks banned IPs in a Python set.

• Open banner.py and add the following imports, constants, and class skeleton:

import subprocess
import logging

logger = logging.getLogger(__name__)

# nftables resource names used across all methods
TABLE_NAME = "ddos_filter"
CHAIN_NAME = "input"
SET_NAME = "blackhole"


class Firewall:
    """Manages nftables rules for banning/unbanning IPs."""

    def __init__(self):
        # Track banned IPs in Python so we don't duplicate nft commands
        self._banned_ips: set[str] = set()

What does this code do?

The three constants define the nftables resources your firewall will create.

• TABLE_NAME is the top-level nftables container that holds all your rules.
• CHAIN_NAME is the rule chain that hooks into the Linux networking input path.
• SET_NAME is the named set where banned IP addresses get stored.

• Save the file. You won't see any output yet since this is just the class skeleton.

Now you need the method that actually creates the nftables infrastructure. The setup() method runs four nft commands in sequence to build the table, chain, set, and drop rule.

• In banner.py, add this setup() method inside the Firewall class, below __init__:

    def setup(self):
        """Create the nftables table, chain, set, and drop rule."""
        commands = [
            # 1. Create the inet table (handles IPv4 and IPv6)
            ["sudo", "nft", "add", "table", "inet", TABLE_NAME],
            # 2. Create the input chain with priority -1 (runs before other chains)
            [
                "sudo", "nft", "add", "chain", "inet", TABLE_NAME, CHAIN_NAME,
                "{ type filter hook input priority -1 ; policy accept ; }"
            ],
            # 3. Create the blackhole set for storing banned IPv4 addresses
            [
                "sudo", "nft", "add", "set", "inet", TABLE_NAME, SET_NAME,
                "{ type ipv4_addr ; }"
            ],
            # 4. Add a rule: drop any packet whose source IP is in the blackhole set
            [
                "sudo", "nft", "add", "rule", "inet", TABLE_NAME, CHAIN_NAME,
                "ip", "saddr", f"@{SET_NAME}", "drop"
            ],
        ]
        for cmd in commands:
            result = subprocess.run(cmd, capture_output=True, text=True)
            if result.returncode != 0:
                logger.warning("nft command failed: %s | %s", " ".join(cmd), result.stderr.strip())

What do these four nft commands do?

• Command 1 creates an inet table, which is the dual-stack family that handles both IPv4 and IPv6 traffic.
• Command 2 creates a base chain hooked into the input path at priority -1, so it runs before other input chains.
• Command 3 creates a named set of type ipv4_addr. This set is your dynamic ban list.
• Command 4 adds the drop rule. Any packet whose source IP matches an address in @blackhole gets silently dropped.

Each command uses subprocess.run() with capture_output=True so errors are captured instead of crashing the program.

• Save banner.py. You won't see a change yet. The next chunk adds the methods that use this infrastructure.

Add ban and unban methods

With the nftables infrastructure in place, you need methods to add and remove individual IPs from the blackhole set. Each method calls a single nft command and updates the Python-side tracking set.

• In banner.py, add the ban_ip() method below setup():

    def ban_ip(self, ip: str):
        """Add an IP to the nftables blackhole set."""
        # Skip if already banned to avoid duplicate nft commands
        if ip in self._banned_ips:
            return
        cmd = [
            "sudo", "nft", "add", "element", "inet", TABLE_NAME, SET_NAME,
            "{", ip, "}"
        ]
        result = subprocess.run(cmd, capture_output=True, text=True)
        if result.returncode == 0:
            self._banned_ips.add(ip)
            logger.info("BANNED IP: %s", ip)
        else:
            logger.error("Failed to ban %s: %s", ip, result.stderr.strip())

What does this code do?

• The early return on if ip in self._banned_ips prevents sending duplicate commands to nftables.
• nft add element inserts the IP address into the blackhole set. Once it is there, the drop rule you created in setup() immediately blocks all traffic from that IP.

• Save the file, then add the unban_ip() method directly below ban_ip():

    def unban_ip(self, ip: str):
        """Remove an IP from the nftables blackhole set."""
        # Skip if not currently banned
        if ip not in self._banned_ips:
            return
        cmd = [
            "sudo", "nft", "delete", "element", "inet", TABLE_NAME, SET_NAME,
            "{", ip, "}"
        ]
        result = subprocess.run(cmd, capture_output=True, text=True)
        if result.returncode == 0:
            self._banned_ips.discard(ip)
            logger.info("UNBANNED IP: %s", ip)
        else:
            logger.error("Failed to unban %s: %s", ip, result.stderr.strip())

What does this code do?

• nft delete element removes the IP from the set, restoring traffic from that address.
• discard() is used instead of remove() because it silently ignores missing keys instead of raising a KeyError.

• Save banner.py. You won't see a change yet. The next chunk adds the remaining utility methods.

Add teardown and utility methods, then verify the firewall

Your firewall needs a clean way to shut down. It also needs two quick helper methods so the rest of the project can check ban status.

• In banner.py, add the teardown() method below unban_ip():

    def teardown(self):
        """Remove the nftables table and all its contents."""
        cmd = ["sudo", "nft", "delete", "table", "inet", TABLE_NAME]
        result = subprocess.run(cmd, capture_output=True, text=True)
        if result.returncode != 0:
            logger.warning("nft teardown failed: %s", result.stderr.strip())
        self._banned_ips.clear()

What does this code do?

nft delete table removes the entire table and everything inside it: the chain, the set, and all rules. This is a single-command cleanup. self._banned_ips.clear() resets the Python-side tracking to match.

• Save the file, then add the two utility methods below teardown():

    def is_banned(self, ip: str) -> bool:
        """Check if an IP is currently banned."""
        return ip in self._banned_ips

    def list_banned(self) -> set[str]:
        """Return the set of currently banned IPs."""
        return self._banned_ips.copy()

What does this code do?

• is_banned() checks whether a specific IP is in the banned set. The detection engine will use this to avoid banning the same IP twice.
• list_banned() returns a copy of the banned set. The copy prevents outside code from accidentally modifying the internal state.

• Save banner.py.

✔️ Awesome, I've got everything!

Great. Double check you've saved your file before moving on to the test.

ⓧ I'd like to double check the full code

import subprocess
import logging

logger = logging.getLogger(__name__)

TABLE_NAME = "ddos_filter"
CHAIN_NAME = "input"
SET_NAME = "blackhole"


class Firewall:
    """Manages nftables rules for banning/unbanning IPs."""

    def __init__(self):
        self._banned_ips: set[str] = set()

    def setup(self):
        """Create the nftables table, chain, set, and drop rule."""
        commands = [
            ["sudo", "nft", "add", "table", "inet", TABLE_NAME],
            [
                "sudo", "nft", "add", "chain", "inet", TABLE_NAME, CHAIN_NAME,
                "{ type filter hook input priority -1 ; policy accept ; }"
            ],
            [
                "sudo", "nft", "add", "set", "inet", TABLE_NAME, SET_NAME,
                "{ type ipv4_addr ; }"
            ],
            [
                "sudo", "nft", "add", "rule", "inet", TABLE_NAME, CHAIN_NAME,
                "ip", "saddr", f"@{SET_NAME}", "drop"
            ],
        ]
        for cmd in commands:
            result = subprocess.run(cmd, capture_output=True, text=True)
            if result.returncode != 0:
                logger.warning("nft command failed: %s | %s", " ".join(cmd), result.stderr.strip())

    def teardown(self):
        """Remove the nftables table and all its contents."""
        cmd = ["sudo", "nft", "delete", "table", "inet", TABLE_NAME]
        result = subprocess.run(cmd, capture_output=True, text=True)
        if result.returncode != 0:
            logger.warning("nft teardown failed: %s", result.stderr.strip())
        self._banned_ips.clear()

    def ban_ip(self, ip: str):
        """Add an IP to the nftables blackhole set."""
        if ip in self._banned_ips:
            return
        cmd = [
            "sudo", "nft", "add", "element", "inet", TABLE_NAME, SET_NAME,
            "{", ip, "}"
        ]
        result = subprocess.run(cmd, capture_output=True, text=True)
        if result.returncode == 0:
            self._banned_ips.add(ip)
            logger.info("BANNED IP: %s", ip)
        else:
            logger.error("Failed to ban %s: %s", ip, result.stderr.strip())

    def unban_ip(self, ip: str):
        """Remove an IP from the nftables blackhole set."""
        if ip not in self._banned_ips:
            return
        cmd = [
            "sudo", "nft", "delete", "element", "inet", TABLE_NAME, SET_NAME,
            "{", ip, "}"
        ]
        result = subprocess.run(cmd, capture_output=True, text=True)
        if result.returncode == 0:
            self._banned_ips.discard(ip)
            logger.info("UNBANNED IP: %s", ip)
        else:
            logger.error("Failed to unban %s: %s", ip, result.stderr.strip())

    def is_banned(self, ip: str) -> bool:
        """Check if an IP is currently banned."""
        return ip in self._banned_ips

    def list_banned(self) -> set[str]:
        """Return the set of currently banned IPs."""
        return self._banned_ips.copy()

Now verify the full lifecycle: setup, ban, check, unban, and teardown. The nft commands require sudo privileges, so you may be prompted for your password.

• Run this test script in your terminal to exercise every Firewall method:

uv run python -c "
from banner import Firewall
fw = Firewall()
fw.setup()
print('Setup complete')
fw.ban_ip('1.2.3.4')
print('Banned:', fw.list_banned())
print('Is 1.2.3.4 banned?', fw.is_banned('1.2.3.4'))
fw.unban_ip('1.2.3.4')
print('After unban:', fw.list_banned())
fw.teardown()
print('Teardown complete')
"

• You should see output like this:

Setup complete
Banned: {'1.2.3.4'}
Is 1.2.3.4 banned? True
After unban: set()
Teardown complete

What should I see?

The output confirms four things. Setup created the nftables table without errors. ban_ip() added 1.2.3.4 to the banned set. unban_ip() removed it. And teardown cleaned everything up.

• Optionally, confirm the nftables side by running this between the ban and unban steps in a separate terminal:

sudo nft list set inet ddos_filter blackhole

You should see 1.2.3.4 listed inside the elements section of the set output. This proves the ban is active at the kernel level, not just in Python memory.

Seeing permission errors or command not found?

• If you see nft: command not found, nftables is not installed. Install it with sudo apt install nftables on Debian/Ubuntu.
• If you see Permission denied or Operation not permitted, your user may not have sudo access. Run sudo echo test first to confirm you can use sudo.
• If you see nft command failed warnings about a table already existing, a previous test run may not have cleaned up. Run sudo nft delete table inet ddos_filter manually, then try again.

Help me debug my nftables Firewall class

Your firewall module is ready. It can create nftables infrastructure, ban IPs with a single method call, and clean up after itself. Next up, you'll wire the tailer, detector, and banner together into a detection engine that monitors your log file in real time.

Build the Detection Engine and Attack Simulator

Your tailer reads new log lines, your detector counts per-IP requests, and your firewall bans offenders at the kernel level. Right now, these three modules sit in separate files with no way to talk to each other.

In this step, you will build the detection engine that wires all three modules into a single monitoring loop. You will also build an attack simulator so you can generate fake traffic and watch the engine detect and ban a malicious IP in real time.

In this step, get ready to:

• Build the detection engine that connects the tailer, detector, and firewall into a monitoring loop.
• Build a traffic simulator that generates normal and attack log entries.
• Test the full pipeline by running the simulator and watching the engine ban the attacker.

Build the detection engine module

The engine is the orchestrator. Each cycle, it asks the tailer for new log lines, extracts IP addresses with a regex, feeds them to the detector, and tells the firewall to ban any offenders.

• Open engine.py and add the imports, the IP regex pattern, and the DetectionEngine class with its constructor and core detection methods:

import re
import time
import logging
import threading

from tailer import LogTailer
from detector import SlidingWindowDetector
from banner import Firewall

logger = logging.getLogger(__name__)

# Regex to extract the first IPv4 address from any log line
IP_PATTERN = re.compile(r"\b(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})\b")


class DetectionEngine:
    """Wires together the tailer, detector, and firewall into a monitoring loop."""

    def __init__(
        self,
        log_file: str = "test.log",
        window_seconds: int = 30,
        max_requests: int = 20,
        poll_interval: float = 1.0,
    ):
        self.log_file = log_file
        self.poll_interval = poll_interval
        # Create instances of each module
        self.tailer = LogTailer(log_file)
        self.detector = SlidingWindowDetector(
            window_seconds=window_seconds,
            max_requests=max_requests,
        )
        self.firewall = Firewall()
        self.start_time = time.time()
        self._running = False
        self._thread: threading.Thread | None = None

    def _extract_ip(self, line: str) -> str | None:
        """Extract the first IPv4 address from a log line."""
        match = IP_PATTERN.search(line)
        return match.group(1) if match else None

    def tick(self):
        """Run one cycle of the detection loop."""
        # Step 1: Read new lines from the log file
        new_lines = self.tailer.read_new_lines()

        # Step 2: Extract IPs and record them in the detector
        for line in new_lines:
            ip = self._extract_ip(line)
            if ip:
                self.detector.record(ip)

        # Step 3: Check for offenders and ban any new ones
        offenders = self.detector.get_offenders()
        for ip in offenders:
            if not self.firewall.is_banned(ip):
                count = self.detector.get_request_count(ip)
                logger.warning(
                    "ANOMALY DETECTED: %s (%d requests in %ds window)",
                    ip, count, self.detector.window_seconds,
                )
                self.firewall.ban_ip(ip)

What does this code do?

IP_PATTERN is a compiled regex that matches any IPv4 address in a log line. Compiling it once at module level avoids recompiling on every call.

The constructor creates one instance each of LogTailer, SlidingWindowDetector, and Firewall. The detection thresholds default to 20 requests in a 30-second window.

The tick() method is the heart of the engine. It reads new log lines, extracts IPs, records each one in the detector, then checks for any IPs that have exceeded the threshold. New offenders get banned through the firewall.

• You won't see a change yet. The next chunk adds the methods that run tick() in a loop.

• Still inside the DetectionEngine class in engine.py, add the lifecycle methods and the metrics method below the tick() method:

    def _run_loop(self):
        """Background loop that calls tick() repeatedly."""
        while self._running:
            self.tick()
            time.sleep(self.poll_interval)

    def start(self):
        """Set up firewall, start tailer, and begin background monitoring."""
        logger.info("Setting up nftables firewall rules...")
        self.firewall.setup()
        self.tailer.start()
        self.start_time = time.time()
        self._running = True
        # Run the detection loop in a background daemon thread
        self._thread = threading.Thread(target=self._run_loop, daemon=True)
        self._thread.start()
        logger.info("Detection engine started on: %s", self.log_file)

    def stop(self):
        """Stop monitoring and clean up firewall rules."""
        self._running = False
        if self._thread:
            self._thread.join(timeout=5)
        self.firewall.teardown()
        logger.info("Detection engine stopped.")

    def get_metrics(self) -> dict:
        """Return current metrics for the dashboard."""
        uptime = time.time() - self.start_time
        hours = int(uptime // 3600)
        minutes = int((uptime % 3600) // 60)
        seconds = int(uptime % 60)

        all_ips = self.detector.get_all_ips()
        # Sort IPs by request count, take top 10
        top_ips = sorted(
            all_ips,
            key=lambda ip: self.detector.get_request_count(ip),
            reverse=True,
        )[:10]

        # Build a list of dicts with ip, rate, and ban status
        top_ips_data = []
        for ip in top_ips:
            count = self.detector.get_request_count(ip)
            rate = count / max(self.detector.window_seconds, 1)
            status = "BANNED" if self.firewall.is_banned(ip) else "OK"
            top_ips_data.append({
                "ip": ip,
                "rate": f"{rate:.2f}",
                "status": status,
            })

        return {
            "uptime": f"{hours:02d}h {minutes:02d}m {seconds:02d}s",
            "active_ips": len(all_ips),
            "banned_count": len(self.firewall.list_banned()),
            "banned_ips": list(self.firewall.list_banned()),
            "global_rate": f"{self.detector.get_total_requests_per_second():.2f}",
            "window_seconds": self.detector.window_seconds,
            "max_requests": self.detector.max_requests,
            "top_ips": top_ips_data,
        }

What does this code do?

• _run_loop() calls tick() in a loop, sleeping for poll_interval seconds between each cycle.
• start() initializes the nftables rules, starts the tailer, then launches _run_loop as a daemon thread so it runs in the background.
• stop() sets _running to False, waits for the thread to finish, then tears down the nftables rules.
• get_metrics() builds a dictionary with uptime, active IPs, banned IPs, global request rate, and the top 10 IPs by traffic. The dashboard will call this method every 2 seconds.

• Save engine.py.

Why a daemon thread?

A daemon thread automatically exits when the main program ends. This means you do not need to worry about the monitoring loop keeping the process alive after the dashboard server shuts down.

Seeing an IndentationError?

All four methods (_run_loop, start, stop, get_metrics) must be indented one level inside the DetectionEngine class, at the same level as tick(). Make sure you did not accidentally paste them outside the class body.

Help me debug an IndentationError in engine.py

✔️ Awesome, I've got everything!

Great. Double check you have saved engine.py before moving on.

ⓧ I'd like to double check the full code

import re
import time
import logging
import threading

from tailer import LogTailer
from detector import SlidingWindowDetector
from banner import Firewall

logger = logging.getLogger(__name__)

IP_PATTERN = re.compile(r"\b(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})\b")


class DetectionEngine:
    """Wires together the tailer, detector, and firewall into a monitoring loop."""

    def __init__(
        self,
        log_file: str = "test.log",
        window_seconds: int = 30,
        max_requests: int = 20,
        poll_interval: float = 1.0,
    ):
        self.log_file = log_file
        self.poll_interval = poll_interval
        self.tailer = LogTailer(log_file)
        self.detector = SlidingWindowDetector(
            window_seconds=window_seconds,
            max_requests=max_requests,
        )
        self.firewall = Firewall()
        self.start_time = time.time()
        self._running = False
        self._thread: threading.Thread | None = None

    def _extract_ip(self, line: str) -> str | None:
        """Extract the first IPv4 address from a log line."""
        match = IP_PATTERN.search(line)
        return match.group(1) if match else None

    def tick(self):
        """Run one cycle of the detection loop."""
        new_lines = self.tailer.read_new_lines()

        for line in new_lines:
            ip = self._extract_ip(line)
            if ip:
                self.detector.record(ip)

        offenders = self.detector.get_offenders()
        for ip in offenders:
            if not self.firewall.is_banned(ip):
                count = self.detector.get_request_count(ip)
                logger.warning(
                    "ANOMALY DETECTED: %s (%d requests in %ds window)",
                    ip, count, self.detector.window_seconds,
                )
                self.firewall.ban_ip(ip)

    def _run_loop(self):
        """Background loop that calls tick() repeatedly."""
        while self._running:
            self.tick()
            time.sleep(self.poll_interval)

    def start(self):
        """Set up firewall, start tailer, and begin background monitoring."""
        logger.info("Setting up nftables firewall rules...")
        self.firewall.setup()
        self.tailer.start()
        self.start_time = time.time()
        self._running = True
        self._thread = threading.Thread(target=self._run_loop, daemon=True)
        self._thread.start()
        logger.info("Detection engine started on: %s", self.log_file)

    def stop(self):
        """Stop monitoring and clean up firewall rules."""
        self._running = False
        if self._thread:
            self._thread.join(timeout=5)
        self.firewall.teardown()
        logger.info("Detection engine stopped.")

    def get_metrics(self) -> dict:
        """Return current metrics for the dashboard."""
        uptime = time.time() - self.start_time
        hours = int(uptime // 3600)
        minutes = int((uptime % 3600) // 60)
        seconds = int(uptime % 60)

        all_ips = self.detector.get_all_ips()
        top_ips = sorted(
            all_ips,
            key=lambda ip: self.detector.get_request_count(ip),
            reverse=True,
        )[:10]

        top_ips_data = []
        for ip in top_ips:
            count = self.detector.get_request_count(ip)
            rate = count / max(self.detector.window_seconds, 1)
            status = "BANNED" if self.firewall.is_banned(ip) else "OK"
            top_ips_data.append({
                "ip": ip,
                "rate": f"{rate:.2f}",
                "status": status,
            })

        return {
            "uptime": f"{hours:02d}h {minutes:02d}m {seconds:02d}s",
            "active_ips": len(all_ips),
            "banned_count": len(self.firewall.list_banned()),
            "banned_ips": list(self.firewall.list_banned()),
            "global_rate": f"{self.detector.get_total_requests_per_second():.2f}",
            "window_seconds": self.detector.window_seconds,
            "max_requests": self.detector.max_requests,
            "top_ips": top_ips_data,
        }

Build the attack simulator

You need traffic to test your engine against. The simulator writes fake log entries to test.log in three phases: normal traffic from five IPs, a burst attack from a single attacker IP, then a return to normal.

• Open simulator.py and add the imports, constants, and the helper function that writes a single log line:

import random
import time
import sys

# The log file the engine will monitor
LOG_FILE = "test.log"

# Five "normal" IPs that generate baseline traffic
NORMAL_IPS = [
    "10.0.0.1",
    "10.0.0.2",
    "10.0.0.3",
    "10.0.0.4",
    "10.0.0.5",
]

# The single IP that will launch a burst attack
ATTACKER_IP = "192.168.66.6"


def write_log_line(f, ip: str):
    """Write a fake log entry to the file."""
    timestamp = time.strftime("%Y-%m-%d %H:%M:%S")
    line = f"{timestamp} server sshd[12345]: Connection from {ip} port {random.randint(1024, 65535)}\n"
    f.write(line)
    f.flush()

What does this code do?

• NORMAL_IPS defines five fake IPs that represent legitimate users generating background traffic.
• ATTACKER_IP is the single IP that will flood the log file during the attack phase.
• write_log_line() writes a line that looks like an SSH connection log. The f.flush() call forces the line to disk immediately so the tailer picks it up right away.

• You won't see a change yet. The next chunk adds the main() function that runs the three simulation phases.

• Below write_log_line() in simulator.py, add the main() function and the entry point:

def main():
    print(f"Simulator writing to: {LOG_FILE}")
    print("Phase 1: Normal traffic (10 seconds)")
    print("Phase 2: Attack burst from", ATTACKER_IP)
    print("Press Ctrl+C to stop\n")

    with open(LOG_FILE, "a") as f:
        # Phase 1: Normal traffic
        for i in range(20):
            ip = random.choice(NORMAL_IPS)
            write_log_line(f, ip)
            print(f"  [normal] {ip}")
            time.sleep(0.5)

        # Phase 2: Attack burst
        print(f"\n--- ATTACK BURST from {ATTACKER_IP} ---\n")
        for i in range(50):
            write_log_line(f, ATTACKER_IP)
            # Mix in some normal traffic too
            if random.random() < 0.3:
                normal_ip = random.choice(NORMAL_IPS)
                write_log_line(f, normal_ip)
                print(f"  [normal] {normal_ip}")
            print(f"  [ATTACK] {ATTACKER_IP} ({i + 1}/50)")
            time.sleep(0.1)

        print("\n--- Attack burst complete ---")
        print("Continuing normal traffic. Press Ctrl+C to stop.")

        # Phase 3: Return to normal
        try:
            while True:
                ip = random.choice(NORMAL_IPS)
                write_log_line(f, ip)
                print(f"  [normal] {ip}")
                time.sleep(1.0)
        except KeyboardInterrupt:
            print("\nSimulator stopped.")


if __name__ == "__main__":
    main()

What does this code do?

The simulator runs three phases to mimic a real attack scenario:

• Phase 1 writes 20 normal log entries at half-second intervals, spread across five IPs. This establishes baseline traffic.
• Phase 2 fires 50 rapid entries from 192.168.66.6 at 0.1-second intervals, with a 30% chance of normal traffic mixed in. This should exceed the engine's 20-request threshold.
• Phase 3 returns to normal traffic at one-second intervals until you press Ctrl+C.

• Save simulator.py.

Seeing a SyntaxError in the f-strings?

Make sure the curly braces inside the f-strings are not escaped or doubled. For example, f" [ATTACK] {ATTACKER_IP}" should have single curly braces around ATTACKER_IP.

Help me fix a SyntaxError in my simulator.py f-strings

✔️ Awesome, I've got everything!

Great. Double check you have saved simulator.py before moving on.

ⓧ I'd like to double check the full code

import random
import time
import sys

LOG_FILE = "test.log"

NORMAL_IPS = [
    "10.0.0.1",
    "10.0.0.2",
    "10.0.0.3",
    "10.0.0.4",
    "10.0.0.5",
]

ATTACKER_IP = "192.168.66.6"


def write_log_line(f, ip: str):
    """Write a fake log entry to the file."""
    timestamp = time.strftime("%Y-%m-%d %H:%M:%S")
    line = f"{timestamp} server sshd[12345]: Connection from {ip} port {random.randint(1024, 65535)}\n"
    f.write(line)
    f.flush()


def main():
    print(f"Simulator writing to: {LOG_FILE}")
    print("Phase 1: Normal traffic (10 seconds)")
    print("Phase 2: Attack burst from", ATTACKER_IP)
    print("Press Ctrl+C to stop\n")

    with open(LOG_FILE, "a") as f:
        # Phase 1: Normal traffic
        for i in range(20):
            ip = random.choice(NORMAL_IPS)
            write_log_line(f, ip)
            print(f"  [normal] {ip}")
            time.sleep(0.5)

        # Phase 2: Attack burst
        print(f"\n--- ATTACK BURST from {ATTACKER_IP} ---\n")
        for i in range(50):
            write_log_line(f, ATTACKER_IP)
            # Mix in some normal traffic too
            if random.random() < 0.3:
                normal_ip = random.choice(NORMAL_IPS)
                write_log_line(f, normal_ip)
                print(f"  [normal] {normal_ip}")
            print(f"  [ATTACK] {ATTACKER_IP} ({i + 1}/50)")
            time.sleep(0.1)

        print("\n--- Attack burst complete ---")
        print("Continuing normal traffic. Press Ctrl+C to stop.")

        # Phase 3: Return to normal
        try:
            while True:
                ip = random.choice(NORMAL_IPS)
                write_log_line(f, ip)
                print(f"  [normal] {ip}")
                time.sleep(1.0)
        except KeyboardInterrupt:
            print("\nSimulator stopped.")


if __name__ == "__main__":
    main()

Test the full detection pipeline

Now it is time to see the engine in action. You will run the simulator in one terminal and a quick test script in another to watch the engine detect the attack and ban the offending IP.

You need two terminal windows open, both inside your project folder. The simulator does not need root access, but the engine does because it runs nft commands.

• If a test.log file exists from earlier testing, clear it by running this command in your first terminal:

> test.log

• Create a small test script by running this command:

touch test_engine.py

• Open test_engine.py and add this temporary test harness:

import time
import logging

# Set up logging so we see the engine's output
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    datefmt="%H:%M:%S",
)

from engine import DetectionEngine

engine = DetectionEngine()
engine.start()

try:
    while True:
        time.sleep(5)
        metrics = engine.get_metrics()
        print(
            f"Active IPs: {metrics['active_ips']}, "
            f"Banned: {metrics['banned_count']}, "
            f"Rate: {metrics['global_rate']} req/s"
        )
except KeyboardInterrupt:
    engine.stop()

What does this script do?

This temporary script starts the detection engine, then prints a metrics summary every 5 seconds. The engine's own logging will print ANOMALY DETECTED and BANNED IP messages as it detects the attack. Press Ctrl+C to stop and clean up the firewall rules.

• Save test_engine.py.
• In your first terminal, start the engine test script by running:

sudo uv run test_engine.py

Why sudo?

The engine calls nft commands through subprocess to create firewall rules. Modifying nftables requires root privileges, so you need sudo.

• You should see the engine start up with messages like Setting up nftables firewall rules... and Detection engine started on: test.log.
• Open a second terminal in the same project folder and start the simulator by running:

uv run simulator.py

• Watch both terminals. The simulator terminal prints [normal] and [ATTACK] labels as it writes log entries.
• In the engine terminal, you should see an ANOMALY DETECTED warning followed by a BANNED IP: 192.168.66.6 message once the attacker exceeds 20 requests in the 30-second window.
• Once you see the BANNED IP message, press Ctrl+C in the simulator terminal first, then Ctrl+C in the engine terminal to clean up the firewall rules.

Not seeing ANOMALY DETECTED?

• Make sure the engine terminal started before the simulator. If the simulator finished its attack burst before the engine began tailing, the engine may have missed the attack lines.
• Check that both scripts use the same log file. The engine defaults to test.log and the simulator writes to test.log.
• If you see nft command failed warnings, make sure you ran the engine script with sudo.

Help me debug why my detection engine is not detecting the simulated attack

Your detection engine is working. It reads log traffic in real time, flags anomalous IPs, and bans them at the kernel level. Next up, you will build a live web dashboard so you can watch all of this happen through a browser instead of raw terminal output.

Build the FastAPI Dashboard Server

Your detection engine can tail logs, detect anomalies, and ban IPs at the kernel level. The missing piece is a way to see all of this happening in real time.

In this step, you'll build a FastAPI web server that hosts a live dashboard. The server collects metrics from your engine and system resources, then serves HTML fragments that the next step will transform into a professional monitoring UI.

In this step, get ready to:

• Create a FastAPI app with Jinja2 template rendering and an index route.
• Add a metrics endpoint that returns live data as an HTML fragment.
• Wire the detection engine into the server's startup and shutdown lifecycle.

Create the FastAPI app with Jinja2 templates

The dashboard follows a server-rendered architecture. FastAPI handles HTTP requests and uses Jinja2 to render HTML templates on the server before sending them to the browser.

• Open dashboard.py and add the imports, logging configuration, and engine instantiation by pasting this code:

import logging
import psutil
import time
from contextlib import asynccontextmanager
from datetime import datetime, timezone

from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse
from fastapi.templating import Jinja2Templates

from engine import DetectionEngine

# Configure logging format for readable terminal output
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    datefmt="%H:%M:%S",
)

# Create the detection engine at module level so routes can access it
engine = DetectionEngine(
    log_file="test.log",
    window_seconds=30,
    max_requests=20,
    poll_interval=1.0,
)

What does this code do?

The imports pull in your DetectionEngine alongside FastAPI, Jinja2 templates, and psutil for CPU and memory monitoring.

logging.basicConfig configures a clean timestamp format so terminal output is easy to read. The engine is created at module level with the same settings you tested earlier, so every route handler can access it directly.

You won't see a change yet. The next chunk uses these to build the web server routes.

• Below the engine instantiation in dashboard.py, add the FastAPI app, templates directory, and the index route by pasting this code:

# Create the FastAPI app and point Jinja2 at the templates folder
app = FastAPI()
templates = Jinja2Templates(directory="templates")


@app.get("/", response_class=HTMLResponse)
async def index(request: Request):
    # Serve the main dashboard HTML page
    return templates.TemplateResponse(request=request, name="index.html")

What does this code do?

• Jinja2Templates(directory="templates") tells FastAPI where to find your HTML template files.
• The GET / route returns the full index.html page. This is the shell that loads once and then auto-refreshes its content via HTMX.
• response_class=HTMLResponse tells FastAPI this route returns HTML instead of JSON.

• Save dashboard.py.

You won't see a visible change yet because the templates are still empty. The next substep adds the metrics endpoint that powers the live data feed.

Why FastAPI and HTMX instead of a JavaScript framework?

A monitoring dashboard needs to refresh data every few seconds. You could build a React or Vue frontend that fetches JSON from an API. That works, but it means maintaining two separate codebases and writing client-side rendering logic.

With FastAPI and HTMX, the server renders HTML fragments and HTMX swaps them into the page automatically. Zero JavaScript to write. The server returns ready-to-display HTML, and HTMX handles the polling and DOM updates.

Add a /partials/metrics endpoint

The dashboard's live data comes from a single endpoint that returns an HTML fragment. Every 2 seconds, HTMX will request this endpoint and swap the new HTML into the page.

• In dashboard.py, add the metrics partial endpoint below the index route by pasting this code:

@app.get("/partials/metrics", response_class=HTMLResponse)
async def metrics_partial(request: Request):
    # Grab detection engine metrics (uptime, IPs, bans, rates)
    metrics = engine.get_metrics()
    # Add system resource metrics from psutil
    metrics["cpu_percent"] = psutil.cpu_percent(interval=None)
    mem = psutil.virtual_memory()
    metrics["mem_percent"] = mem.percent
    metrics["mem_used_mb"] = f"{mem.used / (1024 * 1024):.1f}"
    metrics["mem_total_gb"] = f"{mem.total / (1024 * 1024 * 1024):.1f}"
    # Add current UTC timestamp for the navbar
    metrics["utc_time"] = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
    return templates.TemplateResponse(
        request=request, name="partials/metrics.html", context=metrics
    )

What does this code do?

• engine.get_metrics() returns a dictionary with uptime, active IPs, banned IPs, global request rate, and top talkers. This is the get_metrics() method you built in engine.py.
• psutil.cpu_percent(interval=None) returns the CPU usage as a non-blocking percentage. psutil.virtual_memory() provides memory stats in bytes, which we convert to MB and GB.
• The context=metrics parameter passes the full dictionary into the Jinja2 template. Every key becomes a variable you can use in the HTML with {{ variable_name }} syntax.

• Save dashboard.py.

The endpoint is ready, but you still need to wire the engine's background thread into the server's lifecycle. The next substep handles that.

Wire the detection engine into FastAPI's lifespan

Your engine needs to start its background monitoring thread when the server boots and clean up nftables rules when the server shuts down. FastAPI provides a lifespan context manager for exactly this pattern.

• In dashboard.py, find this line:

app = FastAPI()

• Add the lifespan function directly above that line by pasting this code:

# Lifespan context manager: start engine on boot, stop on shutdown
@asynccontextmanager
async def lifespan(app: FastAPI):
    engine.start()
    yield
    engine.stop()

What is a lifespan context manager?

Code before yield runs when the server starts. Code after yield runs when the server shuts down. This guarantees your nftables rules are always cleaned up, even if you stop the server with Ctrl+C.

• Now update the app = FastAPI() line to pass in the lifespan. Find this line:

app = FastAPI()

• Replace it with:

app = FastAPI(lifespan=lifespan)

• Save dashboard.py.
• Start the server by running this command:

sudo uv run fastapi dev dashboard.py

Why sudo?

The nft commands that set up the firewall table and blackhole set require root privileges. Running with sudo ensures the engine's setup() call can create the nftables rules.

You should see output in the terminal including Setting up nftables firewall rules... and Detection engine started on: test.log. The server is running and the engine is monitoring in the background.

Server not starting?

• If you see ModuleNotFoundError: No module named 'fastapi', make sure you run with sudo uv run (not just sudo python). The uv run command activates the virtual environment where your dependencies are installed.
• If you see nft command failed warnings, confirm that nftables is installed on your system and that you included sudo in the command.
• If the import for engine fails, make sure engine.py is in the same directory as dashboard.py.

Help me debug my FastAPI dashboard server startup

• Press Ctrl+C to stop the server. You should see Detection engine stopped. confirming that the lifespan teardown ran and cleaned up the nftables rules.

✔️ Awesome, I've got everything!

Double check you've saved dashboard.py and move on to the next step.

ⓧ I'd like to double check the full code

Here is the complete dashboard.py file. Compare it with yours to spot any differences.

import logging
import psutil
import time
from contextlib import asynccontextmanager
from datetime import datetime, timezone

from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse
from fastapi.templating import Jinja2Templates

from engine import DetectionEngine

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    datefmt="%H:%M:%S",
)

engine = DetectionEngine(
    log_file="test.log",
    window_seconds=30,
    max_requests=20,
    poll_interval=1.0,
)


@asynccontextmanager
async def lifespan(app: FastAPI):
    engine.start()
    yield
    engine.stop()


app = FastAPI(lifespan=lifespan)
templates = Jinja2Templates(directory="templates")


@app.get("/", response_class=HTMLResponse)
async def index(request: Request):
    return templates.TemplateResponse(request=request, name="index.html")


@app.get("/partials/metrics", response_class=HTMLResponse)
async def metrics_partial(request: Request):
    metrics = engine.get_metrics()
    metrics["cpu_percent"] = psutil.cpu_percent(interval=None)
    mem = psutil.virtual_memory()
    metrics["mem_percent"] = mem.percent
    metrics["mem_used_mb"] = f"{mem.used / (1024 * 1024):.1f}"
    metrics["mem_total_gb"] = f"{mem.total / (1024 * 1024 * 1024):.1f}"
    metrics["utc_time"] = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
    return templates.TemplateResponse(
        request=request, name="partials/metrics.html", context=metrics
    )

Your server is ready to serve live metrics. Next up, you'll build the HTML templates that turn this raw data into a professional dark-themed monitoring dashboard.

Build the Dashboard UI with HTMX

Your detection engine is running and the simulator can trigger bans. Now it's time to give your engine a face.

The dashboard uses two HTML templates. The first is the main page that loads HTMX and defines the visual theme. The second is a partial fragment that HTMX polls every 2 seconds to refresh the metrics on screen.

In this step, get ready to:

• Create the main dashboard page with HTMX and dark-themed CSS.
• Create the metrics partial template with live data cards.
• Run the full stack and watch the dashboard detect an attack in real time.

Create the main dashboard page

The index.html file is the shell page your browser loads. It contains the HTMX script tag, all the CSS for the dark theme, and a single div that HTMX uses to poll the server and swap in fresh content every 2 seconds.

• Open templates/index.html and paste the following code:

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>DDoS Detector - Live Metrics</title>
    <!-- Load HTMX from CDN for auto-polling -->
    <script src="https://cdn.jsdelivr.net/npm/htmx.org@2.0.9/dist/htmx.min.js"></script>
    <style>
        /* Reset and base styles */
        * { margin: 0; padding: 0; box-sizing: border-box; }
        body { font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif; background: #0f1923; color: #c5d0dc; min-height: 100vh; }

        /* Top navigation bar */
        .navbar { background: #1a2736; padding: 12px 24px; display: flex; justify-content: space-between; align-items: center; border-bottom: 1px solid #2a3a4a; }
        .navbar h1 { font-size: 16px; color: #4fc3f7; }
        .navbar .meta { font-size: 13px; color: #7a8a9a; }
        .navbar .meta strong { color: #c5d0dc; }

        /* Dashboard grid layout */
        .grid { display: grid; grid-template-columns: repeat(4, 1fr); gap: 20px; padding: 24px; max-width: 1400px; margin: 0 auto; }

        /* Metric cards */
        .card { background: #1a2736; border: 1px solid #2a3a4a; border-radius: 8px; padding: 20px; }
        .card h2 { font-size: 11px; text-transform: uppercase; letter-spacing: 1px; color: #7a8a9a; margin-bottom: 16px; }
        .card .big-number { font-size: 48px; font-weight: 700; }
        .card .big-number.blue { color: #4fc3f7; }
        .card .big-number.red { color: #ef5350; }
        .card .big-number.green { color: #66bb6a; }
        .card .subtitle { font-size: 13px; color: #7a8a9a; margin-top: 4px; }

        /* Status badges */
        .badge { display: inline-block; padding: 4px 12px; border-radius: 12px; font-size: 12px; font-weight: 600; }
        .badge.active { background: #1b5e20; color: #66bb6a; }
        .badge.banned { background: #b71c1c33; color: #ef5350; }

        /* Key-value rows inside cards */
        .stat-row { display: flex; justify-content: space-between; padding: 8px 0; border-bottom: 1px solid #2a3a4a22; font-size: 14px; }
        .stat-row:last-child { border-bottom: none; }
        .stat-row .label { color: #7a8a9a; }
        .stat-row .value { color: #c5d0dc; font-weight: 600; }
        .stat-row .value.highlight { color: #ef5350; }

        /* Top IPs table */
        table { width: 100%; border-collapse: collapse; font-size: 13px; }
        th { text-align: left; color: #7a8a9a; padding: 8px 4px; border-bottom: 1px solid #2a3a4a; }
        td { padding: 8px 4px; border-bottom: 1px solid #2a3a4a22; }

        /* CPU and memory progress bars */
        .progress-bar { height: 6px; background: #2a3a4a; border-radius: 3px; margin-top: 6px; }
        .progress-fill { height: 100%; border-radius: 3px; transition: width 0.3s; }
        .progress-fill.cpu { background: #7a8a9a; }
        .progress-fill.mem { background: #4fc3f7; }

        /* Utility classes */
        .span-2 { grid-column: span 2; }
        .empty-state { color: #5a6a7a; font-style: italic; padding: 12px 0; font-size: 13px; }
    </style>
</head>
<body>
    <!-- HTMX polls /partials/metrics on load and every 2 seconds -->
    <div id="dashboard"
         hx-get="/partials/metrics"
         hx-trigger="load, every 2s"
         hx-swap="innerHTML">
        <div class="navbar">
            <h1>DDoS Detector - Live Metrics</h1>
            <div class="meta">Loading...</div>
        </div>
    </div>
</body>
</html>

What does this code do?

The <script> tag loads HTMX 2.0.9 from a CDN. HTMX is a tiny JavaScript library that lets you add dynamic behavior using HTML attributes instead of writing JavaScript.

The three hx- attributes on the div#dashboard are where the magic happens:

• hx-get="/partials/metrics" tells HTMX to make a GET request to your FastAPI endpoint.
• hx-trigger="load, every 2s" fires the request on page load and then every 2 seconds.
• hx-swap="innerHTML" replaces the contents of the div with the server's HTML response.

• Save templates/index.html.

You won't see the dashboard yet. The metrics partial template still needs to be created so the server has something to return on each poll.

Seeing a syntax highlighting error in your editor?

Some editors flag the hx- attributes as unknown HTML. This is safe to ignore. HTMX attributes are processed by the library at runtime, not by the browser's HTML parser.

Why is my editor showing warnings on HTMX attributes?

✔️ Awesome, I've got everything!

Great. Double check you've saved templates/index.html before moving on.

ⓧ I'd like to double check the full code

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>DDoS Detector - Live Metrics</title>
    <script src="https://cdn.jsdelivr.net/npm/htmx.org@2.0.9/dist/htmx.min.js"></script>
    <style>
        * { margin: 0; padding: 0; box-sizing: border-box; }
        body { font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif; background: #0f1923; color: #c5d0dc; min-height: 100vh; }
        .navbar { background: #1a2736; padding: 12px 24px; display: flex; justify-content: space-between; align-items: center; border-bottom: 1px solid #2a3a4a; }
        .navbar h1 { font-size: 16px; color: #4fc3f7; }
        .navbar .meta { font-size: 13px; color: #7a8a9a; }
        .navbar .meta strong { color: #c5d0dc; }
        .grid { display: grid; grid-template-columns: repeat(4, 1fr); gap: 20px; padding: 24px; max-width: 1400px; margin: 0 auto; }
        .card { background: #1a2736; border: 1px solid #2a3a4a; border-radius: 8px; padding: 20px; }
        .card h2 { font-size: 11px; text-transform: uppercase; letter-spacing: 1px; color: #7a8a9a; margin-bottom: 16px; }
        .card .big-number { font-size: 48px; font-weight: 700; }
        .card .big-number.blue { color: #4fc3f7; }
        .card .big-number.red { color: #ef5350; }
        .card .big-number.green { color: #66bb6a; }
        .card .subtitle { font-size: 13px; color: #7a8a9a; margin-top: 4px; }
        .badge { display: inline-block; padding: 4px 12px; border-radius: 12px; font-size: 12px; font-weight: 600; }
        .badge.active { background: #1b5e20; color: #66bb6a; }
        .badge.banned { background: #b71c1c33; color: #ef5350; }
        .stat-row { display: flex; justify-content: space-between; padding: 8px 0; border-bottom: 1px solid #2a3a4a22; font-size: 14px; }
        .stat-row:last-child { border-bottom: none; }
        .stat-row .label { color: #7a8a9a; }
        .stat-row .value { color: #c5d0dc; font-weight: 600; }
        .stat-row .value.highlight { color: #ef5350; }
        table { width: 100%; border-collapse: collapse; font-size: 13px; }
        th { text-align: left; color: #7a8a9a; padding: 8px 4px; border-bottom: 1px solid #2a3a4a; }
        td { padding: 8px 4px; border-bottom: 1px solid #2a3a4a22; }
        .progress-bar { height: 6px; background: #2a3a4a; border-radius: 3px; margin-top: 6px; }
        .progress-fill { height: 100%; border-radius: 3px; transition: width 0.3s; }
        .progress-fill.cpu { background: #7a8a9a; }
        .progress-fill.mem { background: #4fc3f7; }
        .span-2 { grid-column: span 2; }
        .empty-state { color: #5a6a7a; font-style: italic; padding: 12px 0; font-size: 13px; }
    </style>
</head>
<body>
    <div id="dashboard"
         hx-get="/partials/metrics"
         hx-trigger="load, every 2s"
         hx-swap="innerHTML">
        <div class="navbar">
            <h1>DDoS Detector - Live Metrics</h1>
            <div class="meta">Loading...</div>
        </div>
    </div>
</body>
</html>

Create the metrics partial template

The partial is the HTML fragment that FastAPI returns every 2 seconds. It is not a full HTML page. It has no <html> or <head> tags because HTMX injects it directly into the existing page's div#dashboard.

This template uses Jinja2 template variables (the {{ }} syntax) to inject the live data that dashboard.py passes from the detection engine.

• Open templates/partials/metrics.html and paste the following navbar and first three dashboard cards:

<!-- Top bar with uptime and UTC clock -->
<div class="navbar">
    <h1>DDoS Detector - Live Metrics</h1>
    <div class="meta">
        Uptime: <strong>{{ uptime }}</strong> &nbsp; UTC: <strong>{{ utc_time }}</strong>
    </div>
</div>
<div class="grid">
    <!-- Card 1: System Status -->
    <div class="card">
        <h2>System Status</h2>
        <span class="badge active">ACTIVE</span>
        <div style="margin-top: 16px;">
            <div class="stat-row"><span class="label">Uptime</span><span class="value">{{ uptime }}</span></div>
            <div class="stat-row"><span class="label">Active IPs (window)</span><span class="value">{{ active_ips }}</span></div>
            <div class="stat-row"><span class="label">Currently Banned</span><span class="value">{{ banned_count }}</span></div>
        </div>
    </div>
    <!-- Card 2: Global Traffic rate -->
    <div class="card">
        <h2>Global Traffic</h2>
        <div class="big-number blue">{{ global_rate }}</div>
        <div class="subtitle">requests / second</div>
        <div style="margin-top: 16px;">
            <div class="stat-row"><span class="label">Window Size</span><span class="value">{{ window_seconds }}s</span></div>
            <div class="stat-row"><span class="label">Ban Threshold</span><span class="value highlight">{{ max_requests }} req/window</span></div>
        </div>
    </div>
    <!-- Card 3: Banned IPs with Jinja2 loop -->
    <div class="card">
        <h2>Banned IPs</h2>
        <div class="big-number red">{{ banned_count }}</div>
        <div class="subtitle">active bans</div>
        <div style="margin-top: 16px;">
            {% if banned_ips %}
                {% for ip in banned_ips %}
                    <div class="stat-row">
                        <span class="label">{{ ip }}</span>
                        <span class="badge banned">BANNED</span>
                    </div>
                {% endfor %}
            {% else %}
                <div class="empty-state">No IPs currently banned</div>
            {% endif %}
        </div>
    </div>

What does this code do?

• The navbar displays the engine's uptime and the current UTC time, both injected via Jinja2 variables.
• The System Status card shows an ACTIVE badge and three stat rows for uptime, active IPs, and banned count.
• The Global Traffic card shows the request rate as a large blue number, plus the detection window size and ban threshold.
• The Banned IPs card uses a Jinja2 {% for %} loop to list each banned IP. If none are banned, it shows a fallback "No IPs currently banned" message.

• Now paste the remaining three cards directly below the code you just added (still inside the same file):

    <!-- Card 4: Top 10 IPs table -->
    <div class="card">
        <h2>Top 10 IPs by Traffic</h2>
        {% if top_ips %}
            <table>
                <tr><th>IP Address</th><th>req/s</th><th>Status</th></tr>
                {% for entry in top_ips %}
                    <tr>
                        <td>{{ entry.ip }}</td>
                        <td>{{ entry.rate }}</td>
                        <td>
                            {% if entry.status == "BANNED" %}
                                <span class="badge banned">BANNED</span>
                            {% else %}
                                <span style="color: #66bb6a;">OK</span>
                            {% endif %}
                        </td>
                    </tr>
                {% endfor %}
            </table>
        {% else %}
            <div class="empty-state">No traffic recorded yet</div>
        {% endif %}
    </div>
    <!-- Card 5: CPU and memory gauges -->
    <div class="card">
        <h2>System Resources</h2>
        <div class="stat-row"><span class="label">CPU Usage</span><span class="value">{{ cpu_percent }}%</span></div>
        <div class="progress-bar"><div class="progress-fill cpu" style="width: {{ cpu_percent }}%"></div></div>
        <div style="margin-top: 12px;">
            <div class="stat-row"><span class="label">Memory Usage</span><span class="value">{{ mem_percent }}%</span></div>
            <div class="progress-bar"><div class="progress-fill mem" style="width: {{ mem_percent }}%"></div></div>
            <div style="margin-top: 8px; font-size: 12px; color: #5a6a7a;">{{ mem_used_mb }} MB used of {{ mem_total_gb }} GB</div>
        </div>
    </div>
    <!-- Card 6: Detection config summary (spans 2 columns) -->
    <div class="card span-2">
        <h2>Detection Configuration</h2>
        <div class="stat-row"><span class="label">Window Size</span><span class="value">{{ window_seconds }}s</span></div>
        <div class="stat-row"><span class="label">Ban Threshold</span><span class="value">{{ max_requests }} requests</span></div>
        <div class="stat-row"><span class="label">Active IPs in Window</span><span class="value">{{ active_ips }}</span></div>
        <div class="stat-row"><span class="label">Global Rate</span><span class="value">{{ global_rate }} req/s</span></div>
    </div>
</div>

What does this code do?

• The Top 10 IPs card renders an HTML table. Each row shows an IP address, its request rate, and a BANNED badge or green OK label.
• The System Resources card displays CPU and memory usage with animated progress bars. The psutil data is injected by dashboard.py before rendering.
• The Detection Configuration card uses span-2 to stretch across two grid columns. It summarizes the engine's current settings.

• Save templates/partials/metrics.html.

Both templates are ready. The next substep brings everything to life by running the dashboard and simulator together.

Seeing Jinja2 syntax errors?

Make sure every {% if %} has a matching {% endif %} and every {% for %} has a matching {% endfor %}. A missing closing tag causes a Jinja2 TemplateSyntaxError at runtime.

Help me fix a Jinja2 template syntax error in my metrics.html partial.

✔️ Awesome, I've got everything!

Great. Double check you've saved templates/partials/metrics.html before moving on.

ⓧ I'd like to double check the full code

<div class="navbar">
    <h1>DDoS Detector - Live Metrics</h1>
    <div class="meta">
        Uptime: <strong>{{ uptime }}</strong> &nbsp; UTC: <strong>{{ utc_time }}</strong>
    </div>
</div>
<div class="grid">
    <div class="card">
        <h2>System Status</h2>
        <span class="badge active">ACTIVE</span>
        <div style="margin-top: 16px;">
            <div class="stat-row"><span class="label">Uptime</span><span class="value">{{ uptime }}</span></div>
            <div class="stat-row"><span class="label">Active IPs (window)</span><span class="value">{{ active_ips }}</span></div>
            <div class="stat-row"><span class="label">Currently Banned</span><span class="value">{{ banned_count }}</span></div>
        </div>
    </div>
    <div class="card">
        <h2>Global Traffic</h2>
        <div class="big-number blue">{{ global_rate }}</div>
        <div class="subtitle">requests / second</div>
        <div style="margin-top: 16px;">
            <div class="stat-row"><span class="label">Window Size</span><span class="value">{{ window_seconds }}s</span></div>
            <div class="stat-row"><span class="label">Ban Threshold</span><span class="value highlight">{{ max_requests }} req/window</span></div>
        </div>
    </div>
    <div class="card">
        <h2>Banned IPs</h2>
        <div class="big-number red">{{ banned_count }}</div>
        <div class="subtitle">active bans</div>
        <div style="margin-top: 16px;">
            {% if banned_ips %}
                {% for ip in banned_ips %}
                    <div class="stat-row">
                        <span class="label">{{ ip }}</span>
                        <span class="badge banned">BANNED</span>
                    </div>
                {% endfor %}
            {% else %}
                <div class="empty-state">No IPs currently banned</div>
            {% endif %}
        </div>
    </div>
    <div class="card">
        <h2>Top 10 IPs by Traffic</h2>
        {% if top_ips %}
            <table>
                <tr><th>IP Address</th><th>req/s</th><th>Status</th></tr>
                {% for entry in top_ips %}
                    <tr>
                        <td>{{ entry.ip }}</td>
                        <td>{{ entry.rate }}</td>
                        <td>
                            {% if entry.status == "BANNED" %}
                                <span class="badge banned">BANNED</span>
                            {% else %}
                                <span style="color: #66bb6a;">OK</span>
                            {% endif %}
                        </td>
                    </tr>
                {% endfor %}
            </table>
        {% else %}
            <div class="empty-state">No traffic recorded yet</div>
        {% endif %}
    </div>
    <div class="card">
        <h2>System Resources</h2>
        <div class="stat-row"><span class="label">CPU Usage</span><span class="value">{{ cpu_percent }}%</span></div>
        <div class="progress-bar"><div class="progress-fill cpu" style="width: {{ cpu_percent }}%"></div></div>
        <div style="margin-top: 12px;">
            <div class="stat-row"><span class="label">Memory Usage</span><span class="value">{{ mem_percent }}%</span></div>
            <div class="progress-bar"><div class="progress-fill mem" style="width: {{ mem_percent }}%"></div></div>
            <div style="margin-top: 8px; font-size: 12px; color: #5a6a7a;">{{ mem_used_mb }} MB used of {{ mem_total_gb }} GB</div>
        </div>
    </div>
    <div class="card span-2">
        <h2>Detection Configuration</h2>
        <div class="stat-row"><span class="label">Window Size</span><span class="value">{{ window_seconds }}s</span></div>
        <div class="stat-row"><span class="label">Ban Threshold</span><span class="value">{{ max_requests }} requests</span></div>
        <div class="stat-row"><span class="label">Active IPs in Window</span><span class="value">{{ active_ips }}</span></div>
        <div class="stat-row"><span class="label">Global Rate</span><span class="value">{{ global_rate }} req/s</span></div>
    </div>
</div>

Why HTMX instead of JavaScript?

A traditional JavaScript approach would require writing fetch calls, parsing JSON responses, and manually updating DOM elements. HTMX flips this pattern. The server returns ready-to-display HTML fragments, and HTMX handles the swap.

For a monitoring dashboard that refreshes on a timer, this is a perfect fit. You get live updates with zero JavaScript written by hand.

Run the full stack and watch it live

You need two terminals running simultaneously. The first runs the dashboard server (which starts the detection engine). The second runs the attack simulator that writes fake log lines.

• In your first terminal, start the dashboard server by running this command:

sudo uv run fastapi dev dashboard.py

Why sudo?

The detection engine calls nft commands to manage nftables firewall rules. These require root privileges. Running with sudo gives the entire process the permissions it needs.

• You should see output showing the server starting on http://127.0.0.1:8000 and log messages like Setting up nftables firewall rules... and Detection engine started on: test.log.
• Open a second terminal in the same project directory.
• Start the attack simulator by running this command:

uv run python simulator.py

• Open your browser and navigate to http://127.0.0.1:8000.

You should see the dark-themed dashboard with all six cards. The metrics update every 2 seconds without you refreshing the page.

Watch the simulator terminal. During the attack burst phase, the attacker IP 192.168.66.6 appears in the Top 10 IPs table. Once it exceeds 20 requests in the 30-second window, the dashboard's Banned IPs card shows it with a red BANNED badge.

Dashboard loads but cards are empty?

Make sure the simulator is running in the second terminal. The dashboard has no data to show until log lines start appearing in test.log.

If you see a TemplateNotFound error in the server terminal, check that your templates/partials/ folder exists and metrics.html is saved inside it.

My dashboard is not showing any live data or I'm seeing an error page.

• When you are done watching, press Ctrl+C in the simulator terminal to stop it.
• Press Ctrl+C in the dashboard terminal to shut down the server. This also triggers the engine's teardown, which cleans up the nftables rules.

Secret mission

Add Auto-Unban TTL and a Manual Unban Button

Right now, once an IP is banned, it stays banned forever. In a real security system, bans should expire after a configurable cooldown period. Can you add auto-unban with a TTL, plus a manual Unban button on the dashboard?

Before you go

Clean Up Your Resources

Decide whether to keep your resources running, pause them to come back later, or delete them entirely. This project runs entirely locally, so there are no ongoing cloud costs.

Resources you used:

• Python project files: tailer.py, detector.py, banner.py, engine.py, simulator.py, dashboard.py.
• Template files: templates/index.html and templates/partials/metrics.html.
• UV project files: pyproject.toml, uv.lock, .venv virtual environment.
• nftables firewall rules: ddos_filter table with blackhole set.
• Log file: test.log generated by the simulator.

Keep everything running

No action needed. Choose this if you are still actively building or experimenting with the detection engine and dashboard.

• Your project files stay on your Desktop inside the project folder.
• The nftables rules persist until the next reboot or until you run the engine's teardown. Check if they are still active by running sudo nft list table inet ddos_filter.
• The test.log file grows each time you run the simulator. Delete it if it gets large.

Pause - I'll come back to this later

Shut down running processes and clean up firewall rules, but keep all your project files so you can pick up where you left off.

• Stop the FastAPI dashboard and the simulator by pressing Ctrl+C in each terminal window.
• Remove the nftables rules by running:

sudo nft delete table inet ddos_filter

• Delete the test.log file to free up space. A fresh one is created the next time you run the simulator.
• Your project folder, Python files, and virtual environment stay intact for next time.

Delete - I don't want to use this again

Remove all project resources and start fresh. This deletes everything you built in this project.

• Stop the FastAPI dashboard and the simulator by pressing Ctrl+C in each terminal window.
• Remove the nftables rules by running:

sudo nft delete table inet ddos_filter

• Navigate to your Desktop and delete the entire project folder by running:

cd ~/Desktop
rm -rf ddos-detector

• Confirm the folder is gone by running ls and checking that the project folder no longer appears.

Mission accomplished

That's a Wrap!

Nice work! You just built a fully functional DDoS detection engine with a live web dashboard from scratch using Python, Linux firewall primitives, and modern web techniques.

This is the exact pattern behind tools like fail2ban, but with a professional live UI on top.

You learned how to:

• Build a real-time log file tailer and a sliding window anomaly detector from scratch using Python's file I/O primitives and standard library.
• Integrate nftables to automatically ban malicious IPs at the Linux kernel level using named sets and subprocess calls.
• Create a professional dark-themed live dashboard with FastAPI and HTMX that auto-refreshes every 2 seconds without writing any JavaScript.
• Secret Mission: Add auto-unban with a configurable TTL and a manual Unban button on the dashboard.

Ready to quiz yourself?