feat(ciba): support RAR requests

Author: cristiandouce

packages/auth0-ai/auth0_ai/authorizers/ciba/ciba_authorizer_base.py

@@ -91,18 +91,18 @@ def __init__(self, params: CIBAAuthorizerParams[ToolInput], auth0: Auth0ClientPa
         self.credentials_store = SubStore[TokenResponse](ciba_store, {
             "get_ttl": lambda credential: credential["expires_in"] * 1000 if "expires_in" in credential else None
         })
-    
+
     def _handle_authorization_interrupts(self, err: Union[AuthorizationPendingInterrupt, AuthorizationPollingInterrupt]) -> None:
         raise err
-    
+
     def _get_instance_id(self, authorize_params) -> str:
         props = {
             "auth0": omit(self.auth0, ["client_secret", "client_assertion_signing_key"]),
             "params": authorize_params
         }
         sh = json.dumps(props, sort_keys=True, separators=(",", ":"))
         return hashlib.md5(sh.encode("utf-8")).hexdigest()
-    
+
     async def _get_authorize_params(self, *args: ToolInput.args, **kwargs: ToolInput.kwargs) -> Dict[str, Any]:
         authorize_params = {
             "scope": _ensure_openid_scope(self.params.get("scope")),
@@ -129,11 +129,18 @@ async def _get_authorize_params(self, *args: ToolInput.args, **kwargs: ToolInput
         else:
             authorize_params["binding_message"] = self.params.get("binding_message")(*args, **kwargs)
 
+        if isinstance(self.params.get("authorization_details"), list):
+            authorize_params["authorization_details"] = self.params.get("authorization_details")
+        elif inspect.iscoroutinefunction(self.params.get("authorization_details")):
+            authorize_params["authorization_details"] = await self.params.get("authorization_details")(*args, **kwargs)
+        else:
+            authorize_params["authorization_details"] = self.params.get("authorization_details")(*args, **kwargs)
+
         return authorize_params
 
     async def _start(self, authorize_params) -> CIBAAuthorizationRequest:
         requested_at = time.time()
-        
+
         try:
             response = self.back_channel_login.back_channel_login(**authorize_params)
             return CIBAAuthorizationRequest(
@@ -189,7 +196,7 @@ def _get_credentials_internal(self, auth_request: CIBAAuthorizationRequest) -> T
 
     def _get_credentials(self, auth_request: CIBAAuthorizationRequest) -> TokenResponse | None:
         return self._get_credentials_internal(auth_request)
-    
+
     async def get_credentials_polling(self, auth_request: CIBAAuthorizationRequest) -> TokenResponse | None:
         credentials: TokenResponse | None = None
 
@@ -200,14 +207,14 @@ async def get_credentials_polling(self, auth_request: CIBAAuthorizationRequest)
                 await asyncio.sleep(err.request["interval"])
             except Exception:
                 raise
-        
+
         return credentials
-    
+
     async def delete_auth_request(self):
         local_store = _get_local_storage()
         auth_request_ns = local_store["auth_request_ns"]
         await self.auth_request_store.delete(auth_request_ns, "auth_request")
-    
+
     def protect(
         self,
         get_context: ContextGetter[ToolInput],
@@ -237,28 +244,28 @@ async def wrapped_execute(*args: ToolInput.args, **kwargs: ToolInput.kwargs):
                                 # initial request
                                 auth_request = await self._start(authorize_params)
                                 await self.auth_request_store.put(auth_request_ns, "auth_request", auth_request)
-                            
+
                             credentials = self._get_credentials(auth_request)
                         else:
                             # block mode
                             auth_request = await self._start(authorize_params)
                             credentials = await self.get_credentials_polling(auth_request)
 
                         await self.delete_auth_request()
-                    
+
                         if credentials is not None:
                             await self.credentials_store.put(credentials_ns, "credential", credentials)
                 except (AuthorizationPendingInterrupt, AuthorizationPollingInterrupt) as interrupt:
                     return self._handle_authorization_interrupts(interrupt)
                 except Exception as err:
                     await self.delete_auth_request()
                     raise
-                
+
                 _update_local_storage({"credentials": credentials})
 
                 if inspect.iscoroutinefunction(execute):
                     return await execute(*args, **kwargs)
                 else:
                     return execute(*args, **kwargs)
-        
+
         return wrapped_execute

packages/auth0-ai/auth0_ai/authorizers/ciba/ciba_authorizer_params.py

@@ -3,13 +3,18 @@
 from auth0_ai.authorizers.types import ToolInput
 from auth0_ai.stores import Store
 
+
 class CIBAAuthorizerParams(TypedDict, Generic[ToolInput]):
     """
     Authorize Options to start CIBA flow.
 
     Attributes:
         scope (list[str]): The scopes to request authorization for.
         binding_message (Union[str, Callable[..., Awaitable[str]]]): A human-readable string to display to the user, or a function that resolves it.
+        authorization_details (Union[list[dict], Callable[..., Awaitable[list[dict]]]]):
+            Authorization details that specify what the user is authorizing. Can be:
+            - A list of dictionaries with authorization details (e.g., [{ type: "custom_type", param: "example", ...}] details)
+            - A function that resolves to a list of dictionaries
         user_id (Union[str, Callable[..., Awaitable[str]]]): The user id string, or a function that resolves it.
         store (Store, optional): An store used to temporarly store the authorization response data while the user is completing the authorization in another device (default: InMemoryStore).
         audience (str, optional): The audience to request authorization for.
@@ -25,6 +30,7 @@ class CIBAAuthorizerParams(TypedDict, Generic[ToolInput]):
     """
     scopes: list[str]
     binding_message: Union[str, Callable[ToolInput, Awaitable[str]]]
+    authorization_details: Union[list[dict], Callable[ToolInput, Awaitable[list[dict]]]]
     user_id: Union[str, Callable[ToolInput, Awaitable[str]]]
     store: Optional[Store]
     audience: Optional[str]