""" Discourse SSO (Connect) integration. """ import base64 import hmac import hashlib import urllib.parse from fastapi import APIRouter, Depends, HTTPException, Request, status from fastapi.responses import RedirectResponse from sqlalchemy.orm import Session from app.database import get_db from app.models import User from app.api.v1.auth import get_current_user from app.config import settings router = APIRouter() @router.get("/discourse") async def discourse_sso( sso: str, sig: str, current_user: User = Depends(get_current_user), ): """ Handle Discourse SSO login request. 1. Validate the signature of the incoming payload. 2. Decode the payload (nonce). 3. Construct a new payload with user data. 4. Sign the new payload. 5. Redirect back to Discourse. """ if not settings.discourse_sso_secret: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Discourse SSO secret not configured" ) # 1. Validate Signature secret = settings.discourse_sso_secret.encode("utf-8") expected_sig = hmac.new(secret, sso.encode("utf-8"), hashlib.sha256).hexdigest() if sig != expected_sig: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid SSO signature" ) # 2. Decode Payload try: decoded_sso = base64.b64decode(sso).decode("utf-8") params = urllib.parse.parse_qs(decoded_sso) nonce = params.get("nonce", [None])[0] if not nonce: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Missing nonce in SSO payload" ) except Exception: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid SSO payload" ) # 3. Construct User Payload # Discourse expects: nonce, email, external_id, username, name, etc. user_params = { "nonce": nonce, "email": current_user.email, "external_id": current_user.id, "username": current_user.profile.display_name.replace(" ", "_") if current_user.profile and current_user.profile.display_name else f"user_{current_user.id[:8]}", "name": current_user.profile.display_name if current_user.profile else "", "require_activation": "false", } # 4. Encode and Sign encoded_params = urllib.parse.urlencode(user_params) base64_params = base64.b64encode(encoded_params.encode("utf-8")).decode("utf-8") new_sig = hmac.new(secret, base64_params.encode("utf-8"), hashlib.sha256).hexdigest() # 5. Redirect # Redirect to Discourse's return URL # Usually: {DISCOURSE_URL}/session/sso_login return_url = f"{settings.discourse_url}/session/sso_login" redirect_url = f"{return_url}?sso={base64_params}&sig={new_sig}" return RedirectResponse(url=redirect_url)