WIP: Async support in pulp-glue

Replaces requests with aiohttp and changes the api.

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

@@ -1,97 +1,147 @@
 import typing as t
-from datetime import datetime, timedelta
+from datetime import datetime
 
-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
+class AuthProviderBase:
     """
+    Base class for auth providers.
 
-    def __init__(
-        self,
-        client_id: str,
-        client_secret: str,
-        token_url: str,
-        scopes: list[str] | None = None,
-        verify_ssl: str | bool | None = None,
-    ):
-        self._token_server_auth = requests.auth.HTTPBasicAuth(client_id, client_secret)
-        self._token_url = token_url
-        self._scopes = scopes
-        self._verify_ssl = verify_ssl
+    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`.
+    """
 
-        self._access_token: str | None = None
-        self._expire_at: datetime | None = None
+    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]]
+    ) -> None:
+        pass
+
+    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 __call__(self, request: requests.PreparedRequest) -> requests.PreparedRequest:
-        if self._expire_at is None or self._expire_at < datetime.now():
-            self._retrieve_token()
+    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
 
-        assert self._access_token is not None
+    def can_complete_http_basic(self) -> bool:
+        return True
 
-        request.headers["Authorization"] = f"Bearer {self._access_token}"
+    async def http_basic_credentials(self) -> tuple[bytes, bytes]:
+        return self.username, self.password
 
-        # Call to untyped function "register_hook" in typed context
-        request.register_hook("response", self._handle401)  # type: ignore[no-untyped-call]
 
-        return request
+class GlueAuthProvider(AuthProviderBase):
+    """
+    AuthProvider allowing to be used with prepared credentials.
+    """
 
-    def _handle401(
+    def __init__(
         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 = {
-            "grant_type": "client_credentials",
-        }
-
-        if self._scopes:
-            data["scope"] = " ".join(self._scopes)
-
-        response: requests.Response = requests.post(
-            self._token_url,
-            data=data,
-            auth=self._token_server_auth,
-            verify=self._verify_ssl,
-        )
-
-        response.raise_for_status()
-
-        token = response.json()
-        self._expire_at = datetime.now() + timedelta(seconds=token["expires_in"])
-        self._access_token = token["access_token"]
+        *,
+        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)

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
@@ -202,6 +203,15 @@ def patch_upstream_pulp_replicate_request_body(api: OpenAPI) -> None:
     operation.pop("requestBody", None)
 
 
+@api_quirk(PluginRequirement("core", specifier="<3.85"))
+def patch_security_scheme_mutual_tls(api: OpenAPI) -> None:
+    api.api_spec["components"]["securitySchemes"]["gluePatchTLS"] = {"type": "mutualTLS"}
+    for method, path in api.operations.values():
+        operation = api.api_spec["paths"][path][method]
+        if "security" in operation:
+            operation["security"].append({"gluePatchTLS": []})
+
+
 class PulpContext:
     """
     Abstract class for the global PulpContext object.
@@ -335,8 +345,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 +400,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 +416,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