Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/requests/utils.py: 16%

1""" 

2requests.utils 

3~~~~~~~~~~~~~~ 

4 

5This module provides utility functions that are used within Requests 

6that are also useful for external consumption. 

7""" 

8 

9import codecs 

10import contextlib 

11import io 

12import os 

13import re 

14import socket 

15import struct 

16import sys 

17import tempfile 

18import warnings 

19import zipfile 

20from collections import OrderedDict 

21 

22from urllib3.util import make_headers, parse_url 

23 

24from . import certs 

25from .__version__ import __version__ 

26 

27# to_native_string is unused here, but imported here for backwards compatibility 

28from ._internal_utils import ( # noqa: F401 

29 _HEADER_VALIDATORS_BYTE, 

30 _HEADER_VALIDATORS_STR, 

31 HEADER_VALIDATORS, 

32 to_native_string, 

33) 

34from .compat import ( 

35 Mapping, 

36 basestring, 

37 bytes, 

38 getproxies, 

39 getproxies_environment, 

40 integer_types, 

41 is_urllib3_1, 

42) 

43from .compat import parse_http_list as _parse_list_header 

44from .compat import ( 

45 proxy_bypass, 

46 proxy_bypass_environment, 

47 quote, 

48 str, 

49 unquote, 

50 urlparse, 

51 urlunparse, 

52) 

53from .cookies import cookiejar_from_dict 

54from .exceptions import ( 

55 FileModeWarning, 

56 InvalidHeader, 

57 InvalidURL, 

58 UnrewindableBodyError, 

59) 

60from .structures import CaseInsensitiveDict 

61 

62NETRC_FILES = (".netrc", "_netrc") 

63 

64DEFAULT_CA_BUNDLE_PATH = certs.where() 

65 

66DEFAULT_PORTS = {"http": 80, "https": 443} 

67 

68# Ensure that ', ' is used to preserve previous delimiter behavior. 

69DEFAULT_ACCEPT_ENCODING = ", ".join( 

70 re.split(r",\s*", make_headers(accept_encoding=True)["accept-encoding"]) 

71) 

72 

73 

74if sys.platform == "win32": 

75 # provide a proxy_bypass version on Windows without DNS lookups 

76 

77 def proxy_bypass_registry(host): 

78 try: 

79 import winreg 

80 except ImportError: 

81 return False 

82 

83 try: 

84 internetSettings = winreg.OpenKey( 

85 winreg.HKEY_CURRENT_USER, 

86 r"Software\Microsoft\Windows\CurrentVersion\Internet Settings", 

87 ) 

88 # ProxyEnable could be REG_SZ or REG_DWORD, normalizing it 

89 proxyEnable = int(winreg.QueryValueEx(internetSettings, "ProxyEnable")[0]) 

90 # ProxyOverride is almost always a string 

91 proxyOverride = winreg.QueryValueEx(internetSettings, "ProxyOverride")[0] 

92 except (OSError, ValueError): 

93 return False 

94 if not proxyEnable or not proxyOverride: 

95 return False 

96 

97 # make a check value list from the registry entry: replace the 

98 # '<local>' string by the localhost entry and the corresponding 

99 # canonical entry. 

100 proxyOverride = proxyOverride.split(";") 

101 # filter out empty strings to avoid re.match return true in the following code. 

102 proxyOverride = filter(None, proxyOverride) 

103 # now check if we match one of the registry values. 

104 for test in proxyOverride: 

105 if test == "<local>": 

106 if "." not in host: 

107 return True 

108 test = test.replace(".", r"\.") # mask dots 

109 test = test.replace("*", r".*") # change glob sequence 

110 test = test.replace("?", r".") # change glob char 

111 if re.match(test, host, re.I): 

112 return True 

113 return False 

114 

115 def proxy_bypass(host): # noqa 

116 """Return True, if the host should be bypassed. 

117 

118 Checks proxy settings gathered from the environment, if specified, 

119 or the registry. 

120 """ 

121 if getproxies_environment(): 

122 return proxy_bypass_environment(host) 

123 else: 

