Security Design Patterns

"""Secure JWT validation middleware with proper error handling."""
import os
import jwt
from functools import wraps
from flask import request, jsonify, g

# RS256 is asymmetric: verify with the IdP's PUBLIC key, never a shared secret.
# Load the PEM from your secret manager and rotate it in step with the IdP.
JWT_PUBLIC_KEY = os.environ["JWT_PUBLIC_KEY"]
JWT_ALGORITHM = "RS256"
JWT_ISSUER = "https://auth.example.com"
JWT_AUDIENCE = "https://api.example.com"

# Prefer the IdP's JWKS so key rotation is automatic — resolve by the token's kid claim:
# jwks_client = jwt.PyJWKClient("https://auth.example.com/.well-known/jwks.json")
# signing_key = jwks_client.get_signing_key_from_jwt(token).key

def require_auth(required_scopes=None):
    """Decorator that validates JWT and checks scopes."""
    def decorator(f):
        @wraps(f)
        def wrapper(*args, **kwargs):
            token = _extract_bearer_token(request)
            if not token:
                return jsonify({"error": "missing_token"}), 401

            try:
                payload = jwt.decode(
                    token,
                    JWT_PUBLIC_KEY,
                    algorithms=[JWT_ALGORITHM],
                    issuer=JWT_ISSUER,
                    audience=JWT_AUDIENCE,
                    options={
                        "require": ["exp", "iat", "iss", "aud", "sub"],
                        "verify_exp": True,
                        "verify_iat": True,
                    }
                )
            except jwt.ExpiredSignatureError:
                return jsonify({"error": "token_expired"}), 401
            except jwt.InvalidTokenError:
                return jsonify({"error": "invalid_token"}), 401

            # Check required scopes
            if required_scopes:
                token_scopes = set(payload.get("scope", "").split())
                if not token_scopes.issuperset(required_scopes):
                    return jsonify({"error": "insufficient_scope"}), 403

            g.current_user = payload["sub"]
            g.token_scopes = payload.get("scope", "").split()
            return f(*args, **kwargs)
        return wrapper
    return decorator

def _extract_bearer_token(req):
    auth_header = req.headers.get("Authorization", "")
    if auth_header.startswith("Bearer "):
        return auth_header[7:]
    return None

# Usage:
@app.route("/api/v1/users/<user_id>", methods=["GET"])
@require_auth(required_scopes={"read:users"})
def get_user(user_id):
    # g.current_user is set by middleware
    if g.current_user != user_id and "admin" not in g.token_scopes:
        return jsonify({"error": "forbidden"}), 403
    return jsonify(get_user_data(user_id))

"""Retrieve secrets from HashiCorp Vault with auto-renewal."""
import hvac
import os
import psycopg2

class VaultSecrets:
    def __init__(self):
        self.client = hvac.Client(
            url=os.environ["VAULT_ADDR"],
            token=os.environ["VAULT_TOKEN"],  # Use AppRole in production
        )

    def get_database_creds(self, role="app-readonly"):
        """Get dynamic, short-lived database credentials."""
        response = self.client.secrets.database.generate_credentials(
            name=role,
            mount_point="database"
        )
        return {
            "username": response["data"]["username"],
            "password": response["data"]["password"],
            "ttl": response["lease_duration"],  # Auto-expires
            "lease_id": response["lease_id"],
        }

    def get_api_key(self, path="api-keys/stripe"):
        """Read a static secret from KV v2. The path is relative to the KV mount
        (which defaults to "secret"), so hvac resolves this to secret/data/..."""
        response = self.client.secrets.kv.v2.read_secret_version(
            path=path
        )
        return response["data"]["data"]["api_key"]

# Usage — credentials auto-rotate, no hardcoded secrets
vault = VaultSecrets()
db_creds = vault.get_database_creds()
conn = psycopg2.connect(
    host="db.internal",
    user=db_creds["username"],
    password=db_creds["password"],  # Short-lived: TTL is set by the Vault role
)