feat: oauth2 client credentials

docs/channels.rst

@@ -60,34 +60,110 @@ Example with oauth2 (like Camunda Identity):
     channel = create_secure_channel(grpc_address="host:port", channel_credentials=channel_credentials)
 
 
-Camunda Cloud
--------------
+Oauth2 Client Credentials Channel
+---------------------------------
 
-Create a grpc channel connected to a Zeebe Gateway running in camunda cloud
+.. autofunction:: pyzeebe.channel.oauth_channel.create_oauth2_client_credentials_channel
 
-.. autofunction:: pyzeebe.create_camunda_cloud_channel
+.. warning:: 
+    Some arguments are Optional and are highly dependent on your Authentication Server configuration,
+    `scope` is usually required and is often optional `audience` .
 
 Example:
 
 .. code-block:: python
 
-    from pyzeebe import create_camunda_cloud_channel
+    import grpc
+    from pyzeebe.channel.oauth_channel import create_oauth2_client_credentials_channel
 
-
-    channel = create_camunda_cloud_channel("client_id", "client_secret", "cluster_id")
+    channel: grpc.aio.Channel = create_oauth2_client_credentials_channel(
+        grpc_address=ZEEBE_ADDRESS,
+        client_id=ZEEBE_CLIENT_ID,
+        client_secret=ZEEBE_CLIENT_SECRET,
+        authorization_server=ZEEBE_AUTHORIZATION_SERVER_URL,
+        scope="profile email",
+        audience="zeebe-api", # NOTE: Can be omitted in some cases.
+    )
+
+Example with custom `channel_options`:
+
+.. code-block:: python
+
+    import grpc
+    from pyzeebe.channel.oauth_channel import create_oauth2_client_credentials_channel
+    from pyzeebe.types import ChannelArgumentType
+
+    channel_options: ChannelArgumentType  = (("grpc.so_reuseport", 0),)
+
+    channel: grpc.aio.Channel = create_oauth2_client_credentials_channel(
+        grpc_address=ZEEBE_ADDRESS,
+        client_id=ZEEBE_CLIENT_ID,
+        client_secret=ZEEBE_CLIENT_SECRET,
+        authorization_server=ZEEBE_AUTHORIZATION_SERVER_URL,
+        scope="profile email",
+        audience="zeebe-api",
+        channel_options=channel_options,
+    )
+
+Example with custom `channel_credentials`:
+
+Useful for self-signed certificates with `grpc.ssl_channel_credentials`.
+
+.. code-block:: python
 
+    import grpc
+    from pyzeebe.channel.oauth_channel import create_oauth2_client_credentials_channel
+    from pyzeebe.types import ChannelArgumentType
+
+    channel_credentials: grpc.ChannelCredentials = grpc.ssl_channel_credentials(
+        certificate_chain=None, private_key=None, root_certificates=None
+    )
+    channel_options: ChannelArgumentType  = (("grpc.so_reuseport", 0),)
+
+    channel: grpc.aio.Channel = create_oauth2_client_credentials_channel(
+        grpc_address=ZEEBE_ADDRESS,
+        client_id=ZEEBE_CLIENT_ID,
+        client_secret=ZEEBE_CLIENT_SECRET,
+        authorization_server=ZEEBE_AUTHORIZATION_SERVER_URL,
+        scope="profile email",
+        audience="zeebe-api",
+        channel_credentials=channel_credentials,
+        channel_options=channel_options,
+    )
+
+Camunda Cloud (Oauth2 Client Credentials Channel)
+-------------------------------------------------
+
+.. autofunction:: pyzeebe.channel.oauth_channel.create_camunda_cloud_channel
+
+.. note::
+    This is a convenience function for creating a channel with the correct parameters for Camunda Cloud.
+    It is equivalent to calling `create_oauth2_client_credentials_channel` with the correct parameters.
+
+Example:
+
+.. code-block:: python
+
+    from pyzeebe.channel.oauth_channel import create_camunda_cloud_channel
 
-Credentials
------------
+    channel: grpc.aio.Channel = create_camunda_cloud_channel(
+        client_id=ZEEBE_CLIENT_ID,
+        client_secret=ZEEBE_CLIENT_SECRET,
+        cluster_id=CAMUNDA_CLUSTER_ID,
+    )
 
