FEAT 222 Add client credential flow

doc/source/api/index.rst

@@ -15,7 +15,7 @@ classes, functions, and attributes.
 
    ApiClient
    ApiClientFactory
-   OIDCSessionBuilder
+   AuthorizationSessionBuilder
    SessionConfiguration
 
 

src/ansys/openapi/common/__init__.py

@@ -10,7 +10,7 @@
 
     __version__ = metadata_backport("ansys-openapi-common")["version"]
 
-from ._session import ApiClientFactory, OIDCSessionBuilder
+from ._session import ApiClientFactory, AuthorizationSessionBuilder
 from ._util import SessionConfiguration, generate_user_agent
 from ._exceptions import ApiConnectionException, ApiException, AuthenticationWarning
 from ._api_client import ApiClient
@@ -26,7 +26,7 @@
     "AuthenticationWarning",
     "create_session_from_granta_stk",
     "generate_user_agent",
-    "OIDCSessionBuilder",
+    "AuthorizationSessionBuilder",
     "ApiBase",
     "ApiClientBase",
     "ModelBase",

src/ansys/openapi/common/_contrib/granta.py

@@ -46,15 +46,15 @@ def create_session_from_granta_stk(
         cached_token_key = auth_settings["token_key"]
         return (
             ApiClientFactory(sl_url, api_session_configuration)
-            .with_oidc(idp_session_configuration)
+            .with_oidc_authorization_flow(idp_session_configuration)
             .with_stored_token(cached_token_key)
             .connect()
         )
     elif mode == "oidc_token":
         refresh_token = auth_settings["refresh_token"]
         return (
             ApiClientFactory(sl_url, api_session_configuration)
-            .with_oidc(idp_session_configuration)
+            .with_oidc_authorization_flow(idp_session_configuration)
             .with_token(refresh_token)
             .connect()
         )

src/ansys/openapi/common/_oidc.py

@@ -4,7 +4,11 @@
 import keyring
 import requests
 from requests.models import CaseInsensitiveDict
-from requests_auth import OAuth2AuthorizationCodePKCE, InvalidGrantRequest  # type: ignore[import]
+from requests_auth import (  # type: ignore[import]
+    OAuth2AuthorizationCodePKCE,
+    OAuth2ClientCredentials,
+    InvalidGrantRequest,
+)
 from requests_auth.authentication import OAuth2  # type: ignore[import]
 
 from ._util import (
@@ -25,6 +29,51 @@
     _log_tokens = False
 
 
+def get_client_credential_auth(
+    token_url: str,
+    client_id: str,
+    client_secret: str,
+    scope: Optional[str] = "",
+    session: Optional[requests.Session] = None,
+) -> OAuth2ClientCredentials:
+
+    """Get a requests auth object for the Client Credential OIDC flow.
+
+    Provides a wrapper around the requests_auth.OAuth2ClientCredentials class.
+
+    Parameters
+    ----------
+    token_url: :class:`str`
+        OAuth 2 token URL.
+    client_id : :class:`str`
+        Resource owner username. Provided by the Identity provider.
+    client_secret : :class:`str`
+        Resource owner password. Provided by the Identity provider.
+    scope : :class:`str`, optional
+        Single scope or list of scopes required by the application.
+    session: :class:`requests.Session`, optional
+        Session used to retrieve the token. Used if a specific configuration is required,
+        e.g. to disable SSL certificate verification.
+
+    Returns
+    -------
+    :class:`requests_auth.OAuth2ClientCredentials`
+        Requests Client Credentials auth object.
+
+    Notes
+    -----
+    OIDC Authentication requires the ``[oidc]`` extra to be installed.
+    """
+
+    return OAuth2ClientCredentials(
+        token_url=token_url,
+        client_id=client_id,
+        client_secret=client_secret,
+        scope=scope,
+        session=session,
+    )
+
+
 class OIDCSessionFactory:
     """
     Creates an OpenID Connect session with the configuration fetched from the API server. This class uses either

src/ansys/openapi/common/_session.py

@@ -1,6 +1,8 @@
 import os
 import warnings
-from typing import Tuple, Union, Container, Optional, Mapping, TypeVar, Any
+from typing import Tuple, Union, Optional, Mapping, TypeVar, Any, Callable
+from functools import wraps
+from copy import copy
 
 import requests
 from urllib3.util.retry import Retry
@@ -31,7 +33,7 @@
     # noinspection PyUnresolvedReferences
     import requests_auth  # type: ignore[import]
     import keyring
-    from ._oidc import OIDCSessionFactory
+    from ._oidc import OIDCSessionFactory, get_client_credential_auth
 except ImportError:
     _oidc_enabled = False
 
@@ -51,6 +53,29 @@
 
     _platform_windows = False
 
+Return_Type = TypeVar("Return_Type")
+
+
+def require_oidc(func: Callable[..., Return_Type]) -> Callable[..., Return_Type]:
+    """Enforce that OIDC features are enabled before executing the wrapped function/method.
+
+    Raises
+    ------
+    ImportError
+        If the OIDC features have not been installed.
+    """
+
+    @wraps(func)
+    def wrapper(*args: Any, **kwargs: Any) -> Return_Type:
+        if not _oidc_enabled:
+            raise ImportError(
+                "OpenID Connect features are not enabled. To use them, run `pip install ansys-openapi-common[oidc]`."
+            )
+        return func(*args, **kwargs)
+
+    return wrapper
+
+
 # Required to allow the ApiClientFactory to be subclassed. This ensures that Pylance
 # understands that the subclass is returned by the builder methods instead of the base class
 Api_Client_Factory = TypeVar("Api_Client_Factory", bound="ApiClientFactory")
@@ -261,11 +286,52 @@ def with_autologon(self: Api_Client_Factory) -> Api_Client_Factory:
                 return self
         raise ConnectionError("Unable to connect with autologon.")
 
-    def with_oidc(
+    @require_oidc
+    def with_oidc_client_credentials_flow(
+        self: Api_Client_Factory,
+        client_id: str,
+        client_secret: str,
+        scope: Optional[str] = "",
+    ) -> Api_Client_Factory:
+        """Set up client authentication for use with OpenID Connect using the Client Credentials flow.
+
+        Parameters
+        ----------
+        client_id : :class:`str`
+            Resource owner username. Provided by the Identity provider.
+        client_secret : :class:`str`
+            Resource owner password. Provided by the Identity provider.
+        scope : Union[:class:`str`, :class:`list`[:class:`str`]], optional
+            Single scope or list of scopes required by the application.
+
+        Returns
+        -------
+        :class:`~ansys.openapi.common.ApiClientFactory`
+            Current client factory object.
+
+        Notes
+        -----
+        OIDC Authentication requires the ``[oidc]`` extra to be installed.
+        """
+
+        auth = get_client_credential_auth(
+            token_url=self._api_url,
+            client_id=client_id,
+            client_secret=client_secret,
+            scope=scope,
+            session=copy(self._session),
+        )
+        self._session.auth = auth
+        self._configured = True
+        return self
+
+    @require_oidc
+    def with_oidc_authorization_flow(
         self,
         idp_session_configuration: Optional[SessionConfiguration] = None,
-    ) -> "OIDCSessionBuilder":
-        """Set up client authentication for use with OpenID Connect.
+    ) -> "AuthorizationSessionBuilder":
+        """Set up client authentication for use with OpenID Connect using the authorization flow. Currently
+        only authorization flow with PKCE is supported.
 
         Parameters
         ----------
@@ -274,20 +340,17 @@ def with_oidc(
 
         Returns
         -------
-        :class:`~ansys.openapi.common.OIDCSessionBuilder`
+        :class:`~ansys.openapi.common.AuthorizationSessionBuilder`
             Builder object to authenticate via OIDC.
 
         Notes
         -----
         OIDC Authentication requires the ``[oidc]`` extra to be installed.
         """
-        if not _oidc_enabled:
-            raise ImportError(
-                "OpenID Connect features are not enabled. To use them, run `pip install ansys-openapi-common[oidc]`."
-            )
+
         initial_response = self._session.get(self._api_url)
         if self.__handle_initial_response(initial_response):
-            return OIDCSessionBuilder(self)
+            return AuthorizationSessionBuilder(self)
 
         session_factory = OIDCSessionFactory(
             self._session,
@@ -296,7 +359,7 @@ def with_oidc(
             idp_session_configuration,
         )
 
-        return OIDCSessionBuilder(self, session_factory)
+        return AuthorizationSessionBuilder(self, session_factory)
 
     def __test_connection(self) -> bool:
         """Attempt to connect to the API server. If this returns a 2XX status code, the method returns
@@ -380,9 +443,9 @@ def __get_authenticate_header(
         return parse_authenticate(response.headers["www-authenticate"])
 
 
-class OIDCSessionBuilder:
-    """Helps create OpenID Connect sessions from different types of input and provides OIDC-specific
-    configuration options.
+class AuthorizationSessionBuilder:
+    """Helps create OpenID Connect Authorize Flow sessions from different types of input and provides
+    configuration options specific to the Authorization Flow.
 
     Parameters
     ----------

src/ansys/openapi/common/_util.py

@@ -3,7 +3,6 @@
 
 import pyparsing as pp
 from collections import OrderedDict
-from http.server import BaseHTTPRequestHandler, HTTPServer
 from itertools import chain
 from typing import Dict, Union, List, Optional, Tuple, Any, Collection, cast
 

tests/contrib/test_granta_integration.py

@@ -62,7 +62,7 @@ def test_provided_token_session():
     builder.with_token.return_value = MagicMock()
 
     with patch.object(
-        ApiClientFactory, "with_oidc", return_value=builder
+        ApiClientFactory, "with_oidc_authorization_flow", return_value=builder
     ) as mock_method:
         _ = create_session_from_granta_stk(stk_token_config)
     mock_method.assert_called_once_with(None)
@@ -79,7 +79,7 @@ def test_stored_token_session():
     builder.with_stored_token.return_value = MagicMock()
 
     with patch.object(
-        ApiClientFactory, "with_oidc", return_value=builder
+        ApiClientFactory, "with_oidc_authorization_flow", return_value=builder
     ) as mock_method:
         _ = create_session_from_granta_stk(stk_token_config)
     mock_method.assert_called_once_with(None)

tests/test_missing_imports.py

@@ -42,7 +42,9 @@ def test_create_oidc_with_no_extra_throws(self, mocker):
         from ansys.openapi.common import ApiClientFactory
 
         with pytest.raises(ImportError) as excinfo:
-            _ = ApiClientFactory("http://www.my-api.com/v1.svc").with_oidc()
+            _ = ApiClientFactory(
+                "http://www.my-api.com/v1.svc"
+            ).with_oidc_authorization_flow()
 
         package_name = get_package_name()
         assert f"`pip install {package_name}[oidc]`" in str(excinfo.value)

tests/test_oidc.py

@@ -4,21 +4,24 @@
 import pytest
 import requests
 import requests_mock
-from requests_auth.authentication import OAuth2
+from requests_auth.authentication import OAuth2, OAuth2ClientCredentials
 from unittest.mock import Mock, MagicMock
 from covertable import make
 
 from ansys.openapi.common import ApiClientFactory
-from ansys.openapi.common._oidc import OIDCSessionFactory
+from ansys.openapi.common._oidc import OIDCSessionFactory, get_client_credential_auth
 
+TOKEN_URL = "https://www.example.com/token"
+CLIENT_ID = "3acde603-9bb9-48e7-9eaa-c624c4fd40ca"
+CLIENT_SECRET = "000a5e5ecdd7e2dbb2d61b4a7291e0b44b719c786ed1d677587e83ab60bf8bbf"
 REQUIRED_HEADERS = {
-    "clientid": "3acde603-9bb9-48e7-9eaa-c624c4fd40ca",
+    "clientid": CLIENT_ID,
     "authority": "authority.com",
     "redirecturi": "http://localhost:1729",
 }
 
 WELL_KNOWN_PARAMETERS = {
-    "token_endpoint": "www.example.com/token",
+    "token_endpoint": TOKEN_URL,
     "authorization_endpoint": "www.example.com/authorization",
 }
 
@@ -233,7 +236,7 @@ def match_token_request(request):
             headers={"WWW-Authenticate": "Bearer error=invalid_token"},
         )
         with pytest.raises(ValueError) as exception_info:
-            ApiClientFactory(api_url).with_oidc().with_token(
+            ApiClientFactory(api_url).with_oidc_authorization_flow().with_token(
                 refresh_token=refresh_token
             )
         assert "refresh token was invalid" in str(exception_info)
@@ -267,8 +270,23 @@ def test_endpoint_with_refresh_configures_correctly():
             headers={"WWW-Authenticate": authenticate_header},
         )
 
-        session = ApiClientFactory(secure_servicelayer_url).with_oidc()
+        session = ApiClientFactory(
+            secure_servicelayer_url
+        ).with_oidc_authorization_flow()
         auth = session._session_factory._auth
 
         assert auth.token_url == f"{authority_url}token"
         assert auth.refresh_data["client_id"] == client_id
+
+
+def test_get_client_credential_auth():
+    result = get_client_credential_auth(TOKEN_URL, CLIENT_ID, CLIENT_SECRET, "scope")
+    assert isinstance(result, OAuth2ClientCredentials)
+
+
+def test_get_client_credential_auth_succeeds_custom_session():
+    session = requests.session()
+    result = get_client_credential_auth(
+        TOKEN_URL, CLIENT_ID, CLIENT_SECRET, "scope", session
+    )
+    assert isinstance(result, OAuth2ClientCredentials)