124 return proxy_bypass_registry(host) 

125 

126 

127def dict_to_sequence(d): 

128 """Returns an internal sequence dictionary update.""" 

129 

130 if hasattr(d, "items"): 

131 d = d.items() 

132 

133 return d 

134 

135 

136def super_len(o): 

137 total_length = None 

138 current_position = 0 

139 

140 if not is_urllib3_1 and isinstance(o, str): 

141 # urllib3 2.x+ treats all strings as utf-8 instead 

142 # of latin-1 (iso-8859-1) like http.client. 

143 o = o.encode("utf-8") 

144 

145 if hasattr(o, "__len__"): 

146 total_length = len(o) 

147 

148 elif hasattr(o, "len"): 

149 total_length = o.len 

150 

151 elif hasattr(o, "fileno"): 

152 try: 

153 fileno = o.fileno() 

154 except (io.UnsupportedOperation, AttributeError): 

155 # AttributeError is a surprising exception, seeing as how we've just checked 

156 # that `hasattr(o, 'fileno')`. It happens for objects obtained via 

157 # `Tarfile.extractfile()`, per issue 5229. 

158 pass 

159 else: 

160 total_length = os.fstat(fileno).st_size 

161 

162 # Having used fstat to determine the file length, we need to 

163 # confirm that this file was opened up in binary mode. 

164 if "b" not in o.mode: 

165 warnings.warn( 

166 ( 

167 "Requests has determined the content-length for this " 

168 "request using the binary size of the file: however, the " 

169 "file has been opened in text mode (i.e. without the 'b' " 

170 "flag in the mode). This may lead to an incorrect " 

171 "content-length. In Requests 3.0, support will be removed " 

172 "for files in text mode." 

173 ), 

174 FileModeWarning, 

175 ) 

176 

177 if hasattr(o, "tell"): 

178 try: 

179 current_position = o.tell() 

180 except OSError: 

181 # This can happen in some weird situations, such as when the file 

182 # is actually a special file descriptor like stdin. In this 

183 # instance, we don't know what the length is, so set it to zero and 

184 # let requests chunk it instead. 

185 if total_length is not None: 

186 current_position = total_length 

187 else: 

188 if hasattr(o, "seek") and total_length is None: 

189 # StringIO and BytesIO have seek but no usable fileno 

190 try: 

191 # seek to end of file 

192 o.seek(0, 2) 

193 total_length = o.tell() 

194 

195 # seek back to current position to support 

196 # partially read file-like objects 

197 o.seek(current_position or 0) 

198 except OSError: 

199 total_length = 0 

200 

201 if total_length is None: 

202 total_length = 0 

203 

204 return max(0, total_length - current_position) 

205 

206 

207def get_netrc_auth(url, raise_errors=False): 

208 """Returns the Requests tuple auth for a given url from netrc.""" 

209 

210 netrc_file = os.environ.get("NETRC") 

211 if netrc_file is not None: 

212 netrc_locations = (netrc_file,) 

213 else: 

214 netrc_locations = (f"~/{f}" for f in NETRC_FILES) 

215 

216 try: 

217 from netrc import NetrcParseError, netrc 

218 

219 netrc_path = None 

220 

221 for f in netrc_locations: 

222 loc = os.path.expanduser(f) 

223 if os.path.exists(loc): 

224 netrc_path = loc 

225 break 

226 

227 # Abort early if there isn't one. 

228 if netrc_path is None: 

229 return 

230 

231 ri = urlparse(url) 

232 host = ri.hostname 

233 

234 try: 

235 _netrc = netrc(netrc_path).authenticators(host) 

236 if _netrc: 

237 # Return with login / password 

238 login_i = 0 if _netrc[0] else 1 

239 return (_netrc[login_i], _netrc[2]) 

240 except (NetrcParseError, OSError): 

241 # If there was a parsing error or a permissions issue reading the file, 

242 # we'll just skip netrc auth unless explicitly asked to raise errors. 

243 if raise_errors: 

244 raise 

245 

246 # App Engine hackiness. 

247 except (ImportError, AttributeError): 

