Add auth to api kwargs

All requests auth class compatible auth mechanisms can now be passed to the
PulpContext.

[noissue]

CHANGES/pulp-glue/+pluggable_auth.feature

@@ -0,0 +1 @@
+Added `auth` to `apikwargs` so you can plug in any `requests.auth.AuthBase`.

pulp-glue/pulp_glue/common/context.py

@@ -4,6 +4,7 @@
 import sys
 import time
 import typing as t
+import warnings
 
 from packaging.specifiers import SpecifierSet
 from requests import HTTPError
@@ -306,8 +307,19 @@ def api(self) -> OpenAPI:
         All calls to the API should be performed via `call`.
         """
         if self._api is None:
-            if self._api_kwargs.get("username") and not self._api_kwargs.get("password"):
-                self._api_kwargs["password"] = self.prompt("password", hide_input=True)
+            if self._api_kwargs.get("username"):
+                # Deprecated for 'auth'.
+                if not self._api_kwargs.get("password"):
+                    self._api_kwargs["password"] = self.prompt("password", hide_input=True)
+                self._api_kwargs["auth"] = (
+                    self._api_kwargs.pop("username"),
+                    self._api_kwargs.pop("password", None),
+                )
+                warnings.warn(
+                    "Using 'username' and 'password' with 'PulpContext' is deprecated. "
+                    "Use a requests auth class with the 'auth' argument instead.",
+                    DeprecationWarning,
+                )
             try:
                 self._api = OpenAPI(
                     doc_path=f"{self._api_root}api/v3/docs/api.json", **self._api_kwargs

pulp-glue/pulp_glue/common/openapi.py

@@ -52,8 +52,7 @@ class OpenAPI:
         base_url: The base URL inlcuding the HTTP scheme, hostname and optional subpaths of the
             served api.
         doc_path: Path of the json api doc schema relative to the `base_url`.
-        username: Username used for basic auth.
-        password: Password used for basic auth.
+        auth: A requests compatible auth object.
         cert: Client certificate used for auth.
         key: Matching key for `cert` if not already included.
         validate_certs: Whether to check server TLS certificates agains a CA.
@@ -68,8 +67,7 @@ def __init__(
         self,
         base_url: str,
         doc_path: str,
-        username: t.Optional[str] = None,
-        password: t.Optional[str] = None,
+        auth: t.Optional[t.Union[t.Tuple[str, str], requests.auth.AuthBase]] = None,
         cert: t.Optional[str] = None,
         key: t.Optional[str] = None,
         validate_certs: bool = True,
@@ -88,20 +86,17 @@ def __init__(
         self.safe_calls_only: bool = safe_calls_only
 
         self._session: requests.Session = requests.session()
-        if username and password:
+        if auth:
             if cert or key:
-                raise OpenAPIError(_("Cannot use both username/password and cert auth."))
-            self._session.auth = (username, password)
-        elif username:
-            raise OpenAPIError(_("Password is required if username is set."))
-        elif password:
-            raise OpenAPIError(_("Username is required if password is set."))
-        elif cert and key:
-            self._session.cert = (cert, key)
-        elif cert:
-            self._session.cert = cert
-        elif key:
-            raise OpenAPIError(_("Cert is required if key is set."))
+                raise OpenAPIError(_("Cannot use both 'auth' and 'cert'."))
+            self._session.auth = auth
+        else:
+            if cert and key:
+                self._session.cert = (cert, key)
+            elif cert:
+                self._session.cert = cert
+            elif key:
+                raise OpenAPIError(_("Cert is required if key is set."))
         self._session.headers.update(
             {
                 "User-Agent": user_agent or f"Pulp-glue openapi parser ({__version__})",

pulpcore/cli/common/config.py

@@ -42,8 +42,8 @@
         help=_("Absolute API base path on server (not including 'api/v3/')"),
     ),
     click.option("--domain", default="default", help=_("Domain to work in if feature is enabled")),
-    click.option("--username", default="", help=_("Username on pulp server")),
-    click.option("--password", default="", help=_("Password on pulp server")),
+    click.option("--username", default=None, help=_("Username on pulp server")),
+    click.option("--password", default=None, help=_("Password on pulp server")),
     click.option("--cert", default="", help=_("Path to client certificate")),
     click.option(
         "--key",
@@ -110,7 +110,7 @@ def validate_config(config: Dict[str, Any], strict: bool = False) -> bool:
     if unknown_settings:
         errors.append(_("Unknown settings: '{}'.").format("','".join(unknown_settings)))
     if strict:
-        missing_settings = set(SETTINGS) - set(config.keys())
+        missing_settings = set(SETTINGS) - set(config.keys()) - {"username", "password"}
         if missing_settings:
             errors.append(_("Missing settings: '{}'.").format("','".join(missing_settings)))
     if errors:

pulpcore/cli/common/debug.py

@@ -62,6 +62,24 @@ def spec(pulp_ctx: PulpCLIContext) -> None:
     pulp_ctx.output_result(pulp_ctx.api.api_spec)
 
 
+@openapi_group.command()
+@pass_pulp_context
+def info(pulp_ctx: PulpCLIContext) -> None:
+    """
+    Print info block.
+    """
+    pulp_ctx.output_result(pulp_ctx.api.api_spec["info"])
+
+
+@openapi_group.command()
+@pass_pulp_context
+def security_schemes(pulp_ctx: PulpCLIContext) -> None:
+    """
+    Print info block.
+    """
+    pulp_ctx.output_result(pulp_ctx.api.api_spec["components"]["securitySchemes"])
+
+
 @openapi_group.command()
 @click.option("--id", "operation_id", required=True, help=_("Operation ID in openapi schema"))
 @pass_pulp_context

pulpcore/cli/common/generic.py

@@ -5,6 +5,7 @@
 from functools import lru_cache, wraps
 
 import click
+import requests
 import schema as s
 import yaml
 from pulp_glue.common.context import (
@@ -93,6 +94,9 @@ def __init__(
         format: str,
         domain: str = "default",
     ) -> None:
+        self.username = api_kwargs.pop("username", None)
+        self.password = api_kwargs.pop("password", None)
+        api_kwargs["auth"] = PulpCLIAuth(pulp_ctx=self)
         super().__init__(
             api_root=api_root,
             api_kwargs=api_kwargs,
@@ -105,9 +109,6 @@ def __init__(
     def echo(self, message: str, nl: bool = True, err: bool = False) -> None:
         click.echo(message, nl=nl, err=err)
 
-    def prompt(self, text: str, hide_input: bool = False) -> t.Any:
-        return click.prompt(text, hide_input=hide_input)
-
     def output_result(self, result: t.Any) -> None:
         """
         Dump the provided result to the console using the selected renderer.
@@ -133,6 +134,20 @@ def output_result(self, result: t.Any) -> None:
             )
 
 
+class PulpCLIAuth(requests.auth.AuthBase):
+    def __init__(self, pulp_ctx: PulpCLIContext) -> None:
+        super().__init__()
+        self.pulp_ctx = pulp_ctx
+
+    def __call__(self, request: requests.PreparedRequest) -> requests.PreparedRequest:
+        if self.pulp_ctx.username is None:
+            return request
+        if self.pulp_ctx.password is None:
+            self.pulp_ctx.password = click.prompt("Password", hide_input=True)
+        wrapped_auth = requests.auth.HTTPBasicAuth(self.pulp_ctx.username, self.pulp_ctx.password)
+        return wrapped_auth(request)  # type: ignore
+
+
 ##############################################################################
 # Decorator to access certain contexts