Refactor internals of openapi class

This will mask attributes of the class private and add a bit of abstraction
from requests.

Author: mdellweg

lint_requirements.txt

@@ -7,6 +7,7 @@ mypy==1.19.0
 shellcheck-py==0.11.0.1
 
 # Type annotation stubs
+types-aiofiles
 types-pygments
 types-PyYAML
 types-requests

lower_bounds_constraints.lock

@@ -1,3 +1,5 @@
+aiofiles==25.1.0
+aiohttp==3.12.0
 click==8.0.0
 packaging==20.0
 PyYAML==5.3

pulp-glue/pulp_glue/common/authentication.py

@@ -4,6 +4,158 @@
 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 auth_success_hook(
+        self, proposal: dict[str, list[str]], security_schemes: dict[str, dict[str, t.Any]]
+    ) -> None:
+        pass
+
+    async def auth_failure_hook(
+        self, proposal: dict[str, list[str]], security_schemes: dict[str, dict[str, t.Any]]
+    ) -> bool:
+        """
+        Hook called on failed authentication.
+        Return True for retrying.
+        """
+        return False
+
+    async def http_basic_credentials(self) -> tuple[bytes, bytes]:
+        raise NotImplementedError()
+
+    async def oauth2_client_credentials(self) -> tuple[bytes, bytes]:
+        raise NotImplementedError()
+
+    def tls_credentials(self) -> tuple[str, str | None]:
+        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,
+        cert: str | None = None,
+        key: str | 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
+        self.cert: str | None = cert
+        self.key: str | None = key
+
+        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
+            )
+
+        if cert is None and key is not None:
+            raise RuntimeError("Key can only be used together with a cert.")
+
+    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
+
+    def can_complete_mutualTLS(self) -> bool:
+        return self.cert 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
+
+    def tls_credentials(self) -> tuple[str, str | None]:
+        assert self.cert is not None
+        return (self.cert, self.key)
+
+
+# ----------------------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 GlueAuthProvider
 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
@@ -335,8 +336,13 @@ def from_config(cls, config: dict[str, t.Any]) -> "t.Self":
         api_kwargs: dict[str, t.Any] = {
             "base_url": config["base_url"],
         }
-        if "username" in config:
-            api_kwargs["auth_provider"] = BasicAuthProvider(config["username"], config["password"])
+        api_kwargs["auth_provider"] = GlueAuthProvider(
+            **{
+                k: v
+                for k, v in config.items()
+                if k in {"username", "password", "client_id", "client_secret", "cert", "key"}
+            }
+        )
         if "headers" in config:
             api_kwargs["headers"] = dict(
                 (header.split(":", maxsplit=1) for header in config["headers"])
@@ -385,7 +391,9 @@ def api(self) -> OpenAPI:
                 # Deprecated for 'auth'.
                 if not password:
                     password = self.prompt("password", hide_input=True)
-                self._api_kwargs["auth_provider"] = BasicAuthProvider(username, password)
+                self._api_kwargs["auth_provider"] = GlueAuthProvider(
+                    username=username, password=password
+                )
                 warnings.warn(
                     "Using 'username' and 'password' with 'PulpContext' is deprecated. "
                     "Use an auth provider with the 'auth_provider' argument instead.",
@@ -399,10 +407,10 @@ def api(self) -> OpenAPI:
                 )
             except OpenAPIError as e:
                 raise PulpException(str(e))
+            self._patch_api_spec()
             # Rerun scheduled version checks
             for plugin_requirement in self._needed_plugins:
                 self.needs_plugin(plugin_requirement)
-            self._patch_api_spec()
         return self._api
 
     @property