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})