Add OAuth2 ClientCredentials flow support

CHANGES/926.feature

@@ -0,0 +1,2 @@
+Added support to OAuth2 ClientCredentials grant flow as authentication method.
+This is tech preview and may change without previous warning.

docs/user/reference/_SUMMARY.md

@@ -1,2 +1,3 @@
 * [Using the CLI](using_the_cli.md)
 * [Supported Workflows](supported_workflows.md)
+* [Authentication Methods](authentication.md)

docs/user/reference/authentication.md

@@ -0,0 +1,15 @@
+# Supported Authentication Methods
+
+Pulp-CLI support some authentication methods to authenticate against a Pulp instance.
+Some very simple and common, like HTTP Basic Auth, and some more complex like OAuth2.
+
+## OAuth2 ClientCredentials grant
+
+!!! warning
+  This is an experimental feature. The support of it could change without any major warning.
+
+More on https://datatracker.ietf.org/doc/html/rfc6749#section-4.4
+
+Using this method the pulp-cli can request a token from an Identity Provider using a pair of 
+credentials (client_id/client_secret). The token is ten sent through using the `Authorization` header.
+The issuer URL and the scope of token must be specified by the Pulp server through the OpenAPI scheme definition.

pulp-glue/pulp_glue/common/authentication.py

@@ -0,0 +1,91 @@
+import typing as t
+from datetime import datetime, timedelta
+
+import requests
+
+
+class OAuth2ClientCredentialsAuth(requests.auth.AuthBase):
+    """
+    This implements the OAuth2 ClientCredentials Grant authentication flow.
+    https://datatracker.ietf.org/doc/html/rfc6749#section-4.4
+    """
+
+    def __init__(
+        self,
+        client_id: str,
+        client_secret: str,
+        token_url: str,
+        scopes: t.List[str],
+    ):
+        self.client_id = client_id
+        self.client_secret = client_secret
+        self.token_url = token_url
+        self.scopes = scopes
+
+        self.access_token: t.Optional[str] = None
+        self.expire_at: t.Optional[datetime] = None
+
+    def __call__(self, request: requests.PreparedRequest) -> requests.PreparedRequest:
+        if self.expire_at is None or self.expire_at < datetime.now():
+            self.retrieve_token()
+
+        assert self.access_token is not None
+
+        request.headers["Authorization"] = f"Bearer {self.access_token}"
+
+        # Call to untyped function "register_hook" in typed context
+        request.register_hook("response", self.handle401)  # type: ignore[no-untyped-call]
+
+        return request
+
+    def handle401(
+        self,
+        response: requests.Response,
+        **kwargs: t.Any,
+    ) -> requests.Response:
+        if response.status_code != 401:
+            return response
+
+        # If we get this far, probably the token is not valid anymore.
+
+        # Try to reach for a new token once.
+        self.retrieve_token()
+
+        assert self.access_token is not None
+
+        # Consume content and release the original connection
+        # to allow our new request to reuse the same one.
+        response.content
+        response.close()
+        prepared_new_request = response.request.copy()
+
+        prepared_new_request.headers["Authorization"] = f"Bearer {self.access_token}"
+
+        # Avoid to enter into an infinity loop.
+        # Call to untyped function "deregister_hook" in typed context
+        prepared_new_request.deregister_hook(  # type: ignore[no-untyped-call]
+            "response", self.handle401
+        )
+
+        # "Response" has no attribute "connection"
+        new_response: requests.Response = response.connection.send(prepared_new_request, **kwargs)
+        new_response.history.append(response)
+        new_response.request = prepared_new_request
+
+        return new_response
+
+    def retrieve_token(self) -> None:
+        data = {
+            "client_id": self.client_id,
+            "client_secret": self.client_secret,
+            "scope": " ".join(self.scopes),
+            "grant_type": "client_credentials",
+        }
+
+        response: requests.Response = requests.post(self.token_url, data=data)
+
+        response.raise_for_status()
+
+        token = response.json()
+        self.expire_at = datetime.now() + timedelta(seconds=token["expires_in"])
+        self.access_token = token["access_token"]

pulp-glue/pulp_glue/common/openapi.py

@@ -24,6 +24,7 @@
 SAFE_METHODS = ["GET", "HEAD", "OPTIONS"]
 ISO_DATE_FORMAT = "%Y-%m-%d"
 ISO_DATETIME_FORMAT = "%Y-%m-%dT%H:%M:%S.%fZ"
