Globus OAuth tools.
ProxyStore provides the proxystore-globus-auth CLI tool to give consent
to the ProxyStore Globus Application.
# basic authentication
proxystore-globus-auth
# delete old tokens
proxystore-globus-auth --delete
# give consent for specific collections
proxystore-globus-auth --collections COLLECTION_UUID COLLECTION_UUID ...
# specify additional scopes
proxystore-globus-auth --scopes SCOPE SCOPE ...
Based on Parsl's implementation
and the Globus examples.
GlobusAuthFileError
Bases: Exception
Exception raised if the Globus Auth token file cannot be read.
load_tokens_from_file
load_tokens_from_file(
filepath: str,
) -> dict[str, dict[str, Any]]
Load a set of saved tokens.
Source code in proxystore/globus.py
| def load_tokens_from_file(filepath: str) -> dict[str, dict[str, Any]]:
"""Load a set of saved tokens."""
with open(filepath) as f:
tokens = json.load(f)
return tokens
|
save_tokens_to_file
save_tokens_to_file(
filepath: str, tokens: globus_sdk.OAuthTokenResponse
) -> None
Save a set of tokens for later use.
Source code in proxystore/globus.py
| def save_tokens_to_file(
filepath: str,
tokens: globus_sdk.OAuthTokenResponse,
) -> None:
"""Save a set of tokens for later use."""
with open(filepath, 'w') as f:
json.dump(tokens.by_resource_server, f, indent=4)
|
authenticate
authenticate(
client_id: str,
redirect_uri: str,
requested_scopes: Iterable[str] | None = None,
) -> globus_sdk.OAuthTokenResponse
Perform Native App auth flow.
Source code in proxystore/globus.py
| def authenticate(
client_id: str,
redirect_uri: str,
requested_scopes: Iterable[str] | None = None,
) -> globus_sdk.OAuthTokenResponse:
"""Perform Native App auth flow."""
client = globus_sdk.NativeAppAuthClient(client_id=client_id)
client.oauth2_start_flow(
redirect_uri=redirect_uri,
refresh_tokens=True,
requested_scopes=requested_scopes,
)
url = client.oauth2_get_authorize_url()
print(f'Please visit the following url to authenticate:\n{url}')
auth_code = input('Enter the auth code: ').strip()
return client.oauth2_exchange_code_for_tokens(auth_code)
|
get_authorizer
get_authorizer(
client_id: str, tokens_file: str, redirect_uri: str
) -> globus_sdk.RefreshTokenAuthorizer
Get an authorizer for the Globus SDK.
Raises:
Source code in proxystore/globus.py
| def get_authorizer(
client_id: str,
tokens_file: str,
redirect_uri: str,
) -> globus_sdk.RefreshTokenAuthorizer:
"""Get an authorizer for the Globus SDK.
Raises:
GlobusAuthFileError: If `tokens_file` cannot be parsed.
"""
try:
tokens = load_tokens_from_file(tokens_file)
except OSError as e:
raise GlobusAuthFileError(
f'Error loading tokens from {tokens_file}: {e}.',
) from e
transfer_tokens = tokens['transfer.api.globus.org']
auth_client = globus_sdk.NativeAppAuthClient(client_id=client_id)
return globus_sdk.RefreshTokenAuthorizer(
transfer_tokens['refresh_token'],
auth_client,
access_token=transfer_tokens['access_token'],
expires_at=transfer_tokens['expires_at_seconds'],
on_refresh=functools.partial(save_tokens_to_file, tokens_file),
)
|
proxystore_authenticate
proxystore_authenticate(
proxystore_dir: str | None = None,
collections: list[str] | None = None,
additional_scopes: list[str] | None = None,
) -> None
Perform auth flow for ProxyStore native app.
Source code in proxystore/globus.py
| def proxystore_authenticate(
proxystore_dir: str | None = None,
collections: list[str] | None = None,
additional_scopes: list[str] | None = None,
) -> None:
"""Perform auth flow for ProxyStore native app."""
proxystore_dir = home_dir() if proxystore_dir is None else proxystore_dir
tokens_file = os.path.join(proxystore_dir, _TOKENS_FILE)
os.makedirs(proxystore_dir, exist_ok=True)
scopes = _get_proxystore_scopes(collections, additional_scopes)
tokens = authenticate(
client_id=_APPLICATION_ID,
redirect_uri=_REDIRECT_URI,
requested_scopes=scopes,
)
save_tokens_to_file(tokens_file, tokens)
|
get_proxystore_authorizer
get_proxystore_authorizer(
proxystore_dir: str | None = None,
) -> globus_sdk.RefreshTokenAuthorizer
Get an authorizer for the ProxyStore native app.
Source code in proxystore/globus.py
| def get_proxystore_authorizer(
proxystore_dir: str | None = None,
) -> globus_sdk.RefreshTokenAuthorizer:
"""Get an authorizer for the ProxyStore native app."""
proxystore_dir = home_dir() if proxystore_dir is None else proxystore_dir
tokens_file = os.path.join(proxystore_dir, _TOKENS_FILE)
return get_authorizer(
client_id=_APPLICATION_ID,
tokens_file=tokens_file,
redirect_uri=_REDIRECT_URI,
)
|
main
main(argv: Sequence[str] | None = None) -> int
Perform Globus authentication.
This function is the entrypoint for the proxystore-globus-auth CLI tool.
Source code in proxystore/globus.py
| def main(argv: Sequence[str] | None = None) -> int:
"""Perform Globus authentication.
This function is the entrypoint for the `proxystore-globus-auth` CLI tool.
"""
parser = argparse.ArgumentParser(
'ProxyStore Globus Auth Tool',
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
)
parser.add_argument(
'--collections',
nargs='+',
help='Collection UUIDs to request scopes for.',
)
parser.add_argument(
'--scopes',
nargs='+',
help='Additional scopes to request.',
)
parser.add_argument(
'--delete',
action='store_true',
help='Delete existing authentication tokens.',
)
args = parser.parse_args(argv)
if args.delete:
tokens_file = os.path.join(home_dir(), _TOKENS_FILE)
if os.path.exists(tokens_file):
os.remove(tokens_file)
print('Deleted tokens file.')
return 0
else:
print('No tokens file found.')
return 1
try:
get_proxystore_authorizer()
except GlobusAuthFileError:
print(
'Performing authentication for the ProxyStore Globus Native app.',
)
proxystore_authenticate(
collections=args.collections,
additional_scopes=args.scopes,
)
get_proxystore_authorizer()
print('Globus authorization complete.')
return 0
else:
print(
'Globus authorization is already completed. To re-authenticate, '
'delete your tokens (proxystore-globus-auth --delete) and try '
'again.',
)
return 1
|