Key Code Examples
1. Password Hashing & Verification (bcrypt)
import bcrypt
class PasswordManager:
"""Secure password hashing using bcrypt."""
@staticmethod
def hash_password(password: str) -> str:
"""Hash a password for storing."""
# bcrypt automatically generates a salt
salt = bcrypt.gensalt(rounds=12) # work factor 12
hashed = bcrypt.hashpw(password.encode('utf-8'), salt)
return hashed.decode('utf-8')
@staticmethod
def verify_password(password: str, hashed: str) -> bool:
"""Verify a password against its hash."""
return bcrypt.checkpw(
password.encode('utf-8'),
hashed.encode('utf-8')
)
@staticmethod
def needs_rehash(hashed: str) -> bool:
"""Check if password needs rehashing (work factor changed)."""
# Extract cost from hash
cost = int(hashed.split('$')[2])
return cost < 12
# Example
pm = PasswordManager()
password = "MyS3cur3P@ssw0rd!"
hashed = pm.hash_password(password)
print(f"Hash: {hashed}")
# $2b$12$xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
print(pm.verify_password(password, hashed)) # True
print(pm.verify_password("wrong", hashed)) # False
# Timing-safe comparison prevents timing attacks
import hmac
def safe_compare(a: str, b: str) -> bool:
return hmac.compare_digest(a.encode(), b.encode())
2. JWT Authentication with Refresh Tokens
import jwt
import datetime
from typing import Optional
class AuthManager:
"""JWT-based authentication with access and refresh tokens."""
def __init__(self, secret_key: str, algorithm: str = 'HS256'):
self.secret_key = secret_key
self.algorithm = algorithm
def create_access_token(self, user_id: str, email: str,
role: str = 'user') -> str:
"""Create short-lived access token (15 minutes)."""
payload = {
'sub': user_id,
'email': email,
'role': role,
'type': 'access',
'iat': datetime.datetime.utcnow(),
'exp': datetime.datetime.utcnow() + datetime.timedelta(minutes=15),
'jti': self._generate_jti(), # Unique token ID
}
return jwt.encode(payload, self.secret_key, algorithm=self.algorithm)
def create_refresh_token(self, user_id: str) -> str:
"""Create long-lived refresh token (7 days)."""
payload = {
'sub': user_id,
'type': 'refresh',
'iat': datetime.datetime.utcnow(),
'exp': datetime.datetime.utcnow() + datetime.timedelta(days=7),
'jti': self._generate_jti(),
}
return jwt.encode(payload, self.secret_key, algorithm=self.algorithm)
def verify_token(self, token: str, expected_type: str = 'access') -> Optional[dict]:
"""Verify and decode a token."""
try:
payload = jwt.decode(token, self.secret_key,
algorithms=[self.algorithm])
# Check token type
if payload.get('type') != expected_type:
return None
# Check expiration (jwt.decode does this automatically)
return payload
except jwt.ExpiredSignatureError:
print("Token has expired")
return None
except jwt.InvalidTokenError:
print("Invalid token")
return None
def _generate_jti(self) -> str:
"""Generate unique token ID for revocation."""
import uuid
return str(uuid.uuid4())
def refresh_access_token(self, refresh_token: str) -> Optional[str]:
"""Issue new access token using refresh token."""
payload = self.verify_token(refresh_token, expected_type='refresh')
if not payload:
return None
# In production: check if refresh token is revoked in DB
# if await self.is_token_revoked(payload['jti']):
# return None
return self.create_access_token(
payload['sub'],
payload.get('email', ''),
payload.get('role', 'user')
)
# Usage
auth = AuthManager(secret_key='super-secret-key-change-in-production')
access = auth.create_access_token('user-123', 'alice@example.com')
refresh = auth.create_refresh_token('user-123')
# Verify
payload = auth.verify_token(access)
print(f"User: {payload['email']}, Role: {payload['role']}")
3. Input Sanitization & XSS Prevention
import re
import html
from urllib.parse import urlparse
class InputValidator:
"""Input validation and sanitization utilities."""
@staticmethod
def sanitize_html(text: str) -> str:
"""Escape HTML special characters."""
return html.escape(text)
@staticmethod
def sanitize_sql_input(value: str) -> str:
"""Basic SQL injection prevention (use parameterized queries!)."""
# This is a LAST RESORT — always use parameterized queries
dangerous = [';', '--', '/*', '*/', 'xp_', 'EXEC', 'DROP',
'DELETE', 'INSERT', 'UPDATE', 'UNION']
sanitized = value
for pattern in dangerous:
sanitized = sanitized.replace(pattern, '')
return sanitized.strip()
@staticmethod
def validate_email(email: str) -> bool:
"""Validate email format."""
pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
return re.match(pattern, email) is not None
@staticmethod
def validate_password(password: str) -> dict:
"""Check password strength."""
result = {
'valid': True,
'errors': []
}
if len(password) < 12:
result['errors'].append('Must be at least 12 characters')
if not re.search(r'[A-Z]', password):
result['errors'].append('Must contain uppercase letter')
if not re.search(r'[a-z]', password):
result['errors'].append('Must contain lowercase letter')
if not re.search(r'[0-9]', password):
result['errors'].append('Must contain digit')
if not re.search(r'[!@#$%^&*(),.?":{}|<>]', password):
result['errors'].append('Must contain special character')
result['valid'] = len(result['errors']) == 0
return result
@staticmethod
def validate_url(url: str, allowed_schemes=None) -> bool:
"""Validate URL to prevent SSRF attacks."""
if allowed_schemes is None:
allowed_schemes = ['https']
try:
parsed = urlparse(url)
return parsed.scheme in allowed_schemes and bool(parsed.netloc)
except Exception:
return False
@staticmethod
def sanitize_filename(filename: str) -> str:
"""Prevent path traversal in filenames."""
# Remove path separators and special chars
filename = re.sub(r'[\\/:*?"<>|]', '', filename)
# Remove leading dots (hidden files)
filename = filename.lstrip('.')
return filename or 'unnamed'
@staticmethod
def strip_javascript(html_content: str) -> str:
"""Remove JavaScript from HTML content."""
# Remove script tags
html_content = re.sub(r']*>.*?', '',
html_content, flags=re.DOTALL | re.IGNORECASE)
# Remove event handlers
html_content = re.sub(r'\son\w+\s*=\s*["\'][^"\']*["\']', '',
html_content, flags=re.IGNORECASE)
# Remove javascript: URLs
html_content = re.sub(r'javascript:', '', html_content,
flags=re.IGNORECASE)
return html_content
# Usage
validator = InputValidator()
print(validator.sanitize_html(''))
# <script>alert("xss")</script>
print(validator.validate_email("user@example.com")) # True
print(validator.validate_email("not-an-email")) # False
print(validator.validate_password("Weak")) # {'valid': False, 'errors': [...]}
print(validator.validate_password("Str0ng!P@ssword")) # {'valid': True, 'errors': []}
4. AES-256 Encryption
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
import os
import base64
class EncryptionManager:
"""AES-256-GCM encryption for data protection."""
def __init__(self, password: str):
"""Derive encryption key from password."""
self.password = password.encode('utf-8')
def _derive_key(self, salt: bytes) -> bytes:
"""Derive a 256-bit key using PBKDF2."""
kdf = PBKDF2HMAC(
algorithm=hashes.SHA256(),
length=32, # 256 bits
salt=salt,
iterations=600_000, # OWASP recommendation
)
return kdf.derive(self.password)
def encrypt(self, plaintext: str) -> str:
"""Encrypt data with AES-256-GCM."""
salt = os.urandom(16) # 128-bit salt
nonce = os.urandom(12) # 96-bit nonce for GCM
key = self._derive_key(salt)
aesgcm = AESGCM(key)
ciphertext = aesgcm.encrypt(nonce, plaintext.encode('utf-8'), None)
# Format: salt + nonce + ciphertext (base64 encoded)
combined = salt + nonce + ciphertext
return base64.b64encode(combined).decode('utf-8')
def decrypt(self, encrypted_data: str) -> str:
"""Decrypt AES-256-GCM encrypted data."""
combined = base64.b64decode(encrypted_data.encode('utf-8'))
# Extract components
salt = combined[:16]
nonce = combined[16:28]
ciphertext = combined[28:]
key = self._derive_key(salt)
aesgcm = AESGCM(key)
plaintext = aesgcm.decrypt(nonce, ciphertext, None)
return plaintext.decode('utf-8')
# Usage
em = EncryptionManager(password="MyStrongMasterPassword123!")
secret_message = "Credit Card: 4111-1111-1111-1111"
encrypted = em.encrypt(secret_message)
print(f"Encrypted: {encrypted[:50]}...")
decrypted = em.decrypt(encrypted)
print(f"Decrypted: {decrypted}")
# Credit Card: 4111-1111-1111-1111
# File encryption example
def encrypt_file(input_path, output_path, password):
em = EncryptionManager(password)
with open(input_path, 'r') as f:
data = f.read()
encrypted = em.encrypt(data)
with open(output_path, 'w') as f:
f.write(encrypted)
5. SQL Injection Prevention
import sqlite3
import psycopg2
from psycopg2 import sql
from contextlib import contextmanager
class SecureDatabase:
"""Database interface with SQL injection prevention."""
def __init__(self, connection_string):
self.conn_string = connection_string
@contextmanager
def get_cursor(self):
conn = psycopg2.connect(self.conn_string)
try:
yield conn.cursor()
conn.commit()
except Exception:
conn.rollback()
raise
finally:
conn.close()
# SECURE: Parameterized queries (ALWAYS USE THIS)
def get_user_by_email(self, email: str):
"""Safely query user by email."""
with self.get_cursor() as cur:
# Parameters are safely escaped by the driver
cur.execute(
"SELECT id, email, name FROM users WHERE email = %s",
(email,) # Tuple of parameters
)
return cur.fetchone()
def search_products(self, search_term: str, category: str = None):
"""Safely search products with filters."""
with self.get_cursor() as cur:
query = "SELECT id, name, price FROM products WHERE name ILIKE %s"
params = [f'%{search_term}%']
if category:
query += " AND category = %s"
params.append(category)
query += " ORDER BY name LIMIT 50"
cur.execute(query, tuple(params))
return cur.fetchall()
# SECURE: Dynamic identifiers using psycopg2.sql
def get_sorted(self, table: str, sort_column: str, sort_order: str = 'ASC'):
"""Safely sort by dynamic column name."""
# Whitelist allowed columns
ALLOWED_COLUMNS = {'id', 'name', 'price', 'created_at', 'email'}
ALLOWED_TABLES = {'users', 'products', 'orders'}
ALLOWED_ORDERS = {'ASC', 'DESC'}
if sort_column not in ALLOWED_COLUMNS:
raise ValueError(f"Invalid column: {sort_column}")
if table not in ALLOWED_TABLES:
raise ValueError(f"Invalid table: {table}")
if sort_order.upper() not in ALLOWED_ORDERS:
raise ValueError(f"Invalid sort order: {sort_order}")
with self.get_cursor() as cur:
# sql.Identifier safely escapes identifiers
query = sql.SQL("SELECT * FROM {} ORDER BY {} {} LIMIT 100").format(
sql.Identifier(table),
sql.Identifier(sort_column),
sql.SQL(sort_order.upper())
)
cur.execute(query)
return cur.fetchall()
# VULNERABLE — Never do this!
def vulnerable_search(username):
conn = sqlite3.connect('app.db')
cur = conn.cursor()
# DANGEROUS: Direct string formatting allows SQL injection
cur.execute(f"SELECT * FROM users WHERE username = '{username}'")
# Attacker: username = "' OR '1'='1" → returns ALL users!
return cur.fetchall()
# SECURE — Always use parameters
def secure_search(username):
conn = sqlite3.connect('app.db')
cur = conn.cursor()
cur.execute("SELECT * FROM users WHERE username = ?", (username,))
return cur.fetchall()
6. Rate Limiting Middleware
import time
from collections import defaultdict
from functools import wraps
class RateLimiter:
"""In-memory rate limiter (use Redis in production)."""
def __init__(self):
# {key: [(timestamp, count), ...]}
self.requests = defaultdict(list)
def is_allowed(self, key: str, max_requests: int,
window_seconds: int) -> bool:
"""Check if request is within rate limit."""
now = time.time()
window_start = now - window_seconds
# Clean old entries
self.requests[key] = [
ts for ts in self.requests[key] if ts > window_start
]
# Check count
if len(self.requests[key]) >= max_requests:
return False
# Record request
self.requests[key].append(now)
return True
def get_remaining(self, key: str, max_requests: int,
window_seconds: int) -> dict:
"""Get rate limit status for response headers."""
now = time.time()
window_start = now - window_seconds
self.requests[key] = [
ts for ts in self.requests[key] if ts > window_start
]
remaining = max(0, max_requests - len(self.requests[key]))
reset_time = int(window_start + window_seconds)
return {
'X-RateLimit-Limit': max_requests,
'X-RateLimit-Remaining': remaining,
'X-RateLimit-Reset': reset_time,
}
# Flask decorator
def rate_limit(max_requests=100, window_seconds=60):
limiter = RateLimiter()
def decorator(f):
@wraps(f)
def wrapped(*args, **kwargs):
# In Flask: key = request.remote_addr
client_key = kwargs.get('client_ip', 'unknown')
headers = limiter.get_remaining(client_key, max_requests,
window_seconds)
if not limiter.is_allowed(client_key, max_requests,
window_seconds):
return {
'error': 'Too many requests',
'retry_after': window_seconds
}, 429, headers
response = f(*args, **kwargs)
# Add rate limit headers
if isinstance(response, tuple):
return (*response, headers)
return response, 200, headers
return wrapped
return decorator
# Usage with Redis (production)
import redis
class RedisRateLimiter:
def __init__(self, redis_url='redis://localhost:6379'):
self.redis = redis.from_url(redis_url, decode_responses=True)
def is_allowed(self, key: str, max_requests: int,
window_seconds: int) -> bool:
"""Sliding window rate limiter using Redis."""
pipe = self.redis.pipeline()
now = time.time()
window_key = f"ratelimit:{key}:{int(now / window_seconds)}"
pipe.incr(window_key)
pipe.expire(window_key, window_seconds + 1)
results = pipe.execute()
return results[0] <= max_requests
7. Web Application Firewall (WAF) Rules
import re
from typing import Optional
class WAF:
"""Simple Web Application Firewall for common attacks."""
# SQL Injection patterns
SQLI_PATTERNS = [
r"(\b(SELECT|INSERT|UPDATE|DELETE|DROP|UNION|ALTER)\b.*\b(FROM|INTO|TABLE|DATABASE)\b)",
r"(--|;|\/\*|\*\/|xp_|sp_)",
r"('(\s)*(OR|AND)(\s)+'(\s)*')",
r"(1\s*=\s*1|'1'\s*=\s*'1')",
r"(EXEC\s*\(|EXECUTE\s*\()",
r"(CONCAT\s*\(|CHAR\s*\()",
r"(INFORMATION_SCHEMA|SYSOBJECTS)",
]
# XSS patterns
XSS_PATTERNS = [
r"]*>.*?",
r"javascript:",
r"on\w+\s*=",
r" Optional[dict]:
"""Inspect incoming request for malicious content."""
# Check URL path
for pattern in self.traversal_regex:
if pattern.search(path):
return self._block('PATH_TRAVERSAL', path)
# Check query parameters and body
check_text = body + path
for pattern in self.sqli_regex:
if pattern.search(check_text):
return self._block('SQL_INJECTION', check_text[:100])
for pattern in self.xss_regex:
if pattern.search(check_text):
return self._block('XSS', check_text[:100])
for pattern in self.cmdi_regex:
if pattern.search(check_text):
return self._block('COMMAND_INJECTION', check_text[:100])
# Check for oversized requests
if len(body) > 10 * 1024 * 1024: # 10 MB limit
return self._block('OVERSIZED_REQUEST', f'Body: {len(body)} bytes')
return None # Request is clean
def _block(self, attack_type: str, detail: str) -> dict:
return {
'action': 'BLOCK',
'attack_type': attack_type,
'detail': detail,
'status_code': 403,
}
# Usage
waf = WAF()
# Malicious request
result = waf.inspect_request(
'GET', "/api/users?search=' OR 1=1--",
{}, ""
)
print(result)
# {'action': 'BLOCK', 'attack_type': 'SQL_INJECTION', ...}
# Clean request
result = waf.inspect_request(
'POST', "/api/search",
{}, '{"query": "laptop", "page": 1}'
)
print(result) # None
8. Secure Session Management
import secrets
import hashlib
import hmac
import time
from typing import Optional
class SessionManager:
"""Secure session management with best practices."""
SESSION_TIMEOUT = 3600 # 1 hour
ABSOLUTE_TIMEOUT = 86400 # 24 hours
def __init__(self, storage_backend):
self.storage = storage_backend # Redis in production
def create_session(self, user_id: str, ip_address: str,
user_agent: str) -> dict:
"""Create a new secure session."""
session_id = secrets.token_urlsafe(32) # 256-bit random
csrf_token = secrets.token_urlsafe(32)
session_data = {
'session_id': session_id,
'user_id': user_id,
'ip_address': ip_address,
'user_agent_hash': hashlib.sha256(
user_agent.encode()
).hexdigest(),
'csrf_token': csrf_token,
'created_at': time.time(),
'last_activity': time.time(),
'is_valid': True,
}
# Store session (use Redis in production)
self.storage.set(
f"session:{session_id}",
session_data,
ex=self.ABSOLUTE_TIMEOUT
)
return {
'session_id': session_id,
'csrf_token': csrf_token,
}
def validate_session(self, session_id: str, ip_address: str,
user_agent: str) -> Optional[dict]:
"""Validate session and check for anomalies."""
session_data = self.storage.get(f"session:{session_id}")
if not session_data or not session_data.get('is_valid'):
return None
# Check session timeout (idle)
if time.time() - session_data['last_activity'] > self.SESSION_TIMEOUT:
self.destroy_session(session_id)
return None
# Check absolute timeout
if time.time() - session_data['created_at'] > self.ABSOLUTE_TIMEOUT:
self.destroy_session(session_id)
return None
# Check for session hijacking (IP + User-Agent)
if session_data['ip_address'] != ip_address:
# Could be legitimate (mobile networks) — flag for review
self._flag_anomaly(session_id, 'IP_CHANGE')
current_ua_hash = hashlib.sha256(user_agent.encode()).hexdigest()
if session_data['user_agent_hash'] != current_ua_hash:
self._flag_anomaly(session_id, 'UA_CHANGE')
# Update last activity
session_data['last_activity'] = time.time()
self.storage.set(f"session:{session_id}", session_data,
ex=self.ABSOLUTE_TIMEOUT)
return session_data
def destroy_session(self, session_id: str):
"""Invalidate a session."""
session_data = self.storage.get(f"session:{session_id}")
if session_data:
session_data['is_valid'] = False
self.storage.set(f"session:{session_id}", session_data, ex=60)
def validate_csrf_token(self, session_id: str,
csrf_token: str) -> bool:
"""Validate CSRF token for state-changing requests."""
session_data = self.storage.get(f"session:{session_id}")
if not session_data:
return False
# Constant-time comparison to prevent timing attacks
return hmac.compare_digest(
session_data['csrf_token'],
csrf_token
)
def _flag_anomaly(self, session_id: str, anomaly_type: str):
"""Flag suspicious session activity."""
print(f"ANOMALY: {anomaly_type} on session {session_id}")
# In production: notify user, log to SIEM, etc.
9. Network Security — Firewall Configuration
#!/bin/bash
# setup_firewall.sh — Secure iptables configuration
# Flush existing rules
iptables -F
iptables -X
# Default policy: DROP everything
iptables -P INPUT DROP
iptables -P FORWARD DROP
iptables -P OUTPUT ACCEPT
# Allow loopback
iptables -A INPUT -i lo -j ACCEPT
iptables -A OUTPUT -o lo -j ACCEPT
# Allow established connections
iptables -A INPUT -m conntrack --ctstate ESTABLISHED,RELATED -j ACCEPT
iptables -A OUTPUT -m conntrack --ctstate ESTABLISHED,RELATED -j ACCEPT
# Allow SSH (rate limited)
iptables -A INPUT -p tcp --dport 22 -m conntrack --ctstate NEW \
-m recent --set --name ssh_attempts
iptables -A INPUT -p tcp --dport 22 -m conntrack --ctstate NEW \
-m recent --update --seconds 60 --hitcount 4 --name ssh_attempts \
-j DROP
iptables -A INPUT -p tcp --dport 22 -m conntrack --ctstate NEW -j ACCEPT
# Allow HTTP/HTTPS
iptables -A INPUT -p tcp --dport 80 -m conntrack --ctstate NEW -j ACCEPT
iptables -A INPUT -p tcp --dport 443 -m conntrack --ctstate NEW -j ACCEPT
# Allow DNS
iptables -A INPUT -p udp --dport 53 -j ACCEPT
iptables -A OUTPUT -p udp --dport 53 -j ACCEPT
# Allow NTP (time sync)
iptables -A OUTPUT -p udp --dport 123 -j ACCEPT
# Drop invalid packets
iptables -A INPUT -m conntrack --ctstate INVALID -j DROP
# SYN flood protection
iptables -A INPUT -p tcp --syn -m limit --limit 1/s --limit-burst 4 -j ACCEPT
iptables -A INPUT -p tcp --syn -j DROP
# Ping flood protection
iptables -A INPUT -p icmp --icmp-type echo-request \
-m limit --limit 1/s --limit-burst 4 -j ACCEPT
iptables -A INPUT -p icmp --icmp-type echo-request -j DROP
# Log dropped packets
iptables -A INPUT -j LOG --log-prefix "IPTables-Dropped: " --log-level 4
iptables -A FORWARD -j LOG --log-prefix "IPTables-Dropped: " --log-level 4
# Save rules
iptables-save > /etc/iptables/rules.v4
echo "Firewall rules applied and saved"
10. Security Headers & CSP Configuration
from flask import Flask, request, make_response
import hashlib
app = Flask(__name__)
def add_security_headers(response):
"""Add comprehensive security headers to all responses."""
# Prevent MIME type sniffing
response.headers['X-Content-Type-Options'] = 'nosniff'
# Prevent clickjacking
response.headers['X-Frame-Options'] = 'DENY'
# XSS protection (legacy, but defense in depth)
response.headers['X-XSS-Protection'] = '1; mode=block'
# Referrer policy
response.headers['Referrer-Policy'] = 'strict-origin-when-cross-origin'
# Permissions policy
response.headers['Permissions-Policy'] = (
'camera=(), microphone=(), geolocation=(self)'
)
# Content Security Policy
csp = (
"default-src 'self'; "
"script-src 'self' 'unsafe-inline' 'nonce-{nonce}'; "
"style-src 'self' 'unsafe-inline'; "
"img-src 'self' data: https:; "
"font-src 'self' data:; "
"connect-src 'self'; "
"frame-ancestors 'none'; "
"base-uri 'self'; "
"form-action 'self'; "
"upgrade-insecure-requests"
)
nonce = hashlib.sha256(
request.endpoint.encode() + str(time.time()).encode()
).hexdigest()[:16]
response.headers['Content-Security-Policy'] = csp.format(nonce=nonce)
# Strict Transport Security (HTTPS only)
response.headers['Strict-Transport-Security'] = (
'max-age=31536000; includeSubDomains; preload'
)
return response
@app.after_request
def after_request(response):
return add_security_headers(response)
# Generate CSP nonce for templates
@app.context_processor
def inject_nonce():
nonce = hashlib.sha256(str(time.time()).encode()).hexdigest()[:16]
return {'csp_nonce': nonce}
# Template usage:
#
Core Concepts
CIA Triad (Security Fundamentals)
Confidentiality: Ensuring data is accessible only to authorized parties. Achieved through encryption, access controls.
Integrity: Ensuring data is accurate and unaltered. Achieved through hashing, digital signatures, checksums.
Availability: Ensuring systems and data are accessible when needed. Achieved through redundancy, failover, DDoS protection.
Non-Repudiation: Preventing denial of actions. Achieved through digital signatures, audit logs.
Threat Modeling (STRIDE)
Definition: A structured approach to identifying and mitigating security threats.
- S — Spoofing: Impersonating someone else → Authentication
- T — Tampering: Modifying data → Integrity checks
- R — Repudiation: Denying actions → Audit logging
- I — Information Disclosure: Leaking sensitive data → Encryption
- D — Denial of Service: Making system unavailable → Rate limiting
- E — Elevation of Privilege: Gaining unauthorized access → Authorization
Common Vulnerabilities (OWASP Top 10)
- Broken Access Control: Users can access unauthorized resources
- Cryptographic Failures: Sensitive data exposed (plaintext passwords, weak encryption)
- Injection: SQL, NoSQL, OS command injection
- Insecure Design: Missing or flawed security architecture
- Security Misconfiguration: Default configs, verbose error messages
- Vulnerable Components: Outdated libraries with known vulnerabilities
- Auth Failures: Weak session management, credential stuffing
- Data Integrity Failures: Insecure deserialization, lack of verification
- Logging Failures: Not logging security events, logs not monitored
- SSRF: Server requests internal resources on behalf of attacker
Encryption Fundamentals
Symmetric Encryption: Same key for encrypt/decrypt (AES-256, ChaCha20). Fast, used for data at rest.
Asymmetric Encryption: Public/private key pair (RSA, ECC). Used for key exchange, digital signatures.
Hashing: One-way function (SHA-256, bcrypt, Argon2). Used for passwords, integrity verification.
Digital Signatures: Hash encrypted with private key. Proves authenticity and integrity.
Authentication vs Authorization
Authentication (AuthN): "Who are you?" — Verifying identity (passwords, MFA, biometrics).
Authorization (AuthZ): "What can you do?" — Checking permissions (RBAC, ABAC, ACLs).
- Multi-Factor Authentication (MFA): Something you know + have + are
- OAuth 2.0: Authorization framework for delegated access
- SSO (Single Sign-On): One login for multiple systems (SAML, OpenID Connect)
Network Security Controls
- Firewall: Filters traffic based on rules (packet, stateful, application)
- IDS/IPS: Detects and prevents intrusions in real-time
- VPN: Encrypted tunnel for remote access (IPSec, SSL/TLS, WireGuard)
- Zero Trust: "Never trust, always verify" — no implicit trust based on network location
- DMZ: Network segment isolating public-facing services from internal network
- Network Segmentation: VLANs, micro-segmentation to limit lateral movement
Incident Response Process
- Preparation: Tools, plans, communication channels ready
- Identification: Detect and classify the incident
- Containment: Short-term (isolate systems) and long-term (fix vulnerabilities)
- Eradication: Remove threat from environment
- Recovery: Restore systems, monitor for reinfection
- Lessons Learned: Post-incident review, update procedures
Security Frameworks & Compliance
- NIST CSF: Identify, Protect, Detect, Respond, Recover
- ISO 27001: Information Security Management System
- SOC 2: Service Organization Controls (Type I, Type II)
- PCI DSS: Payment Card Industry Data Security Standard
- HIPAA: Healthcare data protection (US)
- GDPR: EU data protection and privacy regulation
Worked Examples
Example 1: Analyze a Phishing Email
Problem: Analyze this email header for signs of phishing.
Return-Path: <ceo@eduv3rse.com>
From: "EduVerse CEO" <ceo@eduv3rse.com>
To: victim@company.com
Subject: URGENT: Verify Your Account Now!
Reply-To: support@eduv3rse-verify.com
X-Mailer: PHPMailer 5.2.14
Received: from unknown IP 45.33.32.156 by mx.google.com
Red Flags:
- Domain spoofing: "eduv3rse.com" vs "eduverse.com" (l → 3)
- Reply-To differs from From address
- Urgency tactic: "URGENT: Verify Now!"
- Generic greeting, threatening language
- Suspicious IP from unknown source
- Free email client (PHPMailer) used for "official" communication
Conclusion: This is a phishing attempt. Do not click links or provide credentials.
Example 2: Password Security Analysis
Problem: Compare password storage methods and demonstrate a secure implementation.
Insecure Methods:
- Plaintext storage: Database breach exposes all passwords
- MD5/SHA1: Vulnerable to rainbow table attacks
- Simple SHA-256: Too fast, vulnerable to brute force
Secure Implementation:
import bcrypt
import secrets
def hash_password(password: str) -> str:
"""Secure password hashing with bcrypt."""
# bcrypt automatically handles salt
# Work factor 12 = ~250ms per hash (adjust based on server capacity)
salt = bcrypt.gensalt(rounds=12)
return bcrypt.hashpw(password.encode(), salt).decode()
def verify_password(password: str, hash_str: str) -> bool:
"""Verify password against bcrypt hash."""
return bcrypt.checkpw(password.encode(), hash_str.encode())
# Additional security measures:
# 1. Rate limit login attempts
# 2. Implement account lockout after 5 failed attempts
# 3. Use constant-time comparison
# 4. Add pepper (application-level secret) on top of bcrypt
# 5. Implement password breach checking (Have I Been Pwned API)
def check_password_breach(password: str) -> bool:
"""Check if password has been in a known data breach."""
import hashlib
import requests
sha1 = hashlib.sha1(password.encode()).hexdigest().upper()
prefix, suffix = sha1[:5], sha1[5:]
response = requests.get(f"https://api.pwnedpasswords.com/range/{prefix}")
return suffix in response.text
Example 3: SQL Injection Attack & Defense
Problem: Demonstrate how SQL injection works and how to prevent it.
Attack Scenario:
-- Vulnerable query (Python):
query = f"SELECT * FROM users WHERE username='{username}' AND password='{password}'"
-- Attacker input:
username = admin'--
password = anything
-- Resulting query (commented out password check):
SELECT * FROM users WHERE username='admin'--' AND password='anything'
-- The '--' comments out the rest, bypassing password check!
-- Another attack: Dump all users
username = ' OR '1'='1
password = ' OR '1'='1
-- Result: SELECT * FROM users WHERE username='' OR '1'='1' AND ...
Defense (Parameterized Queries):
# SECURE: Parameters are escaped by the database driver
cursor.execute(
"SELECT * FROM users WHERE username = %s AND password = %s",
(username, password)
)
# The database treats input as DATA, never as SQL code
Example 4: Implement HTTPS with HSTS
Problem: Configure a web server to enforce HTTPS with HSTS.
# Nginx SSL Configuration
server {
listen 443 ssl http2;
server_name eduverse.com;
ssl_certificate /etc/letsencrypt/live/eduverse.com/fullchain.pem;
ssl_certificate_key /etc/letsencrypt/live/eduverse.com/privkey.pem;
# Modern SSL configuration
ssl_protocols TLSv1.2 TLSv1.3;
ssl_ciphers ECDHE-ECDSA-AES128-GCM-SHA256:ECDHE-RSA-AES128-GCM-SHA256;
ssl_prefer_server_ciphers off;
# HSTS (force HTTPS for 2 years)
add_header Strict-Transport-Security "max-age=63072000; includeSubDomains; preload" always;
# SSL session caching
ssl_session_cache shared:SSL:10m;
ssl_session_timeout 10m;
# OCSP Stapling
ssl_stapling on;
ssl_stapling_verify on;
location / {
proxy_pass http://localhost:3000;
proxy_set_header X-Forwarded-Proto $scheme;
}
}
# HTTP → HTTPS redirect
server {
listen 80;
server_name eduverse.com;
return 301 https://$server_name$request_uri;
}
Example 5: Security Audit Checklist
Problem: Perform a security audit of a web application. What do you check?
- Authentication: Password policies, MFA, session management, brute force protection
- Authorization: Access control, privilege escalation, IDOR vulnerabilities
- Input Validation: XSS, SQL injection, CSRF, file upload vulnerabilities
- Encryption: HTTPS, data at rest encryption, key management
- Logging: Security event logging, log protection, monitoring
- Dependencies: Known vulnerabilities in libraries, outdated packages
- Configuration: Security headers, CORS, error messages, debug mode
Practice Problems
Problem 1: Hash a Password Securely
Write a Python function that hashes a password using bcrypt with a work factor of 12.
import bcrypt
def hash_password(password: str) -> str:
salt = bcrypt.gensalt(rounds=12)
return bcrypt.hashpw(password.encode('utf-8'), salt).decode('utf-8')
def verify_password(password: str, hashed: str) -> bool:
return bcrypt.checkpw(
password.encode('utf-8'),
hashed.encode('utf-8')
)
# Test
hashed = hash_password("MySecureP@ssw0rd!")
print(hashed) # $2b$12$...
print(verify_password("MySecureP@ssw0rd!", hashed)) # True
print(verify_password("wrong", hashed)) # False
Problem 2: Implement CSRF Protection
Add CSRF token generation and validation to a Flask application.
import secrets
import hmac
from flask import Flask, session, request, abort
app = Flask(__name__)
def generate_csrf_token() -> str:
if 'csrf_token' not in session:
session['csrf_token'] = secrets.token_urlsafe(32)
return session['csrf_token']
def validate_csrf_token():
token = request.headers.get('X-CSRF-Token') or request.form.get('csrf_token')
if not token or not hmac.compare_digest(token, session.get('csrf_token', '')):
abort(403, 'Invalid CSRF token')
@app.route('/api/transfer', methods=['POST'])
def transfer_funds():
validate_csrf_token()
# Process transfer...
return {'status': 'success'}
# Template: <input type="hidden" name="csrf_token" value="{{ csrf_token() }}">
Problem 3: Detect SQL Injection Attempts
Write a function that detects common SQL injection patterns in user input.
import re
def detect_sqli(input_string: str) -> dict:
patterns = {
'comment': r'(--|#|/\*|\*/)',
'union': r'\bUNION\b.*\bSELECT\b',
'or_true': r"'(\s)*(OR|AND)(\s)+'",
'stacked': r';\s*(DROP|INSERT|UPDATE|DELETE)',
'benchmark': r'(BENCHMARK|SLEEP|WAITFOR\s+DELAY)',
}
detected = []
for name, pattern in patterns.items():
if re.search(pattern, input_string, re.IGNORECASE):
detected.append(name)
return {
'is_malicious': len(detected) > 0,
'detected_patterns': detected,
'sanitized': re.sub(r'[^a-zA-Z0-9\s\.\-\_]', '', input_string)
}
# Test
print(detect_sqli("admin' OR '1'='1"))
# {'is_malicious': True, 'detected_patterns': ['or_true']}
print(detect_sqli("laptop computer"))
# {'is_malicious': False, 'detected_patterns': [], ...}
Problem 4: Implement AES File Encryption
Write functions to encrypt and decrypt files using AES-256-GCM.
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
import os
def encrypt_file(input_path, output_path, key):
nonce = os.urandom(12)
aesgcm = AESGCM(key)
with open(input_path, 'rb') as f:
plaintext = f.read()
ciphertext = aesgcm.encrypt(nonce, plaintext, None)
with open(output_path, 'wb') as f:
f.write(nonce + ciphertext)
def decrypt_file(input_path, output_path, key):
with open(input_path, 'rb') as f:
data = f.read()
nonce, ciphertext = data[:12], data[12:]
aesgcm = AESGCM(key)
plaintext = aesgcm.decrypt(nonce, ciphertext, None)
with open(output_path, 'wb') as f:
f.write(plaintext)
# Generate key: AESGCM.generate_key(bit_length=256)
Problem 5: Configure CORS Securely
Write a Flask CORS configuration that only allows specific origins.
from flask import Flask
from flask_cors import CORS
app = Flask(__name__)
# Whitelist approach — never use allow_origins="*" with credentials
ALLOWED_ORIGINS = [
"https://app.eduverse.com",
"https://admin.eduverse.com",
]
CORS(app,
origins=ALLOWED_ORIGINS,
methods=['GET', 'POST', 'PUT', 'DELETE'],
allow_headers=['Content-Type', 'Authorization'],
supports_credentials=True,
max_age=3600)
Problem 6: Implement Multi-Factor Authentication
Write a TOTP (Time-based One-Time Password) generator for 2FA.
import hmac
import hashlib
import struct
import time
import base64
def generate_totp(secret: str, time_step: int = 30,
digits: int = 6) -> str:
"""Generate TOTP code (compatible with Google Authenticator)."""
# Get current time step
counter = int(time.time() // time_step)
# Decode base32 secret
key = base64.b32decode(secret.upper())
# Create HMAC-SHA1
counter_bytes = struct.pack('>Q', counter)
hmac_digest = hmac.new(key, counter_bytes, hashlib.sha1).digest()
# Dynamic truncation
offset = hmac_digest[-1] & 0x0F
code = struct.unpack('>I', hmac_digest[offset:offset+4])[0]
code &= 0x7FFFFFFF
code %= 10 ** digits
return str(code).zfill(digits)
# Usage
secret = "JBSWY3DPEHPK3PXP" # Base32 encoded secret
code = generate_totp(secret)
print(f"Your 2FA code: {code}") # 6-digit code
Problem 7: Analyze Network Traffic for Anomalies
Write a Python script to detect port scanning from network logs.
from collections import defaultdict
def detect_port_scan(log_entries, threshold=20, time_window=60):
"""Detect port scanning from firewall log entries."""
# Group connections by source IP
connections = defaultdict(list)
for entry in log_entries:
src_ip = entry['src_ip']
dst_port = entry['dst_port']
timestamp = entry['timestamp']
connections[src_ip].append((timestamp, dst_port))
scanners = []
for ip, conns in connections.items():
# Sort by timestamp
conns.sort()
# Sliding window
for i in range(len(conns)):
window_start = conns[i][0]
ports_in_window = set()
for j in range(i, len(conns)):
if conns[j][0] - window_start <= time_window:
ports_in_window.add(conns[j][1])
else:
break
if len(ports_in_window) >= threshold:
scanners.append({
'ip': ip,
'unique_ports': len(ports_in_window),
'time_window': time_window
})
break # Already flagged this IP
return scanners
Problem 8: Implement Content Security Policy
Create a CSP header that allows scripts only from your own domain and a CDN.
# Express.js CSP configuration
const helmet = require('helmet');
app.use(helmet.contentSecurityPolicy({
directives: {
defaultSrc: ["'self'"],
scriptSrc: [
"'self'",
"'nonce-RANDOM_NONCE'",
"https://cdn.jsdelivr.net"
],
styleSrc: ["'self'", "'unsafe-inline'"],
imgSrc: ["'self'", "data:", "https:"],
connectSrc: ["'self'", "https://api.eduverse.com"],
fontSrc: ["'self'", "https://fonts.gstatic.com"],
objectSrc: ["'none'"],
mediaSrc: ["'self'"],
frameSrc: ["'none'"],
baseUri: ["'self'"],
formAction: ["'self'"],
upgradeInsecureRequests: [],
}
}));
Problem 9: Password Strength Checker
Write a function that evaluates password strength and provides feedback.
import re
def check_password_strength(password: str) -> dict:
score = 0
feedback = []
# Length
if len(password) >= 16:
score += 3
elif len(password) >= 12:
score += 2
elif len(password) >= 8:
score += 1
feedback.append('Use at least 12 characters')
else:
feedback.append('Too short — use at least 8 characters')
# Character diversity
if re.search(r'[A-Z]', password): score += 1
else: feedback.append('Add uppercase letters')
if re.search(r'[a-z]', password): score += 1
else: feedback.append('Add lowercase letters')
if re.search(r'[0-9]', password): score += 1
else: feedback.append('Add numbers')
if re.search(r'[!@#$%^&*(),.?":{}|<>]', password): score += 1
else: feedback.append('Add special characters')
# Common patterns to avoid
if re.search(r'(.)\1{2,}', password):
score -= 1
feedback.append('Avoid repeated characters')
if password.lower() in ['password', '123456', 'qwerty']:
score = 0
feedback.append('This is a common password — choose something unique')
levels = ['Very Weak', 'Weak', 'Fair', 'Good', 'Strong', 'Very Strong']
level = levels[min(score, 5)]
return {'score': score, 'level': level, 'feedback': feedback}
Problem 10: Implement Audit Logging
Create a security audit logging system that tracks authentication events.
import logging
import json
import hashlib
from datetime import datetime
class AuditLogger:
def __init__(self, log_file='/var/log/security/audit.log'):
self.logger = logging.getLogger('security_audit')
handler = logging.FileHandler(log_file)
handler.setFormatter(logging.Formatter('%(message)s'))
self.logger.addHandler(handler)
self.logger.setLevel(logging.INFO)
def log_auth_event(self, event_type, user_id, ip_address,
user_agent, success, details=None):
log_entry = {
'timestamp': datetime.utcnow().isoformat(),
'event_type': event_type, # login, logout, password_change
'user_id': user_id,
'ip_address': ip_address,
'user_agent_hash': hashlib.sha256(
user_agent.encode()
).hexdigest()[:16],
'success': success,
'details': details or {}
}
self.logger.info(json.dumps(log_entry))
def log_access(self, user_id, resource, action, allowed):
log_entry = {
'timestamp': datetime.utcnow().isoformat(),
'event_type': 'access',
'user_id': user_id,
'resource': resource,
'action': action,
'allowed': allowed
}
self.logger.info(json.dumps(log_entry))
# Usage
audit = AuditLogger()
audit.log_auth_event('login', 'user-123', '192.168.1.100',
'Mozilla/5.0...', True)
audit.log_auth_event('login', 'user-123', '10.0.0.50',
'curl/7.68.0', False,
details={'reason': 'invalid_password'})