248 pass 

249 

250 

251def guess_filename(obj): 

252 """Tries to guess the filename of the given object.""" 

253 name = getattr(obj, "name", None) 

254 if name and isinstance(name, basestring) and name[0] != "<" and name[-1] != ">": 

255 return os.path.basename(name) 

256 

257 

258def extract_zipped_paths(path): 

259 """Replace nonexistent paths that look like they refer to a member of a zip 

260 archive with the location of an extracted copy of the target, or else 

261 just return the provided path unchanged. 

262 """ 

263 if os.path.exists(path): 

264 # this is already a valid path, no need to do anything further 

265 return path 

266 

267 # find the first valid part of the provided path and treat that as a zip archive 

268 # assume the rest of the path is the name of a member in the archive 

269 archive, member = os.path.split(path) 

270 while archive and not os.path.exists(archive): 

271 archive, prefix = os.path.split(archive) 

272 if not prefix: 

273 # If we don't check for an empty prefix after the split (in other words, archive remains unchanged after the split), 

274 # we _can_ end up in an infinite loop on a rare corner case affecting a small number of users 

275 break 

276 member = "/".join([prefix, member]) 

277 

278 if not zipfile.is_zipfile(archive): 

279 return path 

280 

281 zip_file = zipfile.ZipFile(archive) 

282 if member not in zip_file.namelist(): 

283 return path 

284 

285 # we have a valid zip archive and a valid member of that archive 

286 tmp = tempfile.gettempdir() 

287 extracted_path = os.path.join(tmp, member.split("/")[-1]) 

288 if not os.path.exists(extracted_path): 

289 # use read + write to avoid the creating nested folders, we only want the file, avoids mkdir racing condition 

290 with atomic_open(extracted_path) as file_handler: 

291 file_handler.write(zip_file.read(member)) 

292 return extracted_path 

293 

294 

295@contextlib.contextmanager 

296def atomic_open(filename): 

297 """Write a file to the disk in an atomic fashion""" 

298 tmp_descriptor, tmp_name = tempfile.mkstemp(dir=os.path.dirname(filename)) 

299 try: 

300 with os.fdopen(tmp_descriptor, "wb") as tmp_handler: 

301 yield tmp_handler 

302 os.replace(tmp_name, filename) 

303 except BaseException: 

304 os.remove(tmp_name) 

305 raise 

306 

307 

308def from_key_val_list(value): 

309 """Take an object and test to see if it can be represented as a 

310 dictionary. Unless it can not be represented as such, return an 

311 OrderedDict, e.g., 

312 

313 :: 

314 

315 >>> from_key_val_list([('key', 'val')]) 

316 OrderedDict([('key', 'val')]) 

317 >>> from_key_val_list('string') 

318 Traceback (most recent call last): 

319 ... 

320 ValueError: cannot encode objects that are not 2-tuples 

321 >>> from_key_val_list({'key': 'val'}) 

322 OrderedDict([('key', 'val')]) 

323 

324 :rtype: OrderedDict 

325 """ 

326 if value is None: 

327 return None 

328 

329 if isinstance(value, (str, bytes, bool, int)): 

330 raise ValueError("cannot encode objects that are not 2-tuples") 

331 

332 return OrderedDict(value) 

333 

334 

335def to_key_val_list(value): 

336 """Take an object and test to see if it can be represented as a 

337 dictionary. If it can be, return a list of tuples, e.g., 

338 

339 :: 

340 

341 >>> to_key_val_list([('key', 'val')]) 

342 [('key', 'val')] 

343 >>> to_key_val_list({'key': 'val'}) 

344 [('key', 'val')] 

345 >>> to_key_val_list('string') 

346 Traceback (most recent call last): 

347 ... 

348 ValueError: cannot encode objects that are not 2-tuples 

349 

350 :rtype: list 

351 """ 

352 if value is None: 

353 return None 

354 

355 if isinstance(value, (str, bytes, bool, int)): 

356 raise ValueError("cannot encode objects that are not 2-tuples") 

357 

358 if isinstance(value, Mapping): 

