Build an AI-Powered Self-Learning Firewall

Train an Isolation Forest model to detect network anomalies in real time.

Introduction

30 Second Summary

Traditional firewalls work like bouncers with a guest list. They only block threats someone has already identified and written a rule for. Novel attacks slip right past because no one told the firewall to look for them.

In this project, you will build a Python-based firewall that uses machine learning to learn what normal network traffic looks like on your machine. Instead of relying on predefined rules, it trains itself on your baseline behavior and flags anything that deviates.

What You'll Build

An intelligent firewall system that captures live network packets using Scapy, trains an Isolation Forest model on your traffic baseline, and performs real-time anomaly detection on every incoming packet.

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

A packet capture engine that extracts 8 machine learning features from live network traffic using Scapy.
A trained Isolation Forest model that identifies network anomalies without needing labeled attack data.
A real-time detection engine with JSON-based structured alerting, severity ratings, and an attack simulator to validate it catches port scans and floods.
Secret Mission: Auto-block repeat offender IPs by programmatically creating Windows Firewall rules.

Are there any prerequisites?

This project assumes intermediate Python experience (functions, classes, and file I/O) and basic networking knowledge (TCP/IP, ports, protocols). You will also need a Windows machine, since the project uses Npcap for packet capture and runs commands in an Administrator terminal.

Build the Packet Capture and Feature Extraction Engine

Your environment is set up and Scapy can see your network interface. Now it's time to build the core engine that turns raw network packets into structured data.

Machine learning models can't read raw packets directly. They need numeric features. In this step, you'll build a feature extraction function that pulls 8 meaningful numbers from every packet, then capture 3000 packets of your normal traffic to create a baseline dataset.

In this step, get ready to:

Create a feature extraction function that converts raw packets into numeric dictionaries.
Build a baseline capture script that collects 3000 packets of normal traffic.
Run the capture and save the results to a CSV file.

Create the feature extraction function

Every network packet carries metadata like protocol type, time-to-live, source port, and payload size. These are the raw signals your Isolation Forest model will learn from.

In your project folder, create a new file called capture_baseline.py.
Add the imports, constants, and the extract_features() function by pasting this code:

import sys
import time
from scapy.all import sniff, IP, TCP, UDP
import pandas as pd

PACKET_COUNT = 3000
packets_data = []


def extract_features(pkt):
    """Extract 8 numeric features from an IPv4 packet."""
    if not pkt.haslayer(IP):
        return None

    ip = pkt[IP]
    features = {
        "proto": ip.proto,
        "ttl": ip.ttl,
        "ip_len": ip.len,
        "flags": int(ip.flags),
        "sport": 0,
        "dport": 0,
        "tcp_flags": 0,
        "payload_len": len(ip.payload),
        "src_ip": ip.src,
        "dst_ip": ip.dst,
    }

    if pkt.haslayer(TCP):
        features["sport"] = pkt[TCP].sport
        features["dport"] = pkt[TCP].dport
        features["tcp_flags"] = int(pkt[TCP].flags)
    elif pkt.haslayer(UDP):
        features["sport"] = pkt[UDP].sport
        features["dport"] = pkt[UDP].dport

    return features

What does this code do?

PACKET_COUNT sets how many packets to collect for the baseline (3000 gives the model enough data to learn from). packets_data is the list that will hold every extracted feature dictionary.

The extract_features() function checks if a packet has an IP layer, then pulls 8 numeric values (protocol, TTL, length, flags, ports, TCP flags, payload size) plus the source and destination IPs. It handles TCP and UDP packets separately to extract port information.

Save capture_baseline.py.

You won't see a change yet. The next chunk adds the capture callback that uses this function.

Create the baseline capture script

With feature extraction ready, you need a callback that Scapy calls for every captured packet, plus a main function that kicks off the sniff and saves results to a CSV.

In capture_baseline.py, add the capture_callback() function below extract_features() by pasting this code:


def capture_callback(pkt):
    """Callback for sniff() that collects packet features."""
    features = extract_features(pkt)
    if features is not None:
        packets_data.append(features)
        count = len(packets_data)
        if count % 500 == 0:
            print(f"  Captured {count}/{PACKET_COUNT} packets...")
        if count >= PACKET_COUNT:
            return True

What does this code do?

Scapy's sniff() calls this function once per packet. It extracts features, appends them to the list, prints progress every 500 packets, and returns True once the target count is reached (which triggers the stop_filter to halt capture).

Save capture_baseline.py.

You won't see a change yet. The next chunk adds the main function that starts the capture and writes the CSV.

Still in capture_baseline.py, add the main() function and entry point below capture_callback():


def main():
    print(f"[*] Capturing {PACKET_COUNT} packets of baseline traffic...")
    print("[*] Browse the web, stream music, do normal things.")
    print()

    sniff(filter="ip", prn=capture_callback, store=False,
          stop_filter=lambda p: len(packets_data) >= PACKET_COUNT)

    df = pd.DataFrame(packets_data)
    df.to_csv("baseline_traffic.csv", index=False)
    print(f"\n[+] Saved {len(df)} packets to baseline_traffic.csv")
    print(f"[+] Columns: {list(df.columns)}")