-.. autoclass:: pyzeebe.AuthMetadataPlugin
-   :members:
-   :undoc-members:
+Camunda Cloud (Deprecated)
+--------------------------
+
+Create a grpc channel connected to a Zeebe Gateway running in camunda cloud
 
-.. autoclass:: pyzeebe.CredentialsABC
-   :members:
-   :undoc-members:
+.. autofunction:: pyzeebe.channel.camunda_cloud_channel.create_camunda_cloud_channel
 
-.. autoclass:: pyzeebe.CamundaIdentityCredentials
-   :members:
-   :undoc-members:
+Example:
+
+.. code-block:: python
+
+    from pyzeebe.channel.camunda_cloud_channel import create_camunda_cloud_channel
+
+
+    channel = create_camunda_cloud_channel("client_id", "client_secret", "cluster_id")

docs/credentials.rst

@@ -0,0 +1,57 @@
+===========
+Credentials
+===========
+
+Oauth2 Client Credentials Plugin
+--------------------------------
+
+.. autoclass:: pyzeebe.credentials.Oauth2ClientCredentialsMetadataPlugin
+    :members:
+    :special-members:
+    :private-members:
+
+Example:
+
+.. code-block:: python
+
+   oauth2_client_credentials = Oauth2ClientCredentialsMetadataPlugin(
+      client_id=ZEEBE_CLIENT_ID,
+      client_secret=ZEEBE_CLIENT_SECRET,
+      authorization_server=ZEEBE_AUTHORIZATION_SERVER_URL,
+      scope="profile email",
+      audience="zeebe-api",
+   )
+   call_credentials: grpc.CallCredentials = grpc.metadata_call_credentials(oauth2_client_credentials)
+   channel_credentials: grpc.ChannelCredentials = grpc.ssl_channel_credentials(
+      certificate_chain=None, private_key=None, root_certificates=None
+   )
+   composite_credentials: grpc.ChannelCredentials = grpc.composite_channel_credentials(
+      channel_credentials, call_credentials
+   )
+   options: ChannelArgumentType = (("grpc.so_reuseport", 0),)
+   channel: grpc.aio.Channel = grpc.aio.secure_channel(
+      target=ZEEBE_ADDRESS, credentials=composite_credentials, options=options
+   )
+   client = ZeebeClient(channel)
+
+Oauth2 Plugin
+-------------
+.. autoclass:: pyzeebe.credentials.OAuth2MetadataPlugin
+    :members:
+    :special-members:
+    :private-members:
+
+Internal (Deprecated)
+---------------------
+
+.. autoclass:: pyzeebe.AuthMetadataPlugin
+   :members:
+   :undoc-members:
+
+.. autoclass:: pyzeebe.CredentialsABC
+   :members:
+   :undoc-members:
+
+.. autoclass:: pyzeebe.CamundaIdentityCredentials
+   :members:
+   :undoc-members:

docs/index.rst

@@ -64,6 +64,7 @@ Table Of Contents
     Client <client>
     Worker <worker>
     Channels <channels>
+    Credentials <credentials>
     Decorators <decorators>
     Exceptions <errors>
     Zeebe Adapter <zeebe_adapter>

pyzeebe/channel/__init__.py

@@ -1,3 +1,6 @@
-from pyzeebe.channel.camunda_cloud_channel import create_camunda_cloud_channel
 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

pyzeebe/channel/camunda_cloud_channel.py

@@ -4,6 +4,7 @@
 from oauthlib import oauth2
 from requests import HTTPError
 from requests_oauthlib import OAuth2Session
