feat: Add OAuth2 client credentials channel creation and Camunda Clou…

…d Channel

Based on Oauth2ClientCredentialsMetadataPlugin and a general OAuth2MetadataPlugin to support Oauth2Sessions

pyzeebe/channel/__init__.py

@@ -1,4 +1,7 @@
-from pyzeebe.channel.camunda_cloud_channel import create_camunda_cloud_channel
+# from pyzeebe.channel.camunda_cloud_channel import create_camunda_cloud_channel # FIXME: could be removed
 from pyzeebe.channel.insecure_channel import create_insecure_channel
+from pyzeebe.channel.oauth_channel import (
+    create_camunda_cloud_channel,
+    create_oauth2_client_credentials_channel,
+)
 from pyzeebe.channel.secure_channel import create_secure_channel
-from pyzeebe.channel.oauth_channel import create_oauth2_client_credentials_channel

pyzeebe/channel/oauth_channel.py

@@ -1,9 +1,15 @@
-from typing import Optional
+from functools import partial
+from typing import Callable, Optional
 
 import grpc
 from grpc.aio._typing import ChannelArgumentType
+from oauthlib import oauth2
+from requests_oauthlib import OAuth2Session
 
-from pyzeebe.credentials.oauth import OAuth2ClientCredentials
+from pyzeebe.credentials.oauth import (
+    Oauth2ClientCredentialsMetadataPlugin,
+    OAuth2MetadataPlugin,
+)
 
 
 def create_oauth2_client_credentials_channel(
@@ -13,35 +19,39 @@ def create_oauth2_client_credentials_channel(
     authorization_server: str,
     scope: Optional[str] = None,
     audience: Optional[str] = None,
-    expire_in: Optional[int] = 0,
+    expire_in: Optional[int] = None,
     channel_options: Optional[ChannelArgumentType] = None,
 ) -> grpc.aio.Channel:
     """
-    Create channel connected to a Camunda Cloud cluster
+    Create a gRPC channel for connecting to Camunda 8 (Self-Managed) with OAuth2ClientCredentials.
+
+    https://oauth.net/2/grant-types/client-credentials/
+    https://datatracker.ietf.org/doc/html/rfc6749#section-11.2.2
 
     Args:
         target (str): The target address of the Zeebe Gateway.
 
-        client_id (str): The client id provided by Camunda Cloud
-        client_secret (str): The client secret provided by Camunda Cloud
-        authorization_server (str): The server issuing access tokens
-        to the client after successfully authenticating the resource owner
-        and obtaining authorization.
-        scope (Optional[str]): The Access Token Scope.
-        audience (Optional[str]): The audience for the token.
+        client_id (str): The client id.
+        client_secret (str): The client secret.
+        authorization_server (str): The authorization server issuing access tokens.
+        to the client after successfully authenticating the client.
+        scope (Optional[str]): The scope of the access request. Defaults to None.
+        audience (Optional[str]): The audience for authentication. Defaults to None.
 
-        expire_in (Optional[int]): Only used if the token does not contain an "expires_in" attribute. The number of seconds the token is valid for.
+        expire_in (Optional[int]): The number of seconds the token is valid for. Defaults to None.
+            Should only be used if the token does not contain an "expires_in" attribute.
 
-        channel_options (Optional[Dict], optional): GRPC channel options. See https://grpc.github.io/grpc/python/glossary.html
+        channel_options (Optional[ChannelArgumentType], optional): Additional options for the gRPC channel. Defaults to None.
+                See https://grpc.github.io/grpc/python/glossary.html#term-channel_arguments
 
     Returns:
-        grpc.aio.Channel: A GRPC Channel connected to the Zeebe Gateway.
+        grpc.aio.Channel: A gRPC channel connected to the Zeebe Gateway.
 
     Raises:
         InvalidOAuthCredentialsError: One of the provided camunda credentials is not correct
     """
 
-    oauth2_client_credentials = OAuth2ClientCredentials(
+    oauth2_client_credentials = Oauth2ClientCredentialsMetadataPlugin(
         client_id=client_id,
         client_secret=client_secret,
         authorization_server=authorization_server,
@@ -50,15 +60,63 @@ def create_oauth2_client_credentials_channel(
         expire_in=expire_in,
     )
 
-    call_credentials: grpc.CallCredentials = grpc.metadata_call_credentials(oauth2_client_credentials)
-    channel_credentials: grpc.ChannelCredentials = grpc.ssl_channel_credentials()
+    channel = oauth2_client_credentials.channel(target=target, channel_options=channel_options)
+    return channel
+
+
+def create_camunda_cloud_channel(
+    client_id: str,
+    client_secret: str,
+    cluster_id: str,
+    region: str = "bru-2",
+    scope: str = "Zeebe",
+    authorization_server: str = "https://login.cloud.camunda.io/oauth/token",
+    audience: str = "zeebe.camunda.io",
+    expire_in: Optional[int] = None,
+    channel_options: Optional[ChannelArgumentType] = None,
+) -> grpc.aio.Channel:
+    """
+    Create a gRPC channel for connecting to Camunda 8 Cloud (SaaS).
+
+    Args:
+        client_id (str): The client id.
+        client_secret (str): The client secret.
+        cluster_id (str): The ID of the cluster to connect to.
+        region (Optional[str]): The region of the cluster. Defaults to "bru-2".
+        scope (Optional[str]): The scope of the access request. Defaults to "Zeebe".
+        authorization_server (Optional[str]): The authorization server issuing access tokens.
+        to the client after successfully authenticating the client. Defaults to "https://login.cloud.camunda.io/oauth/token".
+        audience (Optional[str]): The audience for authentication. Defaults to "zeebe.camunda.io".
+
+        expire_in (Optional[int]): The expiration time for the token. Defaults to None.
+
+        channel_options (Optional[ChannelArgumentType], optional): Additional options for the gRPC channel. Defaults to None.
+                See https://grpc.github.io/grpc/python/glossary.html#term-channel_arguments
 
-    composite_credentials: grpc.ChannelCredentials = grpc.composite_channel_credentials(
-        channel_credentials, call_credentials
+    Returns:
+        grpc.aio.Channel: The gRPC channel for connecting to Camunda Cloud.
+    """
+
+    target = f"{cluster_id}.{region}.zeebe.camunda.io:443"
+
+    oauth2_client_credentials = Oauth2ClientCredentialsMetadataPlugin(
+        client_id=client_id,
+        client_secret=client_secret,
+        authorization_server=authorization_server,
+        scope=scope,
+        audience=audience,
+        expire_in=expire_in,
     )
 
-    channel: grpc.aio.Channel = grpc.aio.secure_channel(
-        target=target, credentials=composite_credentials, options=channel_options
+    # NOTE: Overwrite the _oauth.fetch_token method to include client_id, client_secret in the request body
+    func = partial(
+        oauth2_client_credentials._oauth.fetch_token,
+        include_client_id=True,
+        token_url=authorization_server,
+        client_secret=client_secret,
+        audience=audience,
     )
+    oauth2_client_credentials._func_retrieve_token = func
 
+    channel = oauth2_client_credentials.channel(target=target, channel_options=channel_options)
     return channel