Refactor LoginFlowManager interactions with GlobusApp

MANIFEST.in

@@ -1 +1 @@
-include src/globus_sdk/experimental/html_files/*
+include src/globus_sdk/experimental/login_flow_manager/local_server_login_flow_manager/html_files/*

changelog.d/20240730_020618_derek_refactor_login_flow_manager_interactions.rst

@@ -0,0 +1,29 @@
+
+Changed
+~~~~~~~
+
+-   Changed the experimental ``GlobusApp`` class in the following ways (:pr:`NUMBER`):
+
+    -   ``LoginFlowManagers`` now insert ``GlobusApp.app_name`` into any native
+        client login flows as the ``prefill_named_grant``.
+
+    -   ``GlobusAppConfig`` now accepts a ``login_redirect_uri`` parameter to specify
+        the redirect URI for a login flow.
+
+        -   Invalid when used with a ``LocalServerLoginFlowManager``.
+
+        -   Defaults to ``"https://auth.globus.org/v2/web/auth-code"`` for native
+            client flows. Raises an error if not set for confidential ones.
+
+    -   ``UserApp`` now allows for the use of confidential client flows with the use of
+        either a ``LocalServerLoginFlowManager`` or a configured ``login_redirect_uri``.
+
+    -   ``GlobusAppConfig.login_flow_manager`` now accepts shorthand string references
+        ``"command-line"`` to use a ``CommandLineLoginFlowManager`` and
+        ``"local-server"`` to use a ``LocalServerLoginFlowManager``.
+
+    -   ``GlobusAppConfig.login_flow_manager`` also now accepts a
+        ``LoginFlowManagerProvider``, a class with a
+        ``for_globus_app(...) -> LoginFlowManager`` class method.
+
+

src/globus_sdk/experimental/globus_app/globus_app.py

@@ -4,6 +4,7 @@
 import dataclasses
 import os
 import sys
+import typing as t
 from dataclasses import dataclass
 
 from globus_sdk import (
@@ -22,6 +23,7 @@
 )
 from globus_sdk.experimental.login_flow_manager import (
     CommandLineLoginFlowManager,
+    LocalServerLoginFlowManager,
     LoginFlowManager,
 )
 from globus_sdk.experimental.tokenstorage import JSONTokenStorage, TokenStorage
@@ -37,9 +39,9 @@
 from .errors import IdentityMismatchError, TokenValidationError
 
 if sys.version_info < (3, 8):
-    from typing_extensions import Protocol
+    from typing_extensions import Protocol, runtime_checkable
 else:
-    from typing import Protocol
+    from typing import Protocol, runtime_checkable
 
 
 def _default_filename(app_name: str, environment: str) -> str:
@@ -68,6 +70,14 @@ def _default_filename(app_name: str, environment: str) -> str:
         return os.path.expanduser(f"~/.globus/app/{app_name}/{filename}")
 
 
+@runtime_checkable
+class LoginFlowManagerProvider(Protocol):
+    @classmethod
+    def for_globus_app(
+        cls, app_name: str, login_client: AuthLoginClient, config: GlobusAppConfig
+    ) -> LoginFlowManager: ...
+
+
 class TokenValidationErrorHandler(Protocol):
     def __call__(self, app: GlobusApp, error: TokenValidationError) -> None: ...
 
@@ -88,31 +98,44 @@ def resolve_by_login_flow(app: GlobusApp, error: TokenValidationError) -> None:
     app.run_login_flow()
 
 
+KnownLoginFlowManager = t.Literal["command-line", "local-server"]
+KNOWN_LOGIN_FLOW_MANAGERS: dict[KnownLoginFlowManager, LoginFlowManagerProvider] = {
+    "command-line": CommandLineLoginFlowManager,
+    "local-server": LocalServerLoginFlowManager,
+}
+
+
 @dataclass(frozen=True)
 class GlobusAppConfig:
     """
     Various configuration options for controlling the behavior of a ``GlobusApp``.
 
-    :param login_flow_manager: an optional ``LoginFlowManager`` instance or class.
-        An instance will be used directly when driving app login flows. A class will
-        be initialized with the app's ``login_client`` and this config's
-        ``request_refresh_tokens``.
-        If not given the default behavior will depend on the type of ``GlobusApp``.
+    :param login_flow_manager: An optional ``LoginFlowManager`` instance, or provider,
+        or identifier ("command-line" or "local-server").
+        For a ``UserApp``, defaults to "command-line".
+        For a ``ClientApp``, this value is not supported.
+    :param login_redirect_uri: The redirect URI to use for login flows.
+        For a "local-server" login flow manager, this value is not supported.
+        For a native client, this value defaults to a globus-hosted helper page.
+        For a confidential client, this value is required.
+    :param request_refresh_tokens: If True, the ``GlobusApp`` will request refresh
+        tokens for long-lived access.
     :param token_storage: a ``TokenStorage`` instance or class that will
         be used for storing token data. If not passed a ``JSONTokenStorage``
         will be used.
-    :param request_refresh_tokens: If True, the ``GlobusApp`` will request refresh
-        tokens for long-lived access.
     :param token_validation_error_handler: A callable that will be called when a
         token validation error is encountered. The default behavior is to retry the
         login flow automatically.
     :param environment: The Globus environment being targeted by this app. This is
         predominately for internal use and can be ignored in most cases.
     """
 
-    login_flow_manager: LoginFlowManager | type[LoginFlowManager] | None = None
-    token_storage: TokenStorage | None = None
+    login_flow_manager: (
+        KnownLoginFlowManager | LoginFlowManagerProvider | LoginFlowManager | None
+    ) = None  # noqa: E501
+    login_redirect_uri: str | None = None
     request_refresh_tokens: bool = False
+    token_storage: TokenStorage | None = None
     token_validation_error_handler: TokenValidationErrorHandler | None = (
         resolve_by_login_flow
     )
@@ -366,33 +389,35 @@ def __init__(
             config=config,
         )
 
-        if client_secret or isinstance(self._login_client, ConfidentialAppAuthClient):
-            # UserApps need more information to be accepted to properly support
-            # Confidential Clients. Raise an error to avoid a confusing failure later.
-            raise GlobusSDKUsageError(
-                "UserApps don't currently support Confidential Clients"
-            )
+        self._login_flow_manager = self._resolve_login_flow_manager(
+            app_name=self.app_name,
+            login_client=self._login_client,
+            config=config,
+        )
 
-        # get or instantiate config's login_flow_manager
-        if self.config.login_flow_manager:
-            if isinstance(self.config.login_flow_manager, LoginFlowManager):
-                self._login_flow_manager = self.config.login_flow_manager
-            elif isinstance(self.config.login_flow_manager, type(LoginFlowManager)):
-                self._login_flow_manager = self.config.login_flow_manager(
-                    self._login_client,
-                    request_refresh_tokens=self.config.request_refresh_tokens,
-                )
-            else:
-                raise TypeError(
-                    "login_flow_manager must be a LoginFlowManager instance or class"
-                )
-        # or make a default CommandLineLoginFlowManager
+    def _resolve_login_flow_manager(
+        self, app_name: str, login_client: AuthLoginClient, config: GlobusAppConfig
+    ) -> LoginFlowManager:
+        login_flow_manager = config.login_flow_manager
+        if isinstance(login_flow_manager, LoginFlowManager):
+            return login_flow_manager
+
+        elif isinstance(login_flow_manager, LoginFlowManagerProvider):
+            provider = login_flow_manager
+        elif login_flow_manager is None:
+            provider = CommandLineLoginFlowManager
+        elif login_flow_manager in KNOWN_LOGIN_FLOW_MANAGERS:
+            provider = KNOWN_LOGIN_FLOW_MANAGERS[login_flow_manager]
         else:
-            self._login_flow_manager = CommandLineLoginFlowManager(
-                self._login_client,
-                request_refresh_tokens=self.config.request_refresh_tokens,
+            allowed_keys = ", ".join(repr(k) for k in KNOWN_LOGIN_FLOW_MANAGERS.keys())
+            raise GlobusSDKUsageError(
+                f"Unsupported login_flow_manager value: {login_flow_manager!r}. "
+                f"Expected {allowed_keys}, a <LoginFlowManagerProvider>, or a "
+                f"<LoginFlowManager>."
             )
 
+        return provider.for_globus_app(app_name, login_client, config)
+
     def _initialize_login_client(self, client_secret: str | None) -> None:
         if self.client_id is None:
             raise GlobusSDKUsageError(
@@ -479,8 +504,8 @@ def __init__(
         scope_requirements: dict[str, list[Scope]] | None = None,
         config: GlobusAppConfig = _DEFAULT_CONFIG,
     ):
-        if config and config.login_flow_manager is not None:
-            raise ValueError("a ClientApp cannot use a login_flow_manager")
+        if config.login_flow_manager is not None:
+            raise GlobusSDKUsageError("A ClientApp cannot use a login_flow_manager")
 
         if login_client and not isinstance(login_client, ConfidentialAppAuthClient):
             raise GlobusSDKUsageError(

src/globus_sdk/experimental/login_flow_manager/command_line_login_flow_manager.py

@@ -1,10 +1,23 @@
-from globus_sdk import AuthLoginClient, OAuthTokenResponse
+from __future__ import annotations
+
+import textwrap
+import typing as t
+
+from globus_sdk import (
+    AuthLoginClient,
+    ConfidentialAppAuthClient,
+    GlobusSDKUsageError,
+    OAuthTokenResponse,
+)
 from globus_sdk.experimental.auth_requirements_error import (
     GlobusAuthorizationParameters,
 )
 
 from .login_flow_manager import LoginFlowManager
 
+if t.TYPE_CHECKING:
+    from globus_sdk.experimental.globus_app import GlobusAppConfig
+
 
 class CommandLineLoginFlowManager(LoginFlowManager):
     """
@@ -25,7 +38,9 @@ def __init__(
         self,
         login_client: AuthLoginClient,
         *,
+        redirect_uri: str | None = None,
         request_refresh_tokens: bool = False,
+        native_prefill_named_grant: str | None = None,
         login_prompt: str = "Please authenticate with Globus here:",
         code_prompt: str = "Enter the resulting Authorization Code here:",
     ):
@@ -35,15 +50,55 @@ def __init__(
             must either be a NativeAppAuthClient or a templated
             ConfidentialAppAuthClient, standard ConfidentialAppAuthClients cannot
             use the web auth-code flow.
+        :param redirect_uri: The redirect URI to use for the login flow. Defaults to
+            a globus-hosted helper web auth-code URI for NativeAppAuthClients.
         :param request_refresh_tokens: Control whether refresh tokens will be requested.
+        :param native_prefill_named_grant: The named grant label to prefill on the
+            consent page when using a NativeAppAuthClient.
         :param login_prompt: The string that will be output to the command line
             prompting the user to authenticate.
         :param code_prompt: The string that will be output to the command line
             prompting the user to enter their authorization code.
         """
+        super().__init__(
+            login_client,
+            request_refresh_tokens=request_refresh_tokens,
+            native_prefill_named_grant=native_prefill_named_grant,
+        )
         self.login_prompt = login_prompt
         self.code_prompt = code_prompt
-        super().__init__(login_client, request_refresh_tokens=request_refresh_tokens)
+
+        if redirect_uri is None:
+            # Confidential clients must always define their own custom redirect URI.
+            if isinstance(login_client, ConfidentialAppAuthClient):
+                msg = "Use of a Confidential client requires an explicit redirect_uri."
+                raise GlobusSDKUsageError(msg)
+
+            # Native clients may infer the globus-provided helper page if omitted.
+            redirect_uri = login_client.base_url + "v2/web/auth-code"
+        self.redirect_uri = redirect_uri
+
+    @classmethod
+    def for_globus_app(
+        cls, app_name: str, login_client: AuthLoginClient, config: GlobusAppConfig
+    ) -> CommandLineLoginFlowManager:
+        """
+        Create a ``CommandLineLoginFlowManager`` for a given ``GlobusAppConfig``.
+
+        :param app_name: The name of the app to use for prefilling the named grant in
+            native auth flows.
+        :param login_client: The ``AuthLoginClient`` to use to drive Globus Auth flows.
+        :param config: A ``GlobusAppConfig`` to configure the login flow.
+        :returns: A ``CommandLineLoginFlowManager`` instance.
+        :raises: GlobusSDKUsageError if a login_redirect_uri is not set on the config
+            but a ConfidentialAppAuthClient is used.
+        """
+        return cls(
+            login_client,
+            redirect_uri=config.login_redirect_uri,
+            request_refresh_tokens=config.request_refresh_tokens,
+            native_prefill_named_grant=app_name,
+        )
 
     def run_login_flow(
         self,
@@ -55,37 +110,23 @@ def run_login_flow(
         :param auth_parameters: ``GlobusAuthorizationParameters`` passed through
             to the authentication flow to control how the user will authenticate.
         """
+        authorize_url = self._get_authorize_url(auth_parameters, self.redirect_uri)
+        auth_code = self._print_and_prompt(authorize_url)
 
-        # type is ignored here as AuthLoginClient does not provide a signature for
-        # oauth2_start_flow since it has different positional arguments between
-        # NativeAppAuthClient and ConfidentialAppAuthClient
-        self.login_client.oauth2_start_flow(  # type: ignore
-            redirect_uri=self.login_client.base_url + "v2/web/auth-code",
-            refresh_tokens=self.request_refresh_tokens,
-            requested_scopes=auth_parameters.required_scopes,
-        )
+        return self.login_client.oauth2_exchange_code_for_tokens(auth_code)
 
-        # create authorization url and prompt user to follow it to login
+    def _print_and_prompt(self, authorize_url: str) -> str:
+        # Prompt the user to authenticate
         print(
-            "{0}\n{1}\n{2}\n{1}\n".format(
-                self.login_prompt,
-                "-" * len(self.login_prompt),
-                self.login_client.oauth2_get_authorize_url(
-                    session_required_identities=(
-                        auth_parameters.session_required_identities
-                    ),
-                    session_required_single_domain=(
-                        auth_parameters.session_required_single_domain
-                    ),
-                    session_required_policies=auth_parameters.session_required_policies,
-                    session_required_mfa=auth_parameters.session_required_mfa,
-                    prompt=auth_parameters.prompt,  # type: ignore
-                ),
+            textwrap.dedent(
+                f"""
+                {self.login_prompt}
+                {"-" * len(self.login_prompt)}
+                {authorize_url}
+                {"-" * len(self.login_prompt)}
+                """
             )
         )
 
-        # ask user to copy and paste auth code
-        auth_code = input(f"{self.code_prompt}\n").strip()
-
-        # get and return tokens
-        return self.login_client.oauth2_exchange_code_for_tokens(auth_code)
+        # Request they to enter the auth code
+        return input(f"{self.code_prompt}\n").strip()

src/globus_sdk/experimental/login_flow_manager/local_server_login_flow_manager/__init__.py

@@ -0,0 +1,5 @@
+from .local_server_login_flow_manager import LocalServerLoginFlowManager
+
+__all__ = [
+    "LocalServerLoginFlowManager",
+]

src/globus_sdk/experimental/login_flow_manager/_local_server.py → src/globus_sdk/experimental/login_flow_manager/local_server_login_flow_manager/_local_server.py

@@ -22,7 +22,7 @@
 else:  # Python < 3.9
     import importlib_resources
 
-from globus_sdk.experimental import html_files
+import globus_sdk.experimental.login_flow_manager.local_server_login_flow_manager.html_files as html_files  # noqa: E501
 
 _IS_WINDOWS = os.name == "nt"