Basic support for connected accounts/mrrt

Author: sam-muncke

src/auth0_fastapi/auth/auth_client.py

@@ -1,7 +1,7 @@
 
 # Imported from auth0-server-python
 from auth0_server_python.auth_server.server_client import ServerClient
-from auth0_server_python.auth_types import LogoutOptions, StartInteractiveLoginOptions
+from auth0_server_python.auth_types import LogoutOptions, StartInteractiveLoginOptions, ConnectAccountOptions
 from fastapi import HTTPException, Request, Response, status
 
 from auth0_fastapi.config import Auth0Config
@@ -45,6 +45,7 @@ def __init__(
             transaction_store=transaction_store,
             state_store=state_store,
             pushed_authorization_requests=config.pushed_authorization_requests,
+            use_mrrt=config.use_mrrt,
             authorization_params={
                 "audience": config.audience,
                 "redirect_uri": redirect_uri,
@@ -82,6 +83,36 @@ async def complete_login(
         """
         return await self.client.complete_interactive_login(callback_url, store_options=store_options)
 
+    async def start_connect_account(
+        self,
+        connection: str,
+        authorization_params: dict = None,
+        store_options: dict = None,
+    ) -> str:
+        """
+        Initiates the connected account process.
+        Optionally, an app_state dictionary can be passed to persist additional state.
+        Returns the authorization URL to redirect the user.
+        """
+        options = ConnectAccountOptions(
+            connection=connection,
+            authorization_params= authorization_params
+        )
+        return await self.client.start_connect_account(options=options, store_options=store_options)
+
+    async def complete_connect_account(
+        self,
+        connect_code: str,
+        state: str,
+        store_options: dict = None,
+    ) -> str:
+        """
+        Initiates the interactive login process.
+        Optionally, an app_state dictionary can be passed to persist additional state.
+        Returns the authorization URL to redirect the user.
+        """
+        return await self.client.complete_connect_account(connect_code=connect_code, state=state, store_options=store_options)
+
     async def logout(
         self,
         return_to: str = None,

src/auth0_fastapi/config.py

@@ -15,9 +15,11 @@ class Auth0Config(BaseModel):
     audience: Optional[str] = Field(None, description="Target audience for tokens (if applicable)")
     authorization_params: Optional[dict[str, Any]] = Field(None, description="Additional parameters to include in the authorization request")
     pushed_authorization_requests: bool = Field(False, description="Whether to use pushed authorization requests")
+    use_mrrt: bool = Field(False, description="Whether to use Multi-Resource Refresh Tokens (MRRT)")
     # Route-mounting flags with desired defaults
     mount_routes: bool = Field(True, description="Controls /auth/* routes: login, logout, callback, backchannel-logout")
     mount_connect_routes: bool = Field(False, description="Controls /auth/connect routes (account-linking)")
+    mount_connected_account_routes: bool = Field(False, description="Controls /auth/connect-account routes (for connected accounts)")
     #Cookie Settings
     cookie_name: str = Field("_a0_session", description="Name of the cookie storing session data")
     session_expiration: int = Field(259200, description="Session expiration time in seconds (default: 3 days)")

src/auth0_fastapi/server/routes.py

@@ -58,10 +58,20 @@ async def callback(
         ):
             """
             Endpoint to handle the callback after Auth0 authentication.
-            Processes the callback URL and completes the login flow.
+            Processes the callback URL and completes the login or connected account flow.
             Redirects the user to a post-login URL based on appState or a default.
             """
+            connect_code = request.query_params.get("connect_code")
+            if connect_code and config.mount_connected_account_routes:
+                state = request.query_params.get("state")
+                return await auth_client.complete_connect_account(
+                    connect_code=connect_code,
+                    state=state,
+                    store_options={"request": request, "response": response},
+                )
+
             full_callback_url = str(request.url)
+
             try:
                 session_data = await auth_client.complete_login(
                     full_callback_url,
@@ -123,7 +133,27 @@ async def backchannel_logout(
                 raise HTTPException(status_code=400, detail=str(e))
             return Response(status_code=204)
 
+    if config.mount_connected_account_routes:
+        @router.get("/auth/connect-account")
+        async def connect_account(
+            request: Request,
+            response: Response,
+            connection: str = Query(),
+            auth_client: AuthClient = Depends(get_auth_client),
+        ):
+            """
+            Endpoint to initiate the connect account flow for linking a third-party account to the user's profile.
+            Redirects the user to the Auth0 connect account URL.
+            """
+            authorization_params = {k: v for k, v in request.query_params.items() if k not in [
+                "connection"]}
+            connect_account_url = await auth_client.start_connect_account(
+                connection=connection,
+                authorization_params=authorization_params,
+                store_options={"request": request, "response": response},
+            )
 
+            return RedirectResponse(url=connect_account_url, headers=response.headers)
 
     if config.mount_connect_routes: