feat: sso profiles

Author: martastain

backend/api/auth/sso.py

@@ -1,6 +1,5 @@
 from urllib.parse import urlparse
 
-from authlib.integrations.starlette_client import OAuthError
 from fastapi import Request
 from fastapi.responses import RedirectResponse
 
@@ -39,14 +38,34 @@ class SSOLoginCallback(APIRequest):
     methods = ["GET"]
 
     async def handle(self, request: Request, provider: str) -> RedirectResponse:
-        client = NebulaSSO.client(provider)
+        remote = NebulaSSO.client(provider)
+        if not remote:
+            return RedirectResponse("/?error=Invalid provider")
 
-        try:
-            token = await client.authorize_access_token(request)
-        except OAuthError as error:
-            return RedirectResponse(f"/?error={error}")
-        user = token.get("userinfo", {})
-        email = user.get("email")
+        code = request.query_params.get("code")
+        id_token = request.query_params.get("id_token")
+        oauth_verifier = request.query_params.get("oauth_verifier")
+
+        user_info = {}
+
+        if code:
+            token = await remote.authorize_access_token(request)
+            user_info = token.get("userinfo", {})
+
+        if id_token and not user_info:
+            token = {"id_token": id_token}
+            user_info = await remote.parse_id_token(request, token)
+
+        if oauth_verifier and not user_info:
+            token = await remote.authorize_access_token(request)
+
+        if token and not user_info:
+            user_info = await remote.userinfo(token=token)
+
+        if not user_info:
+            return RedirectResponse("/?error=Invalid response from provider")
+
+        email = user_info.get("email")
 
         if not email:
             return RedirectResponse("/?error=User email not found")

backend/nebula/settings/models.py

@@ -94,11 +94,55 @@ class BaseSystemSettings(SettingsModel):
 
 
 class SSOProvider(SettingsModel):
-    name: str
-    title: str | None = None
-    entrypoint: str
-    client_id: str
-    client_secret: str
+    name: Annotated[
+        str,
+        Field(
+            title="Name",
+            example="myoauth",
+        ),
+    ]
+
+    title: Annotated[
+        str,
+        Field(
+            title="Title",
+            description="Used on the SSO button on the login page",
+            example="Log in using MyOauth",
+        ),
+    ]
+
+    profile: Annotated[
+        Literal["google", "github"] | None,
+        Field(
+            title="Profile",
+            description="Configuration preset if entrypoint is not provided",
+        ),
+    ] = None
+
+    entrypoint: Annotated[
+        str | None,
+        Field(
+            title="Entrypoint",
+            description="URL to the SSO provider configuration endpoint",
+            example="https://iam.example.com/realms/nebula/.well-known/openid-configuration",
+        ),
+    ] = None
+
+    client_id: Annotated[
+        str,
+        Field(
+            title="Client ID",
+            example="myclientid",
+        ),
+    ]
+
+    client_secret: Annotated[
+        str,
+        Field(
+            title="Client secret",
+            example="myclientsecret",
+        ),
+    ]
 
 
 class SystemSettings(BaseSystemSettings):

backend/server/sso.py

@@ -1,9 +1,26 @@
+from typing import Any
+
 from authlib.integrations.httpx_client import AsyncOAuth2Client
 from authlib.integrations.starlette_client import OAuth
 
 import nebula
 from server.models import ResponseModel
 
+PROFILES = {
+    "github": {
+        "api_base_url": "https://api.github.com/",
+        "access_token_url": "https://github.com/login/oauth/access_token",
+        "authorize_url": "https://github.com/login/oauth/authorize",
+        "client_kwargs": {"scope": "user:email"},
+        "userinfo_endpoint": "https://api.github.com/user",
+    },
+    "google": {
+        "api_base_url": "https://www.googleapis.com/",
+        "server_metadata_url": "https://accounts.google.com/.well-known/openid-configuration",
+        "client_kwargs": {"scope": "openid email profile"},
+    },
+}
+
 
 class SSOOption(ResponseModel):
     name: str
@@ -37,11 +54,15 @@ def create_oauth(cls) -> OAuth:
         cls._oauth = OAuth(ssoconfig)
         assert cls._oauth is not None, "OAuth is not initialized"
         for provider in nebula.settings.system.sso_providers:
-            cls._oauth.register(
-                name=provider.name,
-                server_metadata_url=provider.entrypoint,
-                client_kwargs={"scope": "openid email profile"},
-            )
+            kw: dict[str, Any]
+            if provider.profile:
+                kw = PROFILES[provider.profile]
+            else:
+                kw = {
+                    "server_metadata_url": provider.entrypoint,
+                    "client_kwargs": {"scope": "openid email profile"},
+                }
+            cls._oauth.register(name=provider.name, **kw)
         return cls._oauth
 
     @classmethod