Add response subtypes for different oauth2 grants

The initial goal of this work is to allow a function handling a token
response, e.g., in a tokenstorage interface, to determine the context
in which the original call was made. It is now possible to check

    if isinstance(response, RefreshTokenResponse): ...

in a generic token handler.

Additionally, this opens the way for distinct token response objects
to implement methods which vary based on the known properties of their
respective grant types. For example, RefreshTokenResponse could
raise a more informative error when `decode_id_token` is called and
no `id_token` is present.

changelog.d/20240917_121333_sirosen_add_token_response_classes.rst

@@ -0,0 +1,17 @@
+Changed
+~~~~~~~
+
+- The response types for different OAuth2 token grants now vary by the grant
+  type. For example, a ``refresh_token`` grant will now produce a
+  ``RefreshTokenResponse``. This allows code handling responses to identify
+  which grant type was used to produce a response. (:pr:`NUMBER`)
+
+  - The following new types have been introduced:
+    ``globus_sdk.RefreshTokenResponse``,
+    ``globus_sdk.AuthorizationCodeTokenResponse``,
+    ``globus_sdk.ClientCredentialsTokenResponse``.
+
+  - The ``RenewingAuthorizer`` class is now a generic over the response type
+    which it handles, and the subtypes of authorizers are specialized for their
+    types of responses. e.g.,
+    ``class RefreshTokenAuthorizer(RenewingAuthorizer[RefreshTokenResponse])``.

docs/services/auth.rst

@@ -65,6 +65,18 @@ Auth Responses
    :members:
    :show-inheritance:
 
+.. autoclass:: AuthorizationCodeTokenResponse
+   :members:
+   :show-inheritance:
+
+.. autoclass:: RefreshTokenResponse
+   :members:
+   :show-inheritance:
+
+.. autoclass:: ClientCredentialsTokenResponse
+   :members:
+   :show-inheritance:
+
 .. autoclass:: GetConsentsResponse
    :members:
    :show-inheritance:

src/globus_sdk/__init__.py

@@ -63,6 +63,9 @@ def _force_eager_imports() -> None:
         "GetIdentitiesResponse",
         "OAuthDependentTokenResponse",
         "OAuthTokenResponse",
+        "AuthorizationCodeTokenResponse",
+        "ClientCredentialsTokenResponse",
+        "RefreshTokenResponse",
         "DependentScopeSpec",
     },
     "services.gcs": {
@@ -185,6 +188,9 @@ def _force_eager_imports() -> None:
     from .services.auth import GetIdentitiesResponse
     from .services.auth import OAuthDependentTokenResponse
     from .services.auth import OAuthTokenResponse
+    from .services.auth import AuthorizationCodeTokenResponse
+    from .services.auth import ClientCredentialsTokenResponse
+    from .services.auth import RefreshTokenResponse
     from .services.auth import DependentScopeSpec
     from .services.gcs import CollectionDocument
     from .services.gcs import GCSAPIError
@@ -292,6 +298,7 @@ def __getattr__(name: str) -> t.Any:
     "AuthAPIError",
     "AuthClient",
     "AuthLoginClient",
+    "AuthorizationCodeTokenResponse",
     "AzureBlobStoragePolicies",
     "BaseClient",
     "BasicAuthorizer",
@@ -300,6 +307,7 @@ def __getattr__(name: str) -> t.Any:
     "BoxStoragePolicies",
     "CephStoragePolicies",
     "ClientCredentialsAuthorizer",
+    "ClientCredentialsTokenResponse",
     "CollectionDocument",
     "CollectionPolicies",
     "ConfidentialAppAuthClient",
@@ -361,6 +369,7 @@ def __getattr__(name: str) -> t.Any:
     "POSIXStoragePolicies",
     "RecurringTimerSchedule",
     "RefreshTokenAuthorizer",
+    "RefreshTokenResponse",
     "RemovedInV4Warning",
     "S3StoragePolicies",
     "Scope",

src/globus_sdk/_generate_init.py

@@ -125,6 +125,9 @@ def __getattr__(name: str) -> t.Any:
             "GetIdentitiesResponse",
             "OAuthDependentTokenResponse",
             "OAuthTokenResponse",
+            "AuthorizationCodeTokenResponse",
+            "ClientCredentialsTokenResponse",
+            "RefreshTokenResponse",
             # API data helpers
             "DependentScopeSpec",
         ),

src/globus_sdk/authorizers/client_credentials.py

@@ -3,18 +3,18 @@
 import logging
 import typing as t
 
+import globus_sdk
 from globus_sdk._types import ScopeCollectionType
 from globus_sdk.scopes import scopes_to_str
 
 from .renewing import RenewingAuthorizer
 
-if t.TYPE_CHECKING:
-    from globus_sdk.services.auth import ConfidentialAppAuthClient, OAuthTokenResponse
-
 log = logging.getLogger(__name__)
 
 
-class ClientCredentialsAuthorizer(RenewingAuthorizer):
+class ClientCredentialsAuthorizer(
+    RenewingAuthorizer["globus_sdk.ClientCredentialsTokenResponse"]
+):
     r"""
     Implementation of a RenewingAuthorizer that renews confidential app client
     Access Tokens using a ConfidentialAppAuthClient and a set of scopes to
@@ -47,22 +47,24 @@ class ClientCredentialsAuthorizer(RenewingAuthorizer):
         POSIX timestamp (i.e. seconds since the epoch)
     :param on_refresh: A callback which is triggered any time this authorizer fetches a
         new access_token. The ``on_refresh`` callable is invoked on the
-        :class:`OAuthTokenResponse <globus_sdk.OAuthTokenResponse>`
-        object resulting from the token being refreshed. It should take only one
-        argument, the token response object.
+        :class:`globus_sdk.ClientCredentialsTokenResponse` object resulting from the
+        token being refreshed. It should take only one argument, the token response
+        object.
         This is useful for implementing storage for Access Tokens, as the
         ``on_refresh`` callback can be used to update the Access Tokens and
         their expiration times.
     """
 
     def __init__(
         self,
-        confidential_client: ConfidentialAppAuthClient,
+        confidential_client: globus_sdk.ConfidentialAppAuthClient,
         scopes: ScopeCollectionType,
         *,
         access_token: str | None = None,
         expires_at: int | None = None,
-        on_refresh: None | t.Callable[[OAuthTokenResponse], t.Any] = None,
+        on_refresh: (
+            None | t.Callable[[globus_sdk.ClientCredentialsTokenResponse], t.Any]
+        ) = None,
     ):
         # values for _get_token_data
         self.confidential_client = confidential_client