+AUTH_PRIORITY = ("oauth2", "basic")
 
 
 class OpenAPIError(Exception):
@@ -44,6 +45,30 @@ class UnsafeCallError(OpenAPIError):
     pass
 
 
+class OpenAPISecurityScheme:
+    def __init__(self, security_scheme: t.Dict[str, t.Any]):
+        self.security_scheme = security_scheme
+
+        self.security_type = self.security_scheme["type"]
+        self.description = self.security_scheme.get("description", "")
+
+        if self.security_type == "oauth2":
+            try:
+                self.flows: t.Dict[str, t.Any] = self.security_scheme["flows"]
+                client_credentials: t.Optional[t.Dict[str, t.Any]] = self.flows.get(
+                    "clientCredentials"
+                )
+                if client_credentials:
+                    self.flow_type: str = "clientCredentials"
+                    self.token_url: str = client_credentials["tokenUrl"]
+                    self.scopes: t.List[str] = list(client_credentials["scopes"].keys())
+            except KeyError:
+                raise OpenAPIValidationError
+
+        if self.security_type == "http":
+            self.scheme = self.security_scheme["scheme"]
+
+
 class AuthProviderBase:
     """
     Base class for auth providers.
@@ -57,18 +82,39 @@ def basic_auth(self) -> t.Optional[t.Union[t.Tuple[str, str], requests.auth.Auth
         """Implement this to provide means of http basic auth."""
         return None
 
+    def oauth2_client_credentials_auth(
+        self, flow: t.Any
+    ) -> t.Optional[t.Union[t.Tuple[str, str], requests.auth.AuthBase]]:
+        """Implement this to provide other authentication methods."""
+        return None
+
     def __call__(
         self,
         security: t.List[t.Dict[str, t.List[str]]],
         security_schemes: t.Dict[str, t.Dict[str, t.Any]],
     ) -> t.Optional[t.Union[t.Tuple[str, str], requests.auth.AuthBase]]:
+
+        authorized_schemes = []
+        authorized_schemes_types = set()
+
         for proposal in security:
-            if [security_schemes[name] for name in proposal] == [
-                {"type": "http", "scheme": "basic"}
-            ]:
-                result = self.basic_auth()
-                if result:
-                    return result
+            for name in proposal:
+                authorized_schemes.append(security_schemes[name])
+                authorized_schemes_types.add(security_schemes[name]["type"])
+
+        if "oauth2" in authorized_schemes_types:
+            for flow in authorized_schemes:
+                if flow["type"] == "oauth2":
+                    oauth2_flow = OpenAPISecurityScheme(flow)
+
+            if oauth2_flow.flow_type == "client_credentials":
+                result = self.oauth2_client_credentials_auth(oauth2_flow)
+            if result:
+                return result
+        elif "http" in authorized_schemes_types:
+            result = self.basic_auth()
+            if result:
+                return result
         raise OpenAPIError(_("No suitable auth scheme found."))
 
 

pulpcore/cli/common/generic.py

@@ -10,6 +10,7 @@
 import requests
 import schema as s
 import yaml
+from pulp_glue.common.authentication import OAuth2ClientCredentialsAuth
 from pulp_glue.common.context import (
     DATETIME_FORMATS,
     DEFAULT_LIMIT,
@@ -231,6 +232,21 @@ def basic_auth(self) -> t.Optional[t.Union[t.Tuple[str, str], requests.auth.Auth
                 self.pulp_ctx.password = click.prompt("Password", hide_input=True)
         return (self.pulp_ctx.username, self.pulp_ctx.password)
 
+    def oauth2_client_credentials_auth(
+        self, oauth2_flow: t.Any
+    ) -> t.Optional[requests.auth.AuthBase]:
+        if self.pulp_ctx.username is None:
+            self.pulp_ctx.username = click.prompt("Username/ClientID")
+        if self.pulp_ctx.password is None:
+            self.pulp_ctx.password = click.prompt("Password/ClientSecret")
+
+        return OAuth2ClientCredentialsAuth(
+            client_id=self.pulp_ctx.username,
+            client_secret=self.pulp_ctx.password,
+            token_url=oauth2_flow.token_url,
+            scopes=oauth2_flow.scopes,
+        )
+
 
 ##############################################################################
 # Decorator to access certain contexts