Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/google/auth/credentials.py: 43%

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. 

14 

15 

16"""Interfaces for credentials.""" 

17 

18import abc 

19from enum import Enum 

20import os 

21from typing import List 

22 

23from google.auth import _helpers, environment_vars 

24from google.auth import exceptions 

25from google.auth import metrics 

26from google.auth._credentials_base import _BaseCredentials 

27from google.auth._default import _LOGGER 

28from google.auth._refresh_worker import RefreshThreadManager 

29 

30DEFAULT_UNIVERSE_DOMAIN = "googleapis.com" 

31NO_OP_TRUST_BOUNDARY_LOCATIONS: List[str] = [] 

32NO_OP_TRUST_BOUNDARY_ENCODED_LOCATIONS = "0x0" 

33 

34 

35class Credentials(_BaseCredentials): 

36 """Base class for all credentials. 

37 

38 All credentials have a :attr:`token` that is used for authentication and 

39 may also optionally set an :attr:`expiry` to indicate when the token will 

40 no longer be valid. 

41 

42 Most credentials will be :attr:`invalid` until :meth:`refresh` is called. 

43 Credentials can do this automatically before the first HTTP request in 

44 :meth:`before_request`. 

45 

46 Although the token and expiration will change as the credentials are 

47 :meth:`refreshed <refresh>` and used, credentials should be considered 

48 immutable. Various credentials will accept configuration such as private 

49 keys, scopes, and other options. These options are not changeable after 

50 construction. Some classes will provide mechanisms to copy the credentials 

51 with modifications such as :meth:`ScopedCredentials.with_scopes`. 

52 """ 

53 

54 def __init__(self): 

55 super(Credentials, self).__init__() 

56 

57 self.expiry = None 

58 """Optional[datetime]: When the token expires and is no longer valid. 

59 If this is None, the token is assumed to never expire.""" 

60 self._quota_project_id = None 

61 """Optional[str]: Project to use for quota and billing purposes.""" 

62 self._trust_boundary = None 

63 """Optional[dict]: Cache of a trust boundary response which has a list 

64 of allowed regions and an encoded string representation of credentials 

65 trust boundary.""" 

66 self._universe_domain = DEFAULT_UNIVERSE_DOMAIN 

67 """Optional[str]: The universe domain value, default is googleapis.com 

68 """ 

69 

70 self._use_non_blocking_refresh = False 

71 self._refresh_worker = RefreshThreadManager() 

72 

73 @property 

74 def expired(self): 

75 """Checks if the credentials are expired. 

76 

77 Note that credentials can be invalid but not expired because 

78 Credentials with :attr:`expiry` set to None is considered to never 

79 expire. 

80 

81 .. deprecated:: v2.24.0 

82 Prefer checking :attr:`token_state` instead. 

83 """ 

84 if not self.expiry: 

85 return False 

86 # Remove some threshold from expiry to err on the side of reporting 

87 # expiration early so that we avoid the 401-refresh-retry loop. 

88 skewed_expiry = self.expiry - _helpers.REFRESH_THRESHOLD 

89 return _helpers.utcnow() >= skewed_expiry 

90 

91 @property 

92 def valid(self): 

93 """Checks the validity of the credentials. 

94 

95 This is True if the credentials have a :attr:`token` and the token 

96 is not :attr:`expired`. 

97 

98 .. deprecated:: v2.24.0 

99 Prefer checking :attr:`token_state` instead. 

100 """ 

101 return self.token is not None and not self.expired 

102 

103 @property 

104 def token_state(self): 

105 """ 

106 See `:obj:`TokenState` 

107 """ 

108 if self.token is None: 

109 return TokenState.INVALID 

110 

111 # Credentials that can't expire are always treated as fresh. 

112 if self.expiry is None: 

113 return TokenState.FRESH 

114 

115 expired = _helpers.utcnow() >= self.expiry 

116 if expired: 

117 return TokenState.INVALID 

118 

119 is_stale = _helpers.utcnow() >= (self.expiry - _helpers.REFRESH_THRESHOLD) 

120 if is_stale: 

121 return TokenState.STALE 

122 

123 return TokenState.FRESH 

124 

125 @property 

126 def quota_project_id(self): 

127 """Project to use for quota and billing purposes.""" 

128 return self._quota_project_id 

129 

130 @property 

131 def universe_domain(self): 

132 """The universe domain value.""" 

133 return self._universe_domain 

134 

135 def get_cred_info(self): 

136 """The credential information JSON. 

137 

138 The credential information will be added to auth related error messages 

139 by client library. 

140 

141 Returns: 

142 Mapping[str, str]: The credential information JSON. 

143 """ 

144 return None 

145 

146 @abc.abstractmethod 

147 def refresh(self, request): 

148 """Refreshes the access token. 

149 

150 Args: 

151 request (google.auth.transport.Request): The object used to make 

152 HTTP requests. 

153 

154 Raises: 

155 google.auth.exceptions.RefreshError: If the credentials could 

156 not be refreshed. 

157 """ 

158 # pylint: disable=missing-raises-doc 

159 # (pylint doesn't recognize that this is abstract) 

160 raise NotImplementedError("Refresh must be implemented") 

161 

162 def _metric_header_for_usage(self): 

163 """The x-goog-api-client header for token usage metric. 

164 

165 This header will be added to the API service requests in before_request 

166 method. For example, "cred-type/sa-jwt" means service account self 

167 signed jwt access token is used in the API service request 

168 authorization header. Children credentials classes need to override 

169 this method to provide the header value, if the token usage metric is 

170 needed. 

171 

172 Returns: 

173 str: The x-goog-api-client header value. 

174 """ 

175 return None 

176 

177 def apply(self, headers, token=None): 

178 """Apply the token to the authentication header. 

179 

180 Args: 

181 headers (Mapping): The HTTP request headers. 

182 token (Optional[str]): If specified, overrides the current access 

183 token. 

184 """ 

185 self._apply(headers, token) 

186 if self.quota_project_id: 

187 headers["x-goog-user-project"] = self.quota_project_id 

188 

189 def _blocking_refresh(self, request): 

190 if not self.valid: 

191 self.refresh(request) 

192 

193 def _non_blocking_refresh(self, request): 

194 use_blocking_refresh_fallback = False 

195 

196 if self.token_state == TokenState.STALE: 

197 use_blocking_refresh_fallback = not self._refresh_worker.start_refresh( 

198 self, request 

199 ) 

200 

201 if self.token_state == TokenState.INVALID or use_blocking_refresh_fallback: 

202 self.refresh(request) 

203 # If the blocking refresh succeeds then we can clear the error info 

204 # on the background refresh worker, and perform refreshes in a 

205 # background thread. 

206 self._refresh_worker.clear_error() 

207 

208 def before_request(self, request, method, url, headers): 

209 """Performs credential-specific before request logic. 

210 

211 Refreshes the credentials if necessary, then calls :meth:`apply` to 

212 apply the token to the authentication header. 

213 

214 Args: 

215 request (google.auth.transport.Request): The object used to make 

216 HTTP requests. 

217 method (str): The request's HTTP method or the RPC method being 

218 invoked. 

219 url (str): The request's URI or the RPC service's URI. 

220 headers (Mapping): The request's headers. 

221 """ 

222 # pylint: disable=unused-argument 

223 # (Subclasses may use these arguments to ascertain information about 

224 # the http request.) 

225 if self._use_non_blocking_refresh: 

226 self._non_blocking_refresh(request) 

227 else: 

228 self._blocking_refresh(request) 

229 

230 metrics.add_metric_header(headers, self._metric_header_for_usage()) 

231 self.apply(headers) 

232 

233 def with_non_blocking_refresh(self): 

234 self._use_non_blocking_refresh = True 

235 

236 

237class CredentialsWithQuotaProject(Credentials): 

238 """Abstract base for credentials supporting ``with_quota_project`` factory""" 

239 

240 def with_quota_project(self, quota_project_id): 

241 """Returns a copy of these credentials with a modified quota project. 

242 

243 Args: 

244 quota_project_id (str): The project to use for quota and 

245 billing purposes 

246 

247 Returns: 

248 google.auth.credentials.Credentials: A new credentials instance. 

249 """ 

250 raise NotImplementedError("This credential does not support quota project.") 

251 

252 def with_quota_project_from_environment(self): 

253 quota_from_env = os.environ.get(environment_vars.GOOGLE_CLOUD_QUOTA_PROJECT) 

254 if quota_from_env: 

255 return self.with_quota_project(quota_from_env) 

256 return self 

257 

258 

259class CredentialsWithTokenUri(Credentials): 

260 """Abstract base for credentials supporting ``with_token_uri`` factory""" 

261 

262 def with_token_uri(self, token_uri): 

263 """Returns a copy of these credentials with a modified token uri. 

264 

265 Args: 

266 token_uri (str): The uri to use for fetching/exchanging tokens 

267 

268 Returns: 

269 google.auth.credentials.Credentials: A new credentials instance. 

270 """ 

271 raise NotImplementedError("This credential does not use token uri.") 

272 

273 

274class CredentialsWithUniverseDomain(Credentials): 

275 """Abstract base for credentials supporting ``with_universe_domain`` factory""" 

276 

277 def with_universe_domain(self, universe_domain): 

278 """Returns a copy of these credentials with a modified universe domain. 

279 

280 Args: 

281 universe_domain (str): The universe domain to use 

282 

283 Returns: 

284 google.auth.credentials.Credentials: A new credentials instance. 

285 """ 

286 raise NotImplementedError( 

287 "This credential does not support with_universe_domain." 

288 ) 

289 

290 

291class CredentialsWithTrustBoundary(Credentials): 

292 """Abstract base for credentials supporting ``with_trust_boundary`` factory""" 

293 

294 @abc.abstractmethod 

295 def _refresh_token(self, request): 

296 """Refreshes the access token. 

297 

298 Args: 

299 request (google.auth.transport.Request): The object used to make 

300 HTTP requests. 

301 

302 Raises: 

303 google.auth.exceptions.RefreshError: If the credentials could 

304 not be refreshed. 

305 """ 

306 raise NotImplementedError("_refresh_token must be implemented") 

307 

308 def with_trust_boundary(self, trust_boundary): 

309 """Returns a copy of these credentials with a modified trust boundary. 

310 

311 Args: 

312 trust_boundary Mapping[str, str]: The trust boundary to use for the 

313 credential. This should be a map with a "locations" key that maps to 

314 a list of GCP regions, and a "encodedLocations" key that maps to a 

315 hex string. 

316 

317 Returns: 

318 google.auth.credentials.Credentials: A new credentials instance. 

319 """ 

320 raise NotImplementedError("This credential does not support trust boundaries.") 

321 

322 def _is_trust_boundary_lookup_required(self): 

323 """Checks if a trust boundary lookup is required. 

324 

325 A lookup is required if the feature is enabled via an environment 

326 variable, the universe domain is supported, and a no-op boundary 

327 is not already cached. 

328 

329 Returns: 

330 bool: True if a trust boundary lookup is required, False otherwise. 

331 """ 

332 # 1. Check if the feature is enabled via environment variable. 

333 if not _helpers.get_bool_from_env( 

334 environment_vars.GOOGLE_AUTH_TRUST_BOUNDARY_ENABLED, default=False 

335 ): 

336 return False 

337 

338 # 2. Skip trust boundary flow for non-default universe domains. 

339 if self.universe_domain != DEFAULT_UNIVERSE_DOMAIN: 

340 return False 

341 

342 # 3. Do not trigger refresh if credential has a cached no-op trust boundary. 

343 return not self._has_no_op_trust_boundary() 

344 

345 def _get_trust_boundary_header(self): 

346 if self._trust_boundary is not None: 

