Security Best Practices

← Back to System Design 101 | ← Previous: Observability

Introduction

Security isn't an afterthought—it's a fundamental part of system design. I've learned that security breaches often result from simple oversights, not sophisticated attacks. This article covers security patterns I implement in every production system.

Authentication & Authorization

JWT-Based Authentication

from jose import JWTError, jwt
from passlib.context import CryptContext
from datetime import datetime, timedelta
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

# Configuration
SECRET_KEY = "your-secret-key-here"  # Store in environment variables!
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
security = HTTPBearer()

class AuthService:
    """Authentication service."""
    
    def hash_password(self, password: str) -> str:
        """Hash password using bcrypt."""
        return pwd_context.hash(password)
    
    def verify_password(self, plain_password: str, hashed_password: str) -> bool:
        """Verify password against hash."""
        return pwd_context.verify(plain_password, hashed_password)
    
    def create_access_token(self, data: dict, expires_delta: timedelta = None) -> str:
        """Create JWT access token."""
        to_encode = data.copy()
        
        if expires_delta:
            expire = datetime.utcnow() + expires_delta
        else:
            expire = datetime.utcnow() + timedelta(minutes=15)
        
        to_encode.update({
            "exp": expire,
            "iat": datetime.utcnow(),
            "type": "access"
        })
        
        encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
        return encoded_jwt
    
    def verify_token(self, token: str) -> dict:
        """Verify and decode JWT token."""
        try:
            payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
            return payload
        except JWTError:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Could not validate credentials"
            )

auth_service = AuthService()

# Dependency for protected routes
async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security)
) -> dict:
    """Get current user from JWT token."""
    token = credentials.credentials
    payload = auth_service.verify_token(token)
    
    user_id = payload.get("sub")
    if not user_id:
        raise HTTPException(status_code=401, detail="Invalid token")
    
    user = db.users.find_one({"id": user_id})
    if not user:
        raise HTTPException(status_code=401, detail="User not found")
    
    return user

# Protected endpoint
@app.get("/api/profile")
async def get_profile(current_user: dict = Depends(get_current_user)):
    """Get current user's profile."""
    return current_user

# Login endpoint
@app.post("/api/auth/login")
async def login(email: str, password: str):
    """Authenticate user and return JWT token."""
    user = db.users.find_one({"email": email})
    
    if not user or not auth_service.verify_password(password, user["password_hash"]):
        raise HTTPException(status_code=401, detail="Invalid credentials")
    
    access_token = auth_service.create_access_token(
        data={"sub": user["id"]},
        expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    )
    
    return {
        "access_token": access_token,
        "token_type": "bearer"
    }

Role-Based Access Control (RBAC)

from enum import Enum
from typing import List

class Permission(Enum):
    READ_USERS = "users:read"
    WRITE_USERS = "users:write"
    DELETE_USERS = "users:delete"
    READ_ORDERS = "orders:read"
    WRITE_ORDERS = "orders:write"

class Role(Enum):
    ADMIN = "admin"
    USER = "user"
    MODERATOR = "moderator"

# Role -> Permissions mapping
ROLE_PERMISSIONS = {
    Role.ADMIN: [p for p in Permission],  # All permissions
    Role.MODERATOR: [
        Permission.READ_USERS,
        Permission.READ_ORDERS,
        Permission.WRITE_ORDERS
    ],
    Role.USER: [
        Permission.READ_ORDERS
    ]
}

def require_permission(permission: Permission):
    """Decorator to require specific permission."""
    def decorator(func):
        async def wrapper(*args, current_user: dict = Depends(get_current_user), **kwargs):
            user_role = Role(current_user.get("role", "user"))
            user_permissions = ROLE_PERMISSIONS.get(user_role, [])
            
            if permission not in user_permissions:
                raise HTTPException(
                    status_code=403,
                    detail=f"Permission denied: {permission.value}"
                )
            
            return await func(*args, current_user=current_user, **kwargs)
        return wrapper
    return decorator

# Usage
@app.delete("/api/users/{user_id}")
@require_permission(Permission.DELETE_USERS)
async def delete_user(user_id: str, current_user: dict):
    """Delete user (admin only)."""
    db.users.delete_one({"id": user_id})
    return {"status": "deleted"}

Encryption

Data at Rest

from cryptography.fernet import Fernet
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2
import base64
import os

