Restructure auth

CHANGES/+memoize_auth.feature

@@ -0,0 +1 @@
+Added memoization to CLI auth provider. This helps to reuse a retrieved oauth token for the lifetime of the process.

CHANGES/+no_username_prompt.removal

@@ -0,0 +1,3 @@
+Removed the prompt for a username.
+Starting with this release the user needs to provide the username in the settings, or via `--username` to allow http basic auth.
+If no authentication is needed or another authentication mechanism should be used, it can be omitted.

CHANGES/+oauth2_client_credentials.feature

@@ -0,0 +1,2 @@
+Changed the way OAuth2 Client Credentials are provided to give the user some choice over the authentication to use.
+The new parameters `--client-id` and `--client-secret` were added and `--username`, `--password` are now restricted to HTTP Basic.

CHANGES/pulp-glue/+oauth2_token_basicauth.bugfix

@@ -0,0 +1 @@
+Use BasicAuth for token retrieval to comply with RFC6749.

pulp-glue/pulp_glue/common/authentication.py

@@ -17,8 +17,7 @@ def __init__(
         token_url: str,
         scopes: t.Optional[t.List[str]] = None,
     ):
-        self.client_id = client_id
-        self.client_secret = client_secret
+        self.token_auth = requests.auth.HTTPBasicAuth(client_id, client_secret)
         self.token_url = token_url
         self.scopes = scopes
 