359 value = value.items() 

360 

361 return list(value) 

362 

363 

364# From mitsuhiko/werkzeug (used with permission). 

365def parse_list_header(value): 

366 """Parse lists as described by RFC 2068 Section 2. 

367 

368 In particular, parse comma-separated lists where the elements of 

369 the list may include quoted-strings. A quoted-string could 

370 contain a comma. A non-quoted string could have quotes in the 

371 middle. Quotes are removed automatically after parsing. 

372 

373 It basically works like :func:`parse_set_header` just that items 

374 may appear multiple times and case sensitivity is preserved. 

375 

376 The return value is a standard :class:`list`: 

377 

378 >>> parse_list_header('token, "quoted value"') 

379 ['token', 'quoted value'] 

380 

381 To create a header from the :class:`list` again, use the 

382 :func:`dump_header` function. 

383 

384 :param value: a string with a list header. 

385 :return: :class:`list` 

386 :rtype: list 

387 """ 

388 result = [] 

389 for item in _parse_list_header(value): 

390 if item[:1] == item[-1:] == '"': 

391 item = unquote_header_value(item[1:-1]) 

392 result.append(item) 

393 return result 

394 

395 

396# From mitsuhiko/werkzeug (used with permission). 

397def parse_dict_header(value): 

398 """Parse lists of key, value pairs as described by RFC 2068 Section 2 and 

399 convert them into a python dict: 

400 

401 >>> d = parse_dict_header('foo="is a fish", bar="as well"') 

402 >>> type(d) is dict 

403 True 

404 >>> sorted(d.items()) 

405 [('bar', 'as well'), ('foo', 'is a fish')] 

406 

407 If there is no value for a key it will be `None`: 

408 

409 >>> parse_dict_header('key_without_value') 

410 {'key_without_value': None} 

411 

412 To create a header from the :class:`dict` again, use the 

413 :func:`dump_header` function. 

414 

415 :param value: a string with a dict header. 

416 :return: :class:`dict` 

417 :rtype: dict 

418 """ 

419 result = {} 

420 for item in _parse_list_header(value): 

421 if "=" not in item: 

422 result[item] = None 

423 continue 

424 name, value = item.split("=", 1) 

425 if value[:1] == value[-1:] == '"': 

426 value = unquote_header_value(value[1:-1]) 

427 result[name] = value 

428 return result 

429 

430 

431# From mitsuhiko/werkzeug (used with permission). 

432def unquote_header_value(value, is_filename=False): 

433 r"""Unquotes a header value. (Reversal of :func:`quote_header_value`). 

434 This does not use the real unquoting but what browsers are actually 

435 using for quoting. 

436 

437 :param value: the header value to unquote. 

438 :rtype: str 

439 """ 

440 if value and value[0] == value[-1] == '"': 

441 # this is not the real unquoting, but fixing this so that the 

442 # RFC is met will result in bugs with internet explorer and 

443 # probably some other browsers as well. IE for example is 

444 # uploading files with "C:\foo\bar.txt" as filename 

445 value = value[1:-1] 

446 

447 # if this is a filename and the starting characters look like 

448 # a UNC path, then just return the value without quotes. Using the 

449 # replace sequence below on a UNC path has the effect of turning 

450 # the leading double slash into a single slash and then 

451 # _fix_ie_filename() doesn't work correctly. See #458. 

452 if not is_filename or value[:2] != "\\\\": 

453 return value.replace("\\\\", "\\").replace('\\"', '"') 

454 return value 

455 

456 

457def dict_from_cookiejar(cj): 

458 """Returns a key/value dictionary from a CookieJar. 

459 

460 :param cj: CookieJar object to extract cookies from. 

461 :rtype: dict 

462 """ 

463 

464 cookie_dict = {cookie.name: cookie.value for cookie in cj} 

465 return cookie_dict 

466 

467 

468def add_dict_to_cookiejar(cj, cookie_dict): 

469 """Returns a CookieJar from a key/value dictionary. 

470 

471 :param cj: CookieJar to insert cookies into. 

472 :param cookie_dict: Dict of key/values to insert into CookieJar. 

473 :rtype: CookieJar 

474 """ 

