Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/google/auth/_helpers.py: 47%

55 statements  

« prev     ^ index     » next       coverage.py v7.2.2, created at 2023-03-26 06:25 +0000

1# Copyright 2015 Google Inc. 

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"""Helper functions for commonly used utilities.""" 

16 

17import base64 

18import calendar 

19import datetime 

20import sys 

21 

22import six 

23from six.moves import urllib 

24 

25from google.auth import exceptions 

26 

27# Token server doesn't provide a new a token when doing refresh unless the 

28# token is expiring within 30 seconds, so refresh threshold should not be 

29# more than 30 seconds. Otherwise auth lib will send tons of refresh requests 

30# until 30 seconds before the expiration, and cause a spike of CPU usage. 

31REFRESH_THRESHOLD = datetime.timedelta(seconds=20) 

32 

33 

34def copy_docstring(source_class): 

35 """Decorator that copies a method's docstring from another class. 

36 

37 Args: 

38 source_class (type): The class that has the documented method. 

39 

40 Returns: 

41 Callable: A decorator that will copy the docstring of the same 

42 named method in the source class to the decorated method. 

43 """ 

44 

45 def decorator(method): 

46 """Decorator implementation. 

47 

48 Args: 

49 method (Callable): The method to copy the docstring to. 

50 

51 Returns: 

52 Callable: the same method passed in with an updated docstring. 

53 

54 Raises: 

55 google.auth.exceptions.InvalidOperation: if the method already has a docstring. 

56 """ 

57 if method.__doc__: 

58 raise exceptions.InvalidOperation("Method already has a docstring.") 

59 

60 source_method = getattr(source_class, method.__name__) 

61 method.__doc__ = source_method.__doc__ 

62 

63 return method 

64 

65 return decorator 

66 

67 

68def utcnow(): 

69 """Returns the current UTC datetime. 

70 

71 Returns: 

72 datetime: The current time in UTC. 

73 """ 

74 return datetime.datetime.utcnow() 

75 

76 

77def datetime_to_secs(value): 

78 """Convert a datetime object to the number of seconds since the UNIX epoch. 

79 

80 Args: 

81 value (datetime): The datetime to convert. 

82 

83 Returns: 

84 int: The number of seconds since the UNIX epoch. 

85 """ 

86 return calendar.timegm(value.utctimetuple()) 

87 

88 

89def to_bytes(value, encoding="utf-8"): 

90 """Converts a string value to bytes, if necessary. 

91 

92 Unfortunately, ``six.b`` is insufficient for this task since in 

93 Python 2 because it does not modify ``unicode`` objects. 

94 

95 Args: 

96 value (Union[str, bytes]): The value to be converted. 

97 encoding (str): The encoding to use to convert unicode to bytes. 

98 Defaults to "utf-8". 

99 

100 Returns: 

101 bytes: The original value converted to bytes (if unicode) or as 

102 passed in if it started out as bytes. 

103 

104 Raises: 

105 google.auth.exceptions.InvalidValue: If the value could not be converted to bytes. 

106 """ 

107 result = value.encode(encoding) if isinstance(value, six.text_type) else value 

108 if isinstance(result, six.binary_type): 

109 return result 

110 else: 

111 raise exceptions.InvalidValue( 

112 "{0!r} could not be converted to bytes".format(value) 

113 ) 

114 

115 

116def from_bytes(value): 

117 """Converts bytes to a string value, if necessary. 

118 

119 Args: 

120 value (Union[str, bytes]): The value to be converted. 

121 

122 Returns: 

123 str: The original value converted to unicode (if bytes) or as passed in 

124 if it started out as unicode. 

125 

126 Raises: 

127 google.auth.exceptions.InvalidValue: If the value could not be converted to unicode. 

128 """ 

129 result = value.decode("utf-8") if isinstance(value, six.binary_type) else value 

130 if isinstance(result, six.text_type): 