class EncryptionService:
    """Encrypt sensitive data at rest."""
    
    def __init__(self, encryption_key: str):
        self.fernet = Fernet(encryption_key.encode())
    
    def encrypt(self, data: str) -> str:
        """Encrypt data."""
        encrypted = self.fernet.encrypt(data.encode())
        return base64.b64encode(encrypted).decode()
    
    def decrypt(self, encrypted_data: str) -> str:
        """Decrypt data."""
        decoded = base64.b64decode(encrypted_data.encode())
        decrypted = self.fernet.decrypt(decoded)
        return decrypted.decode()
    
    @staticmethod
    def generate_key() -> str:
        """Generate new encryption key."""
        return Fernet.generate_key().decode()

# Usage
encryption_service = EncryptionService(os.environ.get("ENCRYPTION_KEY"))

def store_sensitive_data(user_id: str, ssn: str):
    """Store SSN encrypted."""
    encrypted_ssn = encryption_service.encrypt(ssn)
    db.users.update_one(
        {"id": user_id},
        {"$set": {"ssn_encrypted": encrypted_ssn}}
    )

def get_sensitive_data(user_id: str) -> str:
    """Retrieve and decrypt SSN."""
    user = db.users.find_one({"id": user_id})
    encrypted_ssn = user["ssn_encrypted"]
    return encryption_service.decrypt(encrypted_ssn)

Data in Transit (TLS/SSL)

# Always use HTTPS in production
# uvicorn main:app --host 0.0.0.0 --port 443 --ssl-keyfile=key.pem --ssl-certfile=cert.pem

# Enforce HTTPS
from fastapi.middleware.httpsredirect import HTTPSRedirectMiddleware

if os.environ.get("ENV") == "production":
    app.add_middleware(HTTPSRedirectMiddleware)

Input Validation & Sanitization

from pydantic import BaseModel, EmailStr, validator, Field
from typing import Optional
import re

class UserCreate(BaseModel):
    """Validate user input."""
    email: EmailStr
    password: str = Field(..., min_length=8, max_length=100)
    first_name: str = Field(..., min_length=1, max_length=50)
    last_name: str = Field(..., min_length=1, max_length=50)
    
    @validator('password')
    def validate_password_strength(cls, v):
        """Ensure password meets security requirements."""
        if not re.search(r'[A-Z]', v):
            raise ValueError('Password must contain uppercase letter')
        if not re.search(r'[a-z]', v):
            raise ValueError('Password must contain lowercase letter')
        if not re.search(r'[0-9]', v):
            raise ValueError('Password must contain digit')
        if not re.search(r'[!@#$%^&*]', v):
            raise ValueError('Password must contain special character')
        return v
    
    @validator('first_name', 'last_name')
    def sanitize_name(cls, v):
        """Remove potentially dangerous characters."""
        # Only allow letters, spaces, hyphens, apostrophes
        return re.sub(r'[^a-zA-Z\s\-\']', '', v)

# SQL Injection Prevention
from sqlalchemy import text

# ❌ Vulnerable to SQL injection
def get_user_vulnerable(email: str):
    query = f"SELECT * FROM users WHERE email = '{email}'"  # NEVER DO THIS
    return db.execute(query)

# ✅ Safe - using parameterized queries
def get_user_safe(email: str):
    query = text("SELECT * FROM users WHERE email = :email")
    return db.execute(query, {"email": email})

Rate Limiting & DDoS Protection

from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

# Different limits for different endpoints
@app.post("/api/auth/login")
@limiter.limit("5/minute")  # Strict limit on login
async def login(request: Request, credentials: LoginCredentials):
    """Login with rate limiting."""
    pass

@app.get("/api/users")
@limiter.limit("100/minute")  # More generous for reads
async def list_users(request: Request):
    """List users."""
    pass

# IP-based blocking
class IPBlocker:
    """Block suspicious IPs."""
    
    def __init__(self, redis_client):
        self.redis = redis_client
        self.block_duration = 3600  # 1 hour
    
    def is_blocked(self, ip: str) -> bool:
        """Check if IP is blocked."""
        return self.redis.exists(f"blocked_ip:{ip}")
    
    def block_ip(self, ip: str, reason: str):
        """Block IP address."""
        self.redis.setex(
            f"blocked_ip:{ip}",
            self.block_duration,
            reason
        )
    
    def record_failed_login(self, ip: str):
        """Record failed login and block if threshold exceeded."""
        key = f"failed_login:{ip}"
        count = self.redis.incr(key)
        self.redis.expire(key, 300)  # 5 minute window
        
        if count > 10:  # 10 failures in 5 minutes
            self.block_ip(ip, "Too many failed login attempts")

ip_blocker = IPBlocker(redis_client)