475 

476 return cookiejar_from_dict(cookie_dict, cj) 

477 

478 

479def get_encodings_from_content(content): 

480 """Returns encodings from given content string. 

481 

482 :param content: bytestring to extract encodings from. 

483 """ 

484 warnings.warn( 

485 ( 

486 "In requests 3.0, get_encodings_from_content will be removed. For " 

487 "more information, please see the discussion on issue #2266. (This" 

488 " warning should only appear once.)" 

489 ), 

490 DeprecationWarning, 

491 ) 

492 

493 charset_re = re.compile(r'<meta.*?charset=["\']*(.+?)["\'>]', flags=re.I) 

494 pragma_re = re.compile(r'<meta.*?content=["\']*;?charset=(.+?)["\'>]', flags=re.I) 

495 xml_re = re.compile(r'^<\?xml.*?encoding=["\']*(.+?)["\'>]') 

496 

497 return ( 

498 charset_re.findall(content) 

499 + pragma_re.findall(content) 

500 + xml_re.findall(content) 

501 ) 

502 

503 

504def _parse_content_type_header(header): 

505 """Returns content type and parameters from given header 

506 

507 :param header: string 

508 :return: tuple containing content type and dictionary of 

509 parameters 

510 """ 

511 

512 tokens = header.split(";") 

513 content_type, params = tokens[0].strip(), tokens[1:] 

514 params_dict = {} 

515 items_to_strip = "\"' " 

516 

517 for param in params: 

518 param = param.strip() 

519 if param: 

520 key, value = param, True 

521 index_of_equals = param.find("=") 

522 if index_of_equals != -1: 

523 key = param[:index_of_equals].strip(items_to_strip) 

524 value = param[index_of_equals + 1 :].strip(items_to_strip) 

525 params_dict[key.lower()] = value 

526 return content_type, params_dict 

527 

528 

529def get_encoding_from_headers(headers): 

530 """Returns encodings from given HTTP Header Dict. 

531 

532 :param headers: dictionary to extract encoding from. 

533 :rtype: str 

534 """ 

535 

536 content_type = headers.get("content-type") 

537 

538 if not content_type: 

539 return None 

540 

541 content_type, params = _parse_content_type_header(content_type) 

542 

543 if "charset" in params: 

544 return params["charset"].strip("'\"") 

545 

546 if "text" in content_type: 

547 return "ISO-8859-1" 

548 

549 if "application/json" in content_type: 

550 # Assume UTF-8 based on RFC 4627: https://www.ietf.org/rfc/rfc4627.txt since the charset was unset 

551 return "utf-8" 

552 

553 

554def stream_decode_response_unicode(iterator, r): 

555 """Stream decodes an iterator.""" 

556 

557 if r.encoding is None: 

558 yield from iterator 

559 return 

560 

561 decoder = codecs.getincrementaldecoder(r.encoding)(errors="replace") 

562 for chunk in iterator: 

563 rv = decoder.decode(chunk) 

564 if rv: 

565 yield rv 

566 rv = decoder.decode(b"", final=True) 

567 if rv: 

568 yield rv 

569 

570 

571def iter_slices(string, slice_length): 

572 """Iterate over slices of a string.""" 

573 pos = 0 

574 if slice_length is None or slice_length <= 0: 

575 slice_length = len(string) 

576 while pos < len(string): 

577 yield string[pos : pos + slice_length] 

578 pos += slice_length 

579 

580 

581def get_unicode_from_response(r): 

582 """Returns the requested content back in unicode. 

583 

584 :param r: Response object to get unicode content from. 

585 

586 Tried: 

587 

588 1. charset from content-type 

589 2. fall back and replace all unicode characters 

590 

591 :rtype: str 

592 """ 

593 warnings.warn( 

594 ( 

595 "In requests 3.0, get_unicode_from_response will be removed. For " 

596 "more information, please see the discussion on issue #2266. (This" 

597 " warning should only appear once.)" 

598 ), 

599 DeprecationWarning, 

600 ) 

