Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/botocore/compat.py: 49%

1# Copyright 2012-2014 Amazon.com, Inc. or its affiliates. All Rights Reserved. 

2# 

3# Licensed under the Apache License, Version 2.0 (the "License"). You 

4# may not use this file except in compliance with the License. A copy of 

5# the License is located at 

6# 

7# http://aws.amazon.com/apache2.0/ 

8# 

9# or in the "license" file accompanying this file. This file is 

10# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF 

11# ANY KIND, either express or implied. See the License for the specific 

12# language governing permissions and limitations under the License. 

13 

14import copy 

15import datetime 

16import sys 

17import inspect 

18import warnings 

19import hashlib 

20from http.client import HTTPMessage 

21import logging 

22import shlex 

23import re 

24import os 

25from collections import OrderedDict 

26from collections.abc import MutableMapping 

27from math import floor 

28 

29from botocore.vendored import six 

30from botocore.exceptions import MD5UnavailableError 

31from dateutil.tz import tzlocal 

32from urllib3 import exceptions 

33 

34logger = logging.getLogger(__name__) 

35 

36 

37class HTTPHeaders(HTTPMessage): 

38 pass 

39 

40from urllib.parse import ( 

41 quote, 

42 urlencode, 

43 unquote, 

44 unquote_plus, 

45 urlparse, 

46 urlsplit, 

47 urlunsplit, 

48 urljoin, 

49 parse_qsl, 

50 parse_qs, 

51) 

52from http.client import HTTPResponse 

53from io import IOBase as _IOBase 

54from base64 import encodebytes 

55from email.utils import formatdate 

56from itertools import zip_longest 

57file_type = _IOBase 

58zip = zip 

59 

60# In python3, unquote takes a str() object, url decodes it, 

61# then takes the bytestring and decodes it to utf-8. 

62unquote_str = unquote_plus 

63 

64def set_socket_timeout(http_response, timeout): 

65 """Set the timeout of the socket from an HTTPResponse. 

66 

67 :param http_response: An instance of ``httplib.HTTPResponse`` 

68 

69 """ 

70 http_response._fp.fp.raw._sock.settimeout(timeout) 

71 

72def accepts_kwargs(func): 

73 # In python3.4.1, there's backwards incompatible 

74 # changes when using getargspec with functools.partials. 

75 return inspect.getfullargspec(func)[2] 

76 

77def ensure_unicode(s, encoding=None, errors=None): 

78 # NOOP in Python 3, because every string is already unicode 

79 return s 

80 

81def ensure_bytes(s, encoding='utf-8', errors='strict'): 

82 if isinstance(s, str): 

83 return s.encode(encoding, errors) 

84 if isinstance(s, bytes): 

85 return s 

86 raise ValueError(f"Expected str or bytes, received {type(s)}.") 

87 

88 

89try: 

90 import xml.etree.cElementTree as ETree 

91except ImportError: 

92 # cElementTree does not exist from Python3.9+ 

93 import xml.etree.ElementTree as ETree 

94XMLParseError = ETree.ParseError 

95import json 

96 

97 

98def filter_ssl_warnings(): 

99 # Ignore warnings related to SNI as it is not being used in validations. 

100 warnings.filterwarnings( 

101 'ignore', 

102 message="A true SSLContext object is not available.*", 

103 category=exceptions.InsecurePlatformWarning, 

104 module=r".*urllib3\.util\.ssl_", 

105 ) 

106 

107 

108@classmethod 

109def from_dict(cls, d): 

110 new_instance = cls() 

111 for key, value in d.items(): 

112 new_instance[key] = value 

113 return new_instance 

114 

115 

116@classmethod 

117def from_pairs(cls, pairs): 

118 new_instance = cls() 

119 for key, value in pairs: 

120 new_instance[key] = value 

121 return new_instance 

122 

123 

124HTTPHeaders.from_dict = from_dict 

125HTTPHeaders.from_pairs = from_pairs 

126 

127 

128def copy_kwargs(kwargs): 

129 """ 

130 This used to be a compat shim for 2.6 but is now just an alias. 

131 """ 

132 copy_kwargs = copy.copy(kwargs) 

133 return copy_kwargs 

134 

135 

136def total_seconds(delta): 

137 """ 

138 Returns the total seconds in a ``datetime.timedelta``. 

139 

140 This used to be a compat shim for 2.6 but is now just an alias. 

141 

142 :param delta: The timedelta object 

143 :type delta: ``datetime.timedelta`` 

144 """ 

