Build a network traffic analyzer that captures packets, detects anomalies, and generates security alerts for suspicious activity.
Project Overview
Build a network traffic analyzer that captures packets, detects anomalies, and generates security alerts for suspicious activity. This hands-on project will help you apply theoretical security knowledge to build practical skills that are directly applicable in professional security roles.
Learning Objectives
By completing this project, you will:
- Understand the security challenges this project addresses
- Implement security controls using Python
- Apply defense-in-depth principles
- Test and validate your implementation
- Document your work for a portfolio
Project Architecture
Project Architecture:
========================
+------------------+ +------------------+ +------------------+
| User Input | --> | Security Layer | --> | Core Logic |
| (Interface) | | (Validation, | | (Processing) |
| | | Auth, Encrypt) | | |
+------------------+ +------------------+ +------------------+
| |
v v
+------------------+ +------------------+
| Logging & | | Data Storage |
| Monitoring | | (Encrypted) |
+------------------+ +------------------+
Implementation
Step 1: Project Setup
import os
import hashlib
import secrets
import json
from datetime import datetime
from pathlib import Path
# Project configuration
PROJECT_CONFIG = {
"name": "Network Monitoring Project",
"version": "1.0.0",
"security_level": "production",
"encryption": "AES-256-GCM",
"hashing": "PBKDF2-SHA256",
"logging": True
}
class ProjectBase:
"""Base class for the security project."""
def __init__(self, config: dict = None):
self.config = config or PROJECT_CONFIG
self.start_time = datetime.now()
self.log_entries = []
def log_event(self, event_type: str, details: str):
"""Log security-relevant events."""
entry = {
"timestamp": datetime.now().isoformat(),
"type": event_type,
"details": details
}
self.log_entries.append(entry)
if self.config.get("logging"):
print(f"[{event_type}] {details}")
def generate_secure_token(self, length: int = 32) -> str:
"""Generate cryptographically secure random token."""
return secrets.token_urlsafe(length)
def hash_data(self, data: str, salt: bytes = None) -> dict:
"""Securely hash data with salt."""
if salt is None:
salt = os.urandom(32)
key = hashlib.pbkdf2_hmac('sha256', data.encode(), salt, 600000)
return {"hash": key.hex(), "salt": salt.hex()}
Step 2: Core Security Implementation
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
import base64
class SecurityEngine(ProjectBase):
"""Core security engine for the project."""
def __init__(self, master_key: bytes = None):
super().__init__()
self.master_key = master_key or os.urandom(32)
self.aesgcm = AESGCM(self.master_key)
self.log_event("INIT", "Security engine initialized")
def encrypt(self, plaintext: str) -> dict:
"""Encrypt data using AES-256-GCM."""
nonce = os.urandom(12)
ciphertext = self.aesgcm.encrypt(nonce, plaintext.encode(), None)
self.log_event("ENCRYPT", f"Encrypted {len(plaintext)} bytes")
return {
"nonce": base64.b64encode(nonce).decode(),
"ciphertext": base64.b64encode(ciphertext).decode()
}
def decrypt(self, encrypted: dict) -> str:
"""Decrypt AES-256-GCM encrypted data."""
nonce = base64.b64decode(encrypted["nonce"])
ciphertext = base64.b64decode(encrypted["ciphertext"])
plaintext = self.aesgcm.decrypt(nonce, ciphertext, None)
self.log_event("DECRYPT", "Data decrypted successfully")
return plaintext.decode()
def derive_key(self, password=[REDACTED_PASSWORD] salt: bytes = None) -> bytes:
"""Derive encryption key from password using PBKDF2."""
if salt is None:
salt = os.urandom(16)
kdf = PBKDF2HMAC(
algorithm=hashes.SHA256(),
length=32,
salt=salt,
iterations=600000
)
return kdf.derive(password.encode())
# Demonstration
engine = SecurityEngine()
secret_data = "This is sensitive information that needs protection"
encrypted = engine.encrypt(secret_data)
print(f"Encrypted: {encrypted['ciphertext'][:40]}...")
decrypted = engine.decrypt(encrypted)
print(f"Decrypted: {decrypted}")
assert decrypted == secret_data, "Decryption failed!"
Step 3: Testing and Validation
import unittest
class SecurityTests(unittest.TestCase):
"""Test suite for security implementation."""
def setUp(self):
self.engine = SecurityEngine()
def test_encryption_decryption(self):
"""Verify encrypt/decrypt roundtrip."""
original = "Test data for encryption"
encrypted = self.engine.encrypt(original)
decrypted = self.engine.decrypt(encrypted)
self.assertEqual(original, decrypted)
def test_different_nonces(self):
"""Verify each encryption uses unique nonce."""
data = "Same data encrypted twice"
enc1 = self.engine.encrypt(data)
enc2 = self.engine.encrypt(data)
self.assertNotEqual(enc1["nonce"], enc2["nonce"])
self.assertNotEqual(enc1["ciphertext"], enc2["ciphertext"])
def test_tamper_detection(self):
"""Verify tampered ciphertext is rejected."""
encrypted = self.engine.encrypt("Original message")
# Tamper with ciphertext
tampered = encrypted.copy()
ct_bytes = bytearray(base64.b64decode(tampered["ciphertext"]))
ct_bytes[0] ^= 0xFF # Flip bits
tampered["ciphertext"] = base64.b64encode(bytes(ct_bytes)).decode()
with self.assertRaises(Exception):
self.engine.decrypt(tampered)
def test_empty_input(self):
"""Verify empty string handling."""
encrypted = self.engine.encrypt("")
decrypted = self.engine.decrypt(encrypted)
self.assertEqual("", decrypted)
# Run tests
if __name__ == "__main__":
unittest.main(verbosity=2)
Challenges and Extensions
| Challenge Level | Task | Skills Practiced |
|---|
| Beginner | Add input validation for all user inputs | Input sanitization |
| Intermediate | Implement rate limiting for brute force protection | Throttling, timing |
| Advanced | Add audit logging with tamper-evident chain | Integrity, forensics |
| Expert | Implement zero-knowledge proof authentication | Advanced crypto |
Security Considerations
- Key Management: Never hardcode keys; use environment variables or key vaults
- Error Handling: Never expose stack traces or internal details
- Logging: Log security events but never log sensitive data (passwords, keys)
- Testing: Include both functional tests and security-focused tests
- Dependencies: Keep libraries updated and use SCA tools
Interview Questions
- What security controls did you implement in this project and why?
- AES-256-GCM for authenticated encryption (confidentiality + integrity), PBKDF2 for password-to-key derivation (slow = resists brute force), CSPRNG for all random values (unpredictable tokens), and comprehensive logging (accountability without leaking secrets).
- How does your implementation handle key management?
- Keys are derived from user passwords using PBKDF2 with high iteration count, master keys are generated from cryptographic random sources, keys are never written to disk in plaintext, and session keys are ephemeral (discarded after use).
- What would you change for a production deployment?
- Use a dedicated secrets manager (HashiCorp Vault), implement proper certificate management, add monitoring and alerting, deploy in hardened containers, implement rate limiting, add WAF protection, and conduct penetration testing.
- How did you test the security of your implementation?
- Unit tests for crypto correctness, tamper detection tests, boundary value testing, fuzzing inputs, timing attack resistance verification, and code review against OWASP guidelines.
- What are the limitations of your implementation?
- Memory-based key storage (vulnerable to cold boot attacks), single encryption algorithm (no algorithm agility), limited to local use (no network security), and relies on OS-level security for file access control.
Summary
This project demonstrates practical security implementation skills including encryption, authentication, and secure coding practices. Extend it further by adding features, improving error handling, and hardening for production use.