Flask API

try:
    claims = await api_client.verify_access_token(
        access_token=token,
        required_claims=["email_verified", "org_id"]
    )
except BaseAuthError as e:
    return jsonify({"error": "Missing required claims"}), 401

import asyncio


# Configure API client with DPoP enabled (Allowed mode)
api_client = ApiClient(ApiClientOptions(
    domain=os.getenv("AUTH0_DOMAIN"),
    audience=os.getenv("AUTH0_AUDIENCE"),
    dpop_enabled=True,   # Default - enables DPoP support
    dpop_required=False  # Default - allows both Bearer and DPoP tokens
))

# For Required mode (DPoP-only, rejects Bearer tokens)
# dpop_required=True

# DPoP-aware decorator
def require_auth_dpop(f):
    @wraps(f)
    def decorated_function(*args, **kwargs):
        headers = {
            "authorization": request.headers.get("Authorization", ""),
            "dpop": request.headers.get("DPoP", "")  # DPoP proof header
        }
        
        try:
            # verify_request() automatically detects Bearer or DPoP scheme
            claims = asyncio.run(api_client.verify_request(
                headers=headers,
                http_method=request.method,
                http_url=request.url
            ))
            g.user_claims = claims
            return f(*args, **kwargs)
        except BaseAuthError as e:
            return jsonify({"error": str(e)}), e.get_status_code(), e.get_headers()
    
    return decorated_function

@app.route("/api/dpop-protected")
@require_auth_dpop
def dpop_protected():
    return jsonify({
        "message": "Successfully authenticated with DPoP or Bearer",
        "user": g.user_claims.get("sub")
    })

def require_scope(required_scope):
    def decorator(f):
        @wraps(f)
        async def wrapper(*args, **kwargs):
            if not hasattr(g, "user_claims"):
                return jsonify({"error": "Unauthorized"}), 401
            
            scopes = g.user_claims.get("scope", "").split()
            
            if required_scope not in scopes:
                return jsonify({"error": "Insufficient permissions"}), 403
            
            return await f(*args, **kwargs)
        return wrapper
    return decorator


@app.route("/api/admin")
@require_auth
@require_scope("admin:write")
async def admin_endpoint():
    return jsonify({"message": "Admin access granted"})

from auth0_api_python.errors import (
    BaseAuthError,
    VerifyAccessTokenError,
    GetTokenByExchangeProfileError,
    ApiError
)

@app.errorhandler(BaseAuthError)
def handle_auth_error(error):
    """Handle all Auth0 authentication errors"""
    return jsonify({
        "error": error.get_error_code(),
        "error_description": str(error)
    }), error.get_status_code(), error.get_headers()


@app.errorhandler(VerifyAccessTokenError)
def handle_token_verification_error(error):
    """Handle token verification failures specifically"""
    app.logger.warning(f"Token verification failed: {str(error)}")
    return jsonify({
        "error": "invalid_token",
        "error_description": str(error)
    }), 401, error.get_headers()


@app.errorhandler(ApiError)
def handle_api_error(error):
    """Handle Auth0 API errors (network, rate limits, etc.)"""
    app.logger.error(f"Auth0 API error: {error.code} - {error.message}")
    return jsonify({
        "error": error.code,
        "error_description": error.message
    }), error.status_code or 500


@app.errorhandler(Exception)
def handle_generic_error(error):
    """Catch-all for unexpected errors"""
    app.logger.error(f"Unexpected error: {str(error)}", exc_info=True)
    return jsonify({"error": "Internal server error"}), 500

from flask import g

PUBLIC_ROUTES = ["/api/public", "/health"]

@app.before_request
async def verify_token():
    # Skip auth for public routes
    if request.path in PUBLIC_ROUTES:
        return
    
    auth_header = request.headers.get("Authorization", "")
    
    if not auth_header.startswith("Bearer "):
        return jsonify({"error": "Missing authorization"}), 401
    
    token = auth_header.split(" ")[1]
    
    try:
        claims = await api_client.verify_access_token(token)
        g.user_claims = claims
    except BaseAuthError as e:
        return jsonify({"error": str(e)}), e.get_status_code(), e.get_headers()


@app.route("/api/protected-data")
async def protected_data():
    # Claims automatically available via g.user_claims
    return jsonify({"user_id": g.user_claims["sub"]})