145 return delta.total_seconds() 

146 

147 

148# Checks to see if md5 is available on this system. A given system might not 

149# have access to it for various reasons, such as FIPS mode being enabled. 

150try: 

151 hashlib.md5() 

152 MD5_AVAILABLE = True 

153except ValueError: 

154 MD5_AVAILABLE = False 

155 

156 

157def get_md5(*args, **kwargs): 

158 """ 

159 Attempts to get an md5 hashing object. 

160 

161 :param args: Args to pass to the MD5 constructor 

162 :param kwargs: Key word arguments to pass to the MD5 constructor 

163 :return: An MD5 hashing object if available. If it is unavailable, None 

164 is returned if raise_error_if_unavailable is set to False. 

165 """ 

166 if MD5_AVAILABLE: 

167 return hashlib.md5(*args, **kwargs) 

168 else: 

169 raise MD5UnavailableError() 

170 

171 

172def compat_shell_split(s, platform=None): 

173 if platform is None: 

174 platform = sys.platform 

175 

176 if platform == "win32": 

177 return _windows_shell_split(s) 

178 else: 

179 return shlex.split(s) 

180 

181 

182def _windows_shell_split(s): 

183 """Splits up a windows command as the built-in command parser would. 

184 

185 Windows has potentially bizarre rules depending on where you look. When 

186 spawning a process via the Windows C runtime (which is what python does 

187 when you call popen) the rules are as follows: 

188 

189 https://docs.microsoft.com/en-us/cpp/cpp/parsing-cpp-command-line-arguments 

190 

191 To summarize: 

192 

193 * Only space and tab are valid delimiters 

194 * Double quotes are the only valid quotes 

195 * Backslash is interpreted literally unless it is part of a chain that 

196 leads up to a double quote. Then the backslashes escape the backslashes, 

197 and if there is an odd number the final backslash escapes the quote. 

198 

199 :param s: The command string to split up into parts. 

200 :return: A list of command components. 

201 """ 

202 if not s: 

203 return [] 

204 

205 components = [] 

206 buff = [] 

207 is_quoted = False 

208 num_backslashes = 0 

209 for character in s: 

210 if character == '\\': 

211 # We can't simply append backslashes because we don't know if 

212 # they are being used as escape characters or not. Instead we 

213 # keep track of how many we've encountered and handle them when 

214 # we encounter a different character. 

215 num_backslashes += 1 

216 elif character == '"': 

217 if num_backslashes > 0: 

218 # The backslashes are in a chain leading up to a double 

219 # quote, so they are escaping each other. 

220 buff.append('\\' * int(floor(num_backslashes / 2))) 

221 remainder = num_backslashes % 2 

222 num_backslashes = 0 

223 if remainder == 1: 

224 # The number of backslashes is uneven, so they are also 

225 # escaping the double quote, so it needs to be added to 

226 # the current component buffer. 

227 buff.append('"') 

228 continue 

229 

230 # We've encountered a double quote that is not escaped, 

231 # so we toggle is_quoted. 

232 is_quoted = not is_quoted 

233 

234 # If there are quotes, then we may want an empty string. To be 

235 # safe, we add an empty string to the buffer so that we make 

236 # sure it sticks around if there's nothing else between quotes. 

237 # If there is other stuff between quotes, the empty string will 

238 # disappear during the joining process. 

239 buff.append('') 

240 elif character in [' ', '\t'] and not is_quoted: 

241 # Since the backslashes aren't leading up to a quote, we put in 

242 # the exact number of backslashes. 

243 if num_backslashes > 0: 

244 buff.append('\\' * num_backslashes) 

245 num_backslashes = 0 

246 

247 # Excess whitespace is ignored, so only add the components list 

248 # if there is anything in the buffer. 

249 if buff: 

250 components.append(''.join(buff)) 

251 buff = [] 

252 else: 

253 # Since the backslashes aren't leading up to a quote, we put in 

254 # the exact number of backslashes. 

255 if num_backslashes > 0: 

256 buff.append('\\' * num_backslashes) 

257 num_backslashes = 0 

258 buff.append(character) 

259 

260 # Quotes must be terminated. 

261 if is_quoted: 

262 raise ValueError(f"No closing quotation in string: {s}") 

263 

264 # There may be some leftover backslashes, so we need to add them in. 