601 

602 tried_encodings = [] 

603 

604 # Try charset from content-type 

605 encoding = get_encoding_from_headers(r.headers) 

606 

607 if encoding: 

608 try: 

609 return str(r.content, encoding) 

610 except UnicodeError: 

611 tried_encodings.append(encoding) 

612 

613 # Fall back: 

614 try: 

615 return str(r.content, encoding, errors="replace") 

616 except TypeError: 

617 return r.content 

618 

619 

620# The unreserved URI characters (RFC 3986) 

621UNRESERVED_SET = frozenset( 

622 "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz" + "0123456789-._~" 

623) 

624 

625 

626def unquote_unreserved(uri): 

627 """Un-escape any percent-escape sequences in a URI that are unreserved 

628 characters. This leaves all reserved, illegal and non-ASCII bytes encoded. 

629 

630 :rtype: str 

631 """ 

632 parts = uri.split("%") 

633 for i in range(1, len(parts)): 

634 h = parts[i][0:2] 

635 if len(h) == 2 and h.isalnum(): 

636 try: 

637 c = chr(int(h, 16)) 

638 except ValueError: 

639 raise InvalidURL(f"Invalid percent-escape sequence: '{h}'") 

640 

641 if c in UNRESERVED_SET: 

642 parts[i] = c + parts[i][2:] 

643 else: 

644 parts[i] = f"%{parts[i]}" 

645 else: 

646 parts[i] = f"%{parts[i]}" 

647 return "".join(parts) 

648 

649 

650def requote_uri(uri): 

651 """Re-quote the given URI. 

652 

653 This function passes the given URI through an unquote/quote cycle to 

654 ensure that it is fully and consistently quoted. 

655 

656 :rtype: str 

657 """ 

658 safe_with_percent = "!#$%&'()*+,/:;=?@[]~" 

659 safe_without_percent = "!#$&'()*+,/:;=?@[]~" 

660 try: 

661 # Unquote only the unreserved characters 

662 # Then quote only illegal characters (do not quote reserved, 

663 # unreserved, or '%') 

664 return quote(unquote_unreserved(uri), safe=safe_with_percent) 

665 except InvalidURL: 

666 # We couldn't unquote the given URI, so let's try quoting it, but 

667 # there may be unquoted '%'s in the URI. We need to make sure they're 

668 # properly quoted so they do not cause issues elsewhere. 

669 return quote(uri, safe=safe_without_percent) 

670 

671 

672def address_in_network(ip, net): 

673 """This function allows you to check if an IP belongs to a network subnet 

674 

675 Example: returns True if ip = 192.168.1.1 and net = 192.168.1.0/24 

676 returns False if ip = 192.168.1.1 and net = 192.168.100.0/24 

677 

678 :rtype: bool 

679 """ 

680 ipaddr = struct.unpack("=L", socket.inet_aton(ip))[0] 

681 netaddr, bits = net.split("/") 

682 netmask = struct.unpack("=L", socket.inet_aton(dotted_netmask(int(bits))))[0] 

683 network = struct.unpack("=L", socket.inet_aton(netaddr))[0] & netmask 

684 return (ipaddr & netmask) == (network & netmask) 

685 

686 

687def dotted_netmask(mask): 

688 """Converts mask from /xx format to xxx.xxx.xxx.xxx 

689 

690 Example: if mask is 24 function returns 255.255.255.0 

691 

692 :rtype: str 

693 """ 

694 bits = 0xFFFFFFFF ^ (1 << 32 - mask) - 1 

695 return socket.inet_ntoa(struct.pack(">I", bits)) 

696 

697 

698def is_ipv4_address(string_ip): 

699 """ 

700 :rtype: bool 

701 """ 

702 try: 

703 socket.inet_aton(string_ip) 

704 except OSError: 

705 return False 

706 return True 

707 

708 

709def is_valid_cidr(string_network): 

710 """ 

711 Very simple check of the cidr format in no_proxy variable. 

712 

713 :rtype: bool 

714 """ 

715 if string_network.count("/") == 1: 

