Skip to content

academy.exchange.cloud.authenticate

Authenticate users from request headers.

Authenticator

Bases: Protocol

Authenticate users from request headers.

authenticate_user 

authenticate_user(headers: Mapping[str, str]) -> UUID

Authenticate user from request headers.

Warning

This method must be thread safe!

Parameters:

Returns:

  • UUID –

    A user id upon authentication success.

Raises:

Source code in academy/exchange/cloud/authenticate.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
def authenticate_user(self, headers: Mapping[str, str]) -> uuid.UUID:
    """Authenticate user from request headers.

    Warning:
        This method must be thread safe!

    Args:
        headers: Request headers.

    Returns:
        A user id upon authentication success.

    Raises:
        ForbiddenError: user is authenticated but is missing permissions
            or accessing forbidden resources.
        UnauthorizedError: user authentication fails.
    """
    ...

NullAuthenticator

Authenticator that implements no authentication.

authenticate_user 

authenticate_user(headers: Mapping[str, str]) -> UUID

Authenticate user from request headers.

Parameters:

Returns:

  • UUID –

    Null user regardless of provided headers.

Source code in academy/exchange/cloud/authenticate.py 
46
47
48
49
50
51
52
53
54
55
def authenticate_user(self, headers: Mapping[str, str]) -> uuid.UUID:
    """Authenticate user from request headers.

    Args:
        headers: Request headers.

    Returns:
        Null user regardless of provided headers.
    """
    return uuid.UUID(int=0)

GlobusAuthenticator 

GlobusAuthenticator(
    client_id: str | None = None,
    client_secret: str | None = None,
    *,
    audience: str = resource_server,
    auth_client: ConfidentialAppAuthClient | None = None
)

Globus Auth authorizer.

Parameters:

  • client_id (str | None, default: None ) –

    Globus application client ID. If either client_id or client_secret is None, the values will be read from the environment variables as described in get_confidential_app_auth_client. Ignored if auth_client is provided.

  • client_secret (str | None, default: None ) –

    Globus application client secret. See client_id for details. Ignored if auth_client is provided.

  • audience (str, default: resource_server ) –

    Intended audience of the token. This should typically be the resource server of the the token was issued for. E.g., the UUID of the ProxyStore Relay Server application.

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

    Optional confidential application authentication client which is used for introspecting client tokens.

Source code in academy/exchange/cloud/authenticate.py 
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
def __init__(
    self,
    client_id: str | None = None,
    client_secret: str | None = None,
    *,
    audience: str = AcademyExchangeScopes.resource_server,
    auth_client: globus_sdk.ConfidentialAppAuthClient | None = None,
) -> None:
    self.auth_client = (
        globus_sdk.ConfidentialAppAuthClient(
            client_id=str(client_id),
            client_secret=str(client_secret),
        )
        if auth_client is None
        else auth_client
    )
    self.auth_client_lock = Lock()
    self.audience = audience

authenticate_user 

authenticate_user(headers: Mapping[str, str]) -> UUID

Authenticate a Globus Auth user from request header.

This follows from the Globus Sample Data Portal example.

The underlying auth client is not thread safe, but this method is made thread safe using a lock.

Parameters:

  • headers (Mapping[str, str]) –

    Request headers to extract tokens from.

Returns:

Raises:

  • UnauthorizedError –

    if the authorization header is missing or the header is malformed.

  • ForbiddenError –

    if the tokens have expired or been revoked.

  • ForbiddenError –

    if audience is not included in the token's audience.

Source code in academy/exchange/cloud/authenticate.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
def authenticate_user(self, headers: Mapping[str, str]) -> uuid.UUID:
    """Authenticate a Globus Auth user from request header.

    This follows from the [Globus Sample Data Portal](https://github.com/globus/globus-sample-data-portal/blob/30e30cd418ee9b103e04916e19deb9902d3aafd8/service/decorators.py)
    example.

    The underlying auth client is not thread safe, but this method is made
    thread safe using a lock.

    Args:
        headers: Request headers to extract tokens from.

    Returns:
        Globus Auth identity returned via \
        [token introspection](https://docs.globus.org/api/auth/reference/#token-introspect).

    Raises:
        UnauthorizedError: if the authorization header is missing or
            the header is malformed.
        ForbiddenError: if the tokens have expired or been revoked.
        ForbiddenError: if `audience` is not included in the token's
            audience.
    """
    token = get_token_from_headers(headers)
    with self.auth_client_lock:
        token_meta = self.auth_client.oauth2_token_introspect(token)

    if not token_meta.get('active'):
        raise ForbiddenError('Token is expired or has been revoked.')

    if self.audience is not None and self.audience not in token_meta.get(
        'aud',
        [],
    ):
        raise ForbiddenError(
            f'Token audience does not include "{self.audience}". This '
            'could result in a confused deputy attack. Ensure the correct '
            'scopes are requested when the token is created.',
        )

    return uuid.UUID(token_meta.get('client_id'))

get_authenticator 

get_authenticator(
    config: ExchangeAuthConfig,
) -> Authenticator

Create an authenticator from a configuration.

Parameters:

Returns:

Raises:

  • ValueError –

    if the authentication method in the config is unknown.

Source code in academy/exchange/cloud/authenticate.py 
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
def get_authenticator(config: ExchangeAuthConfig) -> Authenticator:
    """Create an authenticator from a configuration.

    Args:
        config: Configuration.

    Returns:
        Authenticator.

    Raises:
        ValueError: if the authentication method in the config is unknown.
    """
    if config.method is None:
        return NullAuthenticator()
    elif config.method == 'globus':
        return GlobusAuthenticator(**config.kwargs)
    else:
        raise ValueError(f'Unknown authentication method "{config.method}."')

get_token_from_headers 

get_token_from_headers(headers: Mapping[str, str]) -> str

Extract token from websockets headers.

The header is expected to have the format Authorization: Bearer <TOKEN>.

Parameters:

  • headers (Mapping[str, str]) –

    Request headers to extract tokens from.

Returns:

  • str –

    String token.

Raises:

Source code in academy/exchange/cloud/authenticate.py 
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
def get_token_from_headers(headers: Mapping[str, str]) -> str:
    """Extract token from websockets headers.

    The header is expected to have the format `Authorization: Bearer <TOKEN>`.

    Args:
         headers: Request headers to extract tokens from.

    Returns:
        String token.

    Raises:
        UnauthorizedError: if the authorization header is missing.
        UnauthorizedError: if the authorization header is malformed.
    """
    if 'Authorization' not in headers:
        raise UnauthorizedError(
            'Request headers are missing authorization header.',
        )

    auth_header_parts = headers['Authorization'].split(' ')

    if len(auth_header_parts) != 2 or auth_header_parts[0] != 'Bearer':  # noqa: PLR2004
        raise UnauthorizedError(
            'Bearer token in authorization header is malformed.',
        )

    return auth_header_parts[1]