@@ -75,15 +77,17 @@ def __init__(
 
         super().__init__(access_token, expires_at, on_refresh)
 
-    def _get_token_response(self) -> OAuthTokenResponse:
+    def _get_token_response(self) -> globus_sdk.ClientCredentialsTokenResponse:
         """
-        Make a client credentials grant
+        Make a client credentials request for new tokens.
         """
         return self.confidential_client.oauth2_client_credentials_tokens(
             requested_scopes=self.scopes
         )
 
-    def _extract_token_data(self, res: OAuthTokenResponse) -> dict[str, t.Any]:
+    def _extract_token_data(
+        self, res: globus_sdk.ClientCredentialsTokenResponse
+    ) -> dict[str, t.Any]:
         """
         Get the tokens .by_resource_server,
         Ensure that only one token was gotten, and return that token.

src/globus_sdk/authorizers/refresh_token.py

@@ -10,7 +10,7 @@
 log = logging.getLogger(__name__)
 
 
-class RefreshTokenAuthorizer(RenewingAuthorizer):
+class RefreshTokenAuthorizer(RenewingAuthorizer["globus_sdk.RefreshTokenResponse"]):
     """
     Implements Authorization using a Refresh Token to periodically fetch
     renewed Access Tokens. It may be initialized with an Access Token, or it
@@ -39,9 +39,8 @@ class RefreshTokenAuthorizer(RenewingAuthorizer):
         POSIX timestamp (i.e. seconds since the epoch)
     :param on_refresh: A callback which is triggered any time this authorizer fetches a
         new access_token. The ``on_refresh`` callable is invoked on the
-        :class:`OAuthTokenResponse <globus_sdk.OAuthTokenResponse>`
-        object resulting from the token being refreshed. It should take only one
-        argument, the token response object.
+        :class:`globus_sdk.RefreshTokenResponse` object resulting from the token being
+        refreshed. It should take only one argument, the token response object.
         This is useful for implementing storage for Access Tokens, as the
         ``on_refresh`` callback can be used to update the Access Tokens and
         their expiration times.
@@ -54,7 +53,7 @@ def __init__(
         *,
         access_token: str | None = None,
         expires_at: int | None = None,
-        on_refresh: None | t.Callable[[globus_sdk.OAuthTokenResponse], t.Any] = None,
+        on_refresh: None | t.Callable[[globus_sdk.RefreshTokenResponse], t.Any] = None,
     ):
         log.info(
             "Setting up RefreshTokenAuthorizer with auth_client="
@@ -78,14 +77,14 @@ def __init__(
 
         super().__init__(access_token, expires_at, on_refresh)
 
-    def _get_token_response(self) -> globus_sdk.OAuthTokenResponse:
+    def _get_token_response(self) -> globus_sdk.RfreshTokenResponse:
         """
         Make a refresh token grant
         """
         return self.auth_client.oauth2_refresh_token(self.refresh_token)
 
     def _extract_token_data(
-        self, res: globus_sdk.OAuthTokenResponse
+        self, res: globus_sdk.RefreshTokenResponse
     ) -> dict[str, t.Any]:
         """
         Get the tokens .by_resource_server,

src/globus_sdk/authorizers/renewing.py

@@ -17,8 +17,12 @@
 # possible delays or clock skew.
 EXPIRES_ADJUST_SECONDS = 60
 
+# the type of the response which is produced by the authorizer, received by it, and
+# passed to the `on_refresh` callback
+ResponseT = t.TypeVar("ResponseT", bound="OAuthTokenResponse")
 
-class RenewingAuthorizer(GlobusAuthorizer, metaclass=abc.ABCMeta):
+
+class RenewingAuthorizer(GlobusAuthorizer, t.Generic[ResponseT], metaclass=abc.ABCMeta):
     r"""
     A ``RenewingAuthorizer`` is an abstract superclass to any authorizer
     that needs to get new Access Tokens in order to form Authorization headers.
@@ -38,8 +42,7 @@ class RenewingAuthorizer(GlobusAuthorizer, metaclass=abc.ABCMeta):
     :param expires_at: Expiration time for the starting ``access_token`` expressed as a
         POSIX timestamp (i.e. seconds since the epoch)
     :param on_refresh: A callback which is triggered any time this authorizer fetches a
-        new access_token. The ``on_refresh`` callable is invoked on the
-        :class:`OAuthTokenResponse <globus_sdk.OAuthTokenResponse>`
+        new access_token. The ``on_refresh`` callable is invoked on the response
         object resulting from the token being refreshed. It should take only one
         argument, the token response object.
         This is useful for implementing storage for Access Tokens, as the
@@ -51,8 +54,8 @@ def __init__(
         self,
         access_token: str | None = None,
         expires_at: int | None = None,
-        on_refresh: None | t.Callable[[OAuthTokenResponse], t.Any] = None,
-    ):
+        on_refresh: None | t.Callable[[ResponseT], t.Any] = None,
+    ) -> None:
         self._access_token = None
         self._access_token_hash = None
 
@@ -97,14 +100,14 @@ def access_token(self, value: str | None) -> None:
             self._access_token_hash = utils.sha256_string(value)
 
     @abc.abstractmethod
-    def _get_token_response(self) -> OAuthTokenResponse:
+    def _get_token_response(self) -> ResponseT:
         """
         Using whatever method the specific authorizer implementing this class
         does, get a new token response.
         """
 
     @abc.abstractmethod
-    def _extract_token_data(self, res: OAuthTokenResponse) -> dict[str, t.Any]:
+    def _extract_token_data(self, res: ResponseT) -> dict[str, t.Any]:
         """
         Given a token response object, get the first element of
         token_response.by_resource_server

src/globus_sdk/services/auth/__init__.py

@@ -12,10 +12,13 @@
 )
 from .identity_map import IdentityMap
 from .response import (
+    AuthorizationCodeTokenResponse,
+    ClientCredentialsTokenResponse,
     GetConsentsResponse,
     GetIdentitiesResponse,
     OAuthDependentTokenResponse,
     OAuthTokenResponse,
+    RefreshTokenResponse,
 )
 
 __all__ = (
@@ -33,8 +36,11 @@
     "GlobusNativeAppFlowManager",
     "GlobusAuthorizationCodeFlowManager",
     # responses
+    "AuthorizationCodeTokenResponse",
+    "ClientCredentialsTokenResponse",
     "GetConsentsResponse",
     "GetIdentitiesResponse",
     "OAuthDependentTokenResponse",
     "OAuthTokenResponse",
+    "RefreshTokenResponse",
 )

src/globus_sdk/services/auth/client/base_login_client.py

@@ -15,7 +15,11 @@
 from .._common import get_jwk_data, pem_decode_jwk_data
 from ..errors import AuthAPIError
 from ..flow_managers import GlobusOAuthFlowManager
-from ..response import OAuthTokenResponse
+from ..response import (
+    AuthorizationCodeTokenResponse,
+    OAuthTokenResponse,
+    RefreshTokenResponse,
+)
 
 if sys.version_info >= (3, 8):
     from typing import Literal
@@ -203,7 +207,9 @@ def oauth2_get_authorize_url(
         log.info(f"Got authorization URL: {auth_url}")
         return auth_url
 
-    def oauth2_exchange_code_for_tokens(self, auth_code: str) -> OAuthTokenResponse:
+    def oauth2_exchange_code_for_tokens(
+        self, auth_code: str
+    ) -> AuthorizationCodeTokenResponse:
         """
         Exchange an authorization code for a token or tokens.
 
@@ -231,7 +237,7 @@ def oauth2_refresh_token(
         refresh_token: str,
         *,
         body_params: dict[str, t.Any] | None = None,
-    ) -> OAuthTokenResponse:
+    ) -> RefreshTokenResponse:
         r"""
         Exchange a refresh token for a
         :class:`OAuthTokenResponse <.OAuthTokenResponse>`, containing
@@ -251,7 +257,9 @@ def oauth2_refresh_token(
         """
         log.info("Executing token refresh; typically requires client credentials")
         form_data = {"refresh_token": refresh_token, "grant_type": "refresh_token"}
-        return self.oauth2_token(form_data, body_params=body_params)
+        return self.oauth2_token(
+            form_data, body_params=body_params, response_class=RefreshTokenResponse
+        )
 
     def oauth2_validate_token(
         self,

src/globus_sdk/services/auth/client/confidential_client.py

@@ -11,9 +11,9 @@
 from .._common import stringify_requested_scopes
 from ..flow_managers import GlobusAuthorizationCodeFlowManager
 from ..response import (
+    ClientCredentialsTokenResponse,
     GetIdentitiesResponse,
     OAuthDependentTokenResponse,
-    OAuthTokenResponse,
 )
 from .base_login_client import AuthLoginClient
 
@@ -109,7 +109,7 @@ def get_identities(
     def oauth2_client_credentials_tokens(
         self,
         requested_scopes: ScopeCollectionType | None = None,
-    ) -> OAuthTokenResponse:
+    ) -> ClientCredentialsTokenResponse:
         r"""
         Perform an OAuth2 Client Credentials Grant to get access tokens which
         directly represent your client and allow it to act on its own
@@ -132,7 +132,8 @@ def oauth2_client_credentials_tokens(
         log.info("Fetching token(s) using client credentials")
         requested_scopes_string = stringify_requested_scopes(requested_scopes)
         return self.oauth2_token(
-            {"grant_type": "client_credentials", "scope": requested_scopes_string}
+            {"grant_type": "client_credentials", "scope": requested_scopes_string},
+            response_class=ClientCredentialsTokenResponse,
         )
 
     def oauth2_start_flow(

src/globus_sdk/services/auth/client/native_client.py

@@ -8,7 +8,7 @@
 from globus_sdk.response import GlobusHTTPResponse
 
 from ..flow_managers import GlobusNativeAppFlowManager
-from ..response import OAuthTokenResponse
+from ..response import RefreshTokenResponse
 from .base_login_client import AuthLoginClient
 
 log = logging.getLogger(__name__)
@@ -115,7 +115,7 @@ def oauth2_refresh_token(
         refresh_token: str,
         *,
         body_params: dict[str, t.Any] | None = None,
-    ) -> OAuthTokenResponse:
+    ) -> RefreshTokenResponse:
         """
         ``NativeAppAuthClient`` specializes the refresh token grant to include
         its client ID as a parameter in the POST body.
@@ -131,7 +131,9 @@ def oauth2_refresh_token(
             "grant_type": "refresh_token",
             "client_id": self.client_id,
         }
-        return self.oauth2_token(form_data, body_params=body_params)
+        return self.oauth2_token(
+            form_data, body_params=body_params, response_class=RefreshTokenResponse
+        )
 
     def create_native_app_instance(
         self,