Skip to content

proxystore.globus.manager

proxystore/globus/manager.py

Globus Auth credential managers.

GlobusAuthManager

Bases: Protocol

Protocol for a Globus Auth manager.

client property 

client: AuthLoginClient

Globus Auth client.

logged_in property 

logged_in: bool

User has valid refresh tokens for necessary scopes.

get_authorizer() 

get_authorizer(resource_server: str) -> GlobusAuthorizer

Get authorizer for a specific resource server.

Raises:

  • LookupError –

    if tokens for the resource server do not exist.

Source code in proxystore/globus/manager.py 
33
34
35
36
37
38
39
40
41
42
def get_authorizer(
    self,
    resource_server: str,
) -> globus_sdk.authorizers.GlobusAuthorizer:
    """Get authorizer for a specific resource server.

    Raises:
        LookupError: if tokens for the resource server do not exist.
    """
    ...

login() 

login(*, additional_scopes: Iterable[str] = ()) -> None

Perform the authentication flow.

This method is idempotent meaning it will be a no-op if the user is already logged in.

Parameters:

  • additional_scopes (Iterable[str], default: () ) –

    Additional scopes to request.

Source code in proxystore/globus/manager.py 
44
45
46
47
48
49
50
51
52
53
def login(self, *, additional_scopes: Iterable[str] = ()) -> None:
    """Perform the authentication flow.

    This method is idempotent meaning it will be a no-op if the user
    is already logged in.

    Args:
        additional_scopes: Additional scopes to request.
    """
    ...

logout() 

logout() -> None

Revoke and remove authentication tokens.

Source code in proxystore/globus/manager.py 
55
56
57
def logout(self) -> None:
    """Revoke and remove authentication tokens."""
    ...

ConfidentialAppAuthManager 

ConfidentialAppAuthManager(
    *,
    client: ConfidentialAppAuthClient | None = None,
    storage: SQLiteAdapter | None = None,
    resource_server_scopes: (
        dict[str, list[str]] | None
    ) = None
)

Globus confidential app (client identity) credential manager.

Parameters:

  • client (ConfidentialAppAuthClient | None, default: None ) –

    Optionally override the standard ProxyStore auth client.

  • storage (SQLiteAdapter | None, default: None ) –

    Optionally override the default token storage.

  • resource_server_scopes (dict[str, list[str]] | None, default: None ) –

    Mapping of resource server URLs to a list of scopes for that resource server. If unspecified, all basic scopes needed by ProxyStore components will be requested. This parameter can be used to request scopes for many resource server when login() is invoked.

Source code in proxystore/globus/manager.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
def __init__(
    self,
    *,
    client: globus_sdk.ConfidentialAppAuthClient | None = None,
    storage: globus_sdk.tokenstorage.SQLiteAdapter | None = None,
    resource_server_scopes: dict[str, list[str]] | None = None,
) -> None:
    self._client = (
        client
        if client is not None
        else get_confidential_app_auth_client()
    )
    self._storage = (
        storage
        if storage is not None
        else get_token_storage_adapter(
            namespace=f'client/{self._client.client_id}',
        )
    )
    self._resource_server_scopes = (
        resource_server_scopes
        if resource_server_scopes is not None
        else get_all_scopes_by_resource_server()
    )

client property 

client: ConfidentialAppAuthClient

Globus Auth client.

logged_in property 

logged_in: bool

User has valid refresh tokens for necessary scopes.

This is always true for client identities.

get_authorizer() 

get_authorizer(resource_server: str) -> GlobusAuthorizer

Get authorizer for a specific resource server.

Source code in proxystore/globus/manager.py 
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
def get_authorizer(
    self,
    resource_server: str,
) -> globus_sdk.authorizers.GlobusAuthorizer:
    """Get authorizer for a specific resource server."""
    scopes = []
    for rs_name, rs_scopes in self._resource_server_scopes.items():
        if rs_name == resource_server:
            scopes.extend(rs_scopes)

    tokens = self._storage.get_token_data(resource_server)
    if tokens is None:
        tokens = {}

    return globus_sdk.ClientCredentialsAuthorizer(
        confidential_client=self.client,
        scopes=scopes,
        access_token=tokens.get('access_token', None),
        expires_at=tokens.get('expires_at_seconds', None),
        on_refresh=self._storage.on_refresh,
    )

login() 

login(*, additional_scopes: Iterable[str] = ()) -> None

Perform the authentication flow.

Client identities do not require a login flow so this is a no-op.

Parameters:

  • additional_scopes (Iterable[str], default: () ) –

    Additional scopes to request.