716 try: 

717 mask = int(string_network.split("/")[1]) 

718 except ValueError: 

719 return False 

720 

721 if mask < 1 or mask > 32: 

722 return False 

723 

724 try: 

725 socket.inet_aton(string_network.split("/")[0]) 

726 except OSError: 

727 return False 

728 else: 

729 return False 

730 return True 

731 

732 

733@contextlib.contextmanager 

734def set_environ(env_name, value): 

735 """Set the environment variable 'env_name' to 'value' 

736 

737 Save previous value, yield, and then restore the previous value stored in 

738 the environment variable 'env_name'. 

739 

740 If 'value' is None, do nothing""" 

741 value_changed = value is not None 

742 if value_changed: 

743 old_value = os.environ.get(env_name) 

744 os.environ[env_name] = value 

745 try: 

746 yield 

747 finally: 

748 if value_changed: 

749 if old_value is None: 

750 del os.environ[env_name] 

751 else: 

752 os.environ[env_name] = old_value 

753 

754 

755def should_bypass_proxies(url, no_proxy): 

756 """ 

757 Returns whether we should bypass proxies or not. 

758 

759 :rtype: bool 

760 """ 

761 

762 # Prioritize lowercase environment variables over uppercase 

763 # to keep a consistent behaviour with other http projects (curl, wget). 

764 def get_proxy(key): 

765 return os.environ.get(key) or os.environ.get(key.upper()) 

766 

767 # First check whether no_proxy is defined. If it is, check that the URL 

768 # we're getting isn't in the no_proxy list. 

769 no_proxy_arg = no_proxy 

770 if no_proxy is None: 

771 no_proxy = get_proxy("no_proxy") 

772 parsed = urlparse(url) 

773 

774 if parsed.hostname is None: 

775 # URLs don't always have hostnames, e.g. file:/// urls. 

776 return True 

777 

778 if no_proxy: 

779 # We need to check whether we match here. We need to see if we match 

780 # the end of the hostname, both with and without the port. 

781 no_proxy = (host for host in no_proxy.replace(" ", "").split(",") if host) 

782 

783 if is_ipv4_address(parsed.hostname): 

784 for proxy_ip in no_proxy: 

785 if is_valid_cidr(proxy_ip): 

786 if address_in_network(parsed.hostname, proxy_ip): 

787 return True 

788 elif parsed.hostname == proxy_ip: 

789 # If no_proxy ip was defined in plain IP notation instead of cidr notation & 

790 # matches the IP of the index 

791 return True 

792 else: 

793 host_with_port = parsed.hostname 

794 if parsed.port: 

795 host_with_port += f":{parsed.port}" 

796 

797 for host in no_proxy: 

798 if parsed.hostname.endswith(host) or host_with_port.endswith(host): 

799 # The URL does match something in no_proxy, so we don't want 

800 # to apply the proxies on this URL. 

801 return True 

802 

803 with set_environ("no_proxy", no_proxy_arg): 

804 # parsed.hostname can be `None` in cases such as a file URI. 

805 try: 

806 bypass = proxy_bypass(parsed.hostname) 

807 except (TypeError, socket.gaierror): 

808 bypass = False 

809 

810 if bypass: 

811 return True 

812 

813 return False 

814 

815 

816def get_environ_proxies(url, no_proxy=None): 

817 """ 

818 Return a dict of environment proxies. 

819 

820 :rtype: dict 

821 """ 

822 if should_bypass_proxies(url, no_proxy=no_proxy): 

823 return {} 

824 else: 

825 return getproxies() 

826 

827 

828def select_proxy(url, proxies): 

829 """Select a proxy for the url, if applicable. 

830 

831 :param url: The url being for the request 

832 :param proxies: A dictionary of schemes or schemes and hosts to proxy URLs 

833 """ 

834 proxies = proxies or {} 

835 urlparts = urlparse(url) 

836 if urlparts.hostname is None: 

837 return proxies.get(urlparts.scheme, proxies.get("all")) 

838 

