diff --git a/changelog.d/20260806_111730_sirosen_mypy_docs_integration.rst b/changelog.d/20260806_111730_sirosen_mypy_docs_integration.rst new file mode 100644 index 000000000..51ff88d52 --- /dev/null +++ b/changelog.d/20260806_111730_sirosen_mypy_docs_integration.rst @@ -0,0 +1,7 @@ +Added +----- + +- Added a new flag, ``sweep``, to ``GlobusApp.logout()``. + Use ``logout(sweep=True)`` to ask the app to clear all tokens found in the + app's storage, not only the ones currently in use by the app. + The default behavior, ``sweep=False``, is unchanged. (:pr:`NUMBER`) diff --git a/src/globus_sdk/globus_app/app.py b/src/globus_sdk/globus_app/app.py index 8df0d33c1..4dc11cc75 100644 --- a/src/globus_sdk/globus_app/app.py +++ b/src/globus_sdk/globus_app/app.py @@ -25,6 +25,7 @@ from globus_sdk.token_storage import ( ScopeRequirementsValidator, TokenStorage, + TokenStorageData, TokenValidationError, ValidatingTokenStorage, ) @@ -385,21 +386,36 @@ def login_required(self) -> bool: return True return False - def logout(self) -> None: + def logout(self, *, sweep: bool = False) -> None: """ Log the current user or client out of the app. - This will remove and revoke all tokens stored for the current app user. + This will remove and revoke tokens stored for the current app user. + + :param sweep: Set to ``True`` in order to clear all tokens in storage. By + default, only tokens for currently in-use services are cleared. """ - # Revoke all tokens, removing them from the underlying token storage + # Revoke tokens, removing them from the underlying token storage inner_token_storage = self.token_storage.token_storage - for resource_server in self._scope_requirements.keys(): - token_data = inner_token_storage.get_token_data(resource_server) - if token_data: - self._login_client.oauth2_revoke_token(token_data.access_token) - if token_data.refresh_token: - self._login_client.oauth2_revoke_token(token_data.refresh_token) - inner_token_storage.remove_token_data(resource_server) + + # collect tokens to clear, either all of them or a subset based on scope reqs + if sweep: + to_clear: t.Iterable[TokenStorageData] = ( + inner_token_storage.get_token_data_by_resource_server().values() + ) + else: + to_clear = [ + token_data + for resource_server in self._scope_requirements.keys() + if (token_data := inner_token_storage.get_token_data(resource_server)) + is not None + ] + + for token_data in to_clear: + self._login_client.oauth2_revoke_token(token_data.access_token) + if token_data.refresh_token: + self._login_client.oauth2_revoke_token(token_data.refresh_token) + inner_token_storage.remove_token_data(token_data.resource_server) # Invalidate any cached authorizers self._authorizer_factory.clear_cache() diff --git a/tests/unit/globus_app/test_globus_app.py b/tests/unit/globus_app/test_globus_app.py index 28b1b2c24..5a496ee3a 100644 --- a/tests/unit/globus_app/test_globus_app.py +++ b/tests/unit/globus_app/test_globus_app.py @@ -561,6 +561,49 @@ def test_client_app_login_logout(): assert memory_storage.get_token_data("auth.globus.org") is None +@pytest.mark.parametrize("app_type", ("UserApp", "ClientApp")) +@pytest.mark.parametrize("sweep", (True, False)) +def test_logout_sweep_flag_controls_clearing_unrecognized_tokens(app_type, sweep): + # setup a storage for Auth + Transfer + memory_storage = MemoryTokenStorage() + memory_storage.store_token_data_by_resource_server(_mock_token_data_by_rs()) + memory_storage.store_token_data_by_resource_server( + _mock_token_data_by_rs( + resource_server="transfer.api.globus.org", + scope=str(globus_sdk.TransferClient.scopes.all), + ) + ) + # setup the app (either type) + config = GlobusAppConfig(token_storage=memory_storage) + if app_type == "UserApp": + app = UserApp("test-app", client_id="mock_client_id", config=config) + load_response(NativeAppAuthClient.oauth2_revoke_token) + elif app_type == "ClientApp": + app = ClientApp( + "test-app", + client_id="mock_client_id", + client_secret="mock_client_secret", + config=config, + ) + load_response(ConfidentialAppAuthClient.oauth2_revoke_token) + else: + raise NotImplementedError(app_type) + + # verify that the data is present + assert memory_storage.get_token_data("auth.globus.org") is not None + assert memory_storage.get_token_data("transfer.api.globus.org") is not None + + app.logout(sweep=sweep) # act! + + # test results + assert memory_storage.get_token_data("auth.globus.org") is None + transfer_token_data = memory_storage.get_token_data("transfer.api.globus.org") + if sweep: + assert transfer_token_data is None + else: + assert transfer_token_data is not None + + @mock.patch.object(globus_sdk.IDTokenDecoder, "decode", _mock_decode) @pytest.mark.parametrize( "login_kwargs,expected_login",