if __name__ == "__main__":
    main()

What does this code do?

main() prints a reminder to generate normal traffic, then calls sniff() with the IP filter and your callback. The store=False flag tells Scapy not to keep raw packets in memory (you only need the extracted features).

Once 3000 packets are collected, it builds a pandas DataFrame from the list of dictionaries and saves it to baseline_traffic.csv. This CSV becomes the training data for your Isolation Forest model.

Save capture_baseline.py.

Seeing a syntax error?

Make sure the sniff() call spans exactly two lines with the second line indented to align with filter=. Check that the if __name__ block is at the very bottom of the file with no extra indentation.

help me fix a syntax error in capture_baseline.py

✔️ Awesome, I've got everything!

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

ⓧ I'd like to double check the full code

import sys
import time
from scapy.all import sniff, IP, TCP, UDP
import pandas as pd

PACKET_COUNT = 3000
packets_data = []


def extract_features(pkt):
    """Extract 8 numeric features from an IPv4 packet."""
    if not pkt.haslayer(IP):
        return None

    ip = pkt[IP]
    features = {
        "proto": ip.proto,
        "ttl": ip.ttl,
        "ip_len": ip.len,
        "flags": int(ip.flags),
        "sport": 0,
        "dport": 0,
        "tcp_flags": 0,
        "payload_len": len(ip.payload),
        "src_ip": ip.src,
        "dst_ip": ip.dst,
    }

    if pkt.haslayer(TCP):
        features["sport"] = pkt[TCP].sport
        features["dport"] = pkt[TCP].dport
        features["tcp_flags"] = int(pkt[TCP].flags)
    elif pkt.haslayer(UDP):
        features["sport"] = pkt[UDP].sport
        features["dport"] = pkt[UDP].dport

    return features


def capture_callback(pkt):
    """Callback for sniff() that collects packet features."""
    features = extract_features(pkt)
    if features is not None:
        packets_data.append(features)
        count = len(packets_data)
        if count % 500 == 0:
            print(f"  Captured {count}/{PACKET_COUNT} packets...")
        if count >= PACKET_COUNT:
            return True


def main():
    print(f"[*] Capturing {PACKET_COUNT} packets of baseline traffic...")
    print("[*] Browse the web, stream music, do normal things.")
    print()

    sniff(filter="ip", prn=capture_callback, store=False,
          stop_filter=lambda p: len(packets_data) >= PACKET_COUNT)

    df = pd.DataFrame(packets_data)
    df.to_csv("baseline_traffic.csv", index=False)
    print(f"\n[+] Saved {len(df)} packets to baseline_traffic.csv")
    print(f"[+] Columns: {list(df.columns)}")


if __name__ == "__main__":
    main()

Run the baseline capture

Your script is ready to capture live traffic. During the capture, you'll generate normal network activity so the model learns what "everyday" traffic looks like on your machine.

In your Administrator terminal, run the capture script:

python capture_baseline.py

While the script is running, browse a few websites, stream some music, or check your email. Do normal internet things for 2-3 minutes until the script finishes.

What should I see?

The terminal prints progress messages every 500 packets. Once it reaches 3000, it saves the CSV and prints a summary like:

[*] Capturing 3000 packets of baseline traffic...
[*] Browse the web, stream music, do normal things.

  Captured 500/3000 packets...
  Captured 1000/3000 packets...
  Captured 1500/3000 packets...
  Captured 2000/3000 packets...
  Captured 2500/3000 packets...
  Captured 3000/3000 packets...

[+] Saved 3000 packets to baseline_traffic.csv
[+] Columns: ['proto', 'ttl', 'ip_len', 'flags', 'sport', 'dport', 'tcp_flags', 'payload_len', 'src_ip', 'dst_ip']

Confirm the file was created by running:

dir baseline_traffic.csv

You should see baseline_traffic.csv listed with a file size of roughly 200-400 KB.

Script hangs or captures slowly?

If capture seems stuck, make sure you're actively generating traffic by browsing the web. Scapy only captures packets that pass through your interface, so idle machines produce very few packets.

If you see a permission error, confirm your terminal is running as Administrator. Scapy on Windows requires elevated privileges for packet capture.

help me troubleshoot why my capture_baseline.py script is not capturing packets

Your baseline dataset is locked in. Next up, you'll feed this CSV into an Isolation Forest model that learns what normal looks like on your network.

Train the Isolation Forest Model

In the previous step, you captured 3000 packets of your network's normal behavior into baseline_traffic.csv. That raw data is the foundation for your firewall's intelligence.

Now you'll train an Isolation Forest model on that baseline data. This unsupervised algorithm learns the shape of normal traffic and flags anything that deviates from it.

