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

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 

27from google.auth import _helpers 

28from google.auth import credentials 

29from google.auth import crypt 

30from google.auth import exceptions 

31 

32IAM_RETRY_CODES = { 

33 http_client.INTERNAL_SERVER_ERROR, 

34 http_client.BAD_GATEWAY, 

35 http_client.SERVICE_UNAVAILABLE, 

36 http_client.GATEWAY_TIMEOUT, 

37} 

38 

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

40 

41_IAM_ENDPOINT = ( 

42 "https://iamcredentials.googleapis.com/v1/projects/-" 

43 + "/serviceAccounts/{}:generateAccessToken" 

44) 

45 

46_IAM_SIGN_ENDPOINT = ( 

47 "https://iamcredentials.googleapis.com/v1/projects/-" 

48 + "/serviceAccounts/{}:signBlob" 

49) 

50 

51_IAM_SIGNJWT_ENDPOINT = ( 

52 "https://iamcredentials.googleapis.com/v1/projects/-" 

53 + "/serviceAccounts/{}:signJwt" 

54) 

55 

56_IAM_IDTOKEN_ENDPOINT = ( 

57 "https://iamcredentials.googleapis.com/v1/" 

58 + "projects/-/serviceAccounts/{}:generateIdToken" 

59) 

60 

61 

62class Signer(crypt.Signer): 

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

64 

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

66 credential's private key file. 

67 

68 .. _signBlob API: 

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

70 /signBlob 

71 """ 

72 

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

74 """ 

75 Args: 

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

77 HTTP requests. 

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

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

80 The credentials must have of one the following scopes: 

81 

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

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

84 service_account_email (str): The service account email identifying 

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

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

87 credentials. 

88 """ 

89 self._request = request 

90 self._credentials = credentials 

91 self._service_account_email = service_account_email 

92 

93 def _make_signing_request(self, message): 

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

95 message = _helpers.to_bytes(message) 

96 

97 method = "POST" 

98 url = _IAM_SIGN_ENDPOINT.replace( 

99 credentials.DEFAULT_UNIVERSE_DOMAIN, self._credentials.universe_domain 

100 ).format(self._service_account_email) 

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

102 body = json.dumps( 

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

104 ).encode("utf-8") 

105 

106 retries = _exponential_backoff.ExponentialBackoff() 

107 for _ in retries: 

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

109 

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

111 

112 if response.status in IAM_RETRY_CODES: 

113 continue 

114 

115 if response.status != http_client.OK: 

116 raise exceptions.TransportError( 

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

118 ) 

119 

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

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

122 

123 @property 

124 def key_id(self): 

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

126 

127 .. warning:: 

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

129 be reliably determined ahead of time. 

130 """ 

131 return None 

132 

133 @_helpers.copy_docstring(crypt.Signer) 

134 def sign(self, message): 

135 response = self._make_signing_request(message) 

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