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

1# Copyright 2020 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"""OAuth 2.0 Token Exchange Spec. 

16 

17This module defines a token exchange utility based on the `OAuth 2.0 Token 

18Exchange`_ spec. This will be mainly used to exchange external credentials 

19for GCP access tokens in workload identity pools to access Google APIs. 

20 

21The implementation will support various types of client authentication as 

22allowed in the spec. 

23 

24A deviation on the spec will be for additional Google specific options that 

25cannot be easily mapped to parameters defined in the RFC. 

26 

27The returned dictionary response will be based on the `rfc8693 section 2.2.1`_ 

28spec JSON response. 

29 

30.. _OAuth 2.0 Token Exchange: https://tools.ietf.org/html/rfc8693 

31.. _rfc8693 section 2.2.1: https://tools.ietf.org/html/rfc8693#section-2.2.1 

32""" 

33 

34import http.client as http_client 

35import json 

36import urllib 

37 

38from google.oauth2 import utils 

39 

40 

41_URLENCODED_HEADERS = {"Content-Type": "application/x-www-form-urlencoded"} 

42 

43 

44class Client(utils.OAuthClientAuthHandler): 

45 """Implements the OAuth 2.0 token exchange spec based on 

46 https://tools.ietf.org/html/rfc8693. 

47 """ 

48 

49 def __init__(self, token_exchange_endpoint, client_authentication=None): 

50 """Initializes an STS client instance. 

51 

52 Args: 

53 token_exchange_endpoint (str): The token exchange endpoint. 

54 client_authentication (Optional(google.oauth2.oauth2_utils.ClientAuthentication)): 

55 The optional OAuth client authentication credentials if available. 

56 """ 

57 super(Client, self).__init__(client_authentication) 

58 self._token_exchange_endpoint = token_exchange_endpoint 

59 

60 def _make_request(self, request, headers, request_body, url=None): 

61 # Initialize request headers. 

62 request_headers = _URLENCODED_HEADERS.copy() 

63 

64 # Inject additional headers. 

65 if headers: 

66 for k, v in dict(headers).items(): 

67 request_headers[k] = v 

68 

69 # Apply OAuth client authentication. 

70 self.apply_client_authentication_options(request_headers, request_body) 

71 

72 # Use default token exchange endpoint if no url is provided. 

73 url = url or self._token_exchange_endpoint 

74 

75 # Execute request. 

76 response = request( 

77 url=url, 

78 method="POST", 

79 headers=request_headers, 

80 body=urllib.parse.urlencode(request_body).encode("utf-8"), 

81 ) 

82 

83 response_body = ( 

84 response.data.decode("utf-8") 

85 if hasattr(response.data, "decode") 

86 else response.data 

87 ) 

88 

89 # If non-200 response received, translate to OAuthError exception. 

90 if response.status != http_client.OK: 

91 utils.handle_error_response(response_body) 

92 

93 # A successful token revocation returns an empty response body. 

94 if not response_body: 

95 return {} 

96 

97 # Other successful responses should be valid JSON. 

98 return json.loads(response_body) 

99 

100 def exchange_token( 

101 self, 

102 request, 

103 grant_type, 

104 subject_token, 

105 subject_token_type, 

106 resource=None, 

107 audience=None, 

108 scopes=None, 

109 requested_token_type=None, 

110 actor_token=None, 

111 actor_token_type=None, 

112 additional_options=None, 

113 additional_headers=None, 

114 ): 

115 """Exchanges the provided token for another type of token based on the 

116 rfc8693 spec. 

117 

118 Args: 

119 request (google.auth.transport.Request): A callable used to make 

120 HTTP requests. 

121 grant_type (str): The OAuth 2.0 token exchange grant type. 

122 subject_token (str): The OAuth 2.0 token exchange subject token. 

123 subject_token_type (str): The OAuth 2.0 token exchange subject token type. 

124 resource (Optional[str]): The optional OAuth 2.0 token exchange resource field. 

125 audience (Optional[str]): The optional OAuth 2.0 token exchange audience field. 

126 scopes (Optional[Sequence[str]]): The optional list of scopes to use. 

127 requested_token_type (Optional[str]): The optional OAuth 2.0 token exchange requested 

128 token type. 

129 actor_token (Optional[str]): The optional OAuth 2.0 token exchange actor token. 

130 actor_token_type (Optional[str]): The optional OAuth 2.0 token exchange actor token type. 

131 additional_options (Optional[Mapping[str, str]]): The optional additional 

132 non-standard Google specific options. 

133 additional_headers (Optional[Mapping[str, str]]): The optional additional 

134 headers to pass to the token exchange endpoint. 

135 

136 Returns: 

137 Mapping[str, str]: The token exchange JSON-decoded response data containing 

138 the requested token and its expiration time. 

139 

140 Raises: 

141 google.auth.exceptions.OAuthError: If the token endpoint returned 

142 an error. 

143 """ 

144 # Initialize request body. 

145 request_body = { 

146 "grant_type": grant_type, 

147 "resource": resource, 

148 "audience": audience, 

149 "scope": " ".join(scopes or []), 

150 "requested_token_type": requested_token_type, 

151 "subject_token": subject_token, 

152 "subject_token_type": subject_token_type, 

153 "actor_token": actor_token, 

154 "actor_token_type": actor_token_type, 

155 "options": None, 

156 } 

157 # Add additional non-standard options. 

158 if additional_options: 

159 request_body["options"] = urllib.parse.quote(json.dumps(additional_options)) 

160 # Remove empty fields in request body. 

161 for k, v in dict(request_body).items(): 

162 if v is None or v == "": 

163 del request_body[k] 

164 

165 return self._make_request(request, additional_headers, request_body) 

166 

167 def refresh_token(self, request, refresh_token): 

168 """Exchanges a refresh token for an access token based on the 

169 RFC6749 spec. 

170 

171 Args: 

172 request (google.auth.transport.Request): A callable used to make 

173 HTTP requests. 

174 subject_token (str): The OAuth 2.0 refresh token. 

175 """ 

176 

177 return self._make_request( 

178 request, 

179 None, 

180 {"grant_type": "refresh_token", "refresh_token": refresh_token}, 

181 ) 

182 

183 def revoke_token(self, request, token, token_type_hint, revoke_url): 

184 """Revokes the provided token based on the RFC7009 spec. 

185 

186 Args: 

187 request (google.auth.transport.Request): A callable used to make 

188 HTTP requests. 

189 token (str): The OAuth 2.0 token to revoke. 

190 token_type_hint (str): Hint for the type of token being revoked. 

191 revoke_url (str): The STS endpoint URL for revoking tokens. 

192 

193 Raises: 

194 google.auth.exceptions.OAuthError: If the token revocation endpoint 

195 returned an error. 

196 """ 

197 request_body = {"token": token} 

198 if token_type_hint: 

199 request_body["token_type_hint"] = token_type_hint 

200 

201 return self._make_request(request, None, request_body, revoke_url)