In this step, get ready to:

Load and normalize your baseline traffic data.
Train an Isolation Forest model on the scaled features.
Save the trained model to disk and verify it works.

Load the baseline data and fit a scaler

Raw feature values have wildly different scales. TTL ranges from 0 to 255, while port numbers go up to 65535. A StandardScaler normalizes everything to a common scale so the model treats each feature fairly.

Create a new file called train_model.py in your project folder.
Add the imports, feature column list, and data loading section by pasting this code:

import pandas as pd
import numpy as np
from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler
import joblib

FEATURE_COLS = ["proto", "ttl", "ip_len", "flags", "sport", "dport",
                "tcp_flags", "payload_len"]


def main():
    print("[*] Loading baseline traffic data...")
    df = pd.read_csv("baseline_traffic.csv")
    print(f"    Loaded {len(df)} samples")

    X = df[FEATURE_COLS].values
    print(f"    Features: {FEATURE_COLS}")

    print("[*] Fitting StandardScaler...")
    scaler = StandardScaler()
    X_scaled = scaler.fit_transform(X)

What does this code do?

FEATURE_COLS defines the 8 numeric columns the model will learn from. The src_ip and dst_ip columns are excluded because IP addresses are metadata, not numeric features.
df[FEATURE_COLS].values extracts just the 8 feature columns as a NumPy array.
StandardScaler transforms each column to have a mean of 0 and standard deviation of 1. This prevents high-range features like port numbers from dominating the model.

Save train_model.py.

You won't run this file yet. The next chunk adds the model training logic.

Train the Isolation Forest on scaled features

The Isolation Forest algorithm works by randomly partitioning data points. Points that require fewer random splits to isolate are more likely to be anomalies. The model learns what "normal" looks like from your baseline, then anything that isolates easily gets flagged.

In train_model.py, add the following code directly below the X_scaled = scaler.fit_transform(X) line, inside the main() function:

    print("[*] Training Isolation Forest...")
    model = IsolationForest(
        n_estimators=100,
        contamination=0.05,
        random_state=42,
        n_jobs=-1
    )
    model.fit(X_scaled)

What do these parameters mean?

n_estimators=100 builds 100 isolation trees. More trees means more stable anomaly scores.
contamination=0.05 tells the model to expect about 5% of the baseline data to be slightly unusual. This sets the decision boundary threshold.
random_state=42 makes training reproducible so you get consistent results each time.
n_jobs=-1 uses all CPU cores during training for faster fitting.

Save train_model.py.

You won't see a change yet. The next chunk saves the model to disk and prints a training summary.

Save the model and run the training script

Once trained, you need to persist the model and scaler to disk so the detection engine can load them later without retraining every time.

In train_model.py, add the following code directly below the model.fit(X_scaled) line, still inside the main() function:

    joblib.dump(model, "model.pkl")
    joblib.dump(scaler, "scaler.pkl")
    print("[+] Model saved to model.pkl")
    print("[+] Scaler saved to scaler.pkl")

    scores = model.decision_function(X_scaled)
    predictions = model.predict(X_scaled)
    n_anomalies = np.sum(predictions == -1)
    print(f"\n[+] Training summary:")
    print(f"    Samples: {len(X)}")
    print(f"    Anomalies detected in baseline: {n_anomalies} "
          f"({n_anomalies/len(X)*100:.1f}%)")
    print(f"    Score range: [{scores.min():.3f}, {scores.max():.3f}]")
    print(f"    Threshold (approx): {np.percentile(scores, 5):.3f}")


if __name__ == "__main__":
    main()

What does the training summary show?

joblib.dump() serializes the trained model and fitted scaler to .pkl files. Your detection engine will load these later.
decision_function() returns a continuous anomaly score for each sample. Lower scores mean more anomalous. Negative scores indicate outliers.
predict() returns 1 for normal traffic (inliers) and -1 for anomalies (outliers).
The threshold approximation shows the score below which packets will be flagged.

Save train_model.py.
Run the training script in your Administrator terminal:

python train_model.py

You should see output showing the model loading 3000 samples, fitting the scaler, training the forest, and printing a summary with the anomaly count and score range.

Confirm the model files were saved by running:

dir *.pkl

You should see model.pkl and scaler.pkl listed in the output.

Training script failing?

If you see FileNotFoundError for baseline_traffic.csv, make sure your terminal is in the same folder where you ran capture_baseline.py. Both scripts need to be in the same directory.
If you see ModuleNotFoundError, make sure your virtual environment is still active. You should see (venv) at the start of your terminal prompt.
If the anomaly percentage is much higher than 5%, your baseline capture may have included unusual traffic. You can re-run capture_baseline.py during calm browsing activity and retrain.

My training script is throwing an error. Help me debug my train_model.py script

Verify the model works by running a quick test in Python:

python -c "import joblib; import numpy as np; model=joblib.load('model.pkl'); scaler=joblib.load('scaler.pkl'); X=np.array([[6,64,52,2,443,55123,18,0]]); print('Prediction:', model.predict(scaler.transform(X)))"