347 if self._has_no_op_trust_boundary(): 

348 # STS expects an empty string if the trust boundary value is no-op. 

349 return {"x-allowed-locations": ""} 

350 else: 

351 return {"x-allowed-locations": self._trust_boundary["encodedLocations"]} 

352 return {} 

353 

354 def apply(self, headers, token=None): 

355 """Apply the token to the authentication header.""" 

356 super().apply(headers, token) 

357 headers.update(self._get_trust_boundary_header()) 

358 

359 def refresh(self, request): 

360 """Refreshes the access token and the trust boundary. 

361 

362 This method calls the subclass's token refresh logic and then 

363 refreshes the trust boundary if applicable. 

364 """ 

365 self._refresh_token(request) 

366 self._refresh_trust_boundary(request) 

367 

368 def _refresh_trust_boundary(self, request): 

369 """Triggers a refresh of the trust boundary and updates the cache if necessary. 

370 

371 Args: 

372 request (google.auth.transport.Request): The object used to make 

373 HTTP requests. 

374 

375 Raises: 

376 google.auth.exceptions.RefreshError: If the trust boundary could 

377 not be refreshed and no cached value is available. 

378 """ 

379 if not self._is_trust_boundary_lookup_required(): 

380 return 

381 try: 

382 self._trust_boundary = self._lookup_trust_boundary(request) 

383 except exceptions.RefreshError as error: 

384 # If the call to the lookup API failed, check if there is a trust boundary 

385 # already cached. If there is, do nothing. If not, then throw the error. 

386 if self._trust_boundary is None: 

387 raise error 

388 if _helpers.is_logging_enabled(_LOGGER): 

389 _LOGGER.debug( 

390 "Using cached trust boundary due to refresh error: %s", error 

391 ) 

392 return 

393 

394 def _lookup_trust_boundary(self, request): 

395 """Calls the trust boundary lookup API to refresh the trust boundary cache. 

396 

397 Args: 

398 request (google.auth.transport.Request): The object used to make 

399 HTTP requests. 

400 

401 Returns: 

402 trust_boundary (dict): The trust boundary object returned by the lookup API. 

403 

404 Raises: 

405 google.auth.exceptions.RefreshError: If the trust boundary could not be 

406 retrieved. 

407 """ 

408 from google.oauth2 import _client 

409 

410 url = self._build_trust_boundary_lookup_url() 

411 if not url: 

412 raise exceptions.InvalidValue("Failed to build trust boundary lookup URL.") 

413 

414 headers = {} 

415 self._apply(headers) 

416 headers.update(self._get_trust_boundary_header()) 

417 return _client._lookup_trust_boundary(request, url, headers=headers) 

418 

419 @abc.abstractmethod 

420 def _build_trust_boundary_lookup_url(self): 

421 """ 

422 Builds and returns the URL for the trust boundary lookup API. 

423 

424 This method should be implemented by subclasses to provide the 

425 specific URL based on the credential type and its properties. 

426 

427 Returns: 

428 str: The URL for the trust boundary lookup endpoint, or None 

429 if lookup should be skipped (e.g., for non-applicable universe domains). 

430 """ 

431 raise NotImplementedError( 

432 "_build_trust_boundary_lookup_url must be implemented" 

433 ) 

434 

435 def _has_no_op_trust_boundary(self): 

436 # A no-op trust boundary is indicated by encodedLocations being "0x0". 

437 # The "locations" list may or may not be present as an empty list. 

438 if self._trust_boundary is None: 

439 return False 

440 return ( 

441 self._trust_boundary.get("encodedLocations") 

442 == NO_OP_TRUST_BOUNDARY_ENCODED_LOCATIONS 

443 ) 

444 

445 

446class AnonymousCredentials(Credentials): 

447 """Credentials that do not provide any authentication information. 

448 

449 These are useful in the case of services that support anonymous access or 

450 local service emulators that do not use credentials. 

451 """ 

452 

453 @property 

454 def expired(self): 

455 """Returns `False`, anonymous credentials never expire.""" 

