Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/google/auth/credentials.py: 52%
104 statements
« prev ^ index » next coverage.py v7.3.2, created at 2023-12-08 06:40 +0000
1# Copyright 2016 Google LLC
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7# http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
16"""Interfaces for credentials."""
18import abc
19import os
21from google.auth import _helpers, environment_vars
22from google.auth import exceptions
23from google.auth import metrics
26class Credentials(metaclass=abc.ABCMeta):
27 """Base class for all credentials.
29 All credentials have a :attr:`token` that is used for authentication and
30 may also optionally set an :attr:`expiry` to indicate when the token will
31 no longer be valid.
33 Most credentials will be :attr:`invalid` until :meth:`refresh` is called.
34 Credentials can do this automatically before the first HTTP request in
35 :meth:`before_request`.
37 Although the token and expiration will change as the credentials are
38 :meth:`refreshed <refresh>` and used, credentials should be considered
39 immutable. Various credentials will accept configuration such as private
40 keys, scopes, and other options. These options are not changeable after
41 construction. Some classes will provide mechanisms to copy the credentials
42 with modifications such as :meth:`ScopedCredentials.with_scopes`.
43 """
45 def __init__(self):
46 self.token = None
47 """str: The bearer token that can be used in HTTP headers to make
48 authenticated requests."""
49 self.expiry = None
50 """Optional[datetime]: When the token expires and is no longer valid.
51 If this is None, the token is assumed to never expire."""
52 self._quota_project_id = None
53 """Optional[str]: Project to use for quota and billing purposes."""
54 self._trust_boundary = None
55 """Optional[dict]: Cache of a trust boundary response which has a list
56 of allowed regions and an encoded string representation of credentials
57 trust boundary."""
58 self._universe_domain = "googleapis.com"
59 """Optional[str]: The universe domain value, default is googleapis.com
60 """
62 @property
63 def expired(self):
64 """Checks if the credentials are expired.
66 Note that credentials can be invalid but not expired because
67 Credentials with :attr:`expiry` set to None is considered to never
68 expire.
69 """
70 if not self.expiry:
71 return False
73 # Remove some threshold from expiry to err on the side of reporting
74 # expiration early so that we avoid the 401-refresh-retry loop.
75 skewed_expiry = self.expiry - _helpers.REFRESH_THRESHOLD
76 return _helpers.utcnow() >= skewed_expiry
78 @property
79 def valid(self):
80 """Checks the validity of the credentials.
82 This is True if the credentials have a :attr:`token` and the token
83 is not :attr:`expired`.
84 """
85 return self.token is not None and not self.expired
87 @property
88 def quota_project_id(self):
89 """Project to use for quota and billing purposes."""
90 return self._quota_project_id
92 @property
93 def universe_domain(self):
94 """The universe domain value."""
95 return self._universe_domain
97 @abc.abstractmethod
98 def refresh(self, request):
99 """Refreshes the access token.
101 Args:
102 request (google.auth.transport.Request): The object used to make
103 HTTP requests.
105 Raises:
106 google.auth.exceptions.RefreshError: If the credentials could
107 not be refreshed.
108 """
109 # pylint: disable=missing-raises-doc
110 # (pylint doesn't recognize that this is abstract)
111 raise NotImplementedError("Refresh must be implemented")
113 def _metric_header_for_usage(self):
114 """The x-goog-api-client header for token usage metric.
116 This header will be added to the API service requests in before_request
117 method. For example, "cred-type/sa-jwt" means service account self
118 signed jwt access token is used in the API service request
119 authorization header. Children credentials classes need to override
120 this method to provide the header value, if the token usage metric is
121 needed.
123 Returns:
124 str: The x-goog-api-client header value.
125 """
126 return None
128 def apply(self, headers, token=None):
129 """Apply the token to the authentication header.
131 Args:
132 headers (Mapping): The HTTP request headers.
133 token (Optional[str]): If specified, overrides the current access
134 token.
135 """
136 headers["authorization"] = "Bearer {}".format(
137 _helpers.from_bytes(token or self.token)
138 )
139 """Trust boundary value will be a cached value from global lookup.
141 The response of trust boundary will be a list of regions and a hex
142 encoded representation.
144 An example of global lookup response:
145 {
146 "locations": [
147 "us-central1", "us-east1", "europe-west1", "asia-east1"
148 ]
149 "encoded_locations": "0xA30"
150 }
151 """
152 if self._trust_boundary is not None:
153 headers["x-allowed-locations"] = self._trust_boundary["encoded_locations"]
154 if self.quota_project_id:
155 headers["x-goog-user-project"] = self.quota_project_id
157 def before_request(self, request, method, url, headers):
158 """Performs credential-specific before request logic.
160 Refreshes the credentials if necessary, then calls :meth:`apply` to
161 apply the token to the authentication header.
163 Args:
164 request (google.auth.transport.Request): The object used to make
165 HTTP requests.
166 method (str): The request's HTTP method or the RPC method being
167 invoked.
168 url (str): The request's URI or the RPC service's URI.
169 headers (Mapping): The request's headers.
170 """
171 # pylint: disable=unused-argument
172 # (Subclasses may use these arguments to ascertain information about
173 # the http request.)
174 if not self.valid:
175 self.refresh(request)
176 metrics.add_metric_header(headers, self._metric_header_for_usage())
177 self.apply(headers)
180class CredentialsWithQuotaProject(Credentials):
181 """Abstract base for credentials supporting ``with_quota_project`` factory"""
183 def with_quota_project(self, quota_project_id):
184 """Returns a copy of these credentials with a modified quota project.
186 Args:
187 quota_project_id (str): The project to use for quota and
188 billing purposes
190 Returns:
191 google.oauth2.credentials.Credentials: A new credentials instance.
192 """
193 raise NotImplementedError("This credential does not support quota project.")
195 def with_quota_project_from_environment(self):
196 quota_from_env = os.environ.get(environment_vars.GOOGLE_CLOUD_QUOTA_PROJECT)
197 if quota_from_env:
198 return self.with_quota_project(quota_from_env)
199 return self
202class CredentialsWithTokenUri(Credentials):
203 """Abstract base for credentials supporting ``with_token_uri`` factory"""
205 def with_token_uri(self, token_uri):
206 """Returns a copy of these credentials with a modified token uri.
208 Args:
209 token_uri (str): The uri to use for fetching/exchanging tokens
211 Returns:
212 google.oauth2.credentials.Credentials: A new credentials instance.
213 """
214 raise NotImplementedError("This credential does not use token uri.")
217class AnonymousCredentials(Credentials):
218 """Credentials that do not provide any authentication information.
220 These are useful in the case of services that support anonymous access or
221 local service emulators that do not use credentials.
222 """
224 @property
225 def expired(self):
226 """Returns `False`, anonymous credentials never expire."""
227 return False
229 @property
230 def valid(self):
231 """Returns `True`, anonymous credentials are always valid."""
232 return True
234 def refresh(self, request):
235 """Raises :class:``InvalidOperation``, anonymous credentials cannot be
236 refreshed."""
237 raise exceptions.InvalidOperation("Anonymous credentials cannot be refreshed.")
239 def apply(self, headers, token=None):
240 """Anonymous credentials do nothing to the request.
242 The optional ``token`` argument is not supported.
244 Raises:
245 google.auth.exceptions.InvalidValue: If a token was specified.
246 """
247 if token is not None:
248 raise exceptions.InvalidValue("Anonymous credentials don't support tokens.")
250 def before_request(self, request, method, url, headers):
251 """Anonymous credentials do nothing to the request."""
254class ReadOnlyScoped(metaclass=abc.ABCMeta):
255 """Interface for credentials whose scopes can be queried.
257 OAuth 2.0-based credentials allow limiting access using scopes as described
258 in `RFC6749 Section 3.3`_.
259 If a credential class implements this interface then the credentials either
260 use scopes in their implementation.
262 Some credentials require scopes in order to obtain a token. You can check
263 if scoping is necessary with :attr:`requires_scopes`::
265 if credentials.requires_scopes:
266 # Scoping is required.
267 credentials = credentials.with_scopes(scopes=['one', 'two'])
269 Credentials that require scopes must either be constructed with scopes::
271 credentials = SomeScopedCredentials(scopes=['one', 'two'])
273 Or must copy an existing instance using :meth:`with_scopes`::
275 scoped_credentials = credentials.with_scopes(scopes=['one', 'two'])
277 Some credentials have scopes but do not allow or require scopes to be set,
278 these credentials can be used as-is.
280 .. _RFC6749 Section 3.3: https://tools.ietf.org/html/rfc6749#section-3.3
281 """
283 def __init__(self):
284 super(ReadOnlyScoped, self).__init__()
285 self._scopes = None
286 self._default_scopes = None
288 @property
289 def scopes(self):
290 """Sequence[str]: the credentials' current set of scopes."""
291 return self._scopes
293 @property
294 def default_scopes(self):
295 """Sequence[str]: the credentials' current set of default scopes."""
296 return self._default_scopes
298 @abc.abstractproperty
299 def requires_scopes(self):
300 """True if these credentials require scopes to obtain an access token.
301 """
302 return False
304 def has_scopes(self, scopes):
305 """Checks if the credentials have the given scopes.
307 .. warning: This method is not guaranteed to be accurate if the
308 credentials are :attr:`~Credentials.invalid`.
310 Args:
311 scopes (Sequence[str]): The list of scopes to check.
313 Returns:
314 bool: True if the credentials have the given scopes.
315 """
316 credential_scopes = (
317 self._scopes if self._scopes is not None else self._default_scopes
318 )
319 return set(scopes).issubset(set(credential_scopes or []))
322class Scoped(ReadOnlyScoped):
323 """Interface for credentials whose scopes can be replaced while copying.
325 OAuth 2.0-based credentials allow limiting access using scopes as described
326 in `RFC6749 Section 3.3`_.
327 If a credential class implements this interface then the credentials either
328 use scopes in their implementation.
330 Some credentials require scopes in order to obtain a token. You can check
331 if scoping is necessary with :attr:`requires_scopes`::
333 if credentials.requires_scopes:
334 # Scoping is required.
335 credentials = credentials.create_scoped(['one', 'two'])
337 Credentials that require scopes must either be constructed with scopes::
339 credentials = SomeScopedCredentials(scopes=['one', 'two'])
341 Or must copy an existing instance using :meth:`with_scopes`::
343 scoped_credentials = credentials.with_scopes(scopes=['one', 'two'])
345 Some credentials have scopes but do not allow or require scopes to be set,
346 these credentials can be used as-is.
348 .. _RFC6749 Section 3.3: https://tools.ietf.org/html/rfc6749#section-3.3
349 """
351 @abc.abstractmethod
352 def with_scopes(self, scopes, default_scopes=None):
353 """Create a copy of these credentials with the specified scopes.
355 Args:
356 scopes (Sequence[str]): The list of scopes to attach to the
357 current credentials.
359 Raises:
360 NotImplementedError: If the credentials' scopes can not be changed.
361 This can be avoided by checking :attr:`requires_scopes` before
362 calling this method.
363 """
364 raise NotImplementedError("This class does not require scoping.")
367def with_scopes_if_required(credentials, scopes, default_scopes=None):
368 """Creates a copy of the credentials with scopes if scoping is required.
370 This helper function is useful when you do not know (or care to know) the
371 specific type of credentials you are using (such as when you use
372 :func:`google.auth.default`). This function will call
373 :meth:`Scoped.with_scopes` if the credentials are scoped credentials and if
374 the credentials require scoping. Otherwise, it will return the credentials
375 as-is.
377 Args:
378 credentials (google.auth.credentials.Credentials): The credentials to
379 scope if necessary.
380 scopes (Sequence[str]): The list of scopes to use.
381 default_scopes (Sequence[str]): Default scopes passed by a
382 Google client library. Use 'scopes' for user-defined scopes.
384 Returns:
385 google.auth.credentials.Credentials: Either a new set of scoped
386 credentials, or the passed in credentials instance if no scoping
387 was required.
388 """
389 if isinstance(credentials, Scoped) and credentials.requires_scopes:
390 return credentials.with_scopes(scopes, default_scopes=default_scopes)
391 else:
392 return credentials
395class Signing(metaclass=abc.ABCMeta):
396 """Interface for credentials that can cryptographically sign messages."""
398 @abc.abstractmethod
399 def sign_bytes(self, message):
400 """Signs the given message.
402 Args:
403 message (bytes): The message to sign.
405 Returns:
406 bytes: The message's cryptographic signature.
407 """
408 # pylint: disable=missing-raises-doc,redundant-returns-doc
409 # (pylint doesn't recognize that this is abstract)
410 raise NotImplementedError("Sign bytes must be implemented.")
412 @abc.abstractproperty
413 def signer_email(self):
414 """Optional[str]: An email address that identifies the signer."""
415 # pylint: disable=missing-raises-doc
416 # (pylint doesn't recognize that this is abstract)
417 raise NotImplementedError("Signer email must be implemented.")
419 @abc.abstractproperty
420 def signer(self):
421 """google.auth.crypt.Signer: The signer used to sign bytes."""
422 # pylint: disable=missing-raises-doc
423 # (pylint doesn't recognize that this is abstract)
424 raise NotImplementedError("Signer must be implemented.")