from dotenv import load_dotenv
from descope import DescopeClient, DeliveryMethod
from fastapi import FastAPI, Request, Depends, HTTPException
from fastapi.responses import RedirectResponse
# ─── NEW: Google Sheets imports ─────────────────────────────────────────────
from google.oauth2.service_account import Credentials
from googleapiclient.discovery import build
# ─────────────────────────────────────────────────────────────────────────────
# ─── DESCOPE SETUP ───────────────────────────────────────────────────────────
descope_client = DescopeClient(project_id=os.getenv("DESCOPE_PROJECT_ID"))
# ─────────────────────────────────────────────────────────────────────────────
# ─── NEW: Google Sheets client setup ────────────────────────────────────────
SCOPES = ["https://www.googleapis.com/auth/spreadsheets.readonly"]
if os.getenv("GOOGLE_CREDS_JSON"):
creds_info = json.loads(os.getenv("GOOGLE_CREDS_JSON"))
creds = Credentials.from_service_account_info(creds_info, scopes=SCOPES)
creds = Credentials.from_service_account_file(
os.getenv("GOOGLE_CREDS_FILE"), scopes=SCOPES
sheets = build("sheets", "v4", credentials=creds).spreadsheets()
SHEET_ID = os.getenv("SHEET_ID")
def get_allowed_emails():
Reads column A (skipping header) from the sheet and returns
a set of normalized email addresses.
RANGE = "AllowedEmails!A2:A" # adjust if your tab/range differs
result = sheets.values().get(spreadsheetId=SHEET_ID, range=RANGE).execute()
rows = result.get("values", [])
return {row[0].strip().lower() for row in rows if row and row[0].strip()}
# ─────────────────────────────────────────────────────────────────────────────
def send_magic_link(email: str) -> str:
email_norm = email.strip().lower()
allowed = get_allowed_emails()
if email_norm not in allowed:
return "❌ This email is not on the allowed list."
masked = descope_client.magiclink.sign_up_or_in(
method=DeliveryMethod.EMAIL,
uri=f"{os.getenv('BASE_URL')}/verify-magic-link"
return f"✅ Magic link sent to {masked}"
# 2️⃣ Dependency to verify cookie
def get_current_user(request: Request):
token = request.cookies.get("session_token")
raise HTTPException(status_code=401, detail="Not authenticated")
info = descope_client.tokens.verify_session(token)
raise HTTPException(status_code=401, detail="Session expired")
gr.Markdown("## 🔐 Login with Magic Link")
email_box = gr.Textbox(label="Your email")
send_btn = gr.Button("Send Magic Link")
status_out = gr.Textbox(label="Status")
send_btn.click(send_magic_link, inputs=email_box, outputs=status_out)
app.mount("/", blocks.server_app)
# 4️⃣ Set cookie on verification
@app.get("/verify-magic-link")
async def verify_magic_link(t: str):
resp = descope_client.magiclink.verify(token=t)
session_jwt = resp["sessionToken"]["jwt"]
response = RedirectResponse(url="/")
return {"status": "error", "detail": str(e)}
# 5️⃣ Example protected endpoint
async def protected(user_email=Depends(get_current_user)):
return {"message": f"Hello {user_email}, this is protected!"}
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=int(os.getenv("PORT", 8000)))