Security Fundamentals

import bcrypt

# Storing passwords -- bcrypt automatically embeds the salt in the hash,
# so you do not need to store the salt separately.
def hash_password(password: str) -> bytes:
    # rounds=12 means 2^12 = 4,096 iterations of the hash function.
    # This makes each hash take ~250ms -- fast enough for login,
    # painfully slow for an attacker trying billions of passwords.
    salt = bcrypt.gensalt(rounds=12)
    return bcrypt.hashpw(password.encode(), salt)

# Verifying passwords -- bcrypt extracts the salt from the stored hash
# and recomputes, so this is a constant-time comparison (prevents timing attacks).
def verify_password(password: str, hashed: bytes) -> bool:
    return bcrypt.checkpw(password.encode(), hashed)

┌─────────────────────────────────────────────────────┐
│                     JWT Token                        │
├─────────────────┬─────────────────┬─────────────────┤
│     Header      │     Payload     │    Signature    │
│  {"alg":"HS256"}│ {"sub":"12345"} │   HMACSHA256(   │
│                 │ {"exp":16234...}│   header.payload│
│                 │                 │   , secret)     │
└─────────────────┴─────────────────┴─────────────────┘
         ↓                ↓                 ↓
    Base64URL        Base64URL         Base64URL
         ↓                ↓                 ↓
    eyJhbGci....   eyJzdWIi....    SflKxwRJS...

import jwt
from datetime import datetime, timedelta

# Creating JWT -- the token is self-contained: it carries the user's identity
# and expiration time, signed by your secret key. The server does not need
# to store session state -- it verifies the signature on every request.
def create_token(user_id: str, secret: str) -> str:
    payload = {
        "sub": user_id,                                # Subject: who this token is for
        "iat": datetime.utcnow(),                      # Issued at: when it was created
        "exp": datetime.utcnow() + timedelta(hours=24) # Expires: auto-reject after 24h
    }
    return jwt.encode(payload, secret, algorithm="HS256")

# Verifying JWT -- the library checks: (1) signature is valid (not tampered),
# (2) token has not expired, (3) required claims are present.
# CRITICAL: always specify the allowed algorithms list to prevent
# the "alg: none" attack where an attacker strips the signature.
def verify_token(token: str, secret: str) -> dict:
    try:
        return jwt.decode(token, secret, algorithms=["HS256"])  # Whitelist algorithms!
    except jwt.ExpiredSignatureError:
        raise Exception("Token expired")
    except jwt.InvalidTokenError:
        raise Exception("Invalid token")

┌──────────┐                              ┌──────────────┐
│   User   │                              │Authorization │
│          │                              │   Server     │
└────┬─────┘                              └──────┬───────┘
     │                                           │
     │  1. User clicks "Login with Google"       │
     │──────────────────────────────────────────►│
     │                                           │
     │  2. Redirect to authorization page        │
     │◄──────────────────────────────────────────│
     │                                           │
     │  3. User grants permission                │
     │──────────────────────────────────────────►│
     │                                           │
     │  4. Redirect with authorization code      │
     │◄──────────────────────────────────────────│
     │                                           │
┌────┴─────┐  5. Exchange code for tokens ┌──────┴───────┐
│   App    │──────────────────────────────►│Auth Server   │
│          │◄──────────────────────────────│              │
└──────────┘  6. Access + Refresh tokens   └──────────────┘

# ❌ Vulnerable: No authorization check -- change user_id in the URL
# and access ANY user's data. This is how many data breaches happen.
# An attacker just iterates: /users/1/data, /users/2/data, ...
@app.get("/users/{user_id}/data")
def get_user_data(user_id: int):
    return db.get_user_data(user_id)  # Anyone can access any user's data!

# ✅ Secure: Verify the authenticated user has permission to access this resource.
# ALWAYS check authorization on the server side, never trust the client.
@app.get("/users/{user_id}/data")
def get_user_data(user_id: int, current_user: User = Depends(get_current_user)):
    if current_user.id != user_id and not current_user.is_admin:
        raise HTTPException(403, "Access denied")
    return db.get_user_data(user_id)

# ❌ Bad: MD5 is NOT a password hashing algorithm -- it was designed to be FAST.
# Fast is exactly what you do NOT want for passwords (helps attackers brute-force).
import hashlib
password_hash = hashlib.md5(password.encode()).hexdigest()  # Crackable in seconds