+from typing_extensions import deprecated
 
 from pyzeebe.channel.channel_options import get_channel_options
 from pyzeebe.errors import (
@@ -13,6 +14,11 @@
 from pyzeebe.types import ChannelArgumentType
 
 
+@deprecated(
+    "Use pyzeebe.channel.oauth_channel.create_camunda_cloud_channel function instead",
+    category=DeprecationWarning,
+    stacklevel=1,
+)
 def create_camunda_cloud_channel(
     client_id: str,
     client_secret: str,
@@ -28,8 +34,7 @@ def create_camunda_cloud_channel(
         client_secret (str): The client secret provided by Camunda Cloud
         cluster_id (str): The zeebe cluster id to connect to
         region (str): The cluster's region. Defaults to bru-2
-        channel_options (Optional[Dict], optional): GRPC channel options.
-            See https://grpc.github.io/grpc/python/glossary.html
+        channel_options (Optional[ChannelArgumentType]): GRPC channel options. See https://grpc.github.io/grpc/python/glossary.html
 
     Returns:
         grpc.aio.Channel: A GRPC Channel connected to the Zeebe gateway.

pyzeebe/channel/oauth_channel.py

@@ -0,0 +1,152 @@
+from functools import partial
+from typing import Optional
+
+import grpc
+
+from pyzeebe.channel.channel_options import get_channel_options
+from pyzeebe.credentials.oauth import Oauth2ClientCredentialsMetadataPlugin
+from pyzeebe.types import ChannelArgumentType
+
+
+def create_oauth2_client_credentials_channel(
+    grpc_address: str,
+    client_id: str,
+    client_secret: str,
+    authorization_server: str,
+    scope: Optional[str] = None,
+    audience: Optional[str] = None,
+    channel_credentials: grpc.ChannelCredentials = grpc.ssl_channel_credentials(),
+    channel_options: Optional[ChannelArgumentType] = None,
+    leeway: int = 60,
+    expire_in: Optional[int] = None,
+) -> grpc.aio.Channel:
+    """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:
+        grpc_address (str): Zeebe Gateway Address.
+
+        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.
+
+        channel_credentials (grpc.ChannelCredentials): The gRPC channel credentials.
+            Defaults to grpc.ssl_channel_credentials().
+        channel_options (Optional[ChannelArgumentType]): Additional options for the gRPC channel.
+            Defaults to None.
+            See https://grpc.github.io/grpc/python/glossary.html#term-channel_arguments
+
+        leeway (int): The number of seconds to consider the token as expired before the actual expiration time.
+            Defaults to 60.
+        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.
+
+    Returns:
+        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 = Oauth2ClientCredentialsMetadataPlugin(
+        client_id=client_id,
+        client_secret=client_secret,
+        authorization_server=authorization_server,
+        scope=scope,
+        audience=audience,
+        leeway=leeway,
+        expire_in=expire_in,
+    )
+
+    call_credentials: grpc.CallCredentials = grpc.metadata_call_credentials(oauth2_client_credentials)
+    # channel_credentials: grpc.ChannelCredentials = channel_credentials or grpc.ssl_channel_credentials()
+    composite_credentials: grpc.ChannelCredentials = grpc.composite_channel_credentials(
+        channel_credentials, call_credentials
+    )
+
+    channel: grpc.aio.Channel = grpc.aio.secure_channel(
+        target=grpc_address, credentials=composite_credentials, options=get_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",
+    channel_credentials: grpc.ChannelCredentials = grpc.ssl_channel_credentials(),
+    channel_options: Optional[ChannelArgumentType] = None,
+    leeway: int = 60,
+    expire_in: Optional[int] = 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".
+
+        channel_credentials (grpc.ChannelCredentials): The gRPC channel credentials.
+            Defaults to grpc.ssl_channel_credentials().
+        channel_options (Optional[ChannelArgumentType]): Additional options for the gRPC channel.
+            Defaults to None.
+            See https://grpc.github.io/grpc/python/glossary.html#term-channel_arguments
+
+        leeway (int): The number of seconds to consider the token as expired before the actual expiration time.
+            Defaults to 60.
+        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.
+
+    Returns:
+        grpc.aio.Channel: The gRPC channel for connecting to Camunda Cloud.
+    """
+
+    grpc_address = 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,
+        leeway=leeway,
+        expire_in=expire_in,
+    )
+
+    # 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
+
+    call_credentials: grpc.CallCredentials = grpc.metadata_call_credentials(oauth2_client_credentials)
+    # channel_credentials: grpc.ChannelCredentials = channel_credentials or grpc.ssl_channel_credentials()
+    composite_credentials: grpc.ChannelCredentials = grpc.composite_channel_credentials(
+        channel_credentials, call_credentials
+    )
+
+    channel: grpc.aio.Channel = grpc.aio.secure_channel(
+        target=grpc_address, credentials=composite_credentials, options=get_channel_options(channel_options)
+    )
+
+    return channel

pyzeebe/credentials/__init__.py

@@ -1,3 +1,4 @@
 from .base import CredentialsABC
 from .camunda_identity import CamundaIdentityCredentials
+from .oauth import Oauth2ClientCredentialsMetadataPlugin, OAuth2MetadataPlugin
 from .plugins import AuthMetadataPlugin