You should see either [1] (normal) or [-1] (anomaly) printed. Both are valid. The important thing is that it runs without errors.

✔️ Awesome, I've got everything!

Great. Double check you've saved train_model.py and that both model.pkl and scaler.pkl exist in your project folder.

ⓧ I'd like to double check the full code

import pandas as pd
import numpy as np
from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler
import joblib

FEATURE_COLS = ["proto", "ttl", "ip_len", "flags", "sport", "dport",
                "tcp_flags", "payload_len"]


def main():
    print("[*] Loading baseline traffic data...")
    df = pd.read_csv("baseline_traffic.csv")
    print(f"    Loaded {len(df)} samples")

    X = df[FEATURE_COLS].values
    print(f"    Features: {FEATURE_COLS}")

    print("[*] Fitting StandardScaler...")
    scaler = StandardScaler()
    X_scaled = scaler.fit_transform(X)

    print("[*] Training Isolation Forest...")
    model = IsolationForest(
        n_estimators=100,
        contamination=0.05,
        random_state=42,
        n_jobs=-1
    )
    model.fit(X_scaled)

    joblib.dump(model, "model.pkl")
    joblib.dump(scaler, "scaler.pkl")
    print("[+] Model saved to model.pkl")
    print("[+] Scaler saved to scaler.pkl")

    scores = model.decision_function(X_scaled)
    predictions = model.predict(X_scaled)
    n_anomalies = np.sum(predictions == -1)
    print(f"\n[+] Training summary:")
    print(f"    Samples: {len(X)}")
    print(f"    Anomalies detected in baseline: {n_anomalies} "
          f"({n_anomalies/len(X)*100:.1f}%)")
    print(f"    Score range: [{scores.min():.3f}, {scores.max():.3f}]")
    print(f"    Threshold (approx): {np.percentile(scores, 5):.3f}")


if __name__ == "__main__":
    main()

Your firewall now has a trained brain that knows what normal network behavior looks like. Next up, you'll build the real-time detection engine that uses this model to score every incoming packet and flag threats.

Build the Real-Time Anomaly Detection Engine

You have a trained Isolation Forest model that knows what "normal" looks like on your network. Now it is time to put that model to work on live traffic.

In this step, you will build a real-time detection engine that scores every packet as it arrives. Packets that deviate from the learned baseline get flagged with a severity level and logged to a structured JSON file.

In this step, get ready to:

Create the detection script that loads your trained model and scaler.
Build the scoring callback that classifies each packet as normal or anomalous.
Add structured JSON logging for every detected anomaly.

Create the detection script and load the trained model

The detection engine needs access to the same model and scaler you trained in the previous step. It also reuses the same feature extraction logic from your baseline capture script.

Create a new file called detect.py in your project folder.
Add the imports, constants, and model loading at the top of the file:

import json
import time
from datetime import datetime
from scapy.all import sniff, IP, TCP, UDP
from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler
import joblib
import numpy as np

FEATURE_COLS = ["proto", "ttl", "ip_len", "flags", "sport", "dport",
                "tcp_flags", "payload_len"]
ALERTS_FILE = "alerts.json"

model = joblib.load("model.pkl")
scaler = joblib.load("scaler.pkl")
anomaly_count = 0
packet_count = 0

What does this code do?

The script loads your trained model and scaler into memory as soon as it starts. This means every packet can be scored instantly without reloading from disk.

The two counters (anomaly_count and packet_count) track session statistics so you can see your anomaly rate when you stop the engine.

Below the model loading code, add the extract_features function. This is the same feature extractor from capture_baseline.py:

def extract_features(pkt):
    """Extract 8 numeric features from an IPv4 packet."""
    if not pkt.haslayer(IP):
        return None

    ip = pkt[IP]
    features = {
        "proto": ip.proto,
        "ttl": ip.ttl,
        "ip_len": ip.len,
        "flags": int(ip.flags),
        "sport": 0,
        "dport": 0,
        "tcp_flags": 0,
        "payload_len": len(ip.payload),
        "src_ip": ip.src,
        "dst_ip": ip.dst,
    }

    if pkt.haslayer(TCP):
        features["sport"] = pkt[TCP].sport
        features["dport"] = pkt[TCP].dport
        features["tcp_flags"] = int(pkt[TCP].flags)
    elif pkt.haslayer(UDP):
        features["sport"] = pkt[UDP].sport
        features["dport"] = pkt[UDP].dport

    return features

Why reuse the same extractor?

The model was trained on features produced by this exact function. If you extract features differently at detection time, the model's predictions become meaningless. Consistency between training and inference is essential.

Save detect.py. You will keep adding functions to this file in the next two substeps.

Build the scoring callback with severity classification

Every packet that arrives needs to be scored by the model in real time. The callback function will extract features, scale them, and decide whether the packet is normal or anomalous.

