ryan
bc3e3d2797
- Add transitional @jwt_access_required (accepts access tokens, and for now also refresh tokens) and use it on every protected route and inside admin_required. /api/user/refresh stays refresh-only. - Move pending OIDC state/PKCE verifiers from an in-process dict to an oidc_states table (10 min TTL, single-use) so logins survive a blue/green flip. Drop the unused redirect_after_login. - Unit tests for the decorator and the state store. Deploy this commit on its own, before the access-token frontend: the old backend answers access tokens with 422, so the new frontend only works once this backend is live on both colors. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
64 lines
2.1 KiB
Python
64 lines
2.1 KiB
Python
"""
|
|
Authentication decorators for role-based access control.
|
|
"""
|
|
|
|
from functools import wraps
|
|
from quart import jsonify
|
|
from quart_jwt_extended import get_jwt_identity
|
|
from quart_jwt_extended.exceptions import WrongTokenError
|
|
from quart_jwt_extended.view_decorators import (
|
|
verify_jwt_in_request,
|
|
verify_jwt_refresh_token_in_request,
|
|
)
|
|
from .models import User
|
|
|
|
|
|
def jwt_access_required(fn):
|
|
"""
|
|
Protect an endpoint with the (1h) access token.
|
|
|
|
TRANSITIONAL: this also accepts the refresh token, so an old frontend
|
|
(which sends the refresh token on every call) keeps working against this
|
|
backend, including tabs that stay open across deploys.
|
|
|
|
It does NOT make a new frontend work against an older backend: routes
|
|
still on `@jwt_refresh_token_required` answer an access token with 422,
|
|
which the frontend does not retry. So this backend must be live on both
|
|
colors before the access-token frontend ships (deploy in two releases).
|
|
|
|
TODO: once the access-token frontend has been live for a while, replace
|
|
the body with `await verify_jwt_in_request()` (i.e. plain `@jwt_required`)
|
|
and map WrongTokenError to 401, so a stale tab still sending the refresh
|
|
token refreshes and retries instead of failing with 422.
|
|
"""
|
|
|
|
@wraps(fn)
|
|
async def wrapper(*args, **kwargs):
|
|
try:
|
|
await verify_jwt_in_request()
|
|
except WrongTokenError:
|
|
# Legacy client sent the refresh token. Expired/invalid tokens
|
|
# still raise from verify_jwt_in_request and are not retried here.
|
|
await verify_jwt_refresh_token_in_request()
|
|
return await fn(*args, **kwargs)
|
|
|
|
return wrapper
|
|
|
|
|
|
def admin_required(fn):
|
|
"""
|
|
Decorator that requires the user to be an admin (member of lldap_admin group).
|
|
Must be used on async route handlers.
|
|
"""
|
|
|
|
@wraps(fn)
|
|
@jwt_access_required
|
|
async def wrapper(*args, **kwargs):
|
|
user_id = get_jwt_identity()
|
|
user = await User.get_or_none(id=user_id)
|
|
if not user or not user.is_admin():
|
|
return jsonify({"error": "Admin access required"}), 403
|
|
return await fn(*args, **kwargs)
|
|
|
|
return wrapper
|