# ✅ Good: bcrypt is designed to be SLOW and includes a built-in salt.
# Cost factor 12 = ~250ms per hash. An attacker computing 4 hashes/second
# would need 7 years to brute-force a decent password.
import bcrypt
password_hash = bcrypt.hashpw(password.encode(), bcrypt.gensalt(12))

# ❌ SQL Injection vulnerable -- the user input becomes part of the SQL command.
# If user_id = "1 OR 1=1" this returns ALL users. If user_id = "1; DROP TABLE users; --"
# this deletes your table. This is not theoretical -- it happens in production.
query = f"SELECT * FROM users WHERE id = {user_id}"

# ✅ Parameterized query -- the database engine treats the parameter as a VALUE,
# never as executable SQL. Even if user_id = "1; DROP TABLE users; --",
# it searches for a user with that literal string as their ID (finds nothing, safely).
query = "SELECT * FROM users WHERE id = ?"
cursor.execute(query, (user_id,))

# ❌ Bad: Security questions are inherently insecure.
# Answers are often public (social media), guessable, or shared
# across services. Sarah Palin's email was hacked in 2008 using
# publicly available answers to her security questions.
def reset_password(email, mothers_maiden_name):
    pass

# ✅ Good: Cryptographically random token sent to the verified email.
# The token is single-use, time-limited, and impossible to guess.
def reset_password(email):
    token = generate_secure_token()              # Cryptographically random, 256-bit
    send_reset_email(email, token)               # Only the email owner receives it
    store_token_with_expiry(email, token, expiry=30_minutes)  # Auto-expires

# ❌ Bad: Debug mode in production -- exposes stack traces, source code,
# and often database credentials to anyone who triggers an error.
app.run(debug=True)

# ❌ Bad: Default/hardcoded credentials -- attackers try these first.
# "admin/admin123" is literally in every brute-force dictionary.
DATABASE_PASSWORD = "admin123"

# ✅ Good: Configuration from environment variables, never hardcoded.
# Debug mode defaults to False (safe by default).
app.run(debug=os.getenv("FLASK_DEBUG", False))
DATABASE_PASSWORD = os.getenv("DATABASE_PASSWORD")  # Injected at deployment time

# Regularly audit dependencies
npm audit
pip-audit
snyk test

# Keep dependencies updated
npm update
pip install --upgrade package_name

# ❌ Bad: No rate limiting -- an attacker can make unlimited login attempts,
# trying every password in the dictionary. This is called "credential stuffing"
# when using credentials leaked from other breached services.
@app.post("/login")
def login(credentials):
    return authenticate(credentials)

# ✅ Good: Rate limiting + account lockout -- limits the blast radius of attacks.
# 5 attempts per minute per IP, plus account-level lockout after 5 failures.
from slowapi import Limiter

limiter = Limiter(key_func=get_remote_address)

@app.post("/login")
@limiter.limit("5/minute")  # IP-level rate limit
def login(credentials):
    if get_failed_attempts(credentials.email) > 5:
        raise HTTPException(429, "Account locked")  # Account-level lockout
    return authenticate(credentials)

┌──────────┐                          ┌──────────┐
│  Client  │                          │  Server  │
└────┬─────┘                          └────┬─────┘
     │  1. ClientHello (supported ciphers) │
     │─────────────────────────────────────►│
     │                                      │
     │  2. ServerHello + Certificate        │
     │◄─────────────────────────────────────│
     │                                      │
     │  3. Key Exchange (encrypted)         │
     │─────────────────────────────────────►│
     │                                      │
     │  4. Secure connection established    │
     │◄────────────────────────────────────►│
     │     (All traffic encrypted)          │
     └──────────────────────────────────────┘

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

# ❌ Bad: Allow all origins -- any website can make requests to your API.
# If combined with allow_credentials=True, attackers can use your users'
# session cookies to make authenticated requests from malicious sites.
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # Dangerous! "The whole internet can talk to me"
    allow_methods=["*"],
    allow_headers=["*"],
)

# ✅ Good: Explicitly list allowed origins -- only YOUR frontends can call your API.
# The browser enforces this: requests from unlisted origins are blocked.
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://myapp.com", "https://admin.myapp.com"],  # Whitelist
    allow_credentials=True,           # Allow cookies/auth headers
    allow_methods=["GET", "POST", "PUT", "DELETE"],  # Only needed methods
    allow_headers=["Authorization", "Content-Type"],  # Only needed headers
)