@@ -76,15 +75,13 @@ def handle401(
 
     def retrieve_token(self) -> None:
         data = {
-            "client_id": self.client_id,
-            "client_secret": self.client_secret,
             "grant_type": "client_credentials",
         }
 
         if self.scopes:
             data["scope"] = " ".join(self.scopes)
 
-        response: requests.Response = requests.post(self.token_url, data=data)
+        response: requests.Response = requests.post(self.token_url, data=data, auth=self.token_auth)
 
         response.raise_for_status()
 

pulp-glue/tests/test_authentication.py

@@ -16,7 +16,7 @@ def raise_for_status(self):
         def json(self):
             return {"expires_in": 1, "access_token": "aaa"}
 
-    def _requests_post_mocked(url: str, data: t.Dict[str, t.Any]):
+    def _requests_post_mocked(url: str, data: t.Dict[str, t.Any], **kwargs: t.Any):
         assert "scope" not in data
         return OAuth2MockResponse()
 

pulp_cli/__init__.py

@@ -188,6 +188,8 @@ def main(
     headers: t.List[str],
     username: t.Optional[str],
     password: t.Optional[str],
+    client_id: t.Optional[str],
+    client_secret: t.Optional[str],
     cert: t.Optional[str],
     key: t.Optional[str],
     verify_ssl: bool,
@@ -206,8 +208,6 @@ def _debug_callback(level: int, x: str) -> None:
     api_kwargs = dict(
         base_url=base_url,
         headers=dict((header.split(":", maxsplit=1) for header in headers)),
-        username=username,
-        password=password,
         cert=cert,
         key=key,
         validate_certs=verify_ssl,
@@ -224,6 +224,10 @@ def _debug_callback(level: int, x: str) -> None:
         format=format,
         background_tasks=background,
         timeout=timeout,
+        username=username,
+        password=password,
+        oauth2_client_id=client_id,
+        oauth2_client_secret=client_secret,
     )
 
 

pulp_cli/config.py

@@ -29,6 +29,8 @@
     "headers",
     "username",
     "password",
+    "client_id",
+    "client_secret",
     "cert",
     "key",
     "verify_ssl",
@@ -59,6 +61,8 @@
     ),
     click.option("--username", default=None, help=_("Username on pulp server")),
     click.option("--password", default=None, help=_("Password on pulp server")),
+    click.option("--client-id", default=None, help=_("OAuth2 client ID")),
+    click.option("--client-secret", default=None, help=_("OAuth2 client secret")),
     click.option("--cert", default="", help=_("Path to client certificate")),
     click.option(
         "--key",
@@ -153,7 +157,11 @@ def validate_config(config: t.Dict[str, t.Any], strict: bool = False) -> None:
     if unknown_settings:
         errors.append(_("Unknown settings: '{}'.").format("','".join(unknown_settings)))
     if strict:
-        missing_settings = set(SETTINGS) - set(config.keys()) - {"plugins", "username", "password"}
+        missing_settings = (
+            set(SETTINGS)
+            - set(config.keys())
+            - {"plugins", "username", "password", "client_id", "client_secret"}
+        )
         if missing_settings:
             errors.append(_("Missing settings: '{}'.").format("','".join(missing_settings)))
     if errors:

pulpcore/cli/common/generic.py

@@ -132,9 +132,15 @@ def __init__(
         timeout: int,
         format: str,
         domain: str = "default",
+        username: t.Optional[str] = None,
+        password: t.Optional[str] = None,
+        oauth2_client_id: t.Optional[str] = None,
+        oauth2_client_secret: t.Optional[str] = None,
     ) -> None:
-        self.username = api_kwargs.pop("username", None)
-        self.password = api_kwargs.pop("password", None)
+        self.username = username
+        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)
         super().__init__(
@@ -170,6 +176,9 @@ def output_result(self, result: t.Any) -> None:
     class SecretStorageBasicAuth(requests.auth.AuthBase):
         def __init__(self, pulp_ctx: PulpCLIContext):
             self.pulp_ctx = pulp_ctx
+            assert self.pulp_ctx.username is not None
+            self.password: t.Optional[str] = None
+
             self.attr: t.Dict[str, str] = {
                 "service": "pulp-cli",
                 "base_url": self.pulp_ctx.api.base_url,
@@ -178,75 +187,97 @@ def __init__(self, pulp_ctx: PulpCLIContext):
             }
 
         def response_hook(self, response: requests.Response, **kwargs: t.Any) -> requests.Response:
+            # Example adapted from:
             # https://docs.python-requests.org/en/latest/_modules/requests/auth/#HTTPDigestAuth
             if 200 <= response.status_code < 300 and not self.password_in_manager:
                 if click.confirm(_("Add password to password manager?")):
-                    assert isinstance(self.pulp_ctx.password, str)
+                    assert isinstance(self.password, str)
 
                     with closing(secretstorage.dbus_init()) as connection:
                         collection = secretstorage.get_default_collection(connection)
                         collection.create_item(
-                            "Pulp CLI", self.attr, self.pulp_ctx.password.encode(), replace=True
+                            "Pulp CLI", self.attr, self.password.encode(), replace=True
                         )
             elif response.status_code == 401 and self.password_in_manager:
-                with closing(secretstorage.dbus_init()) as connection:
-                    collection = secretstorage.get_default_collection(connection)
-                    item = next(collection.search_items(self.attr), None)
-                    if item:
-                        if click.confirm(_("Remove failed password from password manager?")):
+                if click.confirm(_("Remove failed password from password manager?")):
+                    with closing(secretstorage.dbus_init()) as connection:
+                        collection = secretstorage.get_default_collection(connection)
+                        item = next(collection.search_items(self.attr), None)
+                        if item is not None:
                             item.delete()
-                            self.pulp_ctx.password = None
+                self.password = None
             return response
 
         def __call__(self, request: requests.PreparedRequest) -> requests.PreparedRequest:
-            with closing(secretstorage.dbus_init()) as connection:
-                collection = secretstorage.get_default_collection(connection)
-                item = next(collection.search_items(self.attr), None)
-                if item:
-                    self.pulp_ctx.password = item.get_secret().decode()
-                    self.password_in_manager = True
-                else:
-                    self.pulp_ctx.password = str(click.prompt("Password", hide_input=True))
-                    self.password_in_manager = False
+            assert self.pulp_ctx.username is not None
+            if self.password is None:
+                with closing(secretstorage.dbus_init()) as connection:
+                    collection = secretstorage.get_default_collection(connection)
+                    item = next(collection.search_items(self.attr), None)
+                    if item:
+                        self.password = item.get_secret().decode()
+                        self.password_in_manager = True
+                    else:
+                        self.password = str(click.prompt("Password", hide_input=True))
+                        self.password_in_manager = False
             request.register_hook("response", self.response_hook)  # type: ignore [no-untyped-call]
             return requests.auth.HTTPBasicAuth(  # type: ignore [no-any-return]
-                self.pulp_ctx.username,
-                self.pulp_ctx.password,
+                self.pulp_ctx.username, self.password
             )(request)
 
 
 class PulpCLIAuthProvider(AuthProviderBase):
     def __init__(self, pulp_ctx: PulpCLIContext):
         self.pulp_ctx = pulp_ctx
+        self._memoized: t.Dict[str, t.Optional[requests.auth.AuthBase]] = {}
 
     def basic_auth(self, scopes: t.List[str]) -> t.Optional[requests.auth.AuthBase]:
-        if self.pulp_ctx.username is None:
-            self.pulp_ctx.username = click.prompt("Username")
-        if self.pulp_ctx.password is None:
-            # TODO give the user a chance to opt out.
-            if SECRET_STORAGE:
-                # We could just try to fetch the password here,
-                # but we want to get a grip on the response_hook.
-                return SecretStorageBasicAuth(self.pulp_ctx)
+        if "BASIC_AUTH" not in self._memoized:
+            if self.pulp_ctx.username is None:
+                # No username -> No basic auth.
+                self._memoized["BASIC_AUTH"] = None
+            elif self.pulp_ctx.password is None:
+                # TODO give the user a chance to opt out.
+                if SECRET_STORAGE:
+                    # We could just try to fetch the password here,
+                    # but we want to get a grip on the response_hook.
+                    self._memoized["BASIC_AUTH"] = SecretStorageBasicAuth(self.pulp_ctx)
+                else:
+                    self._memoized["BASIC_AUTH"] = requests.auth.HTTPBasicAuth(
+                        self.pulp_ctx.username, click.prompt("Password", hide_input=True)
+                    )
             else:
-                self.pulp_ctx.password = click.prompt("Password", hide_input=True)
-        return requests.auth.HTTPBasicAuth(self.pulp_ctx.username, self.pulp_ctx.password)
+                self._memoized["BASIC_AUTH"] = requests.auth.HTTPBasicAuth(
+                    self.pulp_ctx.username, self.pulp_ctx.password
+                )
+        return self._memoized["BASIC_AUTH"]
 
     def oauth2_client_credentials_auth(
         self, flow: t.Any, scopes: t.List[str]
     ) -> 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=flow["tokenUrl"],
-            # Try to request all possible scopes.
-            scopes=flow["scopes"],
-        )
+        token_url = flow["tokenUrl"]
+        key = "OAUTH2_CLIENT_CREDENTIALS;" + token_url + ";" + ":".join(scopes)
+        if key not in self._memoized:
+            if self.pulp_ctx.oauth2_client_id is None:
+                # No client_id -> No oauth2 client credentials.
+                self._memoized[key] = None
+            elif self.pulp_ctx.oauth2_client_secret is None:
+                self._memoized[key] = OAuth2ClientCredentialsAuth(
+                    client_id=self.pulp_ctx.oauth2_client_id,
+                    client_secret=click.prompt("Client Secret"),
+                    token_url=flow["tokenUrl"],
+                    # Try to request all possible scopes.
+                    scopes=flow["scopes"],
+                )
+            else:
+                self._memoized[key] = OAuth2ClientCredentialsAuth(
+                    client_id=self.pulp_ctx.oauth2_client_id,
+                    client_secret=self.pulp_ctx.oauth2_client_secret,
+                    token_url=flow["tokenUrl"],
+                    # Try to request all possible scopes.
+                    scopes=flow["scopes"],
+                )
+        return self._memoized[key]
 
 
 ##############################################################################