bvandeusen
e8a4ca915a
Checks Authorization: Bearer header first, hashes token, looks up in api_keys. Read-only keys get 403 on non-GET. Admin routes inaccessible to non-admin key owners. Session path unchanged. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
60 lines
2.2 KiB
Python
60 lines
2.2 KiB
Python
import functools
|
|
|
|
from quart import g, jsonify, request, session
|
|
|
|
from fabledassistant.services.auth import get_user_by_id
|
|
from fabledassistant.services.api_keys import lookup_key
|
|
|
|
|
|
def _check_auth(f, required_role: str | None = None):
|
|
@functools.wraps(f)
|
|
async def decorated(*args, **kwargs):
|
|
# --- Bearer token path ---
|
|
auth_header = request.headers.get("Authorization", "")
|
|
if auth_header.startswith("Bearer "):
|
|
raw_key = auth_header[len("Bearer "):]
|
|
api_key = await lookup_key(raw_key)
|
|
if api_key is None:
|
|
return jsonify({"error": "Invalid or revoked API key"}), 401
|
|
user = await get_user_by_id(api_key.user_id)
|
|
if not user:
|
|
return jsonify({"error": "User not found"}), 401
|
|
# Scope enforcement: read-only keys cannot mutate
|
|
if api_key.scope == "read" and request.method not in ("GET", "HEAD", "OPTIONS"):
|
|
return jsonify({"error": "Read-only key cannot perform write operations"}), 403
|
|
# Role check (admin_required routes)
|
|
if required_role and user.role != required_role:
|
|
return jsonify({"error": "Admin access required"}), 403
|
|
g.user = user
|
|
g.api_key = api_key
|
|
return await f(*args, **kwargs)
|
|
|
|
# --- Session path (unchanged) ---
|
|
user_id = session.get("user_id")
|
|
if not user_id:
|
|
return jsonify({"error": "Authentication required"}), 401
|
|
user = await get_user_by_id(user_id)
|
|
if not user:
|
|
session.clear()
|
|
return jsonify({"error": "Authentication required"}), 401
|
|
if session.get("session_version") != user.session_version:
|
|
session.clear()
|
|
return jsonify({"error": "Session expired. Please log in again."}), 401
|
|
if required_role and user.role != required_role:
|
|
return jsonify({"error": "Admin access required"}), 403
|
|
g.user = user
|
|
return await f(*args, **kwargs)
|
|
return decorated
|
|
|
|
|
|
def login_required(f):
|
|
return _check_auth(f)
|
|
|
|
|
|
def admin_required(f):
|
|
return _check_auth(f, required_role="admin")
|
|
|
|
|
|
def get_current_user_id() -> int:
|
|
return g.user.id
|