1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
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
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
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
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
|
# Copyright 2016 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Interfaces for credentials."""
import abc
from enum import Enum
import os
from google.auth import _helpers, environment_vars
from google.auth import exceptions
from google.auth import metrics
from google.auth._refresh_worker import RefreshThreadManager
class Credentials(metaclass=abc.ABCMeta):
"""Base class for all credentials.
All credentials have a :attr:`token` that is used for authentication and
may also optionally set an :attr:`expiry` to indicate when the token will
no longer be valid.
Most credentials will be :attr:`invalid` until :meth:`refresh` is called.
Credentials can do this automatically before the first HTTP request in
:meth:`before_request`.
Although the token and expiration will change as the credentials are
:meth:`refreshed <refresh>` and used, credentials should be considered
immutable. Various credentials will accept configuration such as private
keys, scopes, and other options. These options are not changeable after
construction. Some classes will provide mechanisms to copy the credentials
with modifications such as :meth:`ScopedCredentials.with_scopes`.
"""
def __init__(self):
self.token = None
"""str: The bearer token that can be used in HTTP headers to make
authenticated requests."""
self.expiry = None
"""Optional[datetime]: When the token expires and is no longer valid.
If this is None, the token is assumed to never expire."""
self._quota_project_id = None
"""Optional[str]: Project to use for quota and billing purposes."""
self._trust_boundary = None
"""Optional[dict]: Cache of a trust boundary response which has a list
of allowed regions and an encoded string representation of credentials
trust boundary."""
self._universe_domain = "googleapis.com"
"""Optional[str]: The universe domain value, default is googleapis.com
"""
self._use_non_blocking_refresh = False
self._refresh_worker = RefreshThreadManager()
@property
def expired(self):
"""Checks if the credentials are expired.
Note that credentials can be invalid but not expired because
Credentials with :attr:`expiry` set to None is considered to never
expire.
.. deprecated:: v2.24.0
Prefer checking :attr:`token_state` instead.
"""
if not self.expiry:
return False
# Remove some threshold from expiry to err on the side of reporting
# expiration early so that we avoid the 401-refresh-retry loop.
skewed_expiry = self.expiry - _helpers.REFRESH_THRESHOLD
return _helpers.utcnow() >= skewed_expiry
@property
def valid(self):
"""Checks the validity of the credentials.
This is True if the credentials have a :attr:`token` and the token
is not :attr:`expired`.
.. deprecated:: v2.24.0
Prefer checking :attr:`token_state` instead.
"""
return self.token is not None and not self.expired
@property
def token_state(self):
"""
See `:obj:`TokenState`
"""
if self.token is None:
return TokenState.INVALID
# Credentials that can't expire are always treated as fresh.
if self.expiry is None:
return TokenState.FRESH
expired = _helpers.utcnow() >= self.expiry
if expired:
return TokenState.INVALID
is_stale = _helpers.utcnow() >= (self.expiry - _helpers.REFRESH_THRESHOLD)
if is_stale:
return TokenState.STALE
return TokenState.FRESH
@property
def quota_project_id(self):
"""Project to use for quota and billing purposes."""
return self._quota_project_id
@property
def universe_domain(self):
"""The universe domain value."""
return self._universe_domain
@abc.abstractmethod
def refresh(self, request):
"""Refreshes the access token.
Args:
request (google.auth.transport.Request): The object used to make
HTTP requests.
Raises:
google.auth.exceptions.RefreshError: If the credentials could
not be refreshed.
"""
# pylint: disable=missing-raises-doc
# (pylint doesn't recognize that this is abstract)
raise NotImplementedError("Refresh must be implemented")
def _metric_header_for_usage(self):
"""The x-goog-api-client header for token usage metric.
This header will be added to the API service requests in before_request
method. For example, "cred-type/sa-jwt" means service account self
signed jwt access token is used in the API service request
authorization header. Children credentials classes need to override
this method to provide the header value, if the token usage metric is
needed.
Returns:
str: The x-goog-api-client header value.
"""
return None
def apply(self, headers, token=None):
"""Apply the token to the authentication header.
Args:
headers (Mapping): The HTTP request headers.
token (Optional[str]): If specified, overrides the current access
token.
"""
headers["authorization"] = "Bearer {}".format(
_helpers.from_bytes(token or self.token)
)
"""Trust boundary value will be a cached value from global lookup.
The response of trust boundary will be a list of regions and a hex
encoded representation.
An example of global lookup response:
{
"locations": [
"us-central1", "us-east1", "europe-west1", "asia-east1"
]
"encoded_locations": "0xA30"
}
"""
if self._trust_boundary is not None:
headers["x-allowed-locations"] = self._trust_boundary["encoded_locations"]
if self.quota_project_id:
headers["x-goog-user-project"] = self.quota_project_id
def _blocking_refresh(self, request):
if not self.valid:
self.refresh(request)
def _non_blocking_refresh(self, request):
use_blocking_refresh_fallback = False
if self.token_state == TokenState.STALE:
use_blocking_refresh_fallback = not self._refresh_worker.start_refresh(
self, request
)
if self.token_state == TokenState.INVALID or use_blocking_refresh_fallback:
self.refresh(request)
# If the blocking refresh succeeds then we can clear the error info
# on the background refresh worker, and perform refreshes in a
# background thread.
self._refresh_worker.clear_error()
def before_request(self, request, method, url, headers):
"""Performs credential-specific before request logic.
Refreshes the credentials if necessary, then calls :meth:`apply` to
apply the token to the authentication header.
Args:
request (google.auth.transport.Request): The object used to make
HTTP requests.
method (str): The request's HTTP method or the RPC method being
invoked.
url (str): The request's URI or the RPC service's URI.
headers (Mapping): The request's headers.
"""
# pylint: disable=unused-argument
# (Subclasses may use these arguments to ascertain information about
# the http request.)
if self._use_non_blocking_refresh:
self._non_blocking_refresh(request)
else:
self._blocking_refresh(request)
metrics.add_metric_header(headers, self._metric_header_for_usage())
self.apply(headers)
def with_non_blocking_refresh(self):
self._use_non_blocking_refresh = True
class CredentialsWithQuotaProject(Credentials):
"""Abstract base for credentials supporting ``with_quota_project`` factory"""
def with_quota_project(self, quota_project_id):
"""Returns a copy of these credentials with a modified quota project.
Args:
quota_project_id (str): The project to use for quota and
billing purposes
Returns:
google.auth.credentials.Credentials: A new credentials instance.
"""
raise NotImplementedError("This credential does not support quota project.")
def with_quota_project_from_environment(self):
quota_from_env = os.environ.get(environment_vars.GOOGLE_CLOUD_QUOTA_PROJECT)
if quota_from_env:
return self.with_quota_project(quota_from_env)
return self
class CredentialsWithTokenUri(Credentials):
"""Abstract base for credentials supporting ``with_token_uri`` factory"""
def with_token_uri(self, token_uri):
"""Returns a copy of these credentials with a modified token uri.
Args:
token_uri (str): The uri to use for fetching/exchanging tokens
Returns:
google.auth.credentials.Credentials: A new credentials instance.
"""
raise NotImplementedError("This credential does not use token uri.")
class CredentialsWithUniverseDomain(Credentials):
"""Abstract base for credentials supporting ``with_universe_domain`` factory"""
def with_universe_domain(self, universe_domain):
"""Returns a copy of these credentials with a modified universe domain.
Args:
universe_domain (str): The universe domain to use
Returns:
google.auth.credentials.Credentials: A new credentials instance.
"""
raise NotImplementedError(
"This credential does not support with_universe_domain."
)
class AnonymousCredentials(Credentials):
"""Credentials that do not provide any authentication information.
These are useful in the case of services that support anonymous access or
local service emulators that do not use credentials.
"""
@property
def expired(self):
"""Returns `False`, anonymous credentials never expire."""
return False
@property
def valid(self):
"""Returns `True`, anonymous credentials are always valid."""
return True
def refresh(self, request):
"""Raises :class:``InvalidOperation``, anonymous credentials cannot be
refreshed."""
raise exceptions.InvalidOperation("Anonymous credentials cannot be refreshed.")
def apply(self, headers, token=None):
"""Anonymous credentials do nothing to the request.
The optional ``token`` argument is not supported.
Raises:
google.auth.exceptions.InvalidValue: If a token was specified.
"""
if token is not None:
raise exceptions.InvalidValue("Anonymous credentials don't support tokens.")
def before_request(self, request, method, url, headers):
"""Anonymous credentials do nothing to the request."""
class ReadOnlyScoped(metaclass=abc.ABCMeta):
"""Interface for credentials whose scopes can be queried.
OAuth 2.0-based credentials allow limiting access using scopes as described
in `RFC6749 Section 3.3`_.
If a credential class implements this interface then the credentials either
use scopes in their implementation.
Some credentials require scopes in order to obtain a token. You can check
if scoping is necessary with :attr:`requires_scopes`::
if credentials.requires_scopes:
# Scoping is required.
credentials = credentials.with_scopes(scopes=['one', 'two'])
Credentials that require scopes must either be constructed with scopes::
credentials = SomeScopedCredentials(scopes=['one', 'two'])
Or must copy an existing instance using :meth:`with_scopes`::
scoped_credentials = credentials.with_scopes(scopes=['one', 'two'])
Some credentials have scopes but do not allow or require scopes to be set,
these credentials can be used as-is.
.. _RFC6749 Section 3.3: https://tools.ietf.org/html/rfc6749#section-3.3
"""
def __init__(self):
super(ReadOnlyScoped, self).__init__()
self._scopes = None
self._default_scopes = None
@property
def scopes(self):
"""Sequence[str]: the credentials' current set of scopes."""
return self._scopes
@property
def default_scopes(self):
"""Sequence[str]: the credentials' current set of default scopes."""
return self._default_scopes
@abc.abstractproperty
def requires_scopes(self):
"""True if these credentials require scopes to obtain an access token.
"""
return False
def has_scopes(self, scopes):
"""Checks if the credentials have the given scopes.
.. warning: This method is not guaranteed to be accurate if the
credentials are :attr:`~Credentials.invalid`.
Args:
scopes (Sequence[str]): The list of scopes to check.
Returns:
bool: True if the credentials have the given scopes.
"""
credential_scopes = (
self._scopes if self._scopes is not None else self._default_scopes
)
return set(scopes).issubset(set(credential_scopes or []))
class Scoped(ReadOnlyScoped):
"""Interface for credentials whose scopes can be replaced while copying.
OAuth 2.0-based credentials allow limiting access using scopes as described
in `RFC6749 Section 3.3`_.
If a credential class implements this interface then the credentials either
use scopes in their implementation.
Some credentials require scopes in order to obtain a token. You can check
if scoping is necessary with :attr:`requires_scopes`::
if credentials.requires_scopes:
# Scoping is required.
credentials = credentials.create_scoped(['one', 'two'])
Credentials that require scopes must either be constructed with scopes::
credentials = SomeScopedCredentials(scopes=['one', 'two'])
Or must copy an existing instance using :meth:`with_scopes`::
scoped_credentials = credentials.with_scopes(scopes=['one', 'two'])
Some credentials have scopes but do not allow or require scopes to be set,
these credentials can be used as-is.
.. _RFC6749 Section 3.3: https://tools.ietf.org/html/rfc6749#section-3.3
"""
@abc.abstractmethod
def with_scopes(self, scopes, default_scopes=None):
"""Create a copy of these credentials with the specified scopes.
Args:
scopes (Sequence[str]): The list of scopes to attach to the
current credentials.
Raises:
NotImplementedError: If the credentials' scopes can not be changed.
This can be avoided by checking :attr:`requires_scopes` before
calling this method.
"""
raise NotImplementedError("This class does not require scoping.")
def with_scopes_if_required(credentials, scopes, default_scopes=None):
"""Creates a copy of the credentials with scopes if scoping is required.
This helper function is useful when you do not know (or care to know) the
specific type of credentials you are using (such as when you use
:func:`google.auth.default`). This function will call
:meth:`Scoped.with_scopes` if the credentials are scoped credentials and if
the credentials require scoping. Otherwise, it will return the credentials
as-is.
Args:
credentials (google.auth.credentials.Credentials): The credentials to
scope if necessary.
scopes (Sequence[str]): The list of scopes to use.
default_scopes (Sequence[str]): Default scopes passed by a
Google client library. Use 'scopes' for user-defined scopes.
Returns:
google.auth.credentials.Credentials: Either a new set of scoped
credentials, or the passed in credentials instance if no scoping
was required.
"""
if isinstance(credentials, Scoped) and credentials.requires_scopes:
return credentials.with_scopes(scopes, default_scopes=default_scopes)
else:
return credentials
class Signing(metaclass=abc.ABCMeta):
"""Interface for credentials that can cryptographically sign messages."""
@abc.abstractmethod
def sign_bytes(self, message):
"""Signs the given message.
Args:
message (bytes): The message to sign.
Returns:
bytes: The message's cryptographic signature.
"""
# pylint: disable=missing-raises-doc,redundant-returns-doc
# (pylint doesn't recognize that this is abstract)
raise NotImplementedError("Sign bytes must be implemented.")
@abc.abstractproperty
def signer_email(self):
"""Optional[str]: An email address that identifies the signer."""
# pylint: disable=missing-raises-doc
# (pylint doesn't recognize that this is abstract)
raise NotImplementedError("Signer email must be implemented.")
@abc.abstractproperty
def signer(self):
"""google.auth.crypt.Signer: The signer used to sign bytes."""
# pylint: disable=missing-raises-doc
# (pylint doesn't recognize that this is abstract)
raise NotImplementedError("Signer must be implemented.")
class TokenState(Enum):
"""
Tracks the state of a token.
FRESH: The token is valid. It is not expired or close to expired, or the token has no expiry.
STALE: The token is close to expired, and should be refreshed. The token can be used normally.
INVALID: The token is expired or invalid. The token cannot be used for a normal operation.
"""
FRESH = 1
STALE = 2
INVALID = 3
|