WIP

Author: mdellweg

pulp-glue/pulp_glue/common/authentication.py

@@ -4,6 +4,125 @@
 import requests
 
 
+class AuthProviderBase:
+    """
+    Base class for auth providers.
+
+    This abstract base class will analyze the authentication proposals of the openapi specs.
+    Different authentication schemes should be implemented by subclasses.
+    Returned auth objects need to be compatible with `requests.auth.AuthBase`.
+    """
+
+    def __init__(self) -> None:
+        self._oauth2_token: str | None = None
+        self._oauth2_expires: datetime = datetime.now()
+
+    def can_complete_http_basic(self) -> bool:
+        return False
+
+    def can_complete_mutualTLS(self) -> bool:
+        return False
+
+    def can_complete_oauth2_client_credentials(self, scopes: list[str]) -> bool:
+        return False
+
+    def can_complete_scheme(self, scheme: dict[str, t.Any], scopes: list[str]) -> bool:
+        if scheme["type"] == "http":
+            if scheme["scheme"] == "basic":
+                return self.can_complete_http_basic()
+        elif scheme["type"] == "mutualTLS":
+            return self.can_complete_mutualTLS()
+        elif scheme["type"] == "oauth2":
+            for flow_name, flow in scheme["flows"].items():
+                if (
+                    flow_name == "clientCredentials"
+                    and self.can_complete_oauth2_client_credentials(flow["scopes"])
+                ):
+                    return True
+        return False
+
+    def can_complete(
+        self, proposal: dict[str, list[str]], security_schemes: dict[str, dict[str, t.Any]]
+    ) -> bool:
+        for name, scopes in proposal.items():
+            scheme = security_schemes.get(name)
+            if scheme is None or not self.can_complete_scheme(scheme, scopes):
+                return False
+        # This covers the case where `[]` allows for no auth at all.
+        return True
+
+    async def http_basic_credentials(self) -> tuple[bytes, bytes]:
+        raise NotImplementedError()
+
+    async def oauth2_client_credentials(self) -> tuple[bytes, bytes]:
+        raise NotImplementedError()
+
+
+class BasicAuthProvider(AuthProviderBase):
+    """
+    AuthProvider providing basic auth with fixed `username`, `password`.
+    """
+
+    def __init__(self, username: t.AnyStr, password: t.AnyStr):
+        super().__init__()
+        self.username: bytes = username.encode("latin1") if isinstance(username, str) else username
+        self.password: bytes = password.encode("latin1") if isinstance(password, str) else password
+
+    def can_complete_http_basic(self) -> bool:
+        return True
+
+    async def http_basic_credentials(self) -> tuple[bytes, bytes]:
+        return self.username, self.password
+
+
+class GlueAuthProvider(AuthProviderBase):
+    """
+    AuthProvider allowing to be used with prepared credentials.
+    """
+
+    def __init__(
+        self,
+        username: t.AnyStr | None = None,
+        password: t.AnyStr | None = None,
+        client_id: t.AnyStr | None = None,
+        client_secret: t.AnyStr | None = None,
+    ):
+        super().__init__()
+        self.username: bytes | None = None
+        self.password: bytes | None = None
+        self.client_id: bytes | None = None
+        self.client_secret: bytes | None = None
+        if username is not None:
+            assert password is not None
+            self.username = username.encode("latin1") if isinstance(username, str) else username
+            self.password = password.encode("latin1") if isinstance(password, str) else password
+        if client_id is not None:
+            assert client_secret is not None
+            self.client_id = client_id.encode("latin1") if isinstance(client_id, str) else client_id
+            self.client_secret = (
+                client_secret.encode("latin1") if isinstance(client_secret, str) else client_secret
+            )
+
+    def can_complete_http_basic(self) -> bool:
+        return self.username is not None
+
+    def can_complete_oauth2_client_credentials(self, scopes: list[str]) -> bool:
+        return self.client_id is not None
+
+    async def http_basic_credentials(self) -> tuple[bytes, bytes]:
+        assert self.username is not None
+        assert self.password is not None
+        return self.username, self.password
+
+    async def oauth2_client_credentials(self) -> tuple[bytes, bytes]:
+        assert self.client_id is not None
+        assert self.client_secret is not None
+        return self.client_id, self.client_secret
+
+
+# ----------------------8<----8<------------------------
+
+
 class OAuth2ClientCredentialsAuth(requests.auth.AuthBase):
     """
     This implements the OAuth2 ClientCredentials Grant authentication flow.

pulp-glue/pulp_glue/common/context.py

@@ -9,6 +9,7 @@
 
 from packaging.specifiers import SpecifierSet
 
+from pulp_glue.common.authentication import BasicAuthProvider
 from pulp_glue.common.exceptions import (
     NotImplementedFake,
     OpenAPIError,
@@ -19,7 +20,7 @@
     UnsafeCallError,
 )
 from pulp_glue.common.i18n import get_translation
-from pulp_glue.common.openapi import BasicAuthProvider, OpenAPI
+from pulp_glue.common.openapi import OpenAPI
 
 if sys.version_info >= (3, 11):
     import tomllib

pulp-glue/pulp_glue/common/openapi.py

@@ -6,19 +6,19 @@
 import typing as t
 import warnings
 from base64 import b64encode
-from collections import defaultdict
 from dataclasses import dataclass
+from datetime import datetime, timedelta
 from functools import cached_property
 from io import BufferedReader
 from urllib.parse import urlencode, urljoin
 
 import aiofiles
 import aiofiles.os
 import aiohttp
-import requests
 from multidict import CIMultiDict, CIMultiDictProxy, MutableMultiMapping
 
 from pulp_glue.common import __version__
+from pulp_glue.common.authentication import AuthProviderBase
 from pulp_glue.common.exceptions import (
     OpenAPIError,
     PulpAuthenticationFailed,
@@ -58,156 +58,14 @@ class _Response:
     body: bytes
 
 
-class AuthProviderBase:
-    """
-    Base class for auth providers.
-
-    This abstract base class will analyze the authentication proposals of the openapi specs.
-    Different authentication schemes should be implemented by subclasses.
-    Returned auth objects need to be compatible with `requests.auth.AuthBase`.
-    """
-
-    def can_complete_http_basic(self) -> bool:
-        return False
-
-    def can_complete_mutualTLS(self) -> bool:
-        return False
-
-    def can_complete_oauth2_client_credentials(self, scopes: list[str]) -> bool:
-        return False
-
-    def can_complete_scheme(self, scheme: dict[str, t.Any], scopes: list[str]) -> bool:
-        if scheme["type"] == "http":
-            if scheme["scheme"] == "basic":
-                return self.can_complete_http_basic()
-        elif scheme["type"] == "mutualTLS":
-            return self.can_complete_mutualTLS()
-        elif scheme["type"] == "oauth2":
-            for flow_name, flow in scheme["flows"].items():
-                if (
-                    flow_name == "clientCredentials"
-                    and self.can_complete_oauth2_client_credentials(flow["scopes"])
-                ):
-                    return True
-        return False
-
-    def can_complete(
-        self, proposal: dict[str, list[str]], security_schemes: dict[str, dict[str, t.Any]]
-    ) -> bool:
-        for name, scopes in proposal.items():
-            scheme = security_schemes.get(name)
-            if scheme is None or not self.can_complete_scheme(scheme, scopes):
-                return False
-        # This covers the case where `[]` allows for no auth at all.
-        return True
-
-    async def http_basic_credentials(self) -> tuple[bytes, bytes]:
-        raise NotImplementedError()
-
-    async def oauth2_client_credentials(self) -> tuple[bytes, bytes]:
-        raise NotImplementedError()
-
-    def basic_auth(self, scopes: list[str]) -> requests.auth.AuthBase | None:
-        """Implement this to provide means of http basic auth."""
-        return None
-
-    def http_auth(
-        self, security_scheme: dict[str, t.Any], scopes: list[str]
-    ) -> requests.auth.AuthBase | None:
-        """Select a suitable http auth scheme or return None."""
-        # https://www.iana.org/assignments/http-authschemes/http-authschemes.xhtml
-        if security_scheme["scheme"] == "basic":
-            result = self.basic_auth(scopes)
-            if result:
-                return result
-        return None
-
-    def oauth2_client_credentials_auth(
-        self, flow: t.Any, scopes: list[str]
-    ) -> requests.auth.AuthBase | None:
-        """Implement this to provide other authentication methods."""
-        return None
-
-    def oauth2_auth(
-        self, security_scheme: dict[str, t.Any], scopes: list[str]
-    ) -> requests.auth.AuthBase | None:
-        """Select a suitable oauth2 flow or return None."""
-        # Check flows by preference.
-        if "clientCredentials" in security_scheme["flows"]:
-            flow = security_scheme["flows"]["clientCredentials"]
-            # Select this flow only if it claims to provide all the necessary scopes.
-            # This will allow subsequent auth proposals to be considered.
-            if set(scopes) - set(flow["scopes"]):
-                return None
-
-            result = self.oauth2_client_credentials_auth(flow, scopes)
-            if result:
-                return result
-        return None
-
-    def __call__(
-        self,
-        security: list[dict[str, list[str]]],
-        security_schemes: dict[str, dict[str, t.Any]],
-    ) -> requests.auth.AuthBase | None:
-
-        # Reorder the proposals by their type to prioritize properly.
-        # Select only single mechanism proposals on the way.
-        proposed_schemes: dict[str, dict[str, list[str]]] = defaultdict(dict)
-        for proposal in security:
-            if len(proposal) == 0:
-                # Empty proposal: No authentication needed. Shortcut return.
-                return None
-            if len(proposal) == 1:
-                name, scopes = list(proposal.items())[0]
-                proposed_schemes[security_schemes[name]["type"]][name] = scopes
-            # Ignore all proposals with more than one required auth mechanism.
-
-        # Check for auth schemes by preference.
-        if "oauth2" in proposed_schemes:
-            for name, scopes in proposed_schemes["oauth2"].items():
-                result = self.oauth2_auth(security_schemes[name], scopes)
-                if result:
-                    return result
-
-        # if we get here, either no-oauth2, OR we couldn't find creds
-        if "http" in proposed_schemes:
-            for name, scopes in proposed_schemes["http"].items():
-                result = self.http_auth(security_schemes[name], scopes)
-                if result:
-                    return result
-
-        raise OpenAPIError(_("No suitable auth scheme found."))
-
-
-class BasicAuthProvider(AuthProviderBase):
-    """
-    Implementation for AuthProviderBase providing basic auth with fixed `username`, `password`.
-    """
-
-    def __init__(self, username: t.AnyStr, password: t.AnyStr):
-        self.username: bytes = username.encode("latin1") if isinstance(username, str) else username
-        self.password: bytes = password.encode("latin1") if isinstance(password, str) else password
-        self.auth = requests.auth.HTTPBasicAuth(username, password)
-
-    def can_complete_http_basic(self) -> bool:
-        return True
-
-    async def http_basic_credentials(self) -> tuple[bytes, bytes]:
-        return self.username, self.password
-
-    def basic_auth(self, scopes: list[str]) -> requests.auth.AuthBase | None:
-        return self.auth
-
-
 class _Middleware:
     def __init__(
         self,
         openapi: "OpenAPI",
         security: t.Optional[t.List[t.Dict[str, t.List[str]]]],
     ):
         self._openapi = openapi
-        # self.method_spec  may be more interesting...
+        # Would be nicer to carry this with the request, but found no way:
         self._security = security
 
     async def __call__(
@@ -235,20 +93,20 @@ async def __call__(
                             await self._openapi._auth_provider.http_basic_credentials()
                         )
                         secret = b64encode(username + b":" + password)
-                        request.headers.add("Authorization", "Basic " + secret.decode())
+                        request.headers["Authorization"] = f"Basic {secret.decode()}"
                     else:
                         raise NotImplementedError("Auth scheme: http " + scheme["scheme"])
-                elif scheme["type"] == "mutualTLS":
-                    # At this point, we assume the cert has already been loaded into the sslcontext.
-                    pass
                 elif scheme["type"] == "oauth2":
                     flow = scheme["flows"].get("clientCredentials")
                     if flow is None:
                         raise NotImplementedError(
                             "OAuth2: Only client credential flow is available."
                         )
-                    token = "DEADBEEF"
-                    request.headers.add("Authorization", f"Bearer {token}")
+                    token = await self.oauth2_token(flow, request)
+                    request.headers["Authorization"] = f"Bearer {token}"
+                elif scheme["type"] == "mutualTLS":
+                    # At this point, we assume the cert has already been loaded into the sslcontext.
+                    pass
                 else:
                     raise NotImplementedError("Auth type: " + scheme["type"])
 
@@ -258,6 +116,29 @@ async def __call__(
             self._openapi._set_correlation_id(response.headers["Correlation-Id"])
         return response
 
+    async def oauth2_token(self, flow: dict[str, t.Any], request: aiohttp.ClientRequest) -> str:
+        # TODO implement retry the request with new token.
+        auth_provider = self._openapi._auth_provider
+        assert auth_provider is not None
+
+        now = datetime.now()
+        if auth_provider._oauth2_token is not None and auth_provider._oauth2_expires > now:
+            return auth_provider._oauth2_token
+
+        client_id, client_secret = await auth_provider.oauth2_client_credentials()
+        secret = b64encode(client_id + b":" + client_secret)
+        response = await request.session.post(
+            flow["tokenUrl"],
+            data={"grant_type": "client_credentials"},
+            headers={"Authorization": f"Basic {secret.decode()}"},
+            ssl=request.ssl,
+        )
+        response.raise_for_status()
+        result = await response.json()
+        auth_provider._oauth2_token = result["access_token"]
+        auth_provider._oauth2_expires = now + timedelta(seconds=result["expires_in"])
+        return auth_provider._oauth2_token
+
 
 class OpenAPI:
     """

pulp-glue/tests/conftest.py

@@ -4,8 +4,9 @@
 
 import pytest
 
+from pulp_glue.common.authentication import BasicAuthProvider
 from pulp_glue.common.context import PulpContext
-from pulp_glue.common.openapi import BasicAuthProvider, OpenAPI
+from pulp_glue.common.openapi import OpenAPI
 
 FAKE_OPENAPI_SPEC = json.dumps(
     {