Security Design Patterns

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

# NEVER hardcode — load from environment / vault
JWT_SECRET = os.environ["JWT_SECRET"]
JWT_ALGORITHM = "RS256"  # Use RS256 (asymmetric) in production
JWT_ISSUER = "https://auth.example.com"
JWT_AUDIENCE = "https://api.example.com"

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_SECRET,
                    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))

package api.authz

import rego.v1

# Default deny
default allow := false

# Admins can do anything
allow if {
    input.user.role == "admin"
}

# Users can read their own data
allow if {
    input.method == "GET"
    input.path = ["api", "v1", "users", user_id]
    input.user.id == user_id
}

# Editors can write to resources in their department
allow if {
    input.method in {"PUT", "POST", "PATCH"}
    input.user.role == "editor"
    input.resource.department == input.user.department
}

# Time-based access: only during business hours
allow if {
    input.user.role == "contractor"
    hour := time.clock(time.now_ns())[0]
    hour >= 9
    hour < 17
}

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

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="secret/data/api-keys/stripe"):
        """Read a static secret from KV v2."""
        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"],  # Expires in 1 hour
)