27 KiB
Extension Challenges
You've built the base project. Now make it yours by extending it with new features.
These challenges are ordered by difficulty. Start with the easier ones to build confidence, then tackle the harder ones when you want to dive deeper.
Easy Challenges
Challenge 1: Add IPv6 Support Display
What to build: Enhance the protocol identification to explicitly recognize and display IPv6 packets separately from IPv4. Currently both show as IP addresses but aren't distinguished.
Why it's useful: IPv6 adoption is growing. Network monitoring tools need to track IPv4 vs IPv6 traffic separately to understand migration progress and troubleshoot dual-stack issues.
What you'll learn:
- Working with Scapy's IPv6 layer
- Extending the Protocol enum
- Modifying analyzer logic
Hints:
- Look at
analyzer.py:51-103where extract_packet_info() handles IP layer - Add check for
packet.haslayer(IPv6)alongside IP check - May want to add Protocol.IPv6 to enum or track as metadata
- Don't forget to handle cases where both IPv4 and IPv6 are present (tunnels)
Test it works:
# Generate IPv6 traffic
ping6 ::1
# Capture and verify IPv6 shows separately
sudo netanal capture -i lo -c 10 --verbose
You should see IPv6 addresses in output and statistics tracking both protocol versions.
Challenge 2: MAC Address OUI Lookup
What to build: Add manufacturer identification from MAC addresses using the OUI (first 3 bytes). Display "Apple Inc." instead of just "a4:83:e7:..."
Why it's useful: During incident response, knowing device manufacturers helps identify what's on the network. "Unknown Apple device" is more useful than cryptic MAC address.
What you'll learn:
- MAC address parsing
- OUI database lookups
- Caching strategies
Implementation approach:
-
Download OUI database from IEEE
- File: http://standards-oui.ieee.org/oui/oui.txt
- Parse into dict:
{"A483E7": "Apple, Inc."}
-
Add lookup function in new file
netanal/mac_lookup.pydef lookup_manufacturer(mac_address: str) -> str: oui = mac_address.replace(":", "")[:6].upper() return OUI_DATABASE.get(oui, "Unknown") -
Integrate with output in
output.py:print_packet()- Show manufacturer after MAC address
- Example: "a4:83:e7:12:34:56 (Apple Inc.)"
Test it works: Capture traffic on your local network. Manufacturers should appear for recognized devices.
Challenge 3: Bandwidth Alert Threshold
What to build:
Add command-line argument --alert-threshold that triggers visual alert when bandwidth exceeds specified MB/s.
Why it's useful: Real-time alerting during capture lets operators react immediately to traffic spikes, potential attacks, or misconfigurations.
What you'll learn:
- Real-time monitoring patterns
- Console notification techniques
- Threshold comparisons in streaming data
Hints:
- Add to CaptureConfig in
models.py:135-145 - Check threshold in
statistics.py:112-127when recording bandwidth samples - Use
output.py:print_warning()or print_error() for alerts - Consider using Rich's Panel for prominent alerts
Test it works:
# Alert on >1 MB/s
sudo netanal capture -i eth0 --alert-threshold 1.0
# Generate traffic to trigger
curl -O https://speed.hetzner.de/100MB.bin
Alert should appear in red when threshold exceeded.
Intermediate Challenges
Challenge 4: TCP Connection Tracking
What to build: Track TCP three-way handshakes and connection states (SYN, SYN-ACK, ACK, FIN). Display incomplete handshakes (potential SYN floods) and long-lived connections.
Real world application: SYN flood DDoS attacks send SYN packets without completing handshake. Detecting incomplete handshakes identifies attacks in progress. Long-lived connections might indicate persistent backdoors.
What you'll learn:
- TCP state machine
- Connection tuple tracking (src_ip, src_port, dst_ip, dst_port)
- Time-series data with connection lifetimes
Implementation approach:
-
Create connection state tracker in new file
netanal/tcp_tracker.py@dataclass class TCPConnection: src_ip: str src_port: int dst_ip: str dst_port: int state: str # SYN_SENT, ESTABLISHED, FIN_WAIT, etc. start_time: float last_seen: float class TCPTracker: def __init__(self): self._connections: dict[tuple, TCPConnection] = {} def process_packet(self, packet: PacketInfo): # Extract TCP flags # Update connection state based on flags # Track in dict keyed by (src_ip, src_port, dst_ip, dst_port) -
Integrate with statistics collector
- Add TCPTracker to StatisticsCollector
- Call tracker.process_packet() in record_packet()
-
Add reporting to show:
- Incomplete handshakes (SYN without SYN-ACK)
- Connections in each state
- Longest-lived connections
Gotchas:
- TCP flags in Scapy:
packet[TCP].flagsis bitmask, check specific flags - Connection direction matters: (A→B) is different from (B→A)
- Timeout old connections to prevent memory leak
Extra credit: Detect port scans by tracking many connections from one IP to different ports with SYN flags only.
Challenge 5: DNS Query/Response Correlation
What to build: Match DNS queries with their responses. Track query latency, failed queries, and which domains get queried most.
Real world application: DNS is often the first indicator of compromise. Tracking query patterns detects DNS tunneling, C2 beaconing, and DGA (domain generation algorithm) malware.
What you'll learn:
- Request/response correlation using transaction IDs
- DNS protocol structure (query vs response flag)
- Time-based pattern detection
Implementation approach:
-
Enhance DNS extraction in
analyzer.pydef extract_dns_info(packet: Packet) -> dict | None: if not packet.haslayer(DNS): return None dns = packet[DNS] return { "transaction_id": dns.id, "query": dns.qr == 0, # 0 = query, 1 = response "domain": dns.qd.qname.decode() if dns.qd else None, "response_code": dns.rcode if dns.qr == 1 else None, "answers": [str(rr.rdata) for rr in dns.an] if dns.an else [] } -
Track queries in new DNSTracker class
- Store pending queries keyed by transaction ID
- Match responses to queries
- Calculate latency: response_time - query_time
- Track failures (NXDOMAIN, SERVFAIL)
-
Report statistics:
- Average DNS latency
- Most queried domains
- Failed query percentage
- Suspicious patterns (too many unique domains, high entropy domains)
Testing:
# Generate DNS traffic
nslookup google.com
nslookup nonexistent-domain-12345.com
# Capture and analyze
sudo netanal capture -i lo -c 20
Should show both successful and failed queries with latency measurements.
Challenge 6: Packet Size Distribution Histogram
What to build: Track distribution of packet sizes and generate histogram chart. Useful for detecting unusual traffic patterns.
Real world application: Packet size distributions reveal application types. VoIP has consistent small packets. File transfers have large packets. DDoS attacks often use specific sizes (tiny for amplification, large for volumetric).
What you'll learn:
- Histogram generation with bins
- Statistical distribution analysis
- Custom matplotlib visualizations
Implementation approach:
-
Add size tracking to StatisticsCollector
# In statistics.py def __init__(self): # ... existing init self._size_buckets: dict[int, int] = defaultdict(int) self._size_bins = [64, 128, 256, 512, 1024, 1500, 9000] def _get_size_bucket(self, size: int) -> int: for bin_size in self._size_bins: if size <= bin_size: return bin_size return self._size_bins[-1] -
Create histogram chart in
visualization.pydef create_size_histogram( stats: CaptureStatistics, title: str = "Packet Size Distribution" ) -> Figure: # Create bar chart with size bins # X-axis: packet size ranges # Y-axis: count or percentage -
Add to CLI in
main.py:chart()command- New chart type:
--type size-histogram
- New chart type:
Test it works: Mix traffic types and observe distribution differences:
# HTTP traffic (varied sizes)
curl http://example.com
# DNS traffic (small consistent sizes)
for i in {1..10}; do nslookup google.com; done
# Analyze
sudo netanal capture -i lo -c 100
netanal chart captured.pcap --type size-histogram
Advanced Challenges
Challenge 7: Geolocation IP Mapping
What to build: Add GeoIP lookup to show country/city for external IPs. Display top talkers with geographic location and generate world map visualization.
Why this is hard: Requires external GeoIP database, coordinate transformation for map plotting, and handling edge cases (private IPs, localhost, unknown locations).
What you'll learn:
- GeoIP database integration (MaxMind or IP2Location)
- Geographic data visualization
- Coordinate system transformations
- Database file handling and updates
Architecture changes needed:
New Components:
- netanal/geoip.py → GeoIP lookup class
- Matplotlib basemap → World map plotting
- Database file → GeoLite2 or similar
Modified:
- EndpointStats → Add location fields
- statistics.py → Call GeoIP on new endpoints
- visualization.py → New map chart function
Implementation steps:
-
Research phase
- Read MaxMind GeoLite2 documentation
- Understand database format (MMDB)
- Look at geoip2 Python library
-
Design phase
- Decide: bundled database vs user download?
- Consider: cache lookups to avoid repeated database hits
- Plan: what to do with private IPs (don't lookup)
-
Implementation phase
- Start with simple country lookup
- Add city and lat/long
- Create world map scatter plot with matplotlib
- Size points by traffic volume
-
Testing phase
- Test with public IPs (8.8.8.8, 1.1.1.1)
- Verify private IPs skip lookup
- Check map rendering with actual traffic
Gotchas:
- MaxMind requires free account signup for database download
- Private IPs (192.168.x.x, 10.x.x.x) should skip lookup
- Localhost (127.0.0.1) has no geographic location
- Database files are large (~100MB), consider gitignore
Resources:
- MaxMind GeoLite2 (https://dev.maxmind.com/geoip/geolite2-free-geolocation-data)
- geoip2 Python library documentation
- matplotlib Basemap tutorial
Challenge 8: SSL/TLS Certificate Extraction
What to build: Extract and display SSL/TLS certificate information from HTTPS handshakes. Show certificate subject, issuer, validity dates, and certificate chain.
Why this is hard: Requires parsing TLS handshake protocol, handling multiple TLS versions, extracting X.509 certificates, and dealing with fragmented handshakes.
What you'll learn:
- TLS handshake protocol structure
- X.509 certificate format
- Scapy TLS layer usage
- Certificate chain validation concepts
Implementation steps:
Phase 1: Basic Certificate Extraction (8-10 hours)
# In analyzer.py, add TLS extraction
from scapy.layers.tls.record import TLS
from scapy.layers.tls.handshake import TLSServerHello, TLSCertificate
def extract_tls_info(packet: Packet) -> dict | None:
if not packet.haslayer(TLS):
return None
# Extract certificates from TLS handshake
# Parse X.509 certificate fields
# Return dict with subject, issuer, dates
Phase 2: Certificate Chain Handling (10-12 hours)
- Handle multiple certificates in chain
- Track root, intermediate, and leaf certificates
- Show certificate hierarchy
Phase 3: Integration and Display (6-8 hours)
- Add to statistics tracking
- Create certificate report table
- Flag expired or self-signed certificates
Phase 4: Advanced Features (8-10 hours)
- SNI (Server Name Indication) extraction
- Cipher suite detection
- TLS version tracking
- Weak cipher alerts
Testing strategy:
# Generate HTTPS traffic
curl https://google.com
curl https://expired.badssl.com # Test expired cert
# Capture and analyze
sudo netanal capture -i eth0 --filter "tcp port 443" -c 50
Known challenges:
-
TLS fragmentation
- Problem: Certificates span multiple TCP segments
- Hint: May need TCP stream reassembly (Challenge 4 helps)
-
TLS versions
- Problem: TLS 1.0, 1.1, 1.2, 1.3 have different structures
- Hint: Use Scapy's TLS layer which abstracts versions
Success criteria: Your implementation should:
- Extract certificate subject and issuer
- Show validity dates (not before, not after)
- Handle certificate chains
- Display SNI hostname
- Flag expired certificates
- Work with TLS 1.2 and 1.3
Challenge 9: Real-Time Anomaly Detection
What to build: Implement statistical anomaly detection that alerts on unusual traffic patterns in real time during capture. Detect bandwidth spikes, unusual protocol ratios, and new connections to strange ports.
Estimated time: 2-3 weeks for full implementation
Prerequisites: Complete Challenge 4 (TCP tracking) and Challenge 6 (size distribution) first, as this builds on those patterns.
What you'll learn:
- Statistical process control
- Z-score calculations for outlier detection
- Moving averages and standard deviations
- Real-time alert generation
- Baseline learning vs anomaly detection modes
Planning this feature:
Before you code, think through:
- How do you establish a baseline? (Need "learning mode" period)
- What metrics indicate anomalies? (Bandwidth, protocol ratio, connection rate)
- How do you avoid alert fatigue? (Threshold tuning, cooldown periods)
- What's your false positive tolerance? (Z-score thresholds)
High level architecture:
┌─────────────────────────────────────┐
│ Baseline Establishment │
│ (Learning Mode: 5-10 minutes) │
│ │
│ - Collect normal traffic samples │
│ - Calculate mean & std dev │
│ - Store baseline metrics │
└────────────┬────────────────────────┘
│
▼
┌─────────────────────────────────────┐
│ Real-Time Detection │
│ (Detection Mode: Continuous) │
│ │
│ - Compare current vs baseline │
│ - Calculate Z-scores │
│ - Trigger alerts on outliers │
└────────────┬────────────────────────┘
│
▼
┌─────────────────────────────────────┐
│ Alert Handler │
│ │
│ - Log anomaly details │
│ - Display terminal alert │
│ - Optional: webhook notification │
└─────────────────────────────────────┘
Implementation phases:
Phase 1: Baseline Collection (12-16 hours)
Create new file netanal/anomaly.py:
@dataclass
class BaselineMetrics:
bandwidth_mean: float
bandwidth_stddev: float
protocol_ratios: dict[Protocol, float]
connection_rate_mean: float
connection_rate_stddev: float
class AnomalyDetector:
def __init__(self, learning_period: int = 300): # 5 minutes
self._learning_mode = True
self._samples: list[float] = []
# ... more init
def learn(self, stats: CaptureStatistics):
# Collect samples during learning period
# Calculate statistics
pass
def detect(self, stats: CaptureStatistics) -> list[Anomaly]:
# Compare against baseline
# Return list of detected anomalies
pass
Phase 2: Statistical Detection (16-20 hours) Implement Z-score calculations:
def calculate_zscore(value: float, mean: float, stddev: float) -> float:
if stddev == 0:
return 0
return (value - mean) / stddev
def is_anomaly(zscore: float, threshold: float = 3.0) -> bool:
return abs(zscore) > threshold
Track multiple metrics:
- Bandwidth (bytes/second)
- Packet rate (packets/second)
- Protocol distribution (% TCP vs UDP vs other)
- Connection rate (new connections/second)
- Unique destination ports
Phase 3: Alert Generation (8-10 hours) Create alert types:
@dataclass
class Anomaly:
timestamp: float
metric: str # "bandwidth", "protocol_ratio", etc.
value: float
expected: float
severity: str # "low", "medium", "high"
description: str
def format_alert(anomaly: Anomaly) -> str:
# Create human-readable alert message
# Include what's unusual and by how much
Phase 4: Integration (6-8 hours)
- Add to CaptureEngine
- Call detector during bandwidth sampling
- Display alerts in real-time
- Log anomalies to file
Testing strategy:
Test with synthetic anomalies:
# Test 1: Bandwidth spike
# Baseline: normal traffic
# Inject: large file download
# Expected: bandwidth anomaly alert
# Test 2: Protocol shift
# Baseline: 80% TCP, 20% UDP
# Inject: UDP flood
# Expected: protocol ratio anomaly
# Test 3: Port scan
# Baseline: connections to normal ports
# Inject: nmap scan
# Expected: high connection rate + unusual ports
Known challenges:
-
Cold start problem
- Problem: No baseline on first run
- Hint: Save learned baselines to disk, load on subsequent runs
-
Concept drift
- Problem: Network behavior changes over time (new services, time of day)
- Hint: Implement adaptive baselines that slowly update
Success criteria: Your implementation should:
- Learn baseline in 5-10 minute period
- Detect bandwidth spikes (>3 std devs)
- Detect unusual protocol distributions
- Detect connection rate anomalies
- Display alerts in real time
- Include confidence score/severity
- Avoid excessive false positives (<10% during normal traffic)
- Save and load learned baselines
Performance Challenges
Challenge: Handle 10 Gbps Traffic
The goal: Optimize the capture engine to handle 10 gigabit per second traffic without dropping packets.
Current bottleneck: At 10 Gbps, you're processing ~10 million packets/second. Current code drops packets because:
- Queue fills (10K buffer is too small)
- Statistics lock contention (millions of lock acquisitions/sec)
- Python GIL limits parallelism
- Memory allocations slow garbage collection
Optimization approaches:
Approach 1: Multiple Processing Threads
- How: Use thread pool with N workers pulling from queue
- Gain: N× processing throughput
- Tradeoff: Need lock-free statistics or per-thread aggregation
Approach 2: Lock-Free Statistics
- How: Use per-thread counters, periodic aggregation
- Gain: Eliminates lock contention bottleneck
- Tradeoff: More complex code, eventual consistency
Approach 3: Sampling
- How: Process 1 out of every N packets
- Gain: Reduces CPU load proportionally
- Tradeoff: Statistical approximation, might miss rare events
Approach 4: BPF Aggregation
- How: Use eBPF to aggregate in kernel before userspace
- Gain: Massive performance improvement
- Tradeoff: Requires Linux, eBPF programming skills
Benchmark it:
# Generate high-speed traffic with iperf3
iperf3 -s & # Server
iperf3 -c localhost -b 10G # Client
# Monitor dropped packets
sudo netanal capture -i lo --verbose | grep "Dropped"
Target metrics:
- Dropped packets: <1% at 10 Gbps
- CPU usage: <80% on 8-core system
- Memory: <4 GB
Security Challenges
Challenge: Add PCAP Encryption
What to implement:
Encrypt captured PCAP files to protect sensitive network data. Add --encrypt flag that prompts for password and encrypts output using AES-256.
Threat model: This protects against:
- Unauthorized access to captured files
- Accidental disclosure of sensitive traffic
- Compliance requirements (HIPAA, PCI-DSS)
Implementation: Use cryptography library:
from cryptography.fernet import Fernet
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2
def encrypt_pcap(filepath: Path, password: str):
# Derive key from password using PBKDF2
# Encrypt file with Fernet (AES-128-CBC + HMAC)
# Write encrypted output with .enc extension
Testing the security:
- Try opening encrypted file with Wireshark (should fail)
- Verify password required for decryption
- Check that weak passwords are rejected
Challenge: Pass CIS Benchmark Checklist
The goal: Make this project compliant with relevant CIS (Center for Internet Security) controls for network monitoring tools.
Current gaps:
Gap 1: No audit logging
- Missing: Who ran captures, when, what filters used
- Remediation: Add audit log to ~/.netanal/audit.log
Gap 2: No input sanitization documentation
- Missing: Clear documentation of validated inputs
- Remediation: Document all validation in security.md
Gap 3: No resource limits
- Missing: Unbounded memory if misconfigured
- Remediation: Add hard limits with configuration
Gap 4: Privilege escalation without logging
- Missing: No record of when root privileges used
- Remediation: Log all elevated operations
Gap 5: No secure defaults
- Missing: Defaults allow promiscuous mode, any interface
- Remediation: Require explicit interface specification
Each gap maps to specific CIS controls. Implement fixes and document compliance.
Mix and Match
Combine features for bigger projects:
Project Idea 1: Network Security Dashboard
- Combine Challenge 4 (TCP tracking) + Challenge 9 (anomaly detection) + Challenge 7 (geolocation)
- Add web UI showing real-time map of connections with anomaly highlights
- Result: Visual SOC dashboard for network monitoring
Project Idea 2: DNS Security Monitor
- Combine Challenge 5 (DNS tracking) + Challenge 9 (anomaly detection)
- Add DGA detection (domain entropy analysis)
- Result: DNS-specific security monitoring tool
Project Idea 3: Encrypted Traffic Analysis
- Combine Challenge 8 (TLS extraction) + Challenge 4 (TCP tracking)
- Add JA3 fingerprinting (TLS client identification)
- Result: Detect malware by TLS patterns without decryption
Real World Integration Challenges
Integrate with SIEM (Splunk/ELK)
The goal: Send capture statistics to a SIEM for central logging and correlation.
What you'll need:
- SIEM instance (free Splunk or ELK stack)
- HTTP Event Collector (Splunk) or Logstash endpoint (ELK)
- JSON formatting for events
Implementation plan:
- Create
netanal/siem.pywith SIEM client - Add
--siem-urlCLI argument - Serialize statistics to JSON
- POST to SIEM endpoint every N seconds
- Handle connection failures gracefully
Watch out for:
- Rate limits on SIEM ingestion
- Authentication (API keys)
- Network failures (queue unsent events)
- Data privacy (don't send packet payloads)
Production checklist:
- TLS for SIEM connection
- API key from environment variable
- Retry logic with exponential backoff
- Local queue for offline operation
Deploy to Kubernetes
The goal: Package the tool as a container and deploy as DaemonSet for cluster-wide network monitoring.
What you'll learn:
- Docker containerization
- Kubernetes networking
- Privilege management in containers
- Distributed data collection
Steps:
-
Create Dockerfile
FROM python:3.14-slim RUN apt-get update && apt-get install -y libpcap-dev COPY . /app WORKDIR /app RUN pip install -e . CMD ["netanal", "capture"] -
Build container
docker build -t netanal:latest . -
Create Kubernetes manifest
apiVersion: apps/v1 kind: DaemonSet metadata: name: netanal spec: template: spec: hostNetwork: true # Required for packet capture containers: - name: netanal image: netanal:latest securityContext: capabilities: add: ["NET_RAW", "NET_ADMIN"] -
Deploy
kubectl apply -f netanal-daemonset.yaml
Production checklist:
- Non-root user in container
- Only required capabilities (NET_RAW)
- Resource limits (CPU/memory)
- Log aggregation to central collector
- Health check endpoints
Getting Help
Stuck on a challenge?
-
Debug systematically
- What did you expect to happen?
- What actually happened?
- What's the smallest test case that reproduces it?
- Can you add print statements to narrow down the problem?
-
Read the existing code
- Challenge 4 (TCP tracking) is similar to Challenge 5 (DNS tracking) - use same patterns
- Visualization code already handles multiple chart types - extend it
-
Search for similar problems
- "Scapy TCP flags" → finds examples of flag parsing
- "Python statistical anomaly detection" → finds Z-score algorithms
- "matplotlib world map" → finds basemap examples
-
Ask for help
- Describe what you're trying to build
- Show what you've tried
- Explain what went wrong
- Include relevant code snippets and error messages
Don't just paste "it doesn't work" with a stack trace. Explain your understanding of the problem.
Challenge Completion
Track your progress:
Easy:
- IPv6 Support Display
- MAC Address OUI Lookup
- Bandwidth Alert Threshold
Intermediate:
- TCP Connection Tracking
- DNS Query/Response Correlation
- Packet Size Distribution Histogram
Advanced:
- Geolocation IP Mapping
- SSL/TLS Certificate Extraction
- Real-Time Anomaly Detection
Performance:
- Handle 10 Gbps Traffic
Security:
- PCAP Encryption
- CIS Benchmark Compliance
Integration:
- SIEM Integration
- Kubernetes Deployment
Completed all of them? You've mastered network packet analysis. Consider contributing your solutions back to the project or building something new like an IDS or network forensics tool.
Challenge Yourself Further
Build Something New
Use the concepts you learned here to build:
- Wireless packet analyzer - Extend to monitor 802.11 frames, detect deauth attacks
- Industrial protocol analyzer - Add support for Modbus/SCADA protocols
- Network forensics tool - Reconstruct file transfers, extract artifacts from pcaps
Study Real Implementations
Compare your implementation to production tools:
- Wireshark/tshark - Look at display filters vs BPF, protocol dissectors
- Zeek (formerly Bro) - Study event-driven architecture, script extensibility
- Suricata - Examine multi-threading approach, GPU acceleration
Read their code, understand their tradeoffs, steal their good ideas.
Write About It
Document your extension:
- Blog post explaining "How I added anomaly detection to a packet analyzer"
- Tutorial for others to implement your feature
- Comparison: Your approach vs how Wireshark does it
Teaching others is the best way to verify you understand it.