Below extract_features, add the compute_severity function that maps anomaly scores to severity levels:

def compute_severity(score):
    """Map anomaly score to severity level."""
    if score < -0.5:
        return "CRITICAL"
    elif score < -0.3:
        return "MEDIUM"
    elif score < 0:
        return "LOW"
    return "NONE"

How do the severity thresholds work?

The decision_function in Isolation Forest returns a score where lower values mean more anomalous. A score below -0.5 indicates a packet that looks very different from anything in your training data. Scores between -0.3 and -0.5 are moderately unusual.

Below compute_severity, add the detect_callback function that scores each packet and prints anomaly alerts:

def detect_callback(pkt):
    """Score each packet and alert on anomalies."""
    global anomaly_count, packet_count

    features = extract_features(pkt)
    if features is None:
        return

    packet_count += 1
    src_ip = features["src_ip"]
    dst_ip = features["dst_ip"]

    feature_values = [features[col] for col in FEATURE_COLS]
    X = np.array([feature_values])
    X_scaled = scaler.transform(X)

    prediction = model.predict(X_scaled)[0]
    score = float(model.decision_function(X_scaled)[0])

    if prediction == -1:
        anomaly_count += 1
        severity = compute_severity(score)
        timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

        print(f"[ANOMALY] {timestamp} | {src_ip} -> {dst_ip} | "
              f"proto={features['proto']} dport={features['dport']} "
              f"score={score:.3f} | {severity}")

        alert_data = {
            "timestamp": timestamp,
            "src_ip": src_ip,
            "dst_ip": dst_ip,
            "proto": features["proto"],
            "dport": features["dport"],
            "anomaly_score": round(score, 4),
            "severity": severity,
            "features": {col: features[col] for col in FEATURE_COLS},
        }
        log_alert(alert_data)

What does this code do?

model.predict() returns 1 for normal packets and -1 for anomalies.
model.decision_function() gives a continuous anomaly score. Lower scores indicate more anomalous behavior.
When a packet is flagged (prediction == -1), the function prints a formatted alert to the console and builds a structured dictionary for logging.
The log_alert function called at the end is defined in the next substep.

Save detect.py. The script is not runnable yet because log_alert and main are not defined yet.

Add JSON logging and the main loop

Console output is useful for watching the engine in real time, but you also need a permanent record. Each anomaly gets appended to a JSON log file so you can review detected threats after a session ends.

In detect.py, add the log_alert function between compute_severity and detect_callback:

def log_alert(alert_data):
    """Append an alert record to the JSON log file."""
    with open(ALERTS_FILE, "a") as f:
        f.write(json.dumps(alert_data) + "\n")

Why newline-delimited JSON?

Each alert is written as its own JSON object on a single line. This format (sometimes called JSONL) lets you append new records without loading the entire file into memory. It also makes it easy to pipe through tools like grep or load into pandas with pd.read_json(lines=True).

Below detect_callback, add the main function that starts continuous packet sniffing:

def main():
    print("[*] AI Firewall - Real-Time Anomaly Detection")
    print("[*] Model loaded. Scaler loaded.")
    print("[*] Starting packet capture... (Ctrl+C to stop)")
    print()

    try:
        sniff(filter="ip", prn=detect_callback, store=False)
    except KeyboardInterrupt:
        pass

    print(f"\n[+] Session complete.")
    print(f"    Packets analyzed: {packet_count}")
    print(f"    Anomalies detected: {anomaly_count}")
    if packet_count > 0:
        print(f"    Anomaly rate: {anomaly_count/packet_count*100:.1f}%")


if __name__ == "__main__":
    main()

What does this code do?

sniff(filter="ip", prn=detect_callback, store=False) captures every IPv4 packet and passes it to your detect_callback function. Setting store=False prevents Scapy from accumulating packets in memory.
The try/except KeyboardInterrupt block lets you stop the engine cleanly by pressing Ctrl+C.
After stopping, the script prints a session summary showing how many packets it analyzed and how many anomalies it found.

Save detect.py.
Run the detection engine by running this command in your Administrator terminal:

python detect.py

You should see the startup banner:

[*] AI Firewall - Real-Time Anomaly Detection
[*] Model loaded. Scaler loaded.
[*] Starting packet capture... (Ctrl+C to stop)

Open a browser and visit a few websites. Normal traffic should pass silently with no (or very few) anomaly alerts appearing in the console.
After 30 seconds of normal browsing, press Ctrl+C to stop the engine. You should see a session summary showing the total packets analyzed and a low (or zero) anomaly count.

A few anomalies are normal

Your model was trained with contamination=0.05, meaning it expects about 5% of traffic to be flagged. Seeing a handful of LOW-severity alerts during normal browsing is expected behavior, not a bug.

Seeing an error when running detect.py?

If you see a permissions error, make sure you are running the command from an Administrator terminal. Scapy requires elevated privileges for packet capture on Windows.
If you see a file-not-found error for model.pkl or scaler.pkl, confirm that both files exist in the same folder as detect.py. Re-run train_model.py if they are missing.
If you see a NameError for log_alert, check that the log_alert function is defined above detect_callback in the file.

Why isn't my detect.py script starting?

✔️ Awesome, I've got everything!

Great. Double check you have saved detect.py and confirmed the detection engine starts without errors.

ⓧ I'd like to double check the full code

Here is the complete detect.py file. Make sure your code matches this exactly:

import json
import time
from datetime import datetime
from scapy.all import sniff, IP, TCP, UDP
from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler
import joblib
import numpy as np

FEATURE_COLS = ["proto", "ttl", "ip_len", "flags", "sport", "dport",
                "tcp_flags", "payload_len"]
ALERTS_FILE = "alerts.json"

model = joblib.load("model.pkl")
scaler = joblib.load("scaler.pkl")
anomaly_count = 0
packet_count = 0


def extract_features(pkt):
    """Extract 8 numeric features from an IPv4 packet."""
    if not pkt.haslayer(IP):
        return None

    ip = pkt[IP]
    features = {
        "proto": ip.proto,
        "ttl": ip.ttl,
        "ip_len": ip.len,
        "flags": int(ip.flags),
        "sport": 0,
        "dport": 0,
        "tcp_flags": 0,
        "payload_len": len(ip.payload),
        "src_ip": ip.src,
        "dst_ip": ip.dst,
    }

    if pkt.haslayer(TCP):
        features["sport"] = pkt[TCP].sport
        features["dport"] = pkt[TCP].dport
        features["tcp_flags"] = int(pkt[TCP].flags)
    elif pkt.haslayer(UDP):
        features["sport"] = pkt[UDP].sport
        features["dport"] = pkt[UDP].dport

    return features


def compute_severity(score):
    """Map anomaly score to severity level."""
    if score < -0.5:
        return "CRITICAL"
    elif score < -0.3:
        return "MEDIUM"
    elif score < 0:
        return "LOW"
    return "NONE"


def log_alert(alert_data):
    """Append an alert record to the JSON log file."""
    with open(ALERTS_FILE, "a") as f:
        f.write(json.dumps(alert_data) + "\n")


def detect_callback(pkt):
    """Score each packet and alert on anomalies."""
    global anomaly_count, packet_count

    features = extract_features(pkt)
    if features is None:
        return

    packet_count += 1
    src_ip = features["src_ip"]
    dst_ip = features["dst_ip"]

    feature_values = [features[col] for col in FEATURE_COLS]
    X = np.array([feature_values])
    X_scaled = scaler.transform(X)

    prediction = model.predict(X_scaled)[0]
    score = float(model.decision_function(X_scaled)[0])

    if prediction == -1:
        anomaly_count += 1
        severity = compute_severity(score)
        timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

        print(f"[ANOMALY] {timestamp} | {src_ip} -> {dst_ip} | "
              f"proto={features['proto']} dport={features['dport']} "
              f"score={score:.3f} | {severity}")

        alert_data = {
            "timestamp": timestamp,
            "src_ip": src_ip,
            "dst_ip": dst_ip,
            "proto": features["proto"],
            "dport": features["dport"],
            "anomaly_score": round(score, 4),
            "severity": severity,
            "features": {col: features[col] for col in FEATURE_COLS},
        }
        log_alert(alert_data)


def main():
    print("[*] AI Firewall - Real-Time Anomaly Detection")
    print("[*] Model loaded. Scaler loaded.")
    print("[*] Starting packet capture... (Ctrl+C to stop)")
    print()

    try:
        sniff(filter="ip", prn=detect_callback, store=False)
    except KeyboardInterrupt:
        pass

    print(f"\n[+] Session complete.")
    print(f"    Packets analyzed: {packet_count}")
    print(f"    Anomalies detected: {anomaly_count}")
    if packet_count > 0:
        print(f"    Anomaly rate: {anomaly_count/packet_count*100:.1f}%")


if __name__ == "__main__":
    main()

Your detection engine is live and scoring traffic. Next up, you will build an attack simulator to generate malicious traffic patterns and prove that the model catches them.

Build the Attack Simulator and Validate Detection

Your detection engine is live and scoring every packet that crosses your network interface. Normal browsing passes through silently, which is exactly what you want from a well-trained model.

Now comes the real test. You need to prove that the model actually catches malicious traffic. In this step, you will build an attack simulator that generates two distinct malicious traffic patterns and verify that your detection engine flags them in real time.

In this step, get ready to:

Build a port scan simulator that sends TCP SYN packets to 50 ports.
Build a UDP flood simulator that sends 200 identical packets to a single port.
Run both attack types against your live detection engine and confirm anomaly alerts fire.

Create the port scan simulator

A port scan probes many ports on a target machine in rapid succession, looking for open services. This traffic pattern is distinctive because normal applications connect to one or two ports, not fifty in a row.

