Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
@@ -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`)
36 changes: 26 additions & 10 deletions src/globus_sdk/globus_app/app.py
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
from globus_sdk.token_storage import (
ScopeRequirementsValidator,
TokenStorage,
TokenStorageData,
TokenValidationError,
ValidatingTokenStorage,
)
Expand Down Expand Up @@ -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()
Expand Down
43 changes: 43 additions & 0 deletions tests/unit/globus_app/test_globus_app.py
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand Down