Python Security Patterns
Secrets Management
from pydantic_settings import BaseSettings
from pydantic import SecretStr
import os
# Good: environment variables, never hardcoded
class Settings(BaseSettings):
database_url: str # from DATABASE_URL env var
jwt_secret: SecretStr # hidden from logs/repr
api_key: SecretStr
# Bad: hardcoded
DATABASE_URL = "postgresql://admin:password@prod.db.internal/app"
# Access secret value only when needed
def get_token_hash(settings: Settings) -> bytes:
return hashlib.sha256(settings.jwt_secret.get_secret_value().encode()).digest()
Password Hashing
from passlib.context import CryptContext
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
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 store or log plain passwords
# Never use MD5/SHA1 for passwords — use bcrypt/argon2/scrypt
SQL Injection Prevention
from sqlalchemy import text
# Good: parameterized queries
async def get_user_by_email(session: AsyncSession, email: str) -> User | None:
result = await session.execute(
select(User).where(User.email == email) # ORM handles parameterization
)
return result.scalar_one_or_none()
# If using raw SQL, always use parameters
async def raw_query(session: AsyncSession, email: str) -> None:
await session.execute(
text("SELECT * FROM users WHERE email = :email"),
{"email": email}
)
# Bad: string interpolation
await session.execute(f"SELECT * FROM users WHERE email = '{email}'") # INJECTION
JWT Authentication
from datetime import datetime, timedelta, timezone
import jwt # PyJWT
SECRET_KEY = settings.jwt_secret.get_secret_value()
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
def create_access_token(subject: str, expires_delta: timedelta | None = None) -> str:
expire = datetime.now(timezone.utc) + (expires_delta or timedelta(minutes=15))
return jwt.encode({"sub": subject, "exp": expire}, SECRET_KEY, algorithm=ALGORITHM)
def decode_token(token: str) -> dict:
return jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
# Raises jwt.ExpiredSignatureError, jwt.InvalidTokenError on invalid tokens
Input Validation
import bleach # HTML sanitization
def sanitize_html(html: str) -> str:
"""Allow only safe HTML tags."""
allowed_tags = ["b", "i", "em", "strong", "a", "p", "br"]
allowed_attributes = {"a": ["href", "rel"]}
return bleach.clean(html, tags=allowed_tags, attributes=allowed_attributes)
# Path traversal prevention
from pathlib import Path
def safe_file_path(base_dir: Path, user_path: str) -> Path:
safe = (base_dir / user_path).resolve()
if not str(safe).startswith(str(base_dir.resolve())):
raise ValueError("Path traversal attempt detected")
return safe
Rate Limiting
from slowapi import Limiter
from slowapi.util import get_remote_address
limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
@router.post("/auth/login")
@limiter.limit("5/minute")
async def login(request: Request, body: LoginRequest) -> TokenResponse:
...
Security Checklist