265 # There's no quote so we add the exact number. 

266 if num_backslashes > 0: 

267 buff.append('\\' * num_backslashes) 

268 

269 # Add the final component in if there is anything in the buffer. 

270 if buff: 

271 components.append(''.join(buff)) 

272 

273 return components 

274 

275 

276def get_tzinfo_options(): 

277 # Due to dateutil/dateutil#197, Windows may fail to parse times in the past 

278 # with the system clock. We can alternatively fallback to tzwininfo when 

279 # this happens, which will get time info from the Windows registry. 

280 if sys.platform == 'win32': 

281 from dateutil.tz import tzwinlocal 

282 

283 return (tzlocal, tzwinlocal) 

284 else: 

285 return (tzlocal,) 

286 

287 

288# Detect if CRT is available for use 

289try: 

290 import awscrt.auth 

291 

292 # Allow user opt-out if needed 

293 disabled = os.environ.get('BOTO_DISABLE_CRT', "false") 

294 HAS_CRT = not disabled.lower() == 'true' 

295except ImportError: 

296 HAS_CRT = False 

297 

298 

299######################################################## 

300# urllib3 compat backports # 

301######################################################## 

302 

303# Vendoring IPv6 validation regex patterns from urllib3 

304# https://github.com/urllib3/urllib3/blob/7e856c0/src/urllib3/util/url.py 

305IPV4_PAT = r"(?:[0-9]{1,3}\.){3}[0-9]{1,3}" 

306IPV4_RE = re.compile("^" + IPV4_PAT + "$") 

307HEX_PAT = "[0-9A-Fa-f]{1,4}" 

308LS32_PAT = "(?:{hex}:{hex}|{ipv4})".format(hex=HEX_PAT, ipv4=IPV4_PAT) 

309_subs = {"hex": HEX_PAT, "ls32": LS32_PAT} 

310_variations = [ 

311 # 6( h16 ":" ) ls32 

312 "(?:%(hex)s:){6}%(ls32)s", 

313 # "::" 5( h16 ":" ) ls32 

314 "::(?:%(hex)s:){5}%(ls32)s", 

315 # [ h16 ] "::" 4( h16 ":" ) ls32 

316 "(?:%(hex)s)?::(?:%(hex)s:){4}%(ls32)s", 

317 # [ *1( h16 ":" ) h16 ] "::" 3( h16 ":" ) ls32 

318 "(?:(?:%(hex)s:)?%(hex)s)?::(?:%(hex)s:){3}%(ls32)s", 

319 # [ *2( h16 ":" ) h16 ] "::" 2( h16 ":" ) ls32 

320 "(?:(?:%(hex)s:){0,2}%(hex)s)?::(?:%(hex)s:){2}%(ls32)s", 

321 # [ *3( h16 ":" ) h16 ] "::" h16 ":" ls32 

322 "(?:(?:%(hex)s:){0,3}%(hex)s)?::%(hex)s:%(ls32)s", 

323 # [ *4( h16 ":" ) h16 ] "::" ls32 

324 "(?:(?:%(hex)s:){0,4}%(hex)s)?::%(ls32)s", 

325 # [ *5( h16 ":" ) h16 ] "::" h16 

326 "(?:(?:%(hex)s:){0,5}%(hex)s)?::%(hex)s", 

327 # [ *6( h16 ":" ) h16 ] "::" 

328 "(?:(?:%(hex)s:){0,6}%(hex)s)?::", 

329] 

330 

331UNRESERVED_PAT = ( 

332 r"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789._!\-~" 

333) 

334IPV6_PAT = "(?:" + "|".join([x % _subs for x in _variations]) + ")" 

335ZONE_ID_PAT = "(?:%25|%)(?:[" + UNRESERVED_PAT + "]|%[a-fA-F0-9]{2})+" 

336IPV6_ADDRZ_PAT = r"\[" + IPV6_PAT + r"(?:" + ZONE_ID_PAT + r")?\]" 

337IPV6_ADDRZ_RE = re.compile("^" + IPV6_ADDRZ_PAT + "$") 

338 

339# These are the characters that are stripped by post-bpo-43882 urlparse(). 

340UNSAFE_URL_CHARS = frozenset('\t\r\n') 

341 

342# Detect if gzip is available for use 

343try: 

344 import gzip 

345 HAS_GZIP = True 

346except ImportError: 

347 HAS_GZIP = False