# HTTP Header to prevent XSS -- each directive controls a resource type.
Content-Security-Policy: 
    default-src 'self';                       # Default: only load from same origin
    script-src 'self' https://trusted-cdn.com;  # JS only from self and your CDN
    style-src 'self' 'unsafe-inline';         # CSS from self (unsafe-inline is a trade-off)
    img-src 'self' data: https:;              # Images from self, data URIs, and HTTPS
    connect-src 'self' https://api.myapp.com; # AJAX/WebSocket only to your API
    frame-ancestors 'none';                   # Cannot be embedded in iframes (clickjacking prevention)

from fastapi import FastAPI, Request
from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)
app = FastAPI()

@app.get("/api/search")
@limiter.limit("10/minute")  # 10 requests per minute per IP
async def search(request: Request, query: str):
    return {"results": perform_search(query)}

@app.post("/api/login")
@limiter.limit("5/minute")  # Stricter for auth endpoints
async def login(request: Request, credentials: Credentials):
    return authenticate(credentials)

from fastapi import Security, HTTPException
from fastapi.security import APIKeyHeader

api_key_header = APIKeyHeader(name="X-API-Key")

async def verify_api_key(api_key: str = Security(api_key_header)):
    # Hash the API key before comparing (store hashed in DB)
    hashed_key = hashlib.sha256(api_key.encode()).hexdigest()
    
    if not await db.verify_api_key(hashed_key):
        raise HTTPException(status_code=403, detail="Invalid API key")
    
    return api_key

@app.get("/api/data")
async def get_data(api_key: str = Security(verify_api_key)):
    return {"data": "protected"}

from fastapi import FastAPI
from fastapi.middleware import Middleware
from starlette.middleware.base import BaseHTTPMiddleware

class SecurityHeadersMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        response = await call_next(request)
        
        # Prevent clickjacking -- blocks embedding your site in iframes.
        # Attackers overlay invisible iframes to trick users into clicking.
        response.headers["X-Frame-Options"] = "DENY"
        
        # Prevent MIME sniffing -- stops browsers from interpreting files
        # as a different content type (e.g., treating a text file as JS).
        response.headers["X-Content-Type-Options"] = "nosniff"
        
        # Enable XSS filter -- legacy defense, mostly replaced by CSP,
        # but still worth including for older browsers.
        response.headers["X-XSS-Protection"] = "1; mode=block"
        
        # HSTS: force HTTPS for 1 year -- even if user types http://,
        # browser auto-upgrades to https://. Prevents protocol downgrade attacks.
        response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"
        
        # Control how much referrer information is sent with requests.
        # "strict-origin-when-cross-origin" sends the origin but not the path
        # when navigating to a different site (prevents URL-based data leaks).
        response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
        
        # Restrict access to browser APIs that your app does not need.
        # If your app does not use geolocation or microphone, block them
        # so that injected scripts cannot abuse these APIs.
        response.headers["Permissions-Policy"] = "geolocation=(), microphone=()"
        
        return response

import os
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    """Pydantic validates that all required secrets are present at startup.
    If DATABASE_URL is missing, the app crashes immediately with a clear
    error message instead of failing on the first database query."""
    database_url: str      # Required -- app will not start without this
    jwt_secret: str        # Required -- missing secret = broken auth
    api_key: str           # Required -- missing key = broken integrations
    
    class Config:
        env_file = ".env"
        env_file_encoding = "utf-8"

# CRITICAL: Never commit .env to version control!
# Add to .gitignore BEFORE creating the file. Check git history --
# if .env was ever committed, those secrets are compromised and must be rotated.

import hvac

client = hvac.Client(url='https://vault.example.com')
# Authenticate using Kubernetes service account -- no hardcoded tokens.
# Vault verifies the pod's identity with the Kubernetes API server.
client.auth.kubernetes.login(role='my-app', jwt=service_account_token)

# Read secrets -- Vault logs every access for audit compliance
secret = client.secrets.kv.v2.read_secret_version(path='myapp/config')
db_password = secret['data']['data']['password']

# Dynamic database credentials -- the gold standard for production.
# Vault generates a unique username/password for this specific pod,
# valid for (e.g.) 1 hour. When the lease expires, credentials are revoked.
# If a credential is compromised, the blast radius is one pod for one hour.
creds = client.secrets.database.generate_credentials(name='myapp-db')
# Credentials automatically expire and are never reused