Create a new file called attack_simulator.py in your project folder.
Add the imports, gateway target, and port scan function by pasting the following code:

import time
import sys
from scapy.all import IP, TCP, UDP, send, conf

TARGET_IP = conf.route.route("0.0.0.0")[2]  # Default gateway IP


def port_scan(target_ip, port_range=50):
    """Simulate a port scan by sending SYN packets to many ports."""
    print(f"[*] PORT SCAN: Sending SYN to ports 1-{port_range} on {target_ip}")

    for port in range(1, port_range + 1):
        pkt = IP(dst=target_ip) / TCP(dport=port, flags="S")
        send(pkt, verbose=False)

    print(f"[+] Port scan complete. Sent {port_range} SYN packets.")

What does this code do?

conf.route.route("0.0.0.0")[2] uses Scapy's routing table to auto-detect your default gateway IP. This is the target for the simulated attacks.

The port_scan() function sends a TCP SYN packet (flags="S") to each of the first 50 ports. SYN packets are the opening handshake of a TCP connection. Sending 50 of them to sequential ports in rapid succession is a textbook port scan signature.

Is this safe?

These simulated attacks target your own gateway (your router). The SYN packets never complete a connection, and the UDP flood targets a non-standard port. Neither causes damage to your network or any external system.

You won't see a change yet. The next chunk adds the flood simulator, and then you will run both attacks together.

Create the flood simulator

A UDP flood overwhelms a target by sending a massive volume of identical packets. Your model has never seen 200 packets to the same port with the exact same payload size, so it should flag these as anomalies.

In attack_simulator.py, add the udp_flood() function below the port_scan() function:

def udp_flood(target_ip, count=200, port=9999):
    """Simulate a UDP flood by sending many identical packets."""
    print(f"[*] UDP FLOOD: Sending {count} packets to {target_ip}:{port}")

    payload = b"X" * 100
    pkt = IP(dst=target_ip) / UDP(dport=port) / payload

    for i in range(count):
        send(pkt, verbose=False)

    print(f"[+] UDP flood complete. Sent {count} identical packets.")

What does this code do?

This function creates a single UDP packet with a fixed 100-byte payload (b"X" * 100) and sends it 200 times to port 9999 on the gateway. The repetition and uniformity are what makes this traffic anomalous compared to your normal baseline.

You won't see a change yet. The next chunk adds the main function that orchestrates both attacks.

Run the detection engine and validate attacks

The final piece is a main() function that runs both attacks in sequence. Once added, you will launch the detection engine and the attack simulator in separate terminals to see anomaly alerts fire in real time.

Add the main() function and entry point below udp_flood():

def main():
    print("[*] Attack Simulator for AI Firewall Testing")
    print(f"[*] Target: {TARGET_IP} (default gateway)")
    print()

    print("=" * 50)
    print("ATTACK 1: Port Scan")
    print("=" * 50)
    port_scan(TARGET_IP)

    print()
    time.sleep(2)

    print("=" * 50)
    print("ATTACK 2: UDP Flood")
    print("=" * 50)
    udp_flood(TARGET_IP)

    print()
    print("[+] All attacks complete. Check detection terminal for alerts.")


if __name__ == "__main__":
    main()

What does this code do?

The main() function runs the port scan first, waits 2 seconds, then runs the UDP flood. The pause gives you time to see the two attack types arrive separately in the detection console.

Save attack_simulator.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved your file before moving on.

ⓧ I'd like to double check the full code

import time
import sys
from scapy.all import IP, TCP, UDP, send, conf

TARGET_IP = conf.route.route("0.0.0.0")[2]  # Default gateway IP


