WIP

Author: mdellweg

pulp-glue/pulp_glue/common/openapi.py

@@ -67,6 +67,46 @@ class AuthProviderBase:
     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
@@ -145,11 +185,17 @@ class BasicAuthProvider(AuthProviderBase):
     Implementation for AuthProviderBase providing basic auth with fixed `username`, `password`.
     """
 
-    def __init__(self, username: str, password: str):
-        self.username = username
-        self.password = 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
 
@@ -171,22 +217,41 @@ async def __call__(
     ) -> aiohttp.ClientResponse:
         if self._security:
             assert self._openapi._auth_provider is not None
-            auth = self._openapi._auth_provider(
-                self._security, self._openapi.api_spec["components"]["securitySchemes"]
-            )
-            if isinstance(auth, requests.auth.HTTPBasicAuth):
-                username = (
-                    auth.username.encode("latin1")
-                    if isinstance(auth.username, str)
-                    else auth.username
-                )
-                password = (
-                    auth.password.encode("latin1")
-                    if isinstance(auth.password, str)
-                    else auth.password
-                )
-                secret = b64encode(username + b":" + password)
-                request.headers["Authorization"] = "Basic " + secret.decode()
+            security_schemes: dict[str, dict[str, t.Any]] = self._openapi.api_spec["components"][
+                "securitySchemes"
+            ]
+            for proposal in self._security:
+                if self._openapi._auth_provider.can_complete(proposal, security_schemes):
+                    break
+            else:
+                raise OpenAPIError(_("No suitable auth scheme found."))
+
+            assert proposal is not None
+            for scheme_name, scopes in proposal.items():
+                scheme = security_schemes[scheme_name]
+                if scheme["type"] == "http":
+                    if scheme["scheme"] == "basic":
+                        username, password = (
+                            await self._openapi._auth_provider.http_basic_credentials()
+                        )
+                        secret = b64encode(username + b":" + password)
+                        request.headers.add("Authorization", "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}")
+                else:
+                    raise NotImplementedError("Auth type: " + scheme["type"])
+
         response = await handler(request)
 
         if "Correlation-Id" in response.headers:

pulp-glue/tests/test_auth_provider.py

@@ -1,10 +1,11 @@
 import typing as t
+import asyncio
 
 import pytest
 from requests.auth import AuthBase
 
 from pulp_glue.common.exceptions import OpenAPIError
-from pulp_glue.common.openapi import AuthProviderBase
+from pulp_glue.common.openapi import AuthProviderBase, BasicAuthProvider
 
 pytestmark = pytest.mark.glue
 
@@ -51,9 +52,34 @@
             },
         },
     },
+    "E": {"type": "mutualTLS"},
 }
 
 
+class TestBasicAuthProvider:
+    @pytest.fixture(scope="class")
+    def provider(self) -> AuthProviderBase:
+        return BasicAuthProvider(username="user1", password="password1")
+
+    def test_can_complete_basic(self, provider: AuthProviderBase) -> None:
+        assert provider.can_complete_http_basic()
+
+    def test_provides_username_and_password(self, provider: AuthProviderBase) -> None:
+        assert asyncio.run(provider.http_basic_credentials()) == (b"user1", b"password1")
+
+    def test_cannot_complete_mutualTLS(self, provider: AuthProviderBase) -> None:
+        assert not provider.can_complete_mutualTLS()
+
+    def test_can_complete_basic_proposal(self, provider: AuthProviderBase) -> None:
+        assert provider.can_complete({"B": []}, security_schemes=SECURITY_SCHEMES)
+
+    def test_cannot_complete_bearer_proposal(self, provider: AuthProviderBase) -> None:
+        assert not provider.can_complete({"A": []}, security_schemes=SECURITY_SCHEMES)
+
+    def test_cannot_complete_combined_proposal(self, provider: AuthProviderBase) -> None:
+        assert not provider.can_complete({"A": [], "B": []}, security_schemes=SECURITY_SCHEMES)
+
+
 class MockBasicAuth(AuthBase):
     pass
 

pulp_cli/generic.py

@@ -160,8 +160,7 @@ def __init__(
         self.password = password
         self.oauth2_client_id = oauth2_client_id
         self.oauth2_client_secret = oauth2_client_secret
-        if not api_kwargs.get("cert"):
-            api_kwargs["auth_provider"] = PulpCLIAuthProvider(pulp_ctx=self)
+        api_kwargs["auth_provider"] = PulpCLIAuthProvider(pulp_ctx=self)
 
         verify_ssl: bool | None = api_kwargs.pop("verify_ssl", None)
         super().__init__(
@@ -253,6 +252,28 @@ def __init__(self, pulp_ctx: PulpCLIContext):
         self.pulp_ctx = pulp_ctx
         self._memoized: dict[str, requests.auth.AuthBase | None] = {}
 
+    def can_complete_http_basic(self) -> bool:
+        return self.pulp_ctx.username is not None
+
+    def can_complete_oauth2_client_credentials(self, scopes: list[str]) -> bool:
+        return self.pulp_ctx.oauth2_client_id is not None
+
+    async def http_basic_credentials(self) -> tuple[bytes, bytes]:
+        username = (
+            self.pulp_ctx.username.encode("latin1")
+            if isinstance(self.pulp_ctx.username, str)
+            else self.pulp_ctx.username
+        )
+        password = (
+            self.pulp_ctx.password.encode("latin1")
+            if isinstance(self.pulp_ctx.password, str)
+            else self.pulp_ctx.password
+        )
+        return username, password
+
+    async def oauth2_client_credentials(self, scopes: list[str]) -> tuple[str, str]:
+        return self.pulp_ctx.client_id, self.pulp_ctx.client_secret
+
     def basic_auth(self, scopes: list[str]) -> requests.auth.AuthBase | None:
         if "BASIC_AUTH" not in self._memoized:
             if self.pulp_ctx.username is None: