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

1# Copyright 2017 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"""Tools for using the Google `Cloud Identity and Access Management (IAM) 

16API`_'s auth-related functionality. 

17 

18.. _Cloud Identity and Access Management (IAM) API: 

19 https://cloud.google.com/iam/docs/ 

20""" 

21 

22import base64 

23import http.client as http_client 

24import json 

25 

26from google.auth import _exponential_backoff, _helpers, credentials, crypt, exceptions 

27from google.auth.transport import _mtls_helper 

28 

29IAM_RETRY_CODES = { 

30 http_client.INTERNAL_SERVER_ERROR, 

31 http_client.BAD_GATEWAY, 

32 http_client.SERVICE_UNAVAILABLE, 

33 http_client.GATEWAY_TIMEOUT, 

34} 

35 

36_IAM_SCOPE = ["https://www.googleapis.com/auth/iam"] 

37 

38# Determine if we should use mTLS. 

39if ( 

40 hasattr(_mtls_helper, "check_use_client_cert") 

41 and _mtls_helper.check_use_client_cert() 

42): 

43 # Construct the template domain using the library's DEFAULT_UNIVERSE_DOMAIN constant. 

44 _IAM_DOMAIN = f"iamcredentials.mtls.{credentials.DEFAULT_UNIVERSE_DOMAIN}" 

45else: 

46 _IAM_DOMAIN = f"iamcredentials.{credentials.DEFAULT_UNIVERSE_DOMAIN}" 

47 

48# Create the common base URL template 

49# We use double brackets {{}} so .format() can be called later for the email. 

50_IAM_BASE_URL = f"https://{_IAM_DOMAIN}/v1/projects/-/serviceAccounts/{{}}" 

51 

52# Define the endpoints as static templates 

53_IAM_ENDPOINT = _IAM_BASE_URL + ":generateAccessToken" 

54_IAM_SIGN_ENDPOINT = _IAM_BASE_URL + ":signBlob" 

55_IAM_SIGNJWT_ENDPOINT = _IAM_BASE_URL + ":signJwt" 

56_IAM_IDTOKEN_ENDPOINT = _IAM_BASE_URL + ":generateIdToken" 

57 

58 

59class Signer(crypt.Signer): 

60 """Signs messages using the IAM `signBlob API`_. 

61 

62 This is useful when you need to sign bytes but do not have access to the 

63 credential's private key file. 

64 

65 .. _signBlob API: 

66 https://cloud.google.com/iam/reference/rest/v1/projects.serviceAccounts 

67 /signBlob 

68 """ 

69 

70 def __init__(self, request, credentials, service_account_email): 

71 """ 

72 Args: 

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

74 HTTP requests. 

75 credentials (google.auth.credentials.Credentials): The credentials 

76 that will be used to authenticate the request to the IAM API. 

77 The credentials must have of one the following scopes: 

78 

79 - https://www.googleapis.com/auth/iam 

80 - https://www.googleapis.com/auth/cloud-platform 

81 service_account_email (str): The service account email identifying 

82 which service account to use to sign bytes. Often, this can 

83 be the same as the service account email in the given 

84 credentials. 

85 """ 

86 self._request = request 

87 self._credentials = credentials 

88 self._service_account_email = service_account_email 

89 

90 def _make_signing_request(self, message): 

91 """Makes a request to the API signBlob API.""" 

92 message = _helpers.to_bytes(message) 

93 

94 method = "POST" 

95 url = _IAM_SIGN_ENDPOINT.replace( 

96 credentials.DEFAULT_UNIVERSE_DOMAIN, self._credentials.universe_domain 

97 ).format(self._service_account_email) 

98 headers = {"Content-Type": "application/json"} 

99 body = json.dumps( 

100 {"payload": base64.b64encode(message).decode("utf-8")} 

101 ).encode("utf-8") 

102 

103 retries = _exponential_backoff.ExponentialBackoff() 

104 for _ in retries: 

105 self._credentials.before_request(self._request, method, url, headers) 

106 

107 response = self._request(url=url, method=method, body=body, headers=headers) 

108 

109 if response.status in IAM_RETRY_CODES: 

110 continue 

111 

112 if response.status != http_client.OK: 

113 raise exceptions.TransportError( 

114 "Error calling the IAM signBlob API: {}".format(response.data) 

115 ) 

116 

117 return json.loads(response.data.decode("utf-8")) 

118 raise exceptions.TransportError("exhausted signBlob endpoint retries") 

119 

120 @property 

121 def key_id(self): 

122 """Optional[str]: The key ID used to identify this private key. 

123 

124 .. warning:: 

125 This is always ``None``. The key ID used by IAM can not 

126 be reliably determined ahead of time. 

127 """ 

128 return None 

129 

130 @_helpers.copy_docstring(crypt.Signer) 

131 def sign(self, message): 

132 response = self._make_signing_request(message) 

133 return base64.b64decode(response["signedBlob"])