Source code in proxystore/globus/manager.py 
135
136
137
138
139
140
141
142
143
def login(self, *, additional_scopes: Iterable[str] = ()) -> None:
    """Perform the authentication flow.

    Client identities do not require a login flow so this is a no-op.

    Args:
        additional_scopes: Additional scopes to request.
    """
    return

logout() 

logout() -> None

Revoke and remove authentication tokens.

Source code in proxystore/globus/manager.py 
145
146
147
148
149
150
151
def logout(self) -> None:
    """Revoke and remove authentication tokens."""
    for server, data in self._storage.get_by_resource_server().items():
        for key in ('access_token', 'refresh_token'):
            token = data[key]
            self.client.oauth2_revoke_token(token)
        self._storage.remove_tokens_for_resource_server(server)

NativeAppAuthManager 

NativeAppAuthManager(
    *,
    client: NativeAppAuthClient | None = None,
    storage: SQLiteAdapter | None = None,
    resource_server_scopes: (
        dict[str, list[str]] | None
    ) = None
)

Globus native app credential manager.

Parameters:

  • client (NativeAppAuthClient | None, default: None ) –

    Optionally override the standard ProxyStore auth client.

  • storage (SQLiteAdapter | None, default: None ) –

    Optionally override the default token storage.

  • resource_server_scopes (dict[str, list[str]] | None, default: None ) –

    Mapping of resource server URLs to a list of scopes for that resource server. If unspecified, all basic scopes needed by ProxyStore components will be requested. This parameter can be used to request scopes for many resource server when login() is invoked.

Source code in proxystore/globus/manager.py 
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
def __init__(
    self,
    *,
    client: globus_sdk.NativeAppAuthClient | None = None,
    storage: globus_sdk.tokenstorage.SQLiteAdapter | None = None,
    resource_server_scopes: dict[str, list[str]] | None = None,
) -> None:
    self._client = (
        client if client is not None else get_native_app_auth_client()
    )
    self._storage = (
        storage
        if storage is not None
        else get_token_storage_adapter(
            namespace=f'user/{self._client.client_id}',
        )
    )
    self._resource_server_scopes = (
        resource_server_scopes
        if resource_server_scopes is not None
        else get_all_scopes_by_resource_server()
    )

client property 

client: NativeAppAuthClient

Globus Auth client.

logged_in property 

logged_in: bool

User has valid refresh tokens for necessary scopes.

get_authorizer() 

get_authorizer(resource_server: str) -> GlobusAuthorizer

Get authorizer for a specific resource server.

Raises:

  • LookupError –

    if tokens for the resource server do not exist.

Source code in proxystore/globus/manager.py 
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
def get_authorizer(
    self,
    resource_server: str,
) -> globus_sdk.authorizers.GlobusAuthorizer:
    """Get authorizer for a specific resource server.

    Raises:
        LookupError: if tokens for the resource server do not exist.
    """
    tokens = self._storage.get_token_data(resource_server)
    if tokens is None:
        raise LookupError(f'Could not find tokens for {resource_server}.')
    return globus_sdk.RefreshTokenAuthorizer(
        tokens['refresh_token'],
        self.client,
        access_token=tokens['access_token'],
        expires_at=tokens['expires_at_seconds'],
        on_refresh=self._storage.on_refresh,
    )

login() 

login(*, additional_scopes: Iterable[str] = ()) -> None

Perform the authentication flow.

This method is idempotent meaning it will be a no-op if the user is already logged in.

On log in, the user will be prompted to follow a link to authenticate on globus.org.

Parameters:

  • additional_scopes (Iterable[str], default: () ) –

    Additional scopes to request.

Source code in proxystore/globus/manager.py 
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
def login(self, *, additional_scopes: Iterable[str] = ()) -> None:
    """Perform the authentication flow.

    This method is idempotent meaning it will be a no-op if the user
    is already logged in.

    On log in, the user will be prompted to follow a link to authenticate
    on [globus.org](https://globus.org).

    Args:
        additional_scopes: Additional scopes to request.
    """
    if not self.logged_in:
        token = self._run_login_flow(additional_scopes=additional_scopes)
        self._storage.store(token)

logout() 

logout() -> None

Revoke and remove authentication tokens.

Source code in proxystore/globus/manager.py 
277
278
279
280
281
282
283
def logout(self) -> None:
    """Revoke and remove authentication tokens."""
    for server, data in self._storage.get_by_resource_server().items():
        for key in ('access_token', 'refresh_token'):
            token = data[key]
            self.client.oauth2_revoke_token(token)
        self._storage.remove_tokens_for_resource_server(server)