839 proxy_keys = [ 

840 urlparts.scheme + "://" + urlparts.hostname, 

841 urlparts.scheme, 

842 "all://" + urlparts.hostname, 

843 "all", 

844 ] 

845 proxy = None 

846 for proxy_key in proxy_keys: 

847 if proxy_key in proxies: 

848 proxy = proxies[proxy_key] 

849 break 

850 

851 return proxy 

852 

853 

854def resolve_proxies(request, proxies, trust_env=True): 

855 """This method takes proxy information from a request and configuration 

856 input to resolve a mapping of target proxies. This will consider settings 

857 such as NO_PROXY to strip proxy configurations. 

858 

859 :param request: Request or PreparedRequest 

860 :param proxies: A dictionary of schemes or schemes and hosts to proxy URLs 

861 :param trust_env: Boolean declaring whether to trust environment configs 

862 

863 :rtype: dict 

864 """ 

865 proxies = proxies if proxies is not None else {} 

866 url = request.url 

867 scheme = urlparse(url).scheme 

868 no_proxy = proxies.get("no_proxy") 

869 new_proxies = proxies.copy() 

870 

871 if trust_env and not should_bypass_proxies(url, no_proxy=no_proxy): 

872 environ_proxies = get_environ_proxies(url, no_proxy=no_proxy) 

873 

874 proxy = environ_proxies.get(scheme, environ_proxies.get("all")) 

875 

876 if proxy: 

877 new_proxies.setdefault(scheme, proxy) 

878 return new_proxies 

879 

880 

881def default_user_agent(name="python-requests"): 

882 """ 

883 Return a string representing the default user agent. 

884 

885 :rtype: str 

886 """ 

887 return f"{name}/{__version__}" 

888 

889 

890def default_headers(): 

891 """ 

892 :rtype: requests.structures.CaseInsensitiveDict 

893 """ 

894 return CaseInsensitiveDict( 

895 { 

896 "User-Agent": default_user_agent(), 

897 "Accept-Encoding": DEFAULT_ACCEPT_ENCODING, 

898 "Accept": "*/*", 

899 "Connection": "keep-alive", 

900 } 

901 ) 

902 

903 

904def parse_header_links(value): 

905 """Return a list of parsed link headers proxies. 

906 

907 i.e. Link: <http:/.../front.jpeg>; rel=front; type="image/jpeg",<http://.../back.jpeg>; rel=back;type="image/jpeg" 

908 

909 :rtype: list 

910 """ 

911 

912 links = [] 

913 

914 replace_chars = " '\"" 

915 

916 value = value.strip(replace_chars) 

917 if not value: 

918 return links 

919 

920 for val in re.split(", *<", value): 

921 try: 

922 url, params = val.split(";", 1) 

923 except ValueError: 

924 url, params = val, "" 

925 

926 link = {"url": url.strip("<> '\"")} 

927 

928 for param in params.split(";"): 

929 try: 

930 key, value = param.split("=") 

931 except ValueError: 

932 break 

933 

934 link[key.strip(replace_chars)] = value.strip(replace_chars) 

935 

936 links.append(link) 

937 

938 return links 

939 

940 

941# Null bytes; no need to recreate these on each call to guess_json_utf 

942_null = "\x00".encode("ascii") # encoding to ASCII for Python 3 

943_null2 = _null * 2 

944_null3 = _null * 3 

945 

946 

947def guess_json_utf(data): 

948 """ 

949 :rtype: str 

950 """ 

951 # JSON always starts with two ASCII characters, so detection is as 

952 # easy as counting the nulls and from their location and count 

953 # determine the encoding. Also detect a BOM, if present. 

954 sample = data[:4] 

955 if sample in (codecs.BOM_UTF32_LE, codecs.BOM_UTF32_BE): 

956 return "utf-32" # BOM included 

957 if sample[:3] == codecs.BOM_UTF8: 

958 return "utf-8-sig" # BOM included, MS style (discouraged) 

959 if sample[:2] in (codecs.BOM_UTF16_LE, codecs.BOM_UTF16_BE): 

960 return "utf-16" # BOM included 

961 nullcount = sample.count(_null)