131 return result 

132 else: 

133 raise exceptions.InvalidValue( 

134 "{0!r} could not be converted to unicode".format(value) 

135 ) 

136 

137 

138def update_query(url, params, remove=None): 

139 """Updates a URL's query parameters. 

140 

141 Replaces any current values if they are already present in the URL. 

142 

143 Args: 

144 url (str): The URL to update. 

145 params (Mapping[str, str]): A mapping of query parameter 

146 keys to values. 

147 remove (Sequence[str]): Parameters to remove from the query string. 

148 

149 Returns: 

150 str: The URL with updated query parameters. 

151 

152 Examples: 

153 

154 >>> url = 'http://example.com?a=1' 

155 >>> update_query(url, {'a': '2'}) 

156 http://example.com?a=2 

157 >>> update_query(url, {'b': '3'}) 

158 http://example.com?a=1&b=3 

159 >> update_query(url, {'b': '3'}, remove=['a']) 

160 http://example.com?b=3 

161 

162 """ 

163 if remove is None: 

164 remove = [] 

165 

166 # Split the URL into parts. 

167 parts = urllib.parse.urlparse(url) 

168 # Parse the query string. 

169 query_params = urllib.parse.parse_qs(parts.query) 

170 # Update the query parameters with the new parameters. 

171 query_params.update(params) 

172 # Remove any values specified in remove. 

173 query_params = { 

174 key: value for key, value in six.iteritems(query_params) if key not in remove 

175 } 

176 # Re-encoded the query string. 

177 new_query = urllib.parse.urlencode(query_params, doseq=True) 

178 # Unsplit the url. 

179 new_parts = parts._replace(query=new_query) 

180 return urllib.parse.urlunparse(new_parts) 

181 

182 

183def scopes_to_string(scopes): 

184 """Converts scope value to a string suitable for sending to OAuth 2.0 

185 authorization servers. 

186 

187 Args: 

188 scopes (Sequence[str]): The sequence of scopes to convert. 

189 

190 Returns: 

191 str: The scopes formatted as a single string. 

192 """ 

193 return " ".join(scopes) 

194 

195 

196def string_to_scopes(scopes): 

197 """Converts stringifed scopes value to a list. 

198 

199 Args: 

200 scopes (Union[Sequence, str]): The string of space-separated scopes 

201 to convert. 

202 Returns: 

203 Sequence(str): The separated scopes. 

204 """ 

205 if not scopes: 

206 return [] 

207 

208 return scopes.split(" ") 

209 

210 

211def padded_urlsafe_b64decode(value): 

212 """Decodes base64 strings lacking padding characters. 

213 

214 Google infrastructure tends to omit the base64 padding characters. 

215 

216 Args: 

217 value (Union[str, bytes]): The encoded value. 

218 

219 Returns: 

220 bytes: The decoded value 

221 """ 

222 b64string = to_bytes(value) 

223 padded = b64string + b"=" * (-len(b64string) % 4) 

224 return base64.urlsafe_b64decode(padded) 

225 

226 

227def unpadded_urlsafe_b64encode(value): 

228 """Encodes base64 strings removing any padding characters. 

229 

230 `rfc 7515`_ defines Base64url to NOT include any padding 

231 characters, but the stdlib doesn't do that by default. 

232 

233 _rfc7515: https://tools.ietf.org/html/rfc7515#page-6 

234 

235 Args: 

236 value (Union[str|bytes]): The bytes-like value to encode 

237 

238 Returns: 

239 Union[str|bytes]: The encoded value 

240 """ 

241 return base64.urlsafe_b64encode(value).rstrip(b"=") 

242 

243 

244def is_python_3(): 

245 """Check if the Python interpreter is Python 2 or 3. 

246 

247 Returns: 

248 bool: True if the Python interpreter is Python 3 and False otherwise. 

249 """ 

250 return sys.version_info > (3, 0)