feat: oauth2 client credentials

pyzeebe/channel/__init__.py

@@ -1,3 +1,10 @@
-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

pyzeebe/channel/oauth_channel.py

@@ -0,0 +1,146 @@
+from functools import partial
+from typing import Optional
+
+import grpc
+from grpc.aio._typing import ChannelArgumentType
+
+from pyzeebe.credentials.oauth import Oauth2ClientCredentialsMetadataPlugin
+
+
+def create_oauth2_client_credentials_channel(
+    target: 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:
+        target (str): The target address of the Zeebe Gateway.
+
+        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], optional): 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=target, credentials=composite_credentials, 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], optional): 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.
+    """
+
+    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,
+        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=target, credentials=composite_credentials, options=channel_options
+    )
+
+    return channel

pyzeebe/credentials/oauth.py

@@ -0,0 +1,182 @@
+import json
+import logging
+import time
+import timeit
+from collections.abc import Callable
+from functools import partial
+from typing import Any, Optional
+
+import grpc
+import requests
+from grpc._auth import _sign_request
+from oauthlib import oauth2
+from requests_oauthlib import OAuth2Session
+
+logger = logging.getLogger(__name__)
+
+
+class OAuth2MetadataPlugin(grpc.AuthMetadataPlugin):
+    """AuthMetadataPlugin for OAuth2 Authentication.
+
+    Implements the AuthMetadataPlugin interface for OAuth2 Authentication based on oauthlib and requests_oauthlib.
+
+    https://datatracker.ietf.org/doc/html/rfc6749
+    https://oauthlib.readthedocs.io/en/latest/oauth2/oauth2.html
+    https://requests-oauthlib.readthedocs.io/en/latest/oauth2_workflow.html
+    """
+
+    def __init__(
+        self,
+        oauth2session: OAuth2Session,
+        func_retrieve_token: Callable[[], Any],
+        leeway: int = 60,
+        expire_in: Optional[int] = None,
+    ) -> None:
+        """AuthMetadataPlugin for OAuth2 Authentication.
+
+        https://datatracker.ietf.org/doc/html/rfc6749
+
+        Args:
+            oauth2session (OAuth2Session): The OAuth2Session object.
+            func_fetch_token (Callable): The function to fetch the token.
+
+            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.
+        """
+        self._oauth: OAuth2Session = oauth2session
+        self._func_retrieve_token: Callable[[], Any] = func_retrieve_token
+
+        self._leeway: int = leeway
+        self._expires_in: Optional[int] = expire_in
+        if self._expires_in is not None:
+            # NOTE: "expires_in" is only RECOMMENDED
+            # https://datatracker.ietf.org/doc/html/rfc6749#section-5.1
+            self._oauth.register_compliance_hook("access_token_response", self._no_expiration)
+
+    def __call__(
+        self,
+        context: grpc.AuthMetadataContext,
+        callback: grpc.AuthMetadataPluginCallback,
+    ) -> None:
+        start_time = timeit.default_timer()
+
+        try:
+            if self.is_token_expired():
+                self.retrieve_token()
+
+        except Exception as exception:  # pylint: disable=broad-except
+            _sign_request(callback, None, exception)
+
+        else:
+            _sign_request(callback, self._oauth.access_token, None)
+            logger.debug(
+                "Requesting OAuth2 took: %fs",
+                timeit.default_timer() - start_time,
+            )
+
+            logger.debug(
+                "Token will expire in: %is",
+                (self._oauth.token.get("expires_at", 0) - time.time()),
+            )
+
+    def is_token_expired(self) -> bool:
+        """Check if the token is still valid."""
+        if not self._oauth.authorized:
+            return True
+
+        # NOTE: "expires_at" is not part of the OAuth2 Standard, but very useful
+        # https://datatracker.ietf.org/doc/html/rfc6749#appendix-A
+        # https://oauthlib.readthedocs.io/en/latest/_modules/oauthlib/oauth2/rfc6749/clients/base.html?highlight=expires_at#
+        expires_at = self._oauth.token.get("expires_at", 0)
+        if time.time() > (expires_at - self._leeway):
+            return True
+
+        return False
+
+    def retrieve_token(self) -> None:
+        """Retrieve the access token from the authorization server."""
+
+        try:
+            self._func_retrieve_token()
+
+        except oauth2.OAuth2Error as e:
+            logger.error(str(e))
+            raise e
+
+    def _no_expiration(self, r: requests.Response) -> requests.Response:
+        """
+        Sets the expiration time for the token if it is not provided in the response.
+
+        Args:
+            r (requests.Response): The response object containing the token.
+
+        Returns:
+            requests.Response: The modified response object with the updated token.
+        """
+        token = json.loads(r.text)
+
+        if token.get("expires_in") is None:
+            logger.warning("Token attribute expires_in not found.")
+            token["expires_in"] = self._expires_in
+
+        r._content = json.dumps(token).encode()
+        return r
+
+
+class Oauth2ClientCredentialsMetadataPlugin(OAuth2MetadataPlugin):
+    """AuthMetadataPlugin for OAuth2 Client Credentials Authentication based on Oauth2MetadataPlugin.
+
+    https://oauth.net/2/grant-types/client-credentials/
+    https://datatracker.ietf.org/doc/html/rfc6749#section-11.2.2
+    """
+
+    def __init__(
+        self,
+        client_id: str,
+        client_secret: str,
+        authorization_server: str,
+        scope: Optional[str] = None,
+        audience: Optional[str] = None,
+        leeway: int = 60,
+        expire_in: Optional[int] = None,
+    ):
+        """AuthMetadataPlugin for OAuth2 Client Credentials Authentication based on Oauth2MetadataPlugin.
+
+        https://oauth.net/2/grant-types/client-credentials/
+        https://datatracker.ietf.org/doc/html/rfc6749#section-11.2.2
+
+        Args:
+            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.
+
+            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.
+        """
+
+        self.client_id: str = client_id
+        self.client_secret: str = client_secret
+        self.authorization_server: str = authorization_server
+        self.scope: Optional[str] = scope
+        self.audience: Optional[str] = audience
+        self.leeway: int = leeway
+        self.expire_in: Optional[int] = expire_in
+
+        client = oauth2.BackendApplicationClient(client_id=self.client_id, scope=self.scope)
+        oauth2session = OAuth2Session(client=client)
+
+        func = partial(
+            oauth2session.fetch_token,
+            token_url=self.authorization_server,
+            client_secret=self.client_secret,
+            audience=self.audience,
+        )
+
+        super().__init__(
+            oauth2session=oauth2session, func_retrieve_token=func, leeway=self.leeway, expire_in=self.expire_in
+        )

tests/unit/channel/oauth_channel_test.py

@@ -0,0 +1,46 @@
+from unittest import mock
+
+import grpc
+import pytest
+
+from pyzeebe.channel.oauth_channel import (
+    create_camunda_cloud_channel,
+    create_oauth2_client_credentials_channel,
+)
+
+
+@pytest.fixture
+def mock_oauth2metadataplugin():
+    with mock.patch("pyzeebe.credentials.oauth.OAuth2MetadataPlugin") as mock_credentials:
+        yield mock_credentials
+
+
+def test_create_oauth2_client_credentials_channel(
+    mock_oauth2metadataplugin,
+):
+
+    target = "zeebe-gateway:26500"
+    client_id = "client_id"
+    client_secret = "client_secret"
+    authorization_server = "https://authorization.server"
+    channel = create_oauth2_client_credentials_channel(target, client_id, client_secret, authorization_server)
+
+    assert isinstance(channel, grpc.aio.Channel)
+
+
+def test_create_camunda_cloud_channel(
+    mock_oauth2metadataplugin,
+):
+    client_id = "client_id"
+    client_secret = "client_secret"
+    cluster_id = "cluster_id"
+    region = "bru-2"
+    scope = "Zeebe"
+    authorization_server = "https://login.cloud.camunda.io/oauth/token"
+    audience = "zeebe.camunda.io"
+
+    channel = create_camunda_cloud_channel(
+        client_id, client_secret, cluster_id, region, scope, authorization_server, audience
+    )
+
+    assert isinstance(channel, grpc.aio.Channel)