File size: 1,995 Bytes
015dbc8 a772ebc 015dbc8 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 |
import os
from typing import Any, Dict, Optional, Tuple
import requests
from flask import Blueprint, jsonify, request
import db
auth_blueprint = Blueprint("auth", __name__)
SUPABASE_URL = os.getenv("SUPABASE_URL")
SUPABASE_ANON_KEY = os.getenv("SUPABASE_ANON_KEY")
SUPABASE_SERVICE_ROLE_KEY = os.getenv("SUPABASE_SERVICE_ROLE_KEY")
if not SUPABASE_URL or not SUPABASE_ANON_KEY or not SUPABASE_SERVICE_ROLE_KEY:
raise RuntimeError("Supabase credentials (URL, ANON key, service role key) must be configured.")
AUTH_BASE = f"{SUPABASE_URL}/auth/v1"
def _extract_token_from_header() -> Optional[str]:
auth_header = request.headers.get("Authorization", "")
if not auth_header.startswith("Bearer "):
return None
return auth_header.split(" ", maxsplit=1)[1].strip()
def _fetch_supabase_user(accessToken: str) -> Tuple[Optional[Dict[str, Any]], Optional[Tuple[str, int]]]:
"""Validate the Supabase access token and return the user payload."""
response = requests.get(
f"{AUTH_BASE}/user",
headers={
"Authorization": f"Bearer {accessToken}",
"apikey": SUPABASE_ANON_KEY,
},
timeout=10,
)
if response.status_code != 200:
return None, ("Invalid Supabase access token.", 401)
return response.json(), None
def verify_request_token() -> Tuple[Optional[Dict[str, Any]], Optional[Tuple[str, int]]]:
token = _extract_token_from_header()
if not token:
return None, ("Authorization header missing or invalid.", 401)
return _fetch_supabase_user(token)
@auth_blueprint.route("/session", methods=["GET"])
def session_info():
"""Return the Supabase-authenticated user profile."""
user_payload, error = verify_request_token()
if error:
message, status = error
return jsonify({"error": message}), status
user_id = user_payload.get("id")
profile = db.get_user_profile(user_id)
return jsonify({"user": user_payload, "profile": profile})
|