456 return False 

457 

458 @property 

459 def valid(self): 

460 """Returns `True`, anonymous credentials are always valid.""" 

461 return True 

462 

463 def refresh(self, request): 

464 """Raises :class:``InvalidOperation``, anonymous credentials cannot be 

465 refreshed.""" 

466 raise exceptions.InvalidOperation("Anonymous credentials cannot be refreshed.") 

467 

468 def apply(self, headers, token=None): 

469 """Anonymous credentials do nothing to the request. 

470 

471 The optional ``token`` argument is not supported. 

472 

473 Raises: 

474 google.auth.exceptions.InvalidValue: If a token was specified. 

475 """ 

476 if token is not None: 

477 raise exceptions.InvalidValue("Anonymous credentials don't support tokens.") 

478 

479 def before_request(self, request, method, url, headers): 

480 """Anonymous credentials do nothing to the request.""" 

481 

482 

483class ReadOnlyScoped(metaclass=abc.ABCMeta): 

484 """Interface for credentials whose scopes can be queried. 

485 

486 OAuth 2.0-based credentials allow limiting access using scopes as described 

487 in `RFC6749 Section 3.3`_. 

488 If a credential class implements this interface then the credentials either 

489 use scopes in their implementation. 

490 

491 Some credentials require scopes in order to obtain a token. You can check 

492 if scoping is necessary with :attr:`requires_scopes`:: 

493 

494 if credentials.requires_scopes: 

495 # Scoping is required. 

496 credentials = credentials.with_scopes(scopes=['one', 'two']) 

497 

498 Credentials that require scopes must either be constructed with scopes:: 

499 

500 credentials = SomeScopedCredentials(scopes=['one', 'two']) 

501 

502 Or must copy an existing instance using :meth:`with_scopes`:: 

503 

504 scoped_credentials = credentials.with_scopes(scopes=['one', 'two']) 

505 

506 Some credentials have scopes but do not allow or require scopes to be set, 

507 these credentials can be used as-is. 

508 

509 .. _RFC6749 Section 3.3: https://tools.ietf.org/html/rfc6749#section-3.3 

510 """ 

511 

512 def __init__(self): 

513 super(ReadOnlyScoped, self).__init__() 

514 self._scopes = None 

515 self._default_scopes = None 

516 

517 @property 

518 def scopes(self): 

519 """Sequence[str]: the credentials' current set of scopes.""" 

520 return self._scopes 

521 

522 @property 

523 def default_scopes(self): 

524 """Sequence[str]: the credentials' current set of default scopes.""" 

525 return self._default_scopes 

526 

527 @abc.abstractproperty 

528 def requires_scopes(self): 

529 """True if these credentials require scopes to obtain an access token.""" 

530 return False 

531 

532 def has_scopes(self, scopes): 

533 """Checks if the credentials have the given scopes. 

534 

535 .. warning: This method is not guaranteed to be accurate if the 

536 credentials are :attr:`~Credentials.invalid`. 

537 

538 Args: 

539 scopes (Sequence[str]): The list of scopes to check. 

540 

541 Returns: 

542 bool: True if the credentials have the given scopes. 

543 """ 

544 credential_scopes = ( 

545 self._scopes if self._scopes is not None else self._default_scopes 

546 ) 

547 return set(scopes).issubset(set(credential_scopes or [])) 

548 

549 

550class Scoped(ReadOnlyScoped): 

