Security Fundamentals

import bcrypt

# Storing passwords
def hash_password(password: str) -> bytes:
    salt = bcrypt.gensalt(rounds=12)
    return bcrypt.hashpw(password.encode(), salt)

# Verifying passwords
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
def create_token(user_id: str, secret: str) -> str:
    payload = {
        "sub": user_id,
        "iat": datetime.utcnow(),
        "exp": datetime.utcnow() + timedelta(hours=24)
    }
    return jwt.encode(payload, secret, algorithm="HS256")

# Verifying JWT
def verify_token(token: str, secret: str) -> dict:
    try:
        return jwt.decode(token, secret, algorithms=["HS256"])
    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
@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: Check authorization
@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: Weak hashing
import hashlib
password_hash = hashlib.md5(password.encode()).hexdigest()

# ✅ Good: Proper password hashing
import bcrypt
password_hash = bcrypt.hashpw(password.encode(), bcrypt.gensalt(12))

# ❌ SQL Injection vulnerable
query = f"SELECT * FROM users WHERE id = {user_id}"

# ✅ Parameterized query
query = "SELECT * FROM users WHERE id = ?"
cursor.execute(query, (user_id,))

# ❌ Bad: Security question for password reset
def reset_password(email, mothers_maiden_name):
    # Easily guessable/discoverable answers
    pass

# ✅ Good: Token-based password reset
def reset_password(email):
    token = generate_secure_token()
    send_reset_email(email, token)
    store_token_with_expiry(email, token, expiry=30_minutes)

# ❌ Bad: Debug mode in production
app.run(debug=True)

# ❌ Bad: Default credentials
DATABASE_PASSWORD = "admin123"

# ✅ Good: Secure configuration
app.run(debug=os.getenv("FLASK_DEBUG", False))
DATABASE_PASSWORD = os.getenv("DATABASE_PASSWORD")

# Regularly audit dependencies
npm audit
pip-audit
snyk test

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

# ❌ Bad: No rate limiting
@app.post("/login")
def login(credentials):
    return authenticate(credentials)

# ✅ Good: Rate limiting + account lockout
from slowapi import Limiter

limiter = Limiter(key_func=get_remote_address)

@app.post("/login")
@limiter.limit("5/minute")
def login(credentials):
    if get_failed_attempts(credentials.email) > 5:
        raise HTTPException(429, "Account locked")
    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
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # Dangerous!
    allow_methods=["*"],
    allow_headers=["*"],
)

# ✅ Good: Specific origins
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://myapp.com", "https://admin.myapp.com"],
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE"],
    allow_headers=["Authorization", "Content-Type"],
)

# HTTP Header to prevent XSS
Content-Security-Policy: 
    default-src 'self';
    script-src 'self' https://trusted-cdn.com;
    style-src 'self' 'unsafe-inline';
    img-src 'self' data: https:;
    connect-src 'self' https://api.myapp.com;
    frame-ancestors 'none';

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
        response.headers["X-Frame-Options"] = "DENY"
        
        # Prevent MIME sniffing
        response.headers["X-Content-Type-Options"] = "nosniff"
        
        # Enable XSS filter
        response.headers["X-XSS-Protection"] = "1; mode=block"
        
        # Strict Transport Security (force HTTPS)
        response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"
        
        # Referrer policy
        response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
        
        # Permissions policy
        response.headers["Permissions-Policy"] = "geolocation=(), microphone=()"
        
        return response

import os
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    database_url: str
    jwt_secret: str
    api_key: str
    
    class Config:
        env_file = ".env"
        env_file_encoding = "utf-8"

# Never commit .env to version control!
# .gitignore should include .env

import hvac

client = hvac.Client(url='https://vault.example.com')
client.auth.kubernetes.login(role='my-app', jwt=service_account_token)

# Read secrets
secret = client.secrets.kv.v2.read_secret_version(path='myapp/config')
db_password = secret['data']['data']['password']

# Dynamic database credentials
creds = client.secrets.database.generate_credentials(name='myapp-db')
# Credentials auto-expire and rotate