def port_scan(target_ip, port_range=50):
    """Simulate a port scan by sending SYN packets to many ports."""
    print(f"[*] PORT SCAN: Sending SYN to ports 1-{port_range} on {target_ip}")

    for port in range(1, port_range + 1):
        pkt = IP(dst=target_ip) / TCP(dport=port, flags=\"S\")
        send(pkt, verbose=False)

    print(f"[+] Port scan complete. Sent {port_range} SYN packets.")


def udp_flood(target_ip, count=200, port=9999):
    """Simulate a UDP flood by sending many identical packets."""
    print(f"[*] UDP FLOOD: Sending {count} packets to {target_ip}:{port}")

    payload = b"X" * 100
    pkt = IP(dst=target_ip) / UDP(dport=port) / payload

    for i in range(count):
        send(pkt, verbose=False)

    print(f"[+] UDP flood complete. Sent {count} identical packets.")


def main():
    print("[*] Attack Simulator for AI Firewall Testing")
    print(f"[*] Target: {TARGET_IP} (default gateway)")
    print()

    print("=" * 50)
    print("ATTACK 1: Port Scan")
    print("=" * 50)
    port_scan(TARGET_IP)

    print()
    time.sleep(2)

    print("=" * 50)
    print("ATTACK 2: UDP Flood")
    print("=" * 50)
    udp_flood(TARGET_IP)

    print()
    print("[+] All attacks complete. Check detection terminal for alerts.")


if __name__ == "__main__":
    main()

Now it is time to run the detection engine and attack simulator side by side. You need two Administrator terminals for this.

In your current Administrator terminal, start the detection engine by running:

python detect.py

You should see the startup messages confirming the model and scaler are loaded. The terminal is now listening for packets.

Open a second Administrator terminal. Press the Windows key, type cmd, then select Run as administrator.
Navigate to your project folder and activate the virtual environment in this second terminal by running:

cd %USERPROFILE%\Desktop\ai-firewall
.venv\Scripts\activate

Launch the attack simulator in this second terminal by running:

python attack_simulator.py

Switch back to your first terminal (the one running detect.py) and watch the output.

You should see a stream of [ANOMALY] lines appearing in the detection terminal. Each line shows the source IP, destination port, anomaly score (a negative number), and severity rating.

After the attack simulator finishes, press Ctrl+C in the detection terminal to stop detect.py. It prints a session summary showing total packets analyzed and anomalies detected.
Verify alerts.json contains structured entries by running:

type alerts.json

You should see JSON records with fields like timestamp, src_ip, anomaly_score, and severity. Scores should be negative numbers, and severity should show CRITICAL, MEDIUM, or LOW ratings.

Not seeing any anomaly alerts?

Make sure both terminals are running as Administrator. Scapy cannot send or capture packets without elevated privileges.

Confirm the virtual environment is activated in both terminals (you should see (.venv) in the prompt). If detect.py cannot find the model files, it will crash on startup instead of listening for packets.

If the attack simulator runs but the detection terminal stays silent, check that model.pkl and scaler.pkl exist in your project folder. Re-run train_model.py if they are missing.

Still stuck? Help me troubleshoot why my detection engine isn't flagging the attack simulator's packets.

Secret mission

Auto-Block Malicious IPs with Windows Firewall

Your firewall detects threats, but it doesn't stop them. What if it could automatically block repeat offenders by creating Windows Firewall rules in real time? In this secret mission, you'll add rate-based IP blocking that shuts the door on attackers after just 5 anomaly detections.

Clean Up Your Resources

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

Resources you used:

Project folder on your Desktop (Python scripts, CSV, PKL files, virtual environment).
Npcap 1.88 (system-level packet capture driver).
Windows Firewall rules created by the Secret Mission (names starting with AI_Firewall_Block_).

Keep everything running

No action needed. Choose this if you are still actively building or want to experiment with the firewall further.

Your project folder, trained model, and scripts remain on your Desktop.
Npcap stays installed and available for future packet capture projects.
Any Windows Firewall block rules from the Secret Mission remain active. They only block the specific IPs flagged during testing.

Pause - I'll come back to this later

Free up system resources by deactivating your virtual environment and removing the firewall rules. Your project files stay intact so you can resume later.

Deactivate your virtual environment by running:

deactivate

Remove any Windows Firewall block rules the Secret Mission created. For each blocked IP, run this command in an Administrator terminal:

netsh advfirewall firewall delete rule name="AI_Firewall_Block_[[BLOCKED_IP="the IP that was blocked"]]"

Check alerts.json for entries with "action": "blocked" to find the IP addresses that were blocked.
Your project folder and all files remain on your Desktop for when you come back.

Delete - I don't want to use this again

Remove all project resources and start fresh. This deletes your project files, uninstalls Npcap, and removes any firewall rules created during the project.

Remove any Windows Firewall block rules the Secret Mission created. Check alerts.json for entries with "action": "blocked" to find the blocked IP addresses before deleting your project files.
For each blocked IP, run this command in an Administrator terminal:

netsh advfirewall firewall delete rule name="AI_Firewall_Block_[[BLOCKED_IP="the IP that was blocked"]]"

Deactivate your virtual environment by running:

deactivate

Delete the project folder from your Desktop by running:

cd %USERPROFILE%\Desktop
rmdir /s /q [[PROJECT_FOLDER="your project folder name"]]

Uninstall Npcap using Windows Settings.
Press the Windows key and type Apps & features. Press Enter to open it.
Locate Npcap in the list, click it, and select Uninstall. Follow the on-screen prompts to complete removal.
Restart your computer after Npcap uninstallation to ensure all driver components are fully removed.

Nice Work!

Nice work! You just built a self-learning firewall that detects network anomalies using unsupervised machine learning.

You've learned how to:

Build a packet capture engine using Scapy that extracts ML features from live network traffic and trains an Isolation Forest model on your network's normal behavior baseline.
Create a real-time anomaly detection engine that scores every packet and flags threats with severity-rated JSON alerts.
Build an attack simulator with port scan and UDP flood capabilities to validate detection against your own firewall.
Secret Mission: Added auto-blocking that creates Windows Firewall rules for repeat offenders using a sliding-window threshold.

Ready to quiz yourself?