Security Fundamentals for Software Engineers

# VULNERABLE — never do this
def get_user(username: str):
    query = f"SELECT * FROM users WHERE username = '{username}'"
    return db.execute(query)

# Input: ' OR '1'='1  → returns ALL users

# SAFE — parameterised queries
def get_user(username: str):
    return db.execute(
        "SELECT * FROM users WHERE username = :username",
        {"username": username}
    )

# SQLAlchemy ORM — safe by default
def get_user(username: str):
    return db.query(User).filter(User.username == username).first()

# Also vulnerable: LIKE patterns, ORDER BY, table names
# For dynamic ORDER BY, use an allowlist:
ALLOWED_COLUMNS = {"name", "created_at", "price"}
def list_products(sort_by: str):
    if sort_by not in ALLOWED_COLUMNS:
        raise ValueError(f"Invalid sort column: {sort_by}")
    return db.execute(f"SELECT * FROM products ORDER BY {sort_by}")

# VULNERABLE — rendering user input in HTML
@app.get("/search")
def search(q: str):
    return HTMLResponse(f"<h1>Results for: {q}</h1>")
# Input: <script>document.cookie</script>

# SAFE — escape output (Jinja2 auto-escapes by default)
@app.get("/search")
def search(q: str):
    return templates.TemplateResponse("search.html", {"request": request, "query": q})
# Template: <h1>Results for: {{ query }}</h1>  ← auto-escaped

# Content Security Policy header — defence in depth
@app.middleware("http")
async def add_security_headers(request, call_next):
    response = await call_next(request)
    response.headers["Content-Security-Policy"] = (
        "default-src 'self'; script-src 'self'; object-src 'none'"
    )
    response.headers["X-Content-Type-Options"] = "nosniff"
    response.headers["X-Frame-Options"] = "DENY"
    return response

# Cross-Site Request Forgery — tricks a logged-in user's browser into making a request

# FastAPI with CSRF middleware
from fastapi_csrf_protect import CsrfProtect
from fastapi_csrf_protect.exceptions import CsrfProtectError

@CsrfProtect.load_config
def get_csrf_config():
    return [("secret_key", settings.SECRET_KEY)]

@app.post("/api/transfer")
async def transfer_funds(request: Request, csrf_protect: CsrfProtect = Depends()):
    await csrf_protect.validate_csrf(request)
    # proceed with transfer

# For SPAs using tokens: use SameSite=Strict cookies + custom header
# Browser won't send SameSite=Strict cookies cross-origin
# Custom header (X-Requested-With) can't be set by cross-origin forms

# NEVER: hardcoded secrets in code
DATABASE_URL = "postgresql://admin:mysecretpassword@prod-db/myapp"

# NEVER: secrets in environment variables baked into Docker images
# (visible in docker inspect, process list, logs)

# CORRECT: load at runtime from secrets manager
import boto3

def get_secret(secret_name: str) -> str:
    client = boto3.client("secretsmanager")
    response = client.get_secret_value(SecretId=secret_name)
    return response["SecretString"]

DATABASE_URL = get_secret("myapp/prod/database-url")

# Or: use AWS SSM Parameter Store
ssm = boto3.client("ssm")
param = ssm.get_parameter(Name="/myapp/prod/db-password", WithDecryption=True)
db_password = param["Parameter"]["Value"]

from passlib.context import CryptContext

# bcrypt with a work factor of 12 (2^12 iterations)
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto", bcrypt__rounds=12)

def hash_password(plain: str) -> str:
    return pwd_context.hash(plain)

def verify_password(plain: str, hashed: str) -> bool:
    return pwd_context.verify(plain, hashed)

# NEVER use: MD5, SHA1, SHA256 for passwords — they're fast, so brute-forceable
# ALWAYS use: bcrypt, argon2, or scrypt — deliberately slow

# Enforce HTTPS — reject non-TLS in production
from starlette.middleware.httpsredirect import HTTPSRedirectMiddleware
app.add_middleware(HTTPSRedirectMiddleware)

# HTTP Strict Transport Security
response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"

# Verify TLS in outbound requests — never disable verification
import httpx
# BAD: httpx.get(url, verify=False)  # disables cert verification
# GOOD:
httpx.get(url, verify=True)  # default, verify server cert
httpx.get(url, verify="/path/to/ca-bundle.pem")  # custom CA

from pydantic import BaseModel, validator, constr, conint

class CreateProductRequest(BaseModel):
    name: constr(min_length=1, max_length=200, strip_whitespace=True)
    price: float
    quantity: conint(ge=0, le=10_000)
    category: str

    @validator("price")
    def price_must_be_positive(cls, v):
        if v <= 0:
            raise ValueError("Price must be positive")
        return round(v, 2)

    @validator("category")
    def category_must_be_allowed(cls, v):
        allowed = {"electronics", "clothing", "books", "food"}
        if v not in allowed:
            raise ValueError(f"Category must be one of: {allowed}")
        return v

# Validate at the boundary — HTTP handler, not deep in business logic
@app.post("/api/products")
def create_product(body: CreateProductRequest):  # Pydantic validates on entry
    ...