@app.middleware("http")
async def block_suspicious_ips(request: Request, call_next):
    """Middleware to block suspicious IPs."""
    client_ip = request.client.host
    
    if ip_blocker.is_blocked(client_ip):
        raise HTTPException(status_code=403, detail="IP blocked")
    
    response = await call_next(request)
    return response

Secrets Management

# Never hardcode secrets!
# ❌ Bad
API_KEY = "sk_live_1234567890"

# ✅ Good - Use environment variables
import os
API_KEY = os.environ.get("API_KEY")

# ✅ Better - Use secrets manager (AWS Secrets Manager)
import boto3
from botocore.exceptions import ClientError

class SecretsManager:
    """Manage secrets using AWS Secrets Manager."""
    
    def __init__(self):
        self.client = boto3.client('secretsmanager')
        self._cache = {}
    
    def get_secret(self, secret_name: str) -> dict:
        """Retrieve secret from AWS Secrets Manager."""
        if secret_name in self._cache:
            return self._cache[secret_name]
        
        try:
            response = self.client.get_secret_value(SecretId=secret_name)
            secret = json.loads(response['SecretString'])
            self._cache[secret_name] = secret
            return secret
        except ClientError as e:
            raise Exception(f"Failed to retrieve secret: {e}")

secrets = SecretsManager()
db_credentials = secrets.get_secret("prod/database")

Security Headers

from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.trustedhost import TrustedHostMiddleware

# CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://yourdomain.com"],  # Don't use "*" in production!
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE"],
    allow_headers=["*"],
)

# Trusted hosts
app.add_middleware(
    TrustedHostMiddleware,
    allowed_hosts=["yourdomain.com", "*.yourdomain.com"]
)

# Security headers
@app.middleware("http")
async def add_security_headers(request: Request, call_next):
    """Add security headers to all responses."""
    response = await call_next(request)
    
    response.headers["X-Content-Type-Options"] = "nosniff"
    response.headers["X-Frame-Options"] = "DENY"
    response.headers["X-XSS-Protection"] = "1; mode=block"
    response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"
    response.headers["Content-Security-Policy"] = "default-src 'self'"
    
    return response

Audit Logging

class AuditLogger:
    """Log security-relevant events."""
    
    def __init__(self, logger):
        self.logger = logger
    
    def log_event(
        self,
        event_type: str,
        user_id: str,
        resource: str,
        action: str,
        result: str,
        ip_address: str = None,
        **metadata
    ):
        """Log security event."""
        self.logger.info(
            f"Security Event: {event_type}",
            extra={
                "event_type": event_type,
                "user_id": user_id,
                "resource": resource,
                "action": action,
                "result": result,
                "ip_address": ip_address,
                "timestamp": datetime.utcnow().isoformat(),
                **metadata
            }
        )

audit_logger = AuditLogger(logger)

# Log security events
@app.post("/api/users/{user_id}/permissions")
async def update_permissions(
    user_id: str,
    permissions: List[str],
    current_user: dict = Depends(get_current_user),
    request: Request = None
):
    """Update user permissions with audit logging."""
    
    # Update permissions
    db.users.update_one(
        {"id": user_id},
        {"$set": {"permissions": permissions}}
    )
    
    # Log the change
    audit_logger.log_event(
        event_type="permission_change",
        user_id=current_user["id"],
        resource=f"user:{user_id}",
        action="update_permissions",
        result="success",
        ip_address=request.client.host,
        new_permissions=permissions
    )
    
    return {"status": "updated"}

Lessons Learned

What worked:

1. JWT for stateless authentication

2. RBAC for authorization

3. Encryption for sensitive data

4. Comprehensive input validation

5. Rate limiting on all endpoints

6. Regular security audits

What didn't work:

1. Storing passwords in plain text (obviously!)

2. Using "*" for CORS in production

3. No rate limiting (got DDoSed)

4. Hardcoded secrets in code

5. Insufficient audit logging

Security Checklist

• ✅ Use HTTPS everywhere

• ✅ Hash passwords with bcrypt/argon2

• ✅ Validate all user input

• ✅ Use parameterized queries

• ✅ Implement rate limiting

• ✅ Add security headers

• ✅ Encrypt sensitive data at rest

• ✅ Use secrets manager

• ✅ Log security events

• ✅ Regular security updates

• ✅ Principle of least privilege

What's Next

With security covered, let's put it all together in a real-world case study:

• Real-World Design Case →: Complete system design example

---

Navigation:

• ← Previous: Observability

• Next: Real-World Design Case →

• Back to Overview