Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/google/oauth2/service

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

7# http://www.apache.org/licenses/LICENSE-2.0

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.

15"""Service Accounts: JSON Web Token (JWT) Profile for OAuth 2.0

17This module implements the JWT Profile for OAuth 2.0 Authorization Grants

18as defined by `RFC 7523`_ with particular support for how this RFC is

19implemented in Google's infrastructure. Google refers to these credentials

20as *Service Accounts*.

22Service accounts are used for server-to-server communication, such as

23interactions between a web application server and a Google service. The

24service account belongs to your application instead of to an individual end

25user. In contrast to other OAuth 2.0 profiles, no users are involved and your

26application "acts" as the service account.

28Typically an application uses a service account when the application uses

29Google APIs to work with its own data rather than a user's data. For example,

30an application that uses Google Cloud Datastore for data persistence would use

31a service account to authenticate its calls to the Google Cloud Datastore API.

32However, an application that needs to access a user's Drive documents would

33use the normal OAuth 2.0 profile.

35Additionally, Google Apps domain administrators can grant service accounts

36`domain-wide delegation`_ authority to access user data on behalf of users in

37the domain.

39This profile uses a JWT to acquire an OAuth 2.0 access token. The JWT is used

40in place of the usual authorization token returned during the standard

41OAuth 2.0 Authorization Code grant. The JWT is only used for this purpose, as

42the acquired access token is used as the bearer token when making requests

43using these credentials.

45This profile differs from normal OAuth 2.0 profile because no user consent

46step is required. The use of the private key allows this profile to assert

47identity directly.

49This profile also differs from the :mod:`google.auth.jwt` authentication

50because the JWT credentials use the JWT directly as the bearer token. This

51profile instead only uses the JWT to obtain an OAuth 2.0 access token. The

52obtained OAuth 2.0 access token is used as the bearer token.

54Domain-wide delegation

55----------------------

57Domain-wide delegation allows a service account to access user data on

58behalf of any user in a Google Apps domain without consent from the user.

59For example, an application that uses the Google Calendar API to add events to

60the calendars of all users in a Google Apps domain would use a service account

61to access the Google Calendar API on behalf of users.

63The Google Apps administrator must explicitly authorize the service account to

64do this. This authorization step is referred to as "delegating domain-wide

65authority" to a service account.

67You can use domain-wise delegation by creating a set of credentials with a

68specific subject using :meth:`~Credentials.with_subject`.

70.. _RFC 7523: https://tools.ietf.org/html/rfc7523

71"""

73import copy

74import datetime

76from google.auth import _helpers

77from google.auth import _service_account_info

78from google.auth import credentials

79from google.auth import exceptions

80from google.auth import iam

81from google.auth import jwt

82from google.auth import metrics

83from google.oauth2 import _client

85_DEFAULT_TOKEN_LIFETIME_SECS = 3600 # 1 hour in seconds

86_GOOGLE_OAUTH2_TOKEN_ENDPOINT = "https://oauth2.googleapis.com/token"

89class Credentials(

90 credentials.Signing,

91 credentials.Scoped,

92 credentials.CredentialsWithQuotaProject,

93 credentials.CredentialsWithTokenUri,

94):

95 """Service account credentials

97 Usually, you'll create these credentials with one of the helper

98 constructors. To create credentials using a Google service account

99 private key JSON file::

100

101 credentials = service_account.Credentials.from_service_account_file(

102 'service-account.json')

103

104 Or if you already have the service account file loaded::

105

106 service_account_info = json.load(open('service_account.json'))

107 credentials = service_account.Credentials.from_service_account_info(

108 service_account_info)

109

110 Both helper methods pass on arguments to the constructor, so you can

111 specify additional scopes and a subject if necessary::

112

113 credentials = service_account.Credentials.from_service_account_file(

114 'service-account.json',

115 scopes=['email'],

116 subject='user@example.com')

117

118 The credentials are considered immutable. If you want to modify the scopes

119 or the subject used for delegation, use :meth:`with_scopes` or

120 :meth:`with_subject`::

121

122 scoped_credentials = credentials.with_scopes(['email'])

123 delegated_credentials = credentials.with_subject(subject)

124

125 To add a quota project, use :meth:`with_quota_project`::

126

127 credentials = credentials.with_quota_project('myproject-123')

128 """

129

130 def __init__(

131 self,

132 signer,

133 service_account_email,

134 token_uri,

135 scopes=None,

136 default_scopes=None,

137 subject=None,

138 project_id=None,

139 quota_project_id=None,

140 additional_claims=None,

141 always_use_jwt_access=False,

142 universe_domain=credentials.DEFAULT_UNIVERSE_DOMAIN,

143 trust_boundary=None,

144 ):

145 """

146 Args:

147 signer (google.auth.crypt.Signer): The signer used to sign JWTs.

148 service_account_email (str): The service account's email.

149 scopes (Sequence[str]): User-defined scopes to request during the

150 authorization grant.

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

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

153 token_uri (str): The OAuth 2.0 Token URI.

154 subject (str): For domain-wide delegation, the email address of the

155 user to for which to request delegated access.

156 project_id (str): Project ID associated with the service account

157 credential.

158 quota_project_id (Optional[str]): The project ID used for quota and

159 billing.

160 additional_claims (Mapping[str, str]): Any additional claims for

161 the JWT assertion used in the authorization grant.

162 always_use_jwt_access (Optional[bool]): Whether self signed JWT should

163 be always used.

164 universe_domain (str): The universe domain. The default

165 universe domain is googleapis.com. For default value self

166 signed jwt is used for token refresh.

167 trust_boundary (str): String representation of trust boundary meta.

168

169 .. note:: Typically one of the helper constructors

170 :meth:`from_service_account_file` or

171 :meth:`from_service_account_info` are used instead of calling the

172 constructor directly.

173 """

174 super(Credentials, self).__init__()

175

176 self._cred_file_path = None

177 self._scopes = scopes

178 self._default_scopes = default_scopes

179 self._signer = signer

180 self._service_account_email = service_account_email

181 self._subject = subject

182 self._project_id = project_id

183 self._quota_project_id = quota_project_id

184 self._token_uri = token_uri

185 self._always_use_jwt_access = always_use_jwt_access

186 self._universe_domain = universe_domain or credentials.DEFAULT_UNIVERSE_DOMAIN

187

188 if universe_domain != credentials.DEFAULT_UNIVERSE_DOMAIN:

189 self._always_use_jwt_access = True

190

191 self._jwt_credentials = None

192

193 if additional_claims is not None:

194 self._additional_claims = additional_claims

195 else:

196 self._additional_claims = {}

197 self._trust_boundary = {"locations": [], "encoded_locations": "0x0"}

198

199 @classmethod

200 def _from_signer_and_info(cls, signer, info, **kwargs):

201 """Creates a Credentials instance from a signer and service account

202 info.

203

204 Args:

205 signer (google.auth.crypt.Signer): The signer used to sign JWTs.

206 info (Mapping[str, str]): The service account info.

207 kwargs: Additional arguments to pass to the constructor.

208

209 Returns:

210 google.auth.jwt.Credentials: The constructed credentials.

211

212 Raises:

213 ValueError: If the info is not in the expected format.

214 """

215 return cls(

216 signer,

217 service_account_email=info["client_email"],

218 token_uri=info["token_uri"],

219 project_id=info.get("project_id"),

220 universe_domain=info.get(

221 "universe_domain", credentials.DEFAULT_UNIVERSE_DOMAIN

222 ),

223 trust_boundary=info.get("trust_boundary"),

224 **kwargs,

225 )

226

227 @classmethod

228 def from_service_account_info(cls, info, **kwargs):

229 """Creates a Credentials instance from parsed service account info.

230

231 Args:

232 info (Mapping[str, str]): The service account info in Google

233 format.

234 kwargs: Additional arguments to pass to the constructor.

235

236 Returns:

237 google.auth.service_account.Credentials: The constructed

238 credentials.

239

240 Raises:

241 ValueError: If the info is not in the expected format.

242 """

243 signer = _service_account_info.from_dict(

244 info, require=["client_email", "token_uri"]

245 )

246 return cls._from_signer_and_info(signer, info, **kwargs)

247

248 @classmethod

249 def from_service_account_file(cls, filename, **kwargs):

250 """Creates a Credentials instance from a service account json file.

251

252 Args:

253 filename (str): The path to the service account json file.

254 kwargs: Additional arguments to pass to the constructor.

255

256 Returns:

257 google.auth.service_account.Credentials: The constructed

258 credentials.

259 """

260 info, signer = _service_account_info.from_filename(

261 filename, require=["client_email", "token_uri"]

262 )

263 return cls._from_signer_and_info(signer, info, **kwargs)

264

265 @property

266 def service_account_email(self):

267 """The service account email."""

268 return self._service_account_email

269

270 @property

271 def project_id(self):

272 """Project ID associated with this credential."""

273 return self._project_id

274

275 @property

276 def requires_scopes(self):

277 """Checks if the credentials requires scopes.

278

279 Returns:

280 bool: True if there are no scopes set otherwise False.

281 """

282 return True if not self._scopes else False

283

284 def _make_copy(self):

285 cred = self.__class__(

286 self._signer,

287 service_account_email=self._service_account_email,

288 scopes=copy.copy(self._scopes),

289 default_scopes=copy.copy(self._default_scopes),

290 token_uri=self._token_uri,

291 subject=self._subject,

292 project_id=self._project_id,

293 quota_project_id=self._quota_project_id,

294 additional_claims=self._additional_claims.copy(),

295 always_use_jwt_access=self._always_use_jwt_access,

296 universe_domain=self._universe_domain,

297 )

298 cred._cred_file_path = self._cred_file_path

299 return cred

300

301 @_helpers.copy_docstring(credentials.Scoped)

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

303 cred = self._make_copy()

304 cred._scopes = scopes

305 cred._default_scopes = default_scopes

306 return cred

307

308 def with_always_use_jwt_access(self, always_use_jwt_access):

309 """Create a copy of these credentials with the specified always_use_jwt_access value.

310

311 Args:

312 always_use_jwt_access (bool): Whether always use self signed JWT or not.

313

314 Returns:

315 google.auth.service_account.Credentials: A new credentials

316 instance.

317 Raises:

318 google.auth.exceptions.InvalidValue: If the universe domain is not

319 default and always_use_jwt_access is False.

320 """

321 cred = self._make_copy()

322 if (

323 cred._universe_domain != credentials.DEFAULT_UNIVERSE_DOMAIN

324 and not always_use_jwt_access

325 ):

326 raise exceptions.InvalidValue(

327 "always_use_jwt_access should be True for non-default universe domain"

328 )

329 cred._always_use_jwt_access = always_use_jwt_access

330 return cred

331

332 @_helpers.copy_docstring(credentials.CredentialsWithUniverseDomain)

333 def with_universe_domain(self, universe_domain):

334 cred = self._make_copy()

335 cred._universe_domain = universe_domain

336 if universe_domain != credentials.DEFAULT_UNIVERSE_DOMAIN:

337 cred._always_use_jwt_access = True

338 return cred

339

340 def with_subject(self, subject):

341 """Create a copy of these credentials with the specified subject.

342

343 Args:

344 subject (str): The subject claim.

345

346 Returns:

347 google.auth.service_account.Credentials: A new credentials

348 instance.

349 """

350 cred = self._make_copy()

351 cred._subject = subject

352 return cred

353

354 def with_claims(self, additional_claims):

355 """Returns a copy of these credentials with modified claims.

356

357 Args:

358 additional_claims (Mapping[str, str]): Any additional claims for

359 the JWT payload. This will be merged with the current

360 additional claims.

361

362 Returns:

363 google.auth.service_account.Credentials: A new credentials

364 instance.

365 """

366 new_additional_claims = copy.deepcopy(self._additional_claims)

367 new_additional_claims.update(additional_claims or {})

368 cred = self._make_copy()

369 cred._additional_claims = new_additional_claims

370 return cred

371

372 @_helpers.copy_docstring(credentials.CredentialsWithQuotaProject)

373 def with_quota_project(self, quota_project_id):

374 cred = self._make_copy()

375 cred._quota_project_id = quota_project_id

376 return cred

377

378 @_helpers.copy_docstring(credentials.CredentialsWithTokenUri)

379 def with_token_uri(self, token_uri):

380 cred = self._make_copy()

381 cred._token_uri = token_uri

382 return cred

383

384 def _make_authorization_grant_assertion(self):

385 """Create the OAuth 2.0 assertion.

386

387 This assertion is used during the OAuth 2.0 grant to acquire an

388 access token.

389

390 Returns:

391 bytes: The authorization grant assertion.

392 """

393 now = _helpers.utcnow()

394 lifetime = datetime.timedelta(seconds=_DEFAULT_TOKEN_LIFETIME_SECS)

395 expiry = now + lifetime

396

397 payload = {

398 "iat": _helpers.datetime_to_secs(now),

399 "exp": _helpers.datetime_to_secs(expiry),

400 # The issuer must be the service account email.

401 "iss": self._service_account_email,

402 # The audience must be the auth token endpoint's URI

403 "aud": _GOOGLE_OAUTH2_TOKEN_ENDPOINT,

404 "scope": _helpers.scopes_to_string(self._scopes or ()),

405 }

406

407 payload.update(self._additional_claims)

408

409 # The subject can be a user email for domain-wide delegation.

410 if self._subject:

411 payload.setdefault("sub", self._subject)

412

413 token = jwt.encode(self._signer, payload)

414

415 return token

416

417 def _use_self_signed_jwt(self):

418 # Since domain wide delegation doesn't work with self signed JWT. If

419 # subject exists, then we should not use self signed JWT.

420 return self._subject is None and self._jwt_credentials is not None

421

422 def _metric_header_for_usage(self):

423 if self._use_self_signed_jwt():

424 return metrics.CRED_TYPE_SA_JWT

425 return metrics.CRED_TYPE_SA_ASSERTION

426

427 @_helpers.copy_docstring(credentials.Credentials)

428 def refresh(self, request):

429 if self._always_use_jwt_access and not self._jwt_credentials:

430 # If self signed jwt should be used but jwt credential is not

431 # created, try to create one with scopes

432 self._create_self_signed_jwt(None)

433

434 if (

435 self._universe_domain != credentials.DEFAULT_UNIVERSE_DOMAIN

436 and self._subject

437 ):

438 raise exceptions.RefreshError(

439 "domain wide delegation is not supported for non-default universe domain"

440 )

441

442 if self._use_self_signed_jwt():

443 self._jwt_credentials.refresh(request)

444 self.token = self._jwt_credentials.token.decode()

445 self.expiry = self._jwt_credentials.expiry

446 else:

447 assertion = self._make_authorization_grant_assertion()

448 access_token, expiry, _ = _client.jwt_grant(

449 request, self._token_uri, assertion

450 )

451 self.token = access_token

452 self.expiry = expiry

453

454 def _create_self_signed_jwt(self, audience):

455 """Create a self-signed JWT from the credentials if requirements are met.

456

457 Args:

458 audience (str): The service URL. ``https://[API_ENDPOINT]/``

459 """

460 # https://google.aip.dev/auth/4111

461 if self._always_use_jwt_access:

462 if self._scopes:

463 additional_claims = {"scope": " ".join(self._scopes)}

464 if (

465 self._jwt_credentials is None

466 or self._jwt_credentials.additional_claims != additional_claims

467 ):

468 self._jwt_credentials = jwt.Credentials.from_signing_credentials(

469 self, None, additional_claims=additional_claims

470 )

471 elif audience:

472 if (

473 self._jwt_credentials is None

474 or self._jwt_credentials._audience != audience

475 ):

476

477 self._jwt_credentials = jwt.Credentials.from_signing_credentials(

478 self, audience

479 )

480 elif self._default_scopes:

481 additional_claims = {"scope": " ".join(self._default_scopes)}

482 if (

483 self._jwt_credentials is None

484 or additional_claims != self._jwt_credentials.additional_claims

485 ):

486 self._jwt_credentials = jwt.Credentials.from_signing_credentials(

487 self, None, additional_claims=additional_claims

488 )

489 elif not self._scopes and audience:

490 self._jwt_credentials = jwt.Credentials.from_signing_credentials(

491 self, audience

492 )

493

494 @_helpers.copy_docstring(credentials.Signing)

495 def sign_bytes(self, message):

496 return self._signer.sign(message)

497

498 @property # type: ignore

499 @_helpers.copy_docstring(credentials.Signing)

500 def signer(self):

501 return self._signer

502

503 @property # type: ignore

504 @_helpers.copy_docstring(credentials.Signing)

505 def signer_email(self):

506 return self._service_account_email

507

508 @_helpers.copy_docstring(credentials.Credentials)

509 def get_cred_info(self):

510 if self._cred_file_path:

511 return {

512 "credential_source": self._cred_file_path,

513 "credential_type": "service account credentials",

514 "principal": self.service_account_email,

515 }

516 return None

517

518

519class IDTokenCredentials(

520 credentials.Signing,

521 credentials.CredentialsWithQuotaProject,

522 credentials.CredentialsWithTokenUri,

523):

524 """Open ID Connect ID Token-based service account credentials.

525

526 These credentials are largely similar to :class:`.Credentials`, but instead

527 of using an OAuth 2.0 Access Token as the bearer token, they use an Open

528 ID Connect ID Token as the bearer token. These credentials are useful when

529 communicating to services that require ID Tokens and can not accept access

530 tokens.

531

532 Usually, you'll create these credentials with one of the helper

533 constructors. To create credentials using a Google service account

534 private key JSON file::

535

536 credentials = (

537 service_account.IDTokenCredentials.from_service_account_file(

538 'service-account.json'))

539

540

541 Or if you already have the service account file loaded::

542

543 service_account_info = json.load(open('service_account.json'))

544 credentials = (

545 service_account.IDTokenCredentials.from_service_account_info(

546 service_account_info))

547

548

549 Both helper methods pass on arguments to the constructor, so you can

550 specify additional scopes and a subject if necessary::

551

552 credentials = (

553 service_account.IDTokenCredentials.from_service_account_file(

554 'service-account.json',

555 scopes=['email'],

556 subject='user@example.com'))

557

558

559 The credentials are considered immutable. If you want to modify the scopes

560 or the subject used for delegation, use :meth:`with_scopes` or

561 :meth:`with_subject`::

562

563 scoped_credentials = credentials.with_scopes(['email'])

564 delegated_credentials = credentials.with_subject(subject)

565

566 """

567

568 def __init__(

569 self,

570 signer,

571 service_account_email,

572 token_uri,

573 target_audience,

574 additional_claims=None,

575 quota_project_id=None,

576 universe_domain=credentials.DEFAULT_UNIVERSE_DOMAIN,

577 ):

578 """

579 Args:

580 signer (google.auth.crypt.Signer): The signer used to sign JWTs.

581 service_account_email (str): The service account's email.

582 token_uri (str): The OAuth 2.0 Token URI.

583 target_audience (str): The intended audience for these credentials,

584 used when requesting the ID Token. The ID Token's ``aud`` claim

585 will be set to this string.

586 additional_claims (Mapping[str, str]): Any additional claims for

587 the JWT assertion used in the authorization grant.

588 quota_project_id (Optional[str]): The project ID used for quota and billing.

589 universe_domain (str): The universe domain. The default

590 universe domain is googleapis.com. For default value IAM ID

591 token endponint is used for token refresh. Note that

592 iam.serviceAccountTokenCreator role is required to use the IAM

593 endpoint.

594 .. note:: Typically one of the helper constructors

595 :meth:`from_service_account_file` or

596 :meth:`from_service_account_info` are used instead of calling the

597 constructor directly.

598 """

599 super(IDTokenCredentials, self).__init__()

600 self._signer = signer

601 self._service_account_email = service_account_email

602 self._token_uri = token_uri

603 self._target_audience = target_audience

604 self._quota_project_id = quota_project_id

605 self._use_iam_endpoint = False

606

607 if not universe_domain:

608 self._universe_domain = credentials.DEFAULT_UNIVERSE_DOMAIN

609 else:

610 self._universe_domain = universe_domain

611 self._iam_id_token_endpoint = iam._IAM_IDTOKEN_ENDPOINT.replace(

612 "googleapis.com", self._universe_domain

613 )

614

615 if self._universe_domain != credentials.DEFAULT_UNIVERSE_DOMAIN:

616 self._use_iam_endpoint = True

617

618 if additional_claims is not None:

619 self._additional_claims = additional_claims

620 else:

621 self._additional_claims = {}

622

623 @classmethod

624 def _from_signer_and_info(cls, signer, info, **kwargs):

625 """Creates a credentials instance from a signer and service account

626 info.

627

628 Args:

629 signer (google.auth.crypt.Signer): The signer used to sign JWTs.

630 info (Mapping[str, str]): The service account info.

631 kwargs: Additional arguments to pass to the constructor.

632

633 Returns:

634 google.auth.jwt.IDTokenCredentials: The constructed credentials.

635

636 Raises:

637 ValueError: If the info is not in the expected format.

638 """

639 kwargs.setdefault("service_account_email", info["client_email"])

640 kwargs.setdefault("token_uri", info["token_uri"])

641 if "universe_domain" in info:

642 kwargs["universe_domain"] = info["universe_domain"]

643 return cls(signer, **kwargs)

644

645 @classmethod

646 def from_service_account_info(cls, info, **kwargs):

647 """Creates a credentials instance from parsed service account info.

648

649 Args:

650 info (Mapping[str, str]): The service account info in Google

651 format.

652 kwargs: Additional arguments to pass to the constructor.

653

654 Returns:

655 google.auth.service_account.IDTokenCredentials: The constructed

656 credentials.

657

658 Raises:

659 ValueError: If the info is not in the expected format.

660 """

661 signer = _service_account_info.from_dict(

662 info, require=["client_email", "token_uri"]

663 )

664 return cls._from_signer_and_info(signer, info, **kwargs)

665

666 @classmethod

667 def from_service_account_file(cls, filename, **kwargs):

668 """Creates a credentials instance from a service account json file.

669

670 Args:

671 filename (str): The path to the service account json file.

672 kwargs: Additional arguments to pass to the constructor.

673

674 Returns:

675 google.auth.service_account.IDTokenCredentials: The constructed

676 credentials.

677 """

678 info, signer = _service_account_info.from_filename(

679 filename, require=["client_email", "token_uri"]

680 )

681 return cls._from_signer_and_info(signer, info, **kwargs)

682

683 def _make_copy(self):

684 cred = self.__class__(

685 self._signer,

686 service_account_email=self._service_account_email,

687 token_uri=self._token_uri,

688 target_audience=self._target_audience,

689 additional_claims=self._additional_claims.copy(),

690 quota_project_id=self.quota_project_id,

691 universe_domain=self._universe_domain,

692 )

693 # _use_iam_endpoint is not exposed in the constructor

694 cred._use_iam_endpoint = self._use_iam_endpoint

695 return cred

696

697 def with_target_audience(self, target_audience):

698 """Create a copy of these credentials with the specified target

699 audience.

700

701 Args:

702 target_audience (str): The intended audience for these credentials,

703 used when requesting the ID Token.

704

705 Returns:

706 google.auth.service_account.IDTokenCredentials: A new credentials

707 instance.

708 """

709 cred = self._make_copy()

710 cred._target_audience = target_audience

711 return cred

712

713 def _with_use_iam_endpoint(self, use_iam_endpoint):

714 """Create a copy of these credentials with the use_iam_endpoint value.

715

716 Args:

717 use_iam_endpoint (bool): If True, IAM generateIdToken endpoint will

718 be used instead of the token_uri. Note that

719 iam.serviceAccountTokenCreator role is required to use the IAM

720 endpoint. The default value is False. This feature is currently

721 experimental and subject to change without notice.

722

723 Returns:

724 google.auth.service_account.IDTokenCredentials: A new credentials

725 instance.

726 Raises:

727 google.auth.exceptions.InvalidValue: If the universe domain is not

728 default and use_iam_endpoint is False.

729 """

730 cred = self._make_copy()

731 if (

732 cred._universe_domain != credentials.DEFAULT_UNIVERSE_DOMAIN

733 and not use_iam_endpoint

734 ):

735 raise exceptions.InvalidValue(

736 "use_iam_endpoint should be True for non-default universe domain"

737 )

738 cred._use_iam_endpoint = use_iam_endpoint

739 return cred

740

741 @_helpers.copy_docstring(credentials.CredentialsWithQuotaProject)

742 def with_quota_project(self, quota_project_id):

743 cred = self._make_copy()

744 cred._quota_project_id = quota_project_id

745 return cred

746

747 @_helpers.copy_docstring(credentials.CredentialsWithTokenUri)

748 def with_token_uri(self, token_uri):

749 cred = self._make_copy()

750 cred._token_uri = token_uri

751 return cred

752

753 def _make_authorization_grant_assertion(self):

754 """Create the OAuth 2.0 assertion.

755

756 This assertion is used during the OAuth 2.0 grant to acquire an

757 ID token.

758

759 Returns:

760 bytes: The authorization grant assertion.

761 """

762 now = _helpers.utcnow()

763 lifetime = datetime.timedelta(seconds=_DEFAULT_TOKEN_LIFETIME_SECS)

764 expiry = now + lifetime

765

766 payload = {

767 "iat": _helpers.datetime_to_secs(now),

768 "exp": _helpers.datetime_to_secs(expiry),

769 # The issuer must be the service account email.

770 "iss": self.service_account_email,

771 # The audience must be the auth token endpoint's URI

772 "aud": _GOOGLE_OAUTH2_TOKEN_ENDPOINT,

773 # The target audience specifies which service the ID token is

774 # intended for.

775 "target_audience": self._target_audience,

776 }

777

778 payload.update(self._additional_claims)

779

780 token = jwt.encode(self._signer, payload)

781

782 return token

783

784 def _refresh_with_iam_endpoint(self, request):

785 """Use IAM generateIdToken endpoint to obtain an ID token.

786

787 It works as follows:

788

789 1. First we create a self signed jwt with

790 https://www.googleapis.com/auth/iam being the scope.

791

792 2. Next we use the self signed jwt as the access token, and make a POST

793 request to IAM generateIdToken endpoint. The request body is:

794 {

795 "audience": self._target_audience,

796 "includeEmail": "true",

797 "useEmailAzp": "true",

798 }

799

800 If the request is succesfully, it will return {"token":"the ID token"},

801 and we can extract the ID token and compute its expiry.

802 """

803 jwt_credentials = jwt.Credentials.from_signing_credentials(

804 self,

805 None,

806 additional_claims={"scope": "https://www.googleapis.com/auth/iam"},

807 )

808 jwt_credentials.refresh(request)

809 self.token, self.expiry = _client.call_iam_generate_id_token_endpoint(

810 request,

811 self._iam_id_token_endpoint,

812 self.signer_email,

813 self._target_audience,

814 jwt_credentials.token.decode(),

815 self._universe_domain,

816 )

817

818 @_helpers.copy_docstring(credentials.Credentials)

819 def refresh(self, request):

820 if self._use_iam_endpoint:

821 self._refresh_with_iam_endpoint(request)

822 else:

823 assertion = self._make_authorization_grant_assertion()

824 access_token, expiry, _ = _client.id_token_jwt_grant(

825 request, self._token_uri, assertion

826 )

827 self.token = access_token

828 self.expiry = expiry

829

830 @property

831 def service_account_email(self):

832 """The service account email."""

833 return self._service_account_email

834

835 @_helpers.copy_docstring(credentials.Signing)

836 def sign_bytes(self, message):

837 return self._signer.sign(message)

838

839 @property # type: ignore

840 @_helpers.copy_docstring(credentials.Signing)

841 def signer(self):

842 return self._signer

843

844 @property # type: ignore

845 @_helpers.copy_docstring(credentials.Signing)

846 def signer_email(self):

847 return self._service_account_email

Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/google/oauth2/service_account.py: 34%

250 statements