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

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 return inspect.getfullargspec(func)[2] 

74 

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

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

77 return s 

78 

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

80 if isinstance(s, str): 

81 return s.encode(encoding, errors) 

82 if isinstance(s, bytes): 

83 return s 

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

85 

86 

87import xml.etree.ElementTree as ETree 

88XMLParseError = ETree.ParseError 

89 

90import json 

91 

92 

93def filter_ssl_warnings(): 

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

95 warnings.filterwarnings( 

96 'ignore', 

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

98 category=exceptions.InsecurePlatformWarning, 

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

100 ) 

101 

102 

103@classmethod 

104def from_dict(cls, d): 

105 new_instance = cls() 

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

107 new_instance[key] = value 

108 return new_instance 

109 

110 

111@classmethod 

112def from_pairs(cls, pairs): 

113 new_instance = cls() 

114 for key, value in pairs: 

115 new_instance[key] = value 

116 return new_instance 

117 

118 

119HTTPHeaders.from_dict = from_dict 

120HTTPHeaders.from_pairs = from_pairs 

121 

122 

123def copy_kwargs(kwargs): 

124 """ 

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

126 """ 

127 copy_kwargs = copy.copy(kwargs) 

128 return copy_kwargs 

129 

130 

131def total_seconds(delta): 

132 """ 

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

134 

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

136 

137 :param delta: The timedelta object 

138 :type delta: ``datetime.timedelta`` 

139 """ 

140 return delta.total_seconds() 

141 

142 

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

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

145try: 

146 hashlib.md5(usedforsecurity=False) 

147 MD5_AVAILABLE = True 

148except (AttributeError, ValueError): 

149 MD5_AVAILABLE = False 

150 

151 

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

153 """ 

154 Attempts to get an md5 hashing object. 

155 

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

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

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

159 is returned if raise_error_if_unavailable is set to False. 

160 """ 

161 if MD5_AVAILABLE: 

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

163 else: 

164 raise MD5UnavailableError() 

165 

166 

167def compat_shell_split(s, platform=None): 

168 if platform is None: 

169 platform = sys.platform 

170 

171 if platform == "win32": 

172 return _windows_shell_split(s) 

173 else: 

174 return shlex.split(s) 

175 

176 

177def _windows_shell_split(s): 

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

179 

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

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

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

183 

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

185 

186 To summarize: 

187 

188 * Only space and tab are valid delimiters 

189 * Double quotes are the only valid quotes 

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

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

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

193 

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

195 :return: A list of command components. 

196 """ 

197 if not s: 

198 return [] 

199 

200 components = [] 

201 buff = [] 

202 is_quoted = False 

203 num_backslashes = 0 

204 for character in s: 

205 if character == '\\': 

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

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

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

209 # we encounter a different character. 

210 num_backslashes += 1 

211 elif character == '"': 

212 if num_backslashes > 0: 

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

214 # quote, so they are escaping each other. 

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

216 remainder = num_backslashes % 2 

217 num_backslashes = 0 

218 if remainder == 1: 

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

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

221 # the current component buffer. 

222 buff.append('"') 

223 continue 

224 

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

226 # so we toggle is_quoted. 

227 is_quoted = not is_quoted 

228 

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

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

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

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

233 # disappear during the joining process. 

234 buff.append('') 

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

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

237 # the exact number of backslashes. 

238 if num_backslashes > 0: 

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

240 num_backslashes = 0 

241 

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

243 # if there is anything in the buffer. 

244 if buff: 

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

246 buff = [] 

247 else: 

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

249 # the exact number of backslashes. 

250 if num_backslashes > 0: 

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

252 num_backslashes = 0 

253 buff.append(character) 

254 

255 # Quotes must be terminated. 

256 if is_quoted: 

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

258 

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

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

261 if num_backslashes > 0: 

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

263 

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

265 if buff: 

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

267 

268 return components 

269 

270 

271def get_tzinfo_options(): 

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

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

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

275 if sys.platform == 'win32': 

276 from dateutil.tz import tzwinlocal 

277 

278 return (tzlocal, tzwinlocal) 

279 else: 

280 return (tzlocal,) 

281 

282 

283# Detect if CRT is available for use 

284try: 

285 import awscrt.auth 

286 

287 # Allow user opt-out if needed 

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

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

290except ImportError: 

291 HAS_CRT = False 

292 

293 

294def has_minimum_crt_version(minimum_version): 

295 """Not intended for use outside botocore.""" 

296 if not HAS_CRT: 

297 return False 

298 

299 crt_version_str = awscrt.__version__ 

300 try: 

301 crt_version_ints = map(int, crt_version_str.split(".")) 

302 crt_version_tuple = tuple(crt_version_ints) 

303 except (TypeError, ValueError): 

304 return False 

305 

306 return crt_version_tuple >= minimum_version 

307 

308 

309######################################################## 

310# urllib3 compat backports # 

311######################################################## 

312 

313# Vendoring IPv6 validation regex patterns from urllib3 

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

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

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

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

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

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

320_variations = [ 

321 # 6( h16 ":" ) ls32 

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

339] 

340 

341UNRESERVED_PAT = ( 

342 r"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789._!\-~" 

343) 

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

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

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

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

348 

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

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

351 

352# Detect if gzip is available for use 

353try: 

354 import gzip 

355 HAS_GZIP = True 

356except ImportError: 

357 HAS_GZIP = False