551 """Interface for credentials whose scopes can be replaced while copying. 

552 

553 OAuth 2.0-based credentials allow limiting access using scopes as described 

554 in `RFC6749 Section 3.3`_. 

555 If a credential class implements this interface then the credentials either 

556 use scopes in their implementation. 

557 

558 Some credentials require scopes in order to obtain a token. You can check 

559 if scoping is necessary with :attr:`requires_scopes`:: 

560 

561 if credentials.requires_scopes: 

562 # Scoping is required. 

563 credentials = credentials.create_scoped(['one', 'two']) 

564 

565 Credentials that require scopes must either be constructed with scopes:: 

566 

567 credentials = SomeScopedCredentials(scopes=['one', 'two']) 

568 

569 Or must copy an existing instance using :meth:`with_scopes`:: 

570 

571 scoped_credentials = credentials.with_scopes(scopes=['one', 'two']) 

572 

573 Some credentials have scopes but do not allow or require scopes to be set, 

574 these credentials can be used as-is. 

575 

576 .. _RFC6749 Section 3.3: https://tools.ietf.org/html/rfc6749#section-3.3 

577 """ 

578 

579 @abc.abstractmethod 

580 def with_scopes(self, scopes, default_scopes=None): 

581 """Create a copy of these credentials with the specified scopes. 

582 

583 Args: 

584 scopes (Sequence[str]): The list of scopes to attach to the 

585 current credentials. 

586 

587 Raises: 

588 NotImplementedError: If the credentials' scopes can not be changed. 

589 This can be avoided by checking :attr:`requires_scopes` before 

590 calling this method. 

591 """ 

592 raise NotImplementedError("This class does not require scoping.") 

593 

594 

595def with_scopes_if_required(credentials, scopes, default_scopes=None): 

596 """Creates a copy of the credentials with scopes if scoping is required. 

597 

598 This helper function is useful when you do not know (or care to know) the 

599 specific type of credentials you are using (such as when you use 

600 :func:`google.auth.default`). This function will call 

601 :meth:`Scoped.with_scopes` if the credentials are scoped credentials and if 

602 the credentials require scoping. Otherwise, it will return the credentials 

603 as-is. 

604 

605 Args: 

606 credentials (google.auth.credentials.Credentials): The credentials to 

607 scope if necessary. 

608 scopes (Sequence[str]): The list of scopes to use. 

609 default_scopes (Sequence[str]): Default scopes passed by a 

610 Google client library. Use 'scopes' for user-defined scopes. 

611 

612 Returns: 

613 google.auth.credentials.Credentials: Either a new set of scoped 

614 credentials, or the passed in credentials instance if no scoping 

615 was required. 

616 """ 

617 if isinstance(credentials, Scoped) and credentials.requires_scopes: 

618 return credentials.with_scopes(scopes, default_scopes=default_scopes) 

619 else: 

620 return credentials 

621 

622 

623class Signing(metaclass=abc.ABCMeta): 

624 """Interface for credentials that can cryptographically sign messages.""" 

625 

626 @abc.abstractmethod 

627 def sign_bytes(self, message): 

628 """Signs the given message. 

629 

630 Args: 

631 message (bytes): The message to sign. 

632 

633 Returns: 

634 bytes: The message's cryptographic signature. 

635 """ 

636 # pylint: disable=missing-raises-doc,redundant-returns-doc 

637 # (pylint doesn't recognize that this is abstract) 

638 raise NotImplementedError("Sign bytes must be implemented.") 

639 

640 @abc.abstractproperty 

641 def signer_email(self): 

642 """Optional[str]: An email address that identifies the signer.""" 

643 # pylint: disable=missing-raises-doc 

644 # (pylint doesn't recognize that this is abstract) 

645 raise NotImplementedError("Signer email must be implemented.") 

646 

647 @abc.abstractproperty 

648 def signer(self): 

649 """google.auth.crypt.Signer: The signer used to sign bytes.""" 

650 # pylint: disable=missing-raises-doc 

651 # (pylint doesn't recognize that this is abstract) 

652 raise NotImplementedError("Signer must be implemented.") 

653 

654 

655class TokenState(Enum): 

656 """ 

657 Tracks the state of a token. 

658 FRESH: The token is valid. It is not expired or close to expired, or the token has no expiry. 

659 STALE: The token is close to expired, and should be refreshed. The token can be used normally. 

660 INVALID: The token is expired or invalid. The token cannot be used for a normal operation. 

661 """ 

662 

663 FRESH = 1 

664 STALE = 2 

665 INVALID = 3