Build a Network Intrusion Detection System
Sniff live packets and detect port scans, SYN floods, and ARP spoofing.
Introduction
30 Second Summary
Every device connected to a network is a potential target. Security teams rely on intrusion detection systems to spot malicious traffic the moment it appears, before attackers can do real damage.
In this project, you will build a real-time network intrusion detection system in Python that captures live packets on your Mac, analyzes them against detection rules, and generates structured alerts when attacks are detected.
What You'll Build
A command-line IDS that sniffs live traffic using Scapy, applies rule-based detection with sliding window algorithms, and logs structured JSON alerts for four attack types.
By the end of this project, you'll have:
- A working packet capture engine using Scapy's AsyncSniffer that intercepts live network traffic on your local machine.
- Four signature-based detection rules with sliding window analysis for port scans, SYN floods, ping floods, and ARP spoofing.
- A structured JSON alert system with severity levels, plus attack simulation scripts to verify your IDS catches real threats.
- Secret Mission: Add a live terminal dashboard using the Rich library that displays color-coded alerts in real time.
Are there any prerequisites?
You should have experience writing Python code and a basic understanding of networking concepts like IP addresses and ports. You do not need prior cybersecurity experience.
Set Up Your Environment
Before you can detect attacks, you need the right tools in place. Scapy is a Python library that lets you capture and manipulate live network packets at the lowest level.
In this step, you will install Scapy inside an isolated Python environment, create a configuration file that defines your detection thresholds, and confirm that your Mac can successfully sniff packets from the network.
In this step, get ready to:
- Create a project directory with a Python virtual environment and install Scapy.
- Create a configuration file with detection thresholds and severity levels.
- Verify that Scapy can capture live packets on your network interface.
Create your project directory and install Scapy
Packet sniffing requires root privileges on macOS because it accesses the network interface directly. You will use sudo throughout this project to grant that access.
- Open Terminal by pressing Cmd+Space (macOS), typing Terminal, and pressing Enter.
- Move to your Desktop by running this command:
cd ~/Desktop- Create your project folder and move into it by running these commands:
mkdir network-ids
cd network-ids- Confirm the folder exists by running:
ls ~/Desktop | grep network-idsYou should see network-ids in the output.
- Create a Python virtual environment by running:
python3 -m venv venv- Activate the virtual environment by running:
source venv/bin/activateYour terminal prompt should now show (venv) at the beginning, confirming the virtual environment is active.
What is a virtual environment?
A virtual environment is an isolated Python installation inside your project folder. Packages you install here stay separate from your system Python, so you can manage dependencies per-project without conflicts.
- Install Scapy by running:
pip install scapyYou should see output ending with a message confirming Scapy was successfully installed.
Seeing a permissions error?
Make sure your virtual environment is activated. Your prompt should start with (venv). If it does not, run source venv/bin/activate again.
help me fix a permissions error installing Scapy in my virtual environment
- Verify Scapy can capture packets by running a quick 5-packet test:
sudo venv/bin/python -c "from scapy.all import sniff; pkts = sniff(count=5, iface='en0'); print(f'Captured {len(pkts)} packets')"Why sudo and why venv/bin/python?
Packet capture on macOS requires root access, so you need sudo. However, sudo does not inherit your virtual environment. Using the explicit path venv/bin/python ensures the command runs with Scapy installed.
When prompted for your password, type it and press Enter. No characters will appear on screen as you type. This is normal macOS security behaviour.
You should see Captured 5 packets printed in your terminal.
Seeing 0 packets or a timeout?
Your Wi-Fi interface might not be en0. Check your active interface by running ifconfig and looking for the interface with an assigned IP address. Replace en0 in the command with your actual interface name.
Also make sure you are connected to Wi-Fi. Scapy needs active network traffic to capture packets.
help me troubleshoot why Scapy is not capturing packets on my Mac
Create the configuration file
Your IDS needs to know what thresholds trigger an alert. A configuration file lets you tune these values without editing code.
- Open your code editor and create a new file called config.json inside the network-ids folder.
- Paste the following configuration into config.json:
{
"interface": "en0",
"bpf_filter": "",
"thresholds": {
"port_scan": {
"ports": 5,
"window_seconds": 10
},
"syn_flood": {
"packets": 20,
"window_seconds": 5
},
"ping_flood": {
"packets": 15,
"window_seconds": 5
}
},
"severity_levels": {
"port_scan": "HIGH",
"syn_flood": "CRITICAL",
"ping_flood": "MEDIUM",
"arp_spoof": "CRITICAL"
}
}What do these thresholds mean?
Each threshold defines what triggers a specific alert. For example, port_scan fires when a single IP hits 5 distinct ports within 10 seconds.
The severity_levels section assigns a priority to each attack type. A SYN flood or ARP spoof is CRITICAL because they can take down a network. A ping flood is MEDIUM because it is easier to mitigate.
The bpf_filter field is left empty for now, meaning the IDS will capture all traffic. You will test a BPF filter in the next substep.
- Save config.json.
✔️ Awesome, I've got everything!
Great. Double check you have saved config.json inside the network-ids folder on your Desktop.
ⓧ I'd like to double check the full code
Here is the full config.json for comparison:
{
"interface": "en0",
"bpf_filter": "",
"thresholds": {
"port_scan": {
"ports": 5,
"window_seconds": 10
},
"syn_flood": {
"packets": 20,
"window_seconds": 5
},
"ping_flood": {
"packets": 15,
"window_seconds": 5
}
},
"severity_levels": {
"port_scan": "HIGH",
"syn_flood": "CRITICAL",
"ping_flood": "MEDIUM",
"arp_spoof": "CRITICAL"
}
}Test that Scapy can capture on your interface with a BPF filter
A BPF (Berkeley Packet Filter) filter tells Scapy to capture only specific traffic types instead of everything. This reduces noise and improves performance when you only care about certain protocols.
- Test a TCP-only BPF filter by running this command:
sudo venv/bin/python -c "from scapy.all import sniff; pkts = sniff(count=5, iface='en0', filter='tcp'); print(f'Captured {len(pkts)} TCP packets')"You should see Captured 5 TCP packets after a few seconds. If your Mac has any background network activity (browser tabs, iCloud sync), this will complete quickly.
What is a BPF filter?
BPF stands for Berkeley Packet Filter. It is a low-level packet filtering mechanism built into the operating system kernel. By passing a filter string like tcp to Scapy, the OS itself discards non-matching packets before they even reach your Python code.
Common BPF filter examples include tcp, udp, icmp, and arp. You can combine them too, like tcp and port 80.
Command hangs without capturing?
If the command sits without completing, your machine might not have active TCP traffic. Open a browser tab or run curl https://example.com in a separate terminal window to generate some TCP packets.
help me troubleshoot why my BPF filter capture is hanging in Scapy
Your environment is ready and confirmed working. Next up, you will build the packet capture engine that continuously sniffs traffic and feeds it into your detection system.
Build the Packet Capture Engine
Your IDS needs a way to listen to raw network traffic in real time. The capture engine is responsible for intercepting every packet that flows through your network interface and converting it into structured data the detection engine can analyze.
In this step, you will build a PacketCapture class that uses Scapy's AsyncSniffer for non-blocking packet capture. Captured packets flow through a thread-safe queue so the detection engine can process them without slowing down the sniffer.
In this step, get ready to:
- Create the PacketCapture class with a thread-safe queue and AsyncSniffer control methods.
- Add packet feature extraction for TCP, UDP, ICMP, and ARP protocols.
- Test the capture engine by sniffing live packets on your Mac.
Create capture.py with the PacketCapture class
The capture engine needs three core capabilities: accepting configuration, queuing packets safely across threads, and controlling the sniffer lifecycle.
- Create a new file called capture.py inside your network-ids folder.
- Add the following imports and class definition:
import queue
import threading
from scapy.all import AsyncSniffer, IP, TCP, UDP, ICMP, ARP
class PacketCapture:
def __init__(self, interface, bpf_filter=""):
# Store the network interface name and optional BPF filter
self.interface = interface
self.bpf_filter = bpf_filter
# Thread-safe queue holds parsed packets for the detection engine
self.packet_queue = queue.Queue(maxsize=1000)
self.sniffer = None
self.running = False
def _process_packet(self, packet):
# Callback invoked by AsyncSniffer for each captured packet
features = self._extract_features(packet)
if features:
try:
self.packet_queue.put_nowait(features)
except queue.Full:
# Drop packet silently if queue is full (back-pressure)
passWhat does this code do?
The __init__ method stores the interface name (like en0) and an optional BPF filter. It also creates a queue.Queue with a maximum capacity of 1000 packets to prevent unbounded memory growth.
The _process_packet method is a callback that fires every time the sniffer captures a packet. It extracts features and places them in the queue using put_nowait so the sniffer thread never blocks waiting for a consumer.
- Add the start() and stop() methods below _process_packet to control the sniffer lifecycle:
def start(self):
self.running = True
# Build keyword arguments for AsyncSniffer
kwargs = {
"iface": self.interface,
"prn": self._process_packet,
"store": False,
}
# Only apply BPF filter if one was configured
if self.bpf_filter:
kwargs["filter"] = self.bpf_filter
self.sniffer = AsyncSniffer(**kwargs)
self.sniffer.start()
def stop(self):
self.running = False
if self.sniffer:
self.sniffer.stop()What does this code do?
AsyncSniffer runs packet capture in a background thread, so your main thread stays free to process the queue. Setting store=False tells Scapy not to keep packets in memory after the callback runs.
The stop() method gracefully shuts down the sniffer. Checking if self.sniffer prevents an error if stop() is called before start().
- Save capture.py.
Seeing an import error for scapy?
Make sure your virtual environment is active. You should see (venv) at the start of your terminal prompt. If not, activate it again with source venv/bin/activate.
Help me fix my Scapy import error in capture.py
Add packet parsing to extract features
Each raw packet needs to be broken down into structured fields the detection engine can reason about. Different protocols carry different information, so the parser must handle TCP, UDP, ICMP, and ARP packets separately.
- In capture.py, add the _extract_features method between _process_packet and start:
def _extract_features(self, packet):
features = {"raw_packet": packet}
# Handle ARP packets separately (no IP layer)
if ARP in packet:
features["protocol"] = "ARP"
features["src_ip"] = packet[ARP].psrc
features["dst_ip"] = packet[ARP].pdst
features["src_mac"] = packet[ARP].hwsrc
features["arp_op"] = packet[ARP].op
return features
# Skip non-IP packets (like raw Ethernet frames)
if IP not in packet:
return None
# Extract common IP-layer fields
features["src_ip"] = packet[IP].src
features["dst_ip"] = packet[IP].dst
features["packet_size"] = len(packet)
# Extract protocol-specific fields
if TCP in packet:
features["protocol"] = "TCP"
features["src_port"] = packet[TCP].sport
features["dst_port"] = packet[TCP].dport
features["tcp_flags"] = packet[TCP].flags
elif UDP in packet:
features["protocol"] = "UDP"
features["src_port"] = packet[UDP].sport
features["dst_port"] = packet[UDP].dport
elif ICMP in packet:
features["protocol"] = "ICMP"
features["icmp_type"] = packet[ICMP].type
else:
features["protocol"] = "OTHER"
return featuresWhat does this code do?
The method checks protocol layers from the outside in. ARP packets are handled first because they do not contain an IP layer. Non-IP, non-ARP packets (like raw Ethernet frames) are discarded by returning None.
For IP packets, it extracts source and destination addresses plus the total packet size. Then it checks for TCP (ports and flags), UDP (ports only), or ICMP (type code) to add protocol-specific fields. The tcp_flags field is critical for detecting SYN floods later.
- Save capture.py.
✔️ Awesome, I've got everything!
Great. Double check you have saved capture.py with all three sections: imports and __init__, the start/stop methods, and the _extract_features parser.
ⓧ I'd like to double check the full code
import queue
import threading
from scapy.all import AsyncSniffer, IP, TCP, UDP, ICMP, ARP
class PacketCapture:
def __init__(self, interface, bpf_filter=""):
self.interface = interface
self.bpf_filter = bpf_filter
self.packet_queue = queue.Queue(maxsize=1000)
self.sniffer = None
self.running = False
def _process_packet(self, packet):
features = self._extract_features(packet)
if features:
try:
self.packet_queue.put_nowait(features)
except queue.Full:
pass
def _extract_features(self, packet):
features = {"raw_packet": packet}
if ARP in packet:
features["protocol"] = "ARP"
features["src_ip"] = packet[ARP].psrc
features["dst_ip"] = packet[ARP].pdst
features["src_mac"] = packet[ARP].hwsrc
features["arp_op"] = packet[ARP].op
return features
if IP not in packet:
return None
features["src_ip"] = packet[IP].src
features["dst_ip"] = packet[IP].dst
features["packet_size"] = len(packet)
if TCP in packet:
features["protocol"] = "TCP"
features["src_port"] = packet[TCP].sport
features["dst_port"] = packet[TCP].dport
features["tcp_flags"] = packet[TCP].flags
elif UDP in packet:
features["protocol"] = "UDP"
features["src_port"] = packet[UDP].sport
features["dst_port"] = packet[UDP].dport
elif ICMP in packet:
features["protocol"] = "ICMP"
features["icmp_type"] = packet[ICMP].type
else:
features["protocol"] = "OTHER"
return features
def start(self):
self.running = True
kwargs = {
"iface": self.interface,
"prn": self._process_packet,
"store": False,
}
if self.bpf_filter:
kwargs["filter"] = self.bpf_filter
self.sniffer = AsyncSniffer(**kwargs)
self.sniffer.start()
def stop(self):
self.running = False
if self.sniffer:
self.sniffer.stop()Test the capture engine
You can verify the engine works by running a quick capture and printing the first parsed packet from the queue.
- Test your capture engine by running this command (it captures for 3 seconds then prints one packet):
sudo python -c "
from capture import PacketCapture
import time
c = PacketCapture('en0')
c.start()
time.sleep(3)
c.stop()
while not c.packet_queue.empty():
print(c.packet_queue.get())
break
"You should see a dictionary printed with keys like protocol, src_ip, dst_ip, and protocol-specific fields.
Why sudo?
Packet capture requires raw socket access, which is a privileged operation on macOS. Running with sudo grants the necessary permissions for Scapy to read from your network interface.
Not seeing any output?
Make sure your Mac is connected to Wi-Fi. If you are using Ethernet or a different interface, check your interface name by running ifconfig in the terminal and update en0 in the test command to match your active interface.
If you see a permission error, confirm you included sudo at the start of the command.
Help me troubleshoot my PacketCapture test
Your capture engine is wired up and pulling live packets off the wire. Next up, you will build the detection engine that analyzes these parsed features and spots attack patterns.
Build the Detection Engine
Your capture engine is now producing structured feature dictionaries for every packet that flows through your network interface. The next question is: what do you do with all that data?
In this step, you'll build a detection engine that analyzes each packet's features against a set of rules. The engine uses sliding window algorithms to track activity over time and fires alerts when behaviour crosses the thresholds you defined in config.json.
In this step, get ready to:
- Create a DetectionEngine class with sliding window tracking for each attack type.
- Implement four detection rules: port scan, SYN flood, ping flood, and ARP spoofing.
- Test the engine with mock packet data to confirm it correctly identifies threshold breaches.
Create detector.py with a DetectionEngine class
The detection engine needs to remember recent activity for each source IP. A sliding window keeps a rolling buffer of timestamps so the engine can answer questions like "how many ports has this IP hit in the last 10 seconds?"
- Create a new file called detector.py in your network-ids folder.
- Add the imports and class initialization by pasting the following code:
import time
from collections import defaultdict
class DetectionEngine:
def __init__(self, config):
# Load detection thresholds and severity levels from config
self.thresholds = config["thresholds"]
self.severity_levels = config["severity_levels"]
# Sliding window trackers for each attack type
self.port_tracker = defaultdict(list)
self.syn_tracker = defaultdict(list)
self.ping_tracker = defaultdict(list)
self.arp_table = {}
# Cooldown tracking to prevent duplicate alerts
self.alerted = defaultdict(float)
self.alert_cooldown = 10What does this code do?
The __init__ method sets up the engine's memory. Each tracker is a defaultdict(list) that automatically creates an empty list for any new source IP it encounters.
- port_tracker, syn_tracker, and ping_tracker each store timestamped events per source IP.
- arp_table maps IP addresses to their last-seen MAC address for spoofing detection.
- alerted tracks when the last alert fired for each (attack_type, source_ip) pair, so the 10-second cooldown prevents duplicate alerts.
Now you need the main entry point that routes each packet to the right detection check.
- Add the analyze() method below the __init__ method:
def analyze(self, features):
threats = []
now = time.time()
protocol = features.get("protocol")
# ARP packets get checked separately (no IP layer)
if protocol == "ARP":
threat = self._check_arp_spoof(features, now)
if threat:
threats.append(threat)
return threats
src_ip = features.get("src_ip", "")
# TCP packets can indicate port scans or SYN floods
if protocol == "TCP":
threat = self._check_port_scan(features, src_ip, now)
if threat:
threats.append(threat)
threat = self._check_syn_flood(features, src_ip, now)
if threat:
threats.append(threat)
# ICMP packets can indicate ping floods
elif protocol == "ICMP":
threat = self._check_ping_flood(features, src_ip, now)
if threat:
threats.append(threat)
return threatsWhat does this code do?
The analyze() method is the dispatcher. It reads the packet's protocol and routes it to the appropriate check methods. A single TCP packet gets tested against both port scan and SYN flood rules, since one packet can contribute to either attack pattern.
The method returns a list of threat dictionaries. Most packets won't trigger anything, so the list is usually empty.
- Save detector.py.
Seeing an indentation error?
Make sure the analyze() method is indented one level inside the DetectionEngine class (4 spaces from the left margin). The method body should be indented a further 4 spaces (8 total).
Help me fix indentation in my detector.py file.
Implement four detection rules
Each detection rule follows the same pattern: record the event in a sliding window, expire old entries, count what remains, and fire an alert if the count crosses the threshold.
- Add the port scan detection method below analyze():
def _check_port_scan(self, features, src_ip, now):
config = self.thresholds["port_scan"]
window = config["window_seconds"]
threshold = config["ports"]
dst_port = features.get("dst_port")
if dst_port is None:
return None
# Record this port hit and expire old entries
key = f"port_scan:{src_ip}"
self.port_tracker[src_ip].append((now, dst_port))
self.port_tracker[src_ip] = [
(t, p) for t, p in self.port_tracker[src_ip] if now - t <= window
]
# Count distinct ports within the window
distinct_ports = len(set(p for _, p in self.port_tracker[src_ip]))
if distinct_ports >= threshold and self._can_alert(key, now):
self.alerted[key] = now
return {
"type": "port_scan",
"severity": self.severity_levels["port_scan"],
"src_ip": src_ip,
"details": f"Scanned {distinct_ports} distinct ports in {window}s",
}
return NoneWhat does this code do?
This method tracks which destination ports each source IP has hit. The list comprehension on line 13 is the sliding window cleanup. It keeps only entries whose timestamp is within the configured window (10 seconds by default).
Instead of counting raw packets, it counts distinct ports. An attacker scanning ports 22, 80, 443, 8080, 3306 would hit the threshold of 5. Repeated requests to port 80 alone would not.
- Add the SYN flood detection method below _check_port_scan():
def _check_syn_flood(self, features, src_ip, now):
config = self.thresholds["syn_flood"]
window = config["window_seconds"]
threshold = config["packets"]
# Only count packets with SYN flag only (not SYN-ACK)
flags = features.get("tcp_flags")
if flags is None or str(flags) != "S":
return None
# Record this SYN and expire old entries
key = f"syn_flood:{src_ip}"
self.syn_tracker[src_ip].append(now)
self.syn_tracker[src_ip] = [
t for t in self.syn_tracker[src_ip] if now - t <= window
]
if len(self.syn_tracker[src_ip]) >= threshold and self._can_alert(key, now):
self.alerted[key] = now
return {
"type": "syn_flood",
"severity": self.severity_levels["syn_flood"],
"src_ip": src_ip,
"details": f"{len(self.syn_tracker[src_ip])} SYN packets in {window}s",
}
return NoneWhy check for str(flags) == "S"?
Scapy represents TCP flags as a FlagValue object. Converting to a string gives "S" for SYN-only packets and "SA" for SYN-ACK packets. You only want to count SYN-only packets because SYN-ACK is a normal server response, not an attack.
- Add the ping flood detection method below _check_syn_flood():
def _check_ping_flood(self, features, src_ip, now):
config = self.thresholds["ping_flood"]
window = config["window_seconds"]
threshold = config["packets"]
# Only count ICMP echo requests (type 8)
icmp_type = features.get("icmp_type")
if icmp_type != 8:
return None
# Record this ping and expire old entries
key = f"ping_flood:{src_ip}"
self.ping_tracker[src_ip].append(now)
self.ping_tracker[src_ip] = [
t for t in self.ping_tracker[src_ip] if now - t <= window
]
if len(self.ping_tracker[src_ip]) >= threshold and self._can_alert(key, now):
self.alerted[key] = now
return {
"type": "ping_flood",
"severity": self.severity_levels["ping_flood"],
"src_ip": src_ip,
"details": f"{len(self.ping_tracker[src_ip])} ICMP echo requests in {window}s",
}
return NoneWhy only ICMP type 8?
ICMP type 8 is an echo request (a ping). Type 0 is an echo reply. You only want to detect incoming requests flooding your machine, not the replies your machine sends back.
- Add the ARP spoofing detection and cooldown helper methods below _check_ping_flood():
def _check_arp_spoof(self, features, now):
src_ip = features.get("src_ip", "")
src_mac = features.get("src_mac", "")
if not src_ip or not src_mac or src_ip == "0.0.0.0":
return None
key = f"arp_spoof:{src_ip}"
# Check if this IP was previously seen with a different MAC
if src_ip in self.arp_table:
if self.arp_table[src_ip] != src_mac and self._can_alert(key, now):
self.alerted[key] = now
old_mac = self.arp_table[src_ip]
self.arp_table[src_ip] = src_mac
return {
"type": "arp_spoof",
"severity": self.severity_levels["arp_spoof"],
"src_ip": src_ip,
"details": f"MAC changed from {old_mac} to {src_mac}",
}
else:
self.arp_table[src_ip] = src_mac
return None
def _can_alert(self, key, now):
# Prevent duplicate alerts within the cooldown period
last_alert = self.alerted.get(key, 0)
return now - last_alert > self.alert_cooldownWhat does this code do?
ARP spoofing detection works differently from the other rules. Instead of counting packets over time, it maintains a table of which MAC address belongs to each IP. When an IP suddenly appears with a new MAC address, something is wrong.
The _can_alert() helper enforces a 10-second cooldown per (attack_type, source_ip) pair. Without this, a single port scan would generate hundreds of duplicate alerts as each new port pushes the count further above the threshold.
- Save detector.py.
Getting a NameError on _can_alert?
Make sure _can_alert() is indented at the same level as the other methods (one level inside the class). If it's accidentally outside the class, Python won't find it as a method.
Help me fix a NameError in my detection engine.
Test the detection engine with mock data
Each check method implements the same sliding window pattern: record the event, expire entries outside the time window, then compare the count to the threshold. This cleanup runs on every packet, so the trackers never grow unbounded.
You can verify the engine works correctly without capturing live traffic by feeding it mock feature dictionaries.
- Test your detection engine by running this command in your terminal:
python -c "
import json
from detector import DetectionEngine
config = json.load(open('config.json'))
engine = DetectionEngine(config)
# Simulate a port scan: 6 distinct ports from the same IP
for port in [22, 80, 443, 8080, 3306, 5432]:
threats = engine.analyze({
'protocol': 'TCP',
'src_ip': '192.168.1.100',
'dst_ip': '10.0.0.1',
'dst_port': port,
'tcp_flags': 'S'
})
if threats:
for t in threats:
print(f'ALERT: {t[\"type\"]} from {t[\"src_ip\"]} - {t[\"details\"]}')
print('Test complete.')
"You should see an alert for a port scan after the 5th distinct port is reached. You may also see a SYN flood alert once the SYN count crosses 20.
What should I see?
The output should include a line like:
ALERT: port_scan from 192.168.1.100 - Scanned 5 distinct ports in 10sThis confirms your sliding window is counting distinct ports and the threshold check is working. The exact port count in the message may be 5 or 6 depending on which packet triggers it.
No alert printed?
Check that your config.json has the port scan threshold set to 5 under thresholds.port_scan.ports. If it's higher than 6, the test won't trigger an alert.
Also verify that your virtual environment is active (you should see (venv) in your terminal prompt) and that you're running the command from inside the network-ids directory.
Help me debug why my detection engine test isn't producing alerts.
✔️ Awesome, I've got everything!
Great. Double check you've saved detector.py and your test printed an alert.
ⓧ I'd like to double check the full code
Here is the complete detector.py file for comparison:
import time
from collections import defaultdict
class DetectionEngine:
def __init__(self, config):
self.thresholds = config["thresholds"]
self.severity_levels = config["severity_levels"]
self.port_tracker = defaultdict(list)
self.syn_tracker = defaultdict(list)
self.ping_tracker = defaultdict(list)
self.arp_table = {}
self.alerted = defaultdict(float)
self.alert_cooldown = 10
def analyze(self, features):
threats = []
now = time.time()
protocol = features.get("protocol")
if protocol == "ARP":
threat = self._check_arp_spoof(features, now)
if threat:
threats.append(threat)
return threats
src_ip = features.get("src_ip", "")
if protocol == "TCP":
threat = self._check_port_scan(features, src_ip, now)
if threat:
threats.append(threat)
threat = self._check_syn_flood(features, src_ip, now)
if threat:
threats.append(threat)
elif protocol == "ICMP":
threat = self._check_ping_flood(features, src_ip, now)
if threat:
threats.append(threat)
return threats
def _check_port_scan(self, features, src_ip, now):
config = self.thresholds["port_scan"]
window = config["window_seconds"]
threshold = config["ports"]
dst_port = features.get("dst_port")
if dst_port is None:
return None
key = f"port_scan:{src_ip}"
self.port_tracker[src_ip].append((now, dst_port))
self.port_tracker[src_ip] = [
(t, p) for t, p in self.port_tracker[src_ip] if now - t <= window
]
distinct_ports = len(set(p for _, p in self.port_tracker[src_ip]))
if distinct_ports >= threshold and self._can_alert(key, now):
self.alerted[key] = now
return {
"type": "port_scan",
"severity": self.severity_levels["port_scan"],
"src_ip": src_ip,
"details": f"Scanned {distinct_ports} distinct ports in {window}s",
}
return None
def _check_syn_flood(self, features, src_ip, now):
config = self.thresholds["syn_flood"]
window = config["window_seconds"]
threshold = config["packets"]
flags = features.get("tcp_flags")
if flags is None or str(flags) != "S":
return None
key = f"syn_flood:{src_ip}"
self.syn_tracker[src_ip].append(now)
self.syn_tracker[src_ip] = [
t for t in self.syn_tracker[src_ip] if now - t <= window
]
if len(self.syn_tracker[src_ip]) >= threshold and self._can_alert(key, now):
self.alerted[key] = now
return {
"type": "syn_flood",
"severity": self.severity_levels["syn_flood"],
"src_ip": src_ip,
"details": f"{len(self.syn_tracker[src_ip])} SYN packets in {window}s",
}
return None
def _check_ping_flood(self, features, src_ip, now):
config = self.thresholds["ping_flood"]
window = config["window_seconds"]
threshold = config["packets"]
icmp_type = features.get("icmp_type")
if icmp_type != 8:
return None
key = f"ping_flood:{src_ip}"
self.ping_tracker[src_ip].append(now)
self.ping_tracker[src_ip] = [
t for t in self.ping_tracker[src_ip] if now - t <= window
]
if len(self.ping_tracker[src_ip]) >= threshold and self._can_alert(key, now):
self.alerted[key] = now
return {
"type": "ping_flood",
"severity": self.severity_levels["ping_flood"],
"src_ip": src_ip,
"details": f"{len(self.ping_tracker[src_ip])} ICMP echo requests in {window}s",
}
return None
def _check_arp_spoof(self, features, now):
src_ip = features.get("src_ip", "")
src_mac = features.get("src_mac", "")
if not src_ip or not src_mac or src_ip == "0.0.0.0":
return None
key = f"arp_spoof:{src_ip}"
if src_ip in self.arp_table:
if self.arp_table[src_ip] != src_mac and self._can_alert(key, now):
self.alerted[key] = now
old_mac = self.arp_table[src_ip]
self.arp_table[src_ip] = src_mac
return {
"type": "arp_spoof",
"severity": self.severity_levels["arp_spoof"],
"src_ip": src_ip,
"details": f"MAC changed from {old_mac} to {src_mac}",
}
else:
self.arp_table[src_ip] = src_mac
return None
def _can_alert(self, key, now):
last_alert = self.alerted.get(key, 0)
return now - last_alert > self.alert_cooldownYour detection engine can now identify four different attack patterns from raw packet features. Next up, you'll build the alert system and main orchestrator to wire everything together into a running IDS.
Build the Alert System and Main Orchestrator
Your detection engine can now identify threats from raw packet features. The missing piece is turning those threat dictionaries into actionable output that a security analyst can review.
In this step, you will build the alert system that logs structured JSON alerts to a file and stdout. Then you will create the main orchestrator that wires the capture engine, detection engine, and alert system into a single running process.
In this step, get ready to:
- Create an alert system that generates structured JSON alerts with UUIDs and timestamps.
- Create a main orchestrator that ties all modules together into a live monitoring loop.
- Run the full IDS with sudo and verify it starts monitoring traffic.
Create the alert system
When the detection engine flags a threat, the IDS needs to do three things: log the alert to a file for later review, print a formatted line to the terminal for real-time awareness, and track running totals for the shutdown summary.
- Create a new file called alerts.py in your network-ids/ folder.
- Add the imports, class initialization, and core alert generation method by pasting this code:
import json
import uuid
from datetime import datetime, timezone
class AlertSystem:
def __init__(self, log_file="alerts.log"):
self.log_file = log_file
self.alert_counts = {"total": 0, "by_type": {}, "by_severity": {}}
def generate_alert(self, threat):
# Build structured alert with unique ID and UTC timestamp
alert = {
"alert_id": str(uuid.uuid4()),
"timestamp": datetime.now(timezone.utc).isoformat(),
"type": threat["type"],
"severity": threat["severity"],
"src_ip": threat["src_ip"],
"details": threat["details"],
}
# Output and track the alert
self._log_to_file(alert)
self._print_alert(alert)
self._update_counts(alert)
return alert
def _log_to_file(self, alert):
# Append alert as a single JSON line to the log file
with open(self.log_file, "a") as f:
f.write(json.dumps(alert) + "\n")What does this code do?
The AlertSystem class takes each threat dictionary from the detection engine and enriches it into a full alert. Each alert gets a unique alert_id (UUID) and a UTC timestamp so you can trace alerts across log files.
The _log_to_file method writes one JSON object per line to alerts.log. This "JSON lines" format makes it easy to parse with standard tools later.
- Add the terminal printing, counting, and summary methods below the code you just pasted:
def _print_alert(self, alert):
# Format a short prefix by severity for quick scanning
severity = alert["severity"]
prefix = {
"LOW": "[LOW]",
"MEDIUM": "[MED]",
"HIGH": "[HIG]",
"CRITICAL": "[CRT]",
}.get(severity, "[???]")
timestamp = alert["timestamp"][:19]
print(
f" {prefix} {timestamp} | {alert['type']:12} | "
f"src={alert['src_ip']:15} | {alert['details']}"
)
def _update_counts(self, alert):
# Increment running totals for the shutdown summary
self.alert_counts["total"] += 1
attack_type = alert["type"]
severity = alert["severity"]
self.alert_counts["by_type"][attack_type] = (
self.alert_counts["by_type"].get(attack_type, 0) + 1
)
self.alert_counts["by_severity"][severity] = (
self.alert_counts["by_severity"].get(severity, 0) + 1
)
def get_summary(self):
return self.alert_countsWhat does this code do?
The _print_alert method formats each alert into a single scannable terminal line. The severity prefix ([CRT], [HIG], etc.) makes it easy to spot critical alerts at a glance.
The _update_counts method tracks totals by attack type and severity. When the IDS shuts down, get_summary() returns these counts for a final report.
- Save alerts.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 json
import uuid
from datetime import datetime, timezone
class AlertSystem:
def __init__(self, log_file="alerts.log"):
self.log_file = log_file
self.alert_counts = {"total": 0, "by_type": {}, "by_severity": {}}
def generate_alert(self, threat):
alert = {
"alert_id": str(uuid.uuid4()),
"timestamp": datetime.now(timezone.utc).isoformat(),
"type": threat["type"],
"severity": threat["severity"],
"src_ip": threat["src_ip"],
"details": threat["details"],
}
self._log_to_file(alert)
self._print_alert(alert)
self._update_counts(alert)
return alert
def _log_to_file(self, alert):
with open(self.log_file, "a") as f:
f.write(json.dumps(alert) + "\n")
def _print_alert(self, alert):
severity = alert["severity"]
prefix = {
"LOW": "[LOW]",
"MEDIUM": "[MED]",
"HIGH": "[HIG]",
"CRITICAL": "[CRT]",
}.get(severity, "[???]")
timestamp = alert["timestamp"][:19]
print(
f" {prefix} {timestamp} | {alert['type']:12} | "
f"src={alert['src_ip']:15} | {alert['details']}"
)
def _update_counts(self, alert):
self.alert_counts["total"] += 1
attack_type = alert["type"]
severity = alert["severity"]
self.alert_counts["by_type"][attack_type] = (
self.alert_counts["by_type"].get(attack_type, 0) + 1
)
self.alert_counts["by_severity"][severity] = (
self.alert_counts["by_severity"].get(severity, 0) + 1
)
def get_summary(self):
return self.alert_countsCreate the main orchestrator
The orchestrator is the entry point that loads your configuration, creates instances of each module, and runs the main processing loop. It ties the capture engine, detection engine, and alert system into a single command.
- Create a new file called ids.py in your network-ids/ folder.
- Add the imports, config loader, and startup banner by pasting this code:
import json
import sys
import time
import queue
from capture import PacketCapture
from detector import DetectionEngine
from alerts import AlertSystem
def load_config(path="config.json"):
# Read detection thresholds and interface settings
with open(path, "r") as f:
return json.load(f)
def print_banner(config):
# Display startup info so the operator knows what's active
print("\n" + "=" * 60)
print(" NETWORK INTRUSION DETECTION SYSTEM")
print("=" * 60)
print(f" Interface : {config['interface']}")
print(f" BPF Filter: {config['bpf_filter'] or 'None (all traffic)'}")
print(f" Rules loaded:")
for rule, params in config["thresholds"].items():
severity = config["severity_levels"].get(rule, "N/A")
print(f" - {rule}: {params} [{severity}]")
print(f" - arp_spoof: IP-MAC mismatch [{config['severity_levels']['arp_spoof']}]")
print("=" * 60)
print(" Monitoring... (Ctrl+C to stop)\n")What does this code do?
The imports pull in your three modules (PacketCapture, DetectionEngine, AlertSystem) so the orchestrator can instantiate them.
The print_banner function displays the active interface, BPF filter, and all loaded detection rules with their severity levels. This gives the operator immediate confirmation that the IDS is configured correctly before monitoring begins.
- Add the main function and entry point below the banner function:
def main():
config = load_config()
print_banner(config)
# Initialize all three modules
capture = PacketCapture(
interface=config["interface"],
bpf_filter=config["bpf_filter"],
)
detector = DetectionEngine(config)
alert_system = AlertSystem()
packets_processed = 0
# Start the async packet sniffer
capture.start()
try:
while True:
try:
# Pull parsed packet features from the capture queue
features = capture.packet_queue.get(timeout=0.5)
packets_processed += 1
# Run detection rules against the packet
threats = detector.analyze(features)
for threat in threats:
alert_system.generate_alert(threat)
except queue.Empty:
continue
except KeyboardInterrupt:
# Graceful shutdown with summary statistics
print("\n\n" + "=" * 60)
print(" SHUTTING DOWN")
print("=" * 60)
capture.stop()
summary = alert_system.get_summary()
print(f" Packets processed: {packets_processed}")
print(f" Total alerts: {summary['total']}")
if summary["by_type"]:
print(f" Alerts by type:")
for attack_type, count in summary["by_type"].items():
print(f" - {attack_type}: {count}")
if summary["by_severity"]:
print(f" Alerts by severity:")
for severity, count in summary["by_severity"].items():
print(f" - {severity}: {count}")
print("=" * 60 + "\n")
if __name__ == "__main__":
main()What does this code do?
The main() function runs a continuous loop that pulls packets from the capture queue, passes their features through the detection engine, and generates alerts for any threats found.
The timeout=0.5 on the queue get prevents the loop from busy-waiting. When no packets arrive within half a second, it catches the queue.Empty exception and loops back to check again.
When you press Ctrl+C, the KeyboardInterrupt handler stops the sniffer and prints a summary of everything the IDS detected during the session.
- Save ids.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 json
import sys
import time
import queue
from capture import PacketCapture
from detector import DetectionEngine
from alerts import AlertSystem
def load_config(path="config.json"):
with open(path, "r") as f:
return json.load(f)
def print_banner(config):
print("\n" + "=" * 60)
print(" NETWORK INTRUSION DETECTION SYSTEM")
print("=" * 60)
print(f" Interface : {config['interface']}")
print(f" BPF Filter: {config['bpf_filter'] or 'None (all traffic)'}")
print(f" Rules loaded:")
for rule, params in config["thresholds"].items():
severity = config["severity_levels"].get(rule, "N/A")
print(f" - {rule}: {params} [{severity}]")
print(f" - arp_spoof: IP-MAC mismatch [{config['severity_levels']['arp_spoof']}]")
print("=" * 60)
print(" Monitoring... (Ctrl+C to stop)\n")
def main():
config = load_config()
print_banner(config)
capture = PacketCapture(
interface=config["interface"],
bpf_filter=config["bpf_filter"],
)
detector = DetectionEngine(config)
alert_system = AlertSystem()
packets_processed = 0
capture.start()
try:
while True:
try:
features = capture.packet_queue.get(timeout=0.5)
packets_processed += 1
threats = detector.analyze(features)
for threat in threats:
alert_system.generate_alert(threat)
except queue.Empty:
continue
except KeyboardInterrupt:
print("\n\n" + "=" * 60)
print(" SHUTTING DOWN")
print("=" * 60)
capture.stop()
summary = alert_system.get_summary()
print(f" Packets processed: {packets_processed}")
print(f" Total alerts: {summary['total']}")
if summary["by_type"]:
print(f" Alerts by type:")
for attack_type, count in summary["by_type"].items():
print(f" - {attack_type}: {count}")
if summary["by_severity"]:
print(f" Alerts by severity:")
for severity, count in summary["by_severity"].items():
print(f" - {severity}: {count}")
print("=" * 60 + "\n")
if __name__ == "__main__":
main()Wire everything together and run
All four files are now in place. Time to start the IDS and confirm it can capture live traffic.
- Make sure your virtual environment is still active (you should see (venv) in your terminal prompt).
- Start the IDS with root privileges by running this command:
sudo python ids.pyWhy sudo?
Capturing raw network packets requires root access on macOS. The sudo command elevates your permissions so Scapy can open the network interface for sniffing. You will be prompted for your system password.
- Enter your system password when prompted.
- You should see the startup banner showing your interface (en0), loaded rules, and the "Monitoring... (Ctrl+C to stop)" message.
- Let it run for about 10 seconds to confirm packets are flowing without errors.
- Press Ctrl+C to stop the IDS.
You should see the shutdown summary showing the number of packets processed and any alerts that fired during those 10 seconds.
Seeing a permission error or no packets?
- If you see a permission denied error, make sure you used sudo before the command.
- If no packets appear after 10 seconds, verify your Wi-Fi is connected. Your interface may not be en0. Run ifconfig to find your active interface and update config.json accordingly.
- If you see a ModuleNotFoundError, confirm your virtual environment is active and you ran pip install scapy inside it.
Why isn't my IDS starting or capturing packets?
Your IDS is now fully operational. Next up, you will put it to the test by simulating real attack patterns and confirming your detection rules fire accurately.
Test with Attack Simulations
Your IDS is running and monitoring live traffic. But how do you know it actually catches attacks?
In this step, you will write a dedicated attack simulation script that uses Scapy to craft malicious-looking packets and send them to localhost. You will run the simulations while the IDS is active in a second terminal, then confirm that the correct alerts fire and the log file fills with structured JSON.
In this step, get ready to:
- Create a test script that simulates port scans, ping floods, and SYN floods.
- Verify detection accuracy by running attacks against the live IDS.
- Tune detection thresholds to eliminate false positives.
Create the port scan simulation
To test the IDS, you need to generate traffic that looks like real attacks. Scapy's send() function lets you craft and transmit individual packets with whatever headers you specify.
- Create a new file called test_attacks.py in your network-ids folder.
- Add the imports and the port scan simulation function by pasting this code:
import sys
import time
from scapy.all import IP, TCP, ICMP, send
def simulate_port_scan(target="127.0.0.1", ports=None):
if ports is None:
ports = [22, 80, 443, 8080, 3306, 5432, 6379, 8443]
print(f"[*] Simulating port scan on {target} ({len(ports)} ports)...")
for port in ports:
send(IP(dst=target) / TCP(dport=port, flags="S"), verbose=False)
time.sleep(0.1)
print("[+] Port scan simulation complete.")What does this code do?
The simulate_port_scan function sends a TCP SYN packet to each port in the list. This mimics what a real port scanner (like nmap) does when probing a target for open services.
- IP(dst=target) / TCP(dport=port, flags="S") builds a SYN packet destined for a specific port on localhost.
- The 8 ports cover common services (SSH, HTTP, HTTPS, MySQL, PostgreSQL, Redis) and exceed the detection threshold of 5 distinct ports.
- time.sleep(0.1) adds a short delay between packets so they arrive within the 10-second detection window.
Add the ping flood and SYN flood simulations
- Add the two remaining simulation functions below simulate_port_scan by pasting this code:
def simulate_ping_flood(target="127.0.0.1", count=20):
print(f"[*] Simulating ping flood on {target} ({count} packets)...")
for i in range(count):
send(IP(dst=target) / ICMP(type=8), verbose=False)
time.sleep(0.05)
print("[+] Ping flood simulation complete.")
def simulate_syn_flood(target="127.0.0.1", port=80, count=25):
print(f"[*] Simulating SYN flood on {target}:{port} ({count} packets)...")
for i in range(count):
send(IP(dst=target) / TCP(dport=port, flags="S"), verbose=False)
time.sleep(0.05)
print("[+] SYN flood simulation complete.")What does this code do?
- simulate_ping_flood sends 20 ICMP echo-request packets (type 8) in rapid succession. This exceeds the threshold of 15 packets within 5 seconds.
- simulate_syn_flood sends 25 SYN packets to a single port. Unlike the port scan (which hits many ports), a SYN flood hammers one port repeatedly to overwhelm it.
- Both functions use shorter delays (0.05 seconds) because floods need to be fast to trigger the thresholds.
Add the CLI entry point
The script needs a way to choose which attack to run from the command line.
- Add the main block at the bottom of test_attacks.py by pasting this code:
if __name__ == "__main__":
if len(sys.argv) < 2:
print("Usage: sudo python test_attacks.py [port_scan|ping_flood|syn_flood|all]")
sys.exit(1)
attack = sys.argv[1]
if attack == "port_scan":
simulate_port_scan()
elif attack == "ping_flood":
simulate_ping_flood()
elif attack == "syn_flood":
simulate_syn_flood()
elif attack == "all":
simulate_port_scan()
time.sleep(2)
simulate_ping_flood()
time.sleep(2)
simulate_syn_flood()
else:
print(f"Unknown attack type: {attack}")
sys.exit(1)What does this code do?
- sys.argv[1] reads the attack type from the command line argument.
- The all option runs all three simulations with a 2-second pause between each. This gives the detection engine time to process one attack before the next starts.
- Save test_attacks.py.
- Confirm the file exists by running:
ls test_attacks.pyYou should see test_attacks.py listed in the output.
File not found?
Make sure you are inside the network-ids directory with your virtual environment active. The file should sit alongside ids.py and detector.py.
Help me find my test_attacks.py file
✔️ Awesome, I've got everything!
Great. Make sure you have saved test_attacks.py before moving on.
ⓧ I'd like to double check the full code
import sys
import time
from scapy.all import IP, TCP, ICMP, send
def simulate_port_scan(target="127.0.0.1", ports=None):
if ports is None:
ports = [22, 80, 443, 8080, 3306, 5432, 6379, 8443]
print(f"[*] Simulating port scan on {target} ({len(ports)} ports)...")
for port in ports:
send(IP(dst=target) / TCP(dport=port, flags=\"S\"), verbose=False)
time.sleep(0.1)
print("[+] Port scan simulation complete.")
def simulate_ping_flood(target="127.0.0.1", count=20):
print(f"[*] Simulating ping flood on {target} ({count} packets)...")
for i in range(count):
send(IP(dst=target) / ICMP(type=8), verbose=False)
time.sleep(0.05)
print("[+] Ping flood simulation complete.")
def simulate_syn_flood(target="127.0.0.1", port=80, count=25):
print(f"[*] Simulating SYN flood on {target}:{port} ({count} packets)...")
for i in range(count):
send(IP(dst=target) / TCP(dport=port, flags=\"S\"), verbose=False)
time.sleep(0.05)
print("[+] SYN flood simulation complete.")
if __name__ == "__main__":
if len(sys.argv) < 2:
print("Usage: sudo python test_attacks.py [port_scan|ping_flood|syn_flood|all]")
sys.exit(1)
attack = sys.argv[1]
if attack == "port_scan":
simulate_port_scan()
elif attack == "ping_flood":
simulate_ping_flood()
elif attack == "syn_flood":
simulate_syn_flood()
elif attack == "all":
simulate_port_scan()
time.sleep(2)
simulate_ping_flood()
time.sleep(2)
simulate_syn_flood()
else:
print(f"Unknown attack type: {attack}")
sys.exit(1)Run the attack simulations against the live IDS
You need two terminal windows open at the same time: one running the IDS, and one running the attack script. Both require sudo because raw packet operations need root access.
- In your first terminal (where the IDS was running from the previous step), start the IDS by running:
sudo python ids.py- Open a second terminal window by pressing Cmd+N in Terminal.
- Navigate to your project directory and activate the virtual environment in this second terminal by running:
cd ~/Desktop/network-ids
source venv/bin/activate- Run the port scan simulation by running:
sudo python test_attacks.py port_scan- Switch to your first terminal (where the IDS is running).
You should see a [HIG] alert for port_scan with source IP 127.0.0.1 and a details message showing the number of distinct ports scanned.
- Switch back to your second terminal and run the ping flood simulation:
sudo python test_attacks.py ping_flood- Switch to the IDS terminal and confirm you see a [MED] alert for ping_flood.
- Switch back to your second terminal and run the SYN flood simulation:
sudo python test_attacks.py syn_flood- Switch to the IDS terminal and confirm you see a [CRT] alert for syn_flood.
Why three separate runs instead of using the all option?
Running each attack individually makes it easier to confirm that each detection rule fires correctly. Once you have verified them one by one, you can use all for future testing.
- Stop the IDS by pressing Ctrl+C in the first terminal.
You should see the shutdown summary showing the total packets processed and a breakdown of alerts by type and severity.
- Verify the log file contains structured JSON entries by running:
cat alerts.logEach line should be a valid JSON object with fields for alert_id, timestamp, type, severity, src_ip, and details.
Not seeing any alerts?
- Make sure you ran both the IDS and the attack script with sudo. Without root access, Scapy cannot send or capture raw packets.
- Confirm the IDS was actively monitoring (you should see "Monitoring..." in the banner) before you ran the attack script.
- Check that the loopback interface (lo0) traffic is visible on en0. On some macOS configurations, localhost traffic stays on lo0. If you suspect this, try changing the interface value in config.json to lo0 and restart the IDS.
My IDS is running but no alerts appear when I run test_attacks.py
Tune thresholds and confirm no false positives
A good IDS alerts on real attacks but stays silent during normal traffic. If your thresholds are too low, background traffic (DNS queries, ARP broadcasts) can trigger false positives.
- Start the IDS again with sudo python ids.py and let it run for 30 seconds without generating any attack traffic.
- Watch the terminal. No alerts should appear during this quiet period.
- Stop the IDS with Ctrl+C and check the shutdown summary. The total alerts count should be 0.
What if I see false positives?
If alerts fire during the quiet period, open config.json and increase the relevant threshold. For example, if you see a spurious ping flood alert, raise the packets value under ping_flood from 15 to 25.
After adjusting, re-run the IDS for another 30 seconds to confirm silence. Then re-run sudo python test_attacks.py all to confirm the simulated attacks still trigger alerts at the new threshold.
Delete old alerts before re-testing
Each time you re-run the IDS, new alerts append to alerts.log. If you want a clean slate for threshold tuning, delete the log first with rm alerts.log before starting the IDS.
Your IDS is now tested, tuned, and proven. You have a working intrusion detection system that captures live packets, detects four attack types, and logs structured alerts. In the Secret Mission, you will add a live terminal dashboard that displays these alerts in a color-coded, auto-refreshing table.
Secret mission
Add a Live Terminal Dashboard
Your IDS logs alerts to the terminal, but reading raw text output feels like reading a server log from the 90s. In this secret mission, you'll add a live Rich terminal dashboard that displays alerts in a color-coded, auto-refreshing table with real-time statistics.
Clean Up Your Resources
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 costs.
Resources you used:
- Project directory network-ids/ on your Desktop (source files, config, test scripts).
- Python virtual environment in network-ids/venv/ (Scapy and Rich installed here).
- The alerts.log file inside network-ids/ (grows each time the IDS runs and detects threats).
Keep everything running
No action needed. Choose this if you want to keep the project on your machine as a portfolio piece or continue experimenting with new detection rules.
- Your network-ids/ folder stays on your Desktop with all source files intact.
- The virtual environment preserves your Scapy and Rich installations.
- The alerts.log file will continue growing each time you run the IDS. You can delete it manually at any time without affecting the system.
Pause - I'll come back to this later
There are no cloud resources or background processes to pause. The IDS only runs when you explicitly start it with sudo python ids.py.
- Make sure the IDS is not actively running. If it is, press Ctrl+C in its terminal window to stop it.
- No other action needed. Everything is local and costs nothing while idle.
Delete - I don't want to use this again
Remove all project files, the virtual environment, and the alert log in one step.
- If the IDS is still running, press Ctrl+C to stop it first.
- Navigate to your Desktop and remove the project directory by running:
cd ~/Desktop
rm -rf network-ids/- Confirm deletion by running:
ls | grep network-ids- If nothing is returned, the directory has been successfully removed.
Nice Work!
Nice Work!
You just built a real-time network intrusion detection system from scratch using Python and Scapy. Your IDS captures live packets, analyzes them against configurable detection rules, and generates structured alerts when threats are detected.
You've learned how to:
- Built a real-time packet capture engine using Scapy's AsyncSniffer with thread-safe queues for non-blocking packet processing on macOS.
- Implemented 4 signature-based detection rules with sliding window algorithms to detect port scans, SYN floods, ping floods, and ARP spoofing.
- Tested the IDS against self-generated attack simulations and confirmed detection accuracy with structured JSON alert logging.
- Secret Mission: Added a live Rich terminal dashboard with color-coded alerts and real-time statistics.
Ready to quiz yourself?