jpayne@7: """
jpayne@7: requests.utils
jpayne@7: ~~~~~~~~~~~~~~
jpayne@7: 
jpayne@7: This module provides utility functions that are used within Requests
jpayne@7: that are also useful for external consumption.
jpayne@7: """
jpayne@7: 
jpayne@7: import codecs
jpayne@7: import contextlib
jpayne@7: import io
jpayne@7: import os
jpayne@7: import re
jpayne@7: import socket
jpayne@7: import struct
jpayne@7: import sys
jpayne@7: import tempfile
jpayne@7: import warnings
jpayne@7: import zipfile
jpayne@7: from collections import OrderedDict
jpayne@7: 
jpayne@7: from urllib3.util import make_headers, parse_url
jpayne@7: 
jpayne@7: from . import certs
jpayne@7: from .__version__ import __version__
jpayne@7: 
jpayne@7: # to_native_string is unused here, but imported here for backwards compatibility
jpayne@7: from ._internal_utils import (  # noqa: F401
jpayne@7:     _HEADER_VALIDATORS_BYTE,
jpayne@7:     _HEADER_VALIDATORS_STR,
jpayne@7:     HEADER_VALIDATORS,
jpayne@7:     to_native_string,
jpayne@7: )
jpayne@7: from .compat import (
jpayne@7:     Mapping,
jpayne@7:     basestring,
jpayne@7:     bytes,
jpayne@7:     getproxies,
jpayne@7:     getproxies_environment,
jpayne@7:     integer_types,
jpayne@7: )
jpayne@7: from .compat import parse_http_list as _parse_list_header
jpayne@7: from .compat import (
jpayne@7:     proxy_bypass,
jpayne@7:     proxy_bypass_environment,
jpayne@7:     quote,
jpayne@7:     str,
jpayne@7:     unquote,
jpayne@7:     urlparse,
jpayne@7:     urlunparse,
jpayne@7: )
jpayne@7: from .cookies import cookiejar_from_dict
jpayne@7: from .exceptions import (
jpayne@7:     FileModeWarning,
jpayne@7:     InvalidHeader,
jpayne@7:     InvalidURL,
jpayne@7:     UnrewindableBodyError,
jpayne@7: )
jpayne@7: from .structures import CaseInsensitiveDict
jpayne@7: 
jpayne@7: NETRC_FILES = (".netrc", "_netrc")
jpayne@7: 
jpayne@7: DEFAULT_CA_BUNDLE_PATH = certs.where()
jpayne@7: 
jpayne@7: DEFAULT_PORTS = {"http": 80, "https": 443}
jpayne@7: 
jpayne@7: # Ensure that ', ' is used to preserve previous delimiter behavior.
jpayne@7: DEFAULT_ACCEPT_ENCODING = ", ".join(
jpayne@7:     re.split(r",\s*", make_headers(accept_encoding=True)["accept-encoding"])
jpayne@7: )
jpayne@7: 
jpayne@7: 
jpayne@7: if sys.platform == "win32":
jpayne@7:     # provide a proxy_bypass version on Windows without DNS lookups
jpayne@7: 
jpayne@7:     def proxy_bypass_registry(host):
jpayne@7:         try:
jpayne@7:             import winreg
jpayne@7:         except ImportError:
jpayne@7:             return False
jpayne@7: 
jpayne@7:         try:
jpayne@7:             internetSettings = winreg.OpenKey(
jpayne@7:                 winreg.HKEY_CURRENT_USER,
jpayne@7:                 r"Software\Microsoft\Windows\CurrentVersion\Internet Settings",
jpayne@7:             )
jpayne@7:             # ProxyEnable could be REG_SZ or REG_DWORD, normalizing it
jpayne@7:             proxyEnable = int(winreg.QueryValueEx(internetSettings, "ProxyEnable")[0])
jpayne@7:             # ProxyOverride is almost always a string
jpayne@7:             proxyOverride = winreg.QueryValueEx(internetSettings, "ProxyOverride")[0]
jpayne@7:         except (OSError, ValueError):
jpayne@7:             return False
jpayne@7:         if not proxyEnable or not proxyOverride:
jpayne@7:             return False
jpayne@7: 
jpayne@7:         # make a check value list from the registry entry: replace the
jpayne@7:         # '<local>' string by the localhost entry and the corresponding
jpayne@7:         # canonical entry.
jpayne@7:         proxyOverride = proxyOverride.split(";")
jpayne@7:         # now check if we match one of the registry values.
jpayne@7:         for test in proxyOverride:
jpayne@7:             if test == "<local>":
jpayne@7:                 if "." not in host:
jpayne@7:                     return True
jpayne@7:             test = test.replace(".", r"\.")  # mask dots
jpayne@7:             test = test.replace("*", r".*")  # change glob sequence
jpayne@7:             test = test.replace("?", r".")  # change glob char
jpayne@7:             if re.match(test, host, re.I):
jpayne@7:                 return True
jpayne@7:         return False
jpayne@7: 
jpayne@7:     def proxy_bypass(host):  # noqa
jpayne@7:         """Return True, if the host should be bypassed.
jpayne@7: 
jpayne@7:         Checks proxy settings gathered from the environment, if specified,
jpayne@7:         or the registry.
jpayne@7:         """
jpayne@7:         if getproxies_environment():
jpayne@7:             return proxy_bypass_environment(host)
jpayne@7:         else:
jpayne@7:             return proxy_bypass_registry(host)
jpayne@7: 
jpayne@7: 
jpayne@7: def dict_to_sequence(d):
jpayne@7:     """Returns an internal sequence dictionary update."""
jpayne@7: 
jpayne@7:     if hasattr(d, "items"):
jpayne@7:         d = d.items()
jpayne@7: 
jpayne@7:     return d
jpayne@7: 
jpayne@7: 
jpayne@7: def super_len(o):
jpayne@7:     total_length = None
jpayne@7:     current_position = 0
jpayne@7: 
jpayne@7:     if hasattr(o, "__len__"):
jpayne@7:         total_length = len(o)
jpayne@7: 
jpayne@7:     elif hasattr(o, "len"):
jpayne@7:         total_length = o.len
jpayne@7: 
jpayne@7:     elif hasattr(o, "fileno"):
jpayne@7:         try:
jpayne@7:             fileno = o.fileno()
jpayne@7:         except (io.UnsupportedOperation, AttributeError):
jpayne@7:             # AttributeError is a surprising exception, seeing as how we've just checked
jpayne@7:             # that `hasattr(o, 'fileno')`.  It happens for objects obtained via
jpayne@7:             # `Tarfile.extractfile()`, per issue 5229.
jpayne@7:             pass
jpayne@7:         else:
jpayne@7:             total_length = os.fstat(fileno).st_size
jpayne@7: 
jpayne@7:             # Having used fstat to determine the file length, we need to
jpayne@7:             # confirm that this file was opened up in binary mode.
jpayne@7:             if "b" not in o.mode:
jpayne@7:                 warnings.warn(
jpayne@7:                     (
jpayne@7:                         "Requests has determined the content-length for this "
jpayne@7:                         "request using the binary size of the file: however, the "
jpayne@7:                         "file has been opened in text mode (i.e. without the 'b' "
jpayne@7:                         "flag in the mode). This may lead to an incorrect "
jpayne@7:                         "content-length. In Requests 3.0, support will be removed "
jpayne@7:                         "for files in text mode."
jpayne@7:                     ),
jpayne@7:                     FileModeWarning,
jpayne@7:                 )
jpayne@7: 
jpayne@7:     if hasattr(o, "tell"):
jpayne@7:         try:
jpayne@7:             current_position = o.tell()
jpayne@7:         except OSError:
jpayne@7:             # This can happen in some weird situations, such as when the file
jpayne@7:             # is actually a special file descriptor like stdin. In this
jpayne@7:             # instance, we don't know what the length is, so set it to zero and
jpayne@7:             # let requests chunk it instead.
jpayne@7:             if total_length is not None:
jpayne@7:                 current_position = total_length
jpayne@7:         else:
jpayne@7:             if hasattr(o, "seek") and total_length is None:
jpayne@7:                 # StringIO and BytesIO have seek but no usable fileno
jpayne@7:                 try:
jpayne@7:                     # seek to end of file
jpayne@7:                     o.seek(0, 2)
jpayne@7:                     total_length = o.tell()
jpayne@7: 
jpayne@7:                     # seek back to current position to support
jpayne@7:                     # partially read file-like objects
jpayne@7:                     o.seek(current_position or 0)
jpayne@7:                 except OSError:
jpayne@7:                     total_length = 0
jpayne@7: 
jpayne@7:     if total_length is None:
jpayne@7:         total_length = 0
jpayne@7: 
jpayne@7:     return max(0, total_length - current_position)
jpayne@7: 
jpayne@7: 
jpayne@7: def get_netrc_auth(url, raise_errors=False):
jpayne@7:     """Returns the Requests tuple auth for a given url from netrc."""
jpayne@7: 
jpayne@7:     netrc_file = os.environ.get("NETRC")
jpayne@7:     if netrc_file is not None:
jpayne@7:         netrc_locations = (netrc_file,)
jpayne@7:     else:
jpayne@7:         netrc_locations = (f"~/{f}" for f in NETRC_FILES)
jpayne@7: 
jpayne@7:     try:
jpayne@7:         from netrc import NetrcParseError, netrc
jpayne@7: 
jpayne@7:         netrc_path = None
jpayne@7: 
jpayne@7:         for f in netrc_locations:
jpayne@7:             try:
jpayne@7:                 loc = os.path.expanduser(f)
jpayne@7:             except KeyError:
jpayne@7:                 # os.path.expanduser can fail when $HOME is undefined and
jpayne@7:                 # getpwuid fails. See https://bugs.python.org/issue20164 &
jpayne@7:                 # https://github.com/psf/requests/issues/1846
jpayne@7:                 return
jpayne@7: 
jpayne@7:             if os.path.exists(loc):
jpayne@7:                 netrc_path = loc
jpayne@7:                 break
jpayne@7: 
jpayne@7:         # Abort early if there isn't one.
jpayne@7:         if netrc_path is None:
jpayne@7:             return
jpayne@7: 
jpayne@7:         ri = urlparse(url)
jpayne@7: 
jpayne@7:         # Strip port numbers from netloc. This weird `if...encode`` dance is
jpayne@7:         # used for Python 3.2, which doesn't support unicode literals.
jpayne@7:         splitstr = b":"
jpayne@7:         if isinstance(url, str):
jpayne@7:             splitstr = splitstr.decode("ascii")
jpayne@7:         host = ri.netloc.split(splitstr)[0]
jpayne@7: 
jpayne@7:         try:
jpayne@7:             _netrc = netrc(netrc_path).authenticators(host)
jpayne@7:             if _netrc:
jpayne@7:                 # Return with login / password
jpayne@7:                 login_i = 0 if _netrc[0] else 1
jpayne@7:                 return (_netrc[login_i], _netrc[2])
jpayne@7:         except (NetrcParseError, OSError):
jpayne@7:             # If there was a parsing error or a permissions issue reading the file,
jpayne@7:             # we'll just skip netrc auth unless explicitly asked to raise errors.
jpayne@7:             if raise_errors:
jpayne@7:                 raise
jpayne@7: 
jpayne@7:     # App Engine hackiness.
jpayne@7:     except (ImportError, AttributeError):
jpayne@7:         pass
jpayne@7: 
jpayne@7: 
jpayne@7: def guess_filename(obj):
jpayne@7:     """Tries to guess the filename of the given object."""
jpayne@7:     name = getattr(obj, "name", None)
jpayne@7:     if name and isinstance(name, basestring) and name[0] != "<" and name[-1] != ">":
jpayne@7:         return os.path.basename(name)
jpayne@7: 
jpayne@7: 
jpayne@7: def extract_zipped_paths(path):
jpayne@7:     """Replace nonexistent paths that look like they refer to a member of a zip
jpayne@7:     archive with the location of an extracted copy of the target, or else
jpayne@7:     just return the provided path unchanged.
jpayne@7:     """
jpayne@7:     if os.path.exists(path):
jpayne@7:         # this is already a valid path, no need to do anything further
jpayne@7:         return path
jpayne@7: 
jpayne@7:     # find the first valid part of the provided path and treat that as a zip archive
jpayne@7:     # assume the rest of the path is the name of a member in the archive
jpayne@7:     archive, member = os.path.split(path)
jpayne@7:     while archive and not os.path.exists(archive):
jpayne@7:         archive, prefix = os.path.split(archive)
jpayne@7:         if not prefix:
jpayne@7:             # If we don't check for an empty prefix after the split (in other words, archive remains unchanged after the split),
jpayne@7:             # we _can_ end up in an infinite loop on a rare corner case affecting a small number of users
jpayne@7:             break
jpayne@7:         member = "/".join([prefix, member])
jpayne@7: 
jpayne@7:     if not zipfile.is_zipfile(archive):
jpayne@7:         return path
jpayne@7: 
jpayne@7:     zip_file = zipfile.ZipFile(archive)
jpayne@7:     if member not in zip_file.namelist():
jpayne@7:         return path
jpayne@7: 
jpayne@7:     # we have a valid zip archive and a valid member of that archive
jpayne@7:     tmp = tempfile.gettempdir()
jpayne@7:     extracted_path = os.path.join(tmp, member.split("/")[-1])
jpayne@7:     if not os.path.exists(extracted_path):
jpayne@7:         # use read + write to avoid the creating nested folders, we only want the file, avoids mkdir racing condition
jpayne@7:         with atomic_open(extracted_path) as file_handler:
jpayne@7:             file_handler.write(zip_file.read(member))
jpayne@7:     return extracted_path
jpayne@7: 
jpayne@7: 
jpayne@7: @contextlib.contextmanager
jpayne@7: def atomic_open(filename):
jpayne@7:     """Write a file to the disk in an atomic fashion"""
jpayne@7:     tmp_descriptor, tmp_name = tempfile.mkstemp(dir=os.path.dirname(filename))
jpayne@7:     try:
jpayne@7:         with os.fdopen(tmp_descriptor, "wb") as tmp_handler:
jpayne@7:             yield tmp_handler
jpayne@7:         os.replace(tmp_name, filename)
jpayne@7:     except BaseException:
jpayne@7:         os.remove(tmp_name)
jpayne@7:         raise
jpayne@7: 
jpayne@7: 
jpayne@7: def from_key_val_list(value):
jpayne@7:     """Take an object and test to see if it can be represented as a
jpayne@7:     dictionary. Unless it can not be represented as such, return an
jpayne@7:     OrderedDict, e.g.,
jpayne@7: 
jpayne@7:     ::
jpayne@7: 
jpayne@7:         >>> from_key_val_list([('key', 'val')])
jpayne@7:         OrderedDict([('key', 'val')])
jpayne@7:         >>> from_key_val_list('string')
jpayne@7:         Traceback (most recent call last):
jpayne@7:         ...
jpayne@7:         ValueError: cannot encode objects that are not 2-tuples
jpayne@7:         >>> from_key_val_list({'key': 'val'})
jpayne@7:         OrderedDict([('key', 'val')])
jpayne@7: 
jpayne@7:     :rtype: OrderedDict
jpayne@7:     """
jpayne@7:     if value is None:
jpayne@7:         return None
jpayne@7: 
jpayne@7:     if isinstance(value, (str, bytes, bool, int)):
jpayne@7:         raise ValueError("cannot encode objects that are not 2-tuples")
jpayne@7: 
jpayne@7:     return OrderedDict(value)
jpayne@7: 
jpayne@7: 
jpayne@7: def to_key_val_list(value):
jpayne@7:     """Take an object and test to see if it can be represented as a
jpayne@7:     dictionary. If it can be, return a list of tuples, e.g.,
jpayne@7: 
jpayne@7:     ::
jpayne@7: 
jpayne@7:         >>> to_key_val_list([('key', 'val')])
jpayne@7:         [('key', 'val')]
jpayne@7:         >>> to_key_val_list({'key': 'val'})
jpayne@7:         [('key', 'val')]
jpayne@7:         >>> to_key_val_list('string')
jpayne@7:         Traceback (most recent call last):
jpayne@7:         ...
jpayne@7:         ValueError: cannot encode objects that are not 2-tuples
jpayne@7: 
jpayne@7:     :rtype: list
jpayne@7:     """
jpayne@7:     if value is None:
jpayne@7:         return None
jpayne@7: 
jpayne@7:     if isinstance(value, (str, bytes, bool, int)):
jpayne@7:         raise ValueError("cannot encode objects that are not 2-tuples")
jpayne@7: 
jpayne@7:     if isinstance(value, Mapping):
jpayne@7:         value = value.items()
jpayne@7: 
jpayne@7:     return list(value)
jpayne@7: 
jpayne@7: 
jpayne@7: # From mitsuhiko/werkzeug (used with permission).
jpayne@7: def parse_list_header(value):
jpayne@7:     """Parse lists as described by RFC 2068 Section 2.
jpayne@7: 
jpayne@7:     In particular, parse comma-separated lists where the elements of
jpayne@7:     the list may include quoted-strings.  A quoted-string could
jpayne@7:     contain a comma.  A non-quoted string could have quotes in the
jpayne@7:     middle.  Quotes are removed automatically after parsing.
jpayne@7: 
jpayne@7:     It basically works like :func:`parse_set_header` just that items
jpayne@7:     may appear multiple times and case sensitivity is preserved.
jpayne@7: 
jpayne@7:     The return value is a standard :class:`list`:
jpayne@7: 
jpayne@7:     >>> parse_list_header('token, "quoted value"')
jpayne@7:     ['token', 'quoted value']
jpayne@7: 
jpayne@7:     To create a header from the :class:`list` again, use the
jpayne@7:     :func:`dump_header` function.
jpayne@7: 
jpayne@7:     :param value: a string with a list header.
jpayne@7:     :return: :class:`list`
jpayne@7:     :rtype: list
jpayne@7:     """
jpayne@7:     result = []
jpayne@7:     for item in _parse_list_header(value):
jpayne@7:         if item[:1] == item[-1:] == '"':
jpayne@7:             item = unquote_header_value(item[1:-1])
jpayne@7:         result.append(item)
jpayne@7:     return result
jpayne@7: 
jpayne@7: 
jpayne@7: # From mitsuhiko/werkzeug (used with permission).
jpayne@7: def parse_dict_header(value):
jpayne@7:     """Parse lists of key, value pairs as described by RFC 2068 Section 2 and
jpayne@7:     convert them into a python dict:
jpayne@7: 
jpayne@7:     >>> d = parse_dict_header('foo="is a fish", bar="as well"')
jpayne@7:     >>> type(d) is dict
jpayne@7:     True
jpayne@7:     >>> sorted(d.items())
jpayne@7:     [('bar', 'as well'), ('foo', 'is a fish')]
jpayne@7: 
jpayne@7:     If there is no value for a key it will be `None`:
jpayne@7: 
jpayne@7:     >>> parse_dict_header('key_without_value')
jpayne@7:     {'key_without_value': None}
jpayne@7: 
jpayne@7:     To create a header from the :class:`dict` again, use the
jpayne@7:     :func:`dump_header` function.
jpayne@7: 
jpayne@7:     :param value: a string with a dict header.
jpayne@7:     :return: :class:`dict`
jpayne@7:     :rtype: dict
jpayne@7:     """
jpayne@7:     result = {}
jpayne@7:     for item in _parse_list_header(value):
jpayne@7:         if "=" not in item:
jpayne@7:             result[item] = None
jpayne@7:             continue
jpayne@7:         name, value = item.split("=", 1)
jpayne@7:         if value[:1] == value[-1:] == '"':
jpayne@7:             value = unquote_header_value(value[1:-1])
jpayne@7:         result[name] = value
jpayne@7:     return result
jpayne@7: 
jpayne@7: 
jpayne@7: # From mitsuhiko/werkzeug (used with permission).
jpayne@7: def unquote_header_value(value, is_filename=False):
jpayne@7:     r"""Unquotes a header value.  (Reversal of :func:`quote_header_value`).
jpayne@7:     This does not use the real unquoting but what browsers are actually
jpayne@7:     using for quoting.
jpayne@7: 
jpayne@7:     :param value: the header value to unquote.
jpayne@7:     :rtype: str
jpayne@7:     """
jpayne@7:     if value and value[0] == value[-1] == '"':
jpayne@7:         # this is not the real unquoting, but fixing this so that the
jpayne@7:         # RFC is met will result in bugs with internet explorer and
jpayne@7:         # probably some other browsers as well.  IE for example is
jpayne@7:         # uploading files with "C:\foo\bar.txt" as filename
jpayne@7:         value = value[1:-1]
jpayne@7: 
jpayne@7:         # if this is a filename and the starting characters look like
jpayne@7:         # a UNC path, then just return the value without quotes.  Using the
jpayne@7:         # replace sequence below on a UNC path has the effect of turning
jpayne@7:         # the leading double slash into a single slash and then
jpayne@7:         # _fix_ie_filename() doesn't work correctly.  See #458.
jpayne@7:         if not is_filename or value[:2] != "\\\\":
jpayne@7:             return value.replace("\\\\", "\\").replace('\\"', '"')
jpayne@7:     return value
jpayne@7: 
jpayne@7: 
jpayne@7: def dict_from_cookiejar(cj):
jpayne@7:     """Returns a key/value dictionary from a CookieJar.
jpayne@7: 
jpayne@7:     :param cj: CookieJar object to extract cookies from.
jpayne@7:     :rtype: dict
jpayne@7:     """
jpayne@7: 
jpayne@7:     cookie_dict = {}
jpayne@7: 
jpayne@7:     for cookie in cj:
jpayne@7:         cookie_dict[cookie.name] = cookie.value
jpayne@7: 
jpayne@7:     return cookie_dict
jpayne@7: 
jpayne@7: 
jpayne@7: def add_dict_to_cookiejar(cj, cookie_dict):
jpayne@7:     """Returns a CookieJar from a key/value dictionary.
jpayne@7: 
jpayne@7:     :param cj: CookieJar to insert cookies into.
jpayne@7:     :param cookie_dict: Dict of key/values to insert into CookieJar.
jpayne@7:     :rtype: CookieJar
jpayne@7:     """
jpayne@7: 
jpayne@7:     return cookiejar_from_dict(cookie_dict, cj)
jpayne@7: 
jpayne@7: 
jpayne@7: def get_encodings_from_content(content):
jpayne@7:     """Returns encodings from given content string.
jpayne@7: 
jpayne@7:     :param content: bytestring to extract encodings from.
jpayne@7:     """
jpayne@7:     warnings.warn(
jpayne@7:         (
jpayne@7:             "In requests 3.0, get_encodings_from_content will be removed. For "
jpayne@7:             "more information, please see the discussion on issue #2266. (This"
jpayne@7:             " warning should only appear once.)"
jpayne@7:         ),
jpayne@7:         DeprecationWarning,
jpayne@7:     )
jpayne@7: 
jpayne@7:     charset_re = re.compile(r'<meta.*?charset=["\']*(.+?)["\'>]', flags=re.I)
jpayne@7:     pragma_re = re.compile(r'<meta.*?content=["\']*;?charset=(.+?)["\'>]', flags=re.I)
jpayne@7:     xml_re = re.compile(r'^<\?xml.*?encoding=["\']*(.+?)["\'>]')
jpayne@7: 
jpayne@7:     return (
jpayne@7:         charset_re.findall(content)
jpayne@7:         + pragma_re.findall(content)
jpayne@7:         + xml_re.findall(content)
jpayne@7:     )
jpayne@7: 
jpayne@7: 
jpayne@7: def _parse_content_type_header(header):
jpayne@7:     """Returns content type and parameters from given header
jpayne@7: 
jpayne@7:     :param header: string
jpayne@7:     :return: tuple containing content type and dictionary of
jpayne@7:          parameters
jpayne@7:     """
jpayne@7: 
jpayne@7:     tokens = header.split(";")
jpayne@7:     content_type, params = tokens[0].strip(), tokens[1:]
jpayne@7:     params_dict = {}
jpayne@7:     items_to_strip = "\"' "
jpayne@7: 
jpayne@7:     for param in params:
jpayne@7:         param = param.strip()
jpayne@7:         if param:
jpayne@7:             key, value = param, True
jpayne@7:             index_of_equals = param.find("=")
jpayne@7:             if index_of_equals != -1:
jpayne@7:                 key = param[:index_of_equals].strip(items_to_strip)
jpayne@7:                 value = param[index_of_equals + 1 :].strip(items_to_strip)
jpayne@7:             params_dict[key.lower()] = value
jpayne@7:     return content_type, params_dict
jpayne@7: 
jpayne@7: 
jpayne@7: def get_encoding_from_headers(headers):
jpayne@7:     """Returns encodings from given HTTP Header Dict.
jpayne@7: 
jpayne@7:     :param headers: dictionary to extract encoding from.
jpayne@7:     :rtype: str
jpayne@7:     """
jpayne@7: 
jpayne@7:     content_type = headers.get("content-type")
jpayne@7: 
jpayne@7:     if not content_type:
jpayne@7:         return None
jpayne@7: 
jpayne@7:     content_type, params = _parse_content_type_header(content_type)
jpayne@7: 
jpayne@7:     if "charset" in params:
jpayne@7:         return params["charset"].strip("'\"")
jpayne@7: 
jpayne@7:     if "text" in content_type:
jpayne@7:         return "ISO-8859-1"
jpayne@7: 
jpayne@7:     if "application/json" in content_type:
jpayne@7:         # Assume UTF-8 based on RFC 4627: https://www.ietf.org/rfc/rfc4627.txt since the charset was unset
jpayne@7:         return "utf-8"
jpayne@7: 
jpayne@7: 
jpayne@7: def stream_decode_response_unicode(iterator, r):
jpayne@7:     """Stream decodes an iterator."""
jpayne@7: 
jpayne@7:     if r.encoding is None:
jpayne@7:         yield from iterator
jpayne@7:         return
jpayne@7: 
jpayne@7:     decoder = codecs.getincrementaldecoder(r.encoding)(errors="replace")
jpayne@7:     for chunk in iterator:
jpayne@7:         rv = decoder.decode(chunk)
jpayne@7:         if rv:
jpayne@7:             yield rv
jpayne@7:     rv = decoder.decode(b"", final=True)
jpayne@7:     if rv:
jpayne@7:         yield rv
jpayne@7: 
jpayne@7: 
jpayne@7: def iter_slices(string, slice_length):
jpayne@7:     """Iterate over slices of a string."""
jpayne@7:     pos = 0
jpayne@7:     if slice_length is None or slice_length <= 0:
jpayne@7:         slice_length = len(string)
jpayne@7:     while pos < len(string):
jpayne@7:         yield string[pos : pos + slice_length]
jpayne@7:         pos += slice_length
jpayne@7: 
jpayne@7: 
jpayne@7: def get_unicode_from_response(r):
jpayne@7:     """Returns the requested content back in unicode.
jpayne@7: 
jpayne@7:     :param r: Response object to get unicode content from.
jpayne@7: 
jpayne@7:     Tried:
jpayne@7: 
jpayne@7:     1. charset from content-type
jpayne@7:     2. fall back and replace all unicode characters
jpayne@7: 
jpayne@7:     :rtype: str
jpayne@7:     """
jpayne@7:     warnings.warn(
jpayne@7:         (
jpayne@7:             "In requests 3.0, get_unicode_from_response will be removed. For "
jpayne@7:             "more information, please see the discussion on issue #2266. (This"
jpayne@7:             " warning should only appear once.)"
jpayne@7:         ),
jpayne@7:         DeprecationWarning,
jpayne@7:     )
jpayne@7: 
jpayne@7:     tried_encodings = []
jpayne@7: 
jpayne@7:     # Try charset from content-type
jpayne@7:     encoding = get_encoding_from_headers(r.headers)
jpayne@7: 
jpayne@7:     if encoding:
jpayne@7:         try:
jpayne@7:             return str(r.content, encoding)
jpayne@7:         except UnicodeError:
jpayne@7:             tried_encodings.append(encoding)
jpayne@7: 
jpayne@7:     # Fall back:
jpayne@7:     try:
jpayne@7:         return str(r.content, encoding, errors="replace")
jpayne@7:     except TypeError:
jpayne@7:         return r.content
jpayne@7: 
jpayne@7: 
jpayne@7: # The unreserved URI characters (RFC 3986)
jpayne@7: UNRESERVED_SET = frozenset(
jpayne@7:     "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz" + "0123456789-._~"
jpayne@7: )
jpayne@7: 
jpayne@7: 
jpayne@7: def unquote_unreserved(uri):
jpayne@7:     """Un-escape any percent-escape sequences in a URI that are unreserved
jpayne@7:     characters. This leaves all reserved, illegal and non-ASCII bytes encoded.
jpayne@7: 
jpayne@7:     :rtype: str
jpayne@7:     """
jpayne@7:     parts = uri.split("%")
jpayne@7:     for i in range(1, len(parts)):
jpayne@7:         h = parts[i][0:2]
jpayne@7:         if len(h) == 2 and h.isalnum():
jpayne@7:             try:
jpayne@7:                 c = chr(int(h, 16))
jpayne@7:             except ValueError:
jpayne@7:                 raise InvalidURL(f"Invalid percent-escape sequence: '{h}'")
jpayne@7: 
jpayne@7:             if c in UNRESERVED_SET:
jpayne@7:                 parts[i] = c + parts[i][2:]
jpayne@7:             else:
jpayne@7:                 parts[i] = f"%{parts[i]}"
jpayne@7:         else:
jpayne@7:             parts[i] = f"%{parts[i]}"
jpayne@7:     return "".join(parts)
jpayne@7: 
jpayne@7: 
jpayne@7: def requote_uri(uri):
jpayne@7:     """Re-quote the given URI.
jpayne@7: 
jpayne@7:     This function passes the given URI through an unquote/quote cycle to
jpayne@7:     ensure that it is fully and consistently quoted.
jpayne@7: 
jpayne@7:     :rtype: str
jpayne@7:     """
jpayne@7:     safe_with_percent = "!#$%&'()*+,/:;=?@[]~"
jpayne@7:     safe_without_percent = "!#$&'()*+,/:;=?@[]~"
jpayne@7:     try:
jpayne@7:         # Unquote only the unreserved characters
jpayne@7:         # Then quote only illegal characters (do not quote reserved,
jpayne@7:         # unreserved, or '%')
jpayne@7:         return quote(unquote_unreserved(uri), safe=safe_with_percent)
jpayne@7:     except InvalidURL:
jpayne@7:         # We couldn't unquote the given URI, so let's try quoting it, but
jpayne@7:         # there may be unquoted '%'s in the URI. We need to make sure they're
jpayne@7:         # properly quoted so they do not cause issues elsewhere.
jpayne@7:         return quote(uri, safe=safe_without_percent)
jpayne@7: 
jpayne@7: 
jpayne@7: def address_in_network(ip, net):
jpayne@7:     """This function allows you to check if an IP belongs to a network subnet
jpayne@7: 
jpayne@7:     Example: returns True if ip = 192.168.1.1 and net = 192.168.1.0/24
jpayne@7:              returns False if ip = 192.168.1.1 and net = 192.168.100.0/24
jpayne@7: 
jpayne@7:     :rtype: bool
jpayne@7:     """
jpayne@7:     ipaddr = struct.unpack("=L", socket.inet_aton(ip))[0]
jpayne@7:     netaddr, bits = net.split("/")
jpayne@7:     netmask = struct.unpack("=L", socket.inet_aton(dotted_netmask(int(bits))))[0]
jpayne@7:     network = struct.unpack("=L", socket.inet_aton(netaddr))[0] & netmask
jpayne@7:     return (ipaddr & netmask) == (network & netmask)
jpayne@7: 
jpayne@7: 
jpayne@7: def dotted_netmask(mask):
jpayne@7:     """Converts mask from /xx format to xxx.xxx.xxx.xxx
jpayne@7: 
jpayne@7:     Example: if mask is 24 function returns 255.255.255.0
jpayne@7: 
jpayne@7:     :rtype: str
jpayne@7:     """
jpayne@7:     bits = 0xFFFFFFFF ^ (1 << 32 - mask) - 1
jpayne@7:     return socket.inet_ntoa(struct.pack(">I", bits))
jpayne@7: 
jpayne@7: 
jpayne@7: def is_ipv4_address(string_ip):
jpayne@7:     """
jpayne@7:     :rtype: bool
jpayne@7:     """
jpayne@7:     try:
jpayne@7:         socket.inet_aton(string_ip)
jpayne@7:     except OSError:
jpayne@7:         return False
jpayne@7:     return True
jpayne@7: 
jpayne@7: 
jpayne@7: def is_valid_cidr(string_network):
jpayne@7:     """
jpayne@7:     Very simple check of the cidr format in no_proxy variable.
jpayne@7: 
jpayne@7:     :rtype: bool
jpayne@7:     """
jpayne@7:     if string_network.count("/") == 1:
jpayne@7:         try:
jpayne@7:             mask = int(string_network.split("/")[1])
jpayne@7:         except ValueError:
jpayne@7:             return False
jpayne@7: 
jpayne@7:         if mask < 1 or mask > 32:
jpayne@7:             return False
jpayne@7: 
jpayne@7:         try:
jpayne@7:             socket.inet_aton(string_network.split("/")[0])
jpayne@7:         except OSError:
jpayne@7:             return False
jpayne@7:     else:
jpayne@7:         return False
jpayne@7:     return True
jpayne@7: 
jpayne@7: 
jpayne@7: @contextlib.contextmanager
jpayne@7: def set_environ(env_name, value):
jpayne@7:     """Set the environment variable 'env_name' to 'value'
jpayne@7: 
jpayne@7:     Save previous value, yield, and then restore the previous value stored in
jpayne@7:     the environment variable 'env_name'.
jpayne@7: 
jpayne@7:     If 'value' is None, do nothing"""
jpayne@7:     value_changed = value is not None
jpayne@7:     if value_changed:
jpayne@7:         old_value = os.environ.get(env_name)
jpayne@7:         os.environ[env_name] = value
jpayne@7:     try:
jpayne@7:         yield
jpayne@7:     finally:
jpayne@7:         if value_changed:
jpayne@7:             if old_value is None:
jpayne@7:                 del os.environ[env_name]
jpayne@7:             else:
jpayne@7:                 os.environ[env_name] = old_value
jpayne@7: 
jpayne@7: 
jpayne@7: def should_bypass_proxies(url, no_proxy):
jpayne@7:     """
jpayne@7:     Returns whether we should bypass proxies or not.
jpayne@7: 
jpayne@7:     :rtype: bool
jpayne@7:     """
jpayne@7:     # Prioritize lowercase environment variables over uppercase
jpayne@7:     # to keep a consistent behaviour with other http projects (curl, wget).
jpayne@7:     def get_proxy(key):
jpayne@7:         return os.environ.get(key) or os.environ.get(key.upper())
jpayne@7: 
jpayne@7:     # First check whether no_proxy is defined. If it is, check that the URL
jpayne@7:     # we're getting isn't in the no_proxy list.
jpayne@7:     no_proxy_arg = no_proxy
jpayne@7:     if no_proxy is None:
jpayne@7:         no_proxy = get_proxy("no_proxy")
jpayne@7:     parsed = urlparse(url)
jpayne@7: 
jpayne@7:     if parsed.hostname is None:
jpayne@7:         # URLs don't always have hostnames, e.g. file:/// urls.
jpayne@7:         return True
jpayne@7: 
jpayne@7:     if no_proxy:
jpayne@7:         # We need to check whether we match here. We need to see if we match
jpayne@7:         # the end of the hostname, both with and without the port.
jpayne@7:         no_proxy = (host for host in no_proxy.replace(" ", "").split(",") if host)
jpayne@7: 
jpayne@7:         if is_ipv4_address(parsed.hostname):
jpayne@7:             for proxy_ip in no_proxy:
jpayne@7:                 if is_valid_cidr(proxy_ip):
jpayne@7:                     if address_in_network(parsed.hostname, proxy_ip):
jpayne@7:                         return True
jpayne@7:                 elif parsed.hostname == proxy_ip:
jpayne@7:                     # If no_proxy ip was defined in plain IP notation instead of cidr notation &
jpayne@7:                     # matches the IP of the index
jpayne@7:                     return True
jpayne@7:         else:
jpayne@7:             host_with_port = parsed.hostname
jpayne@7:             if parsed.port:
jpayne@7:                 host_with_port += f":{parsed.port}"
jpayne@7: 
jpayne@7:             for host in no_proxy:
jpayne@7:                 if parsed.hostname.endswith(host) or host_with_port.endswith(host):
jpayne@7:                     # The URL does match something in no_proxy, so we don't want
jpayne@7:                     # to apply the proxies on this URL.
jpayne@7:                     return True
jpayne@7: 
jpayne@7:     with set_environ("no_proxy", no_proxy_arg):
jpayne@7:         # parsed.hostname can be `None` in cases such as a file URI.
jpayne@7:         try:
jpayne@7:             bypass = proxy_bypass(parsed.hostname)
jpayne@7:         except (TypeError, socket.gaierror):
jpayne@7:             bypass = False
jpayne@7: 
jpayne@7:     if bypass:
jpayne@7:         return True
jpayne@7: 
jpayne@7:     return False
jpayne@7: 
jpayne@7: 
jpayne@7: def get_environ_proxies(url, no_proxy=None):
jpayne@7:     """
jpayne@7:     Return a dict of environment proxies.
jpayne@7: 
jpayne@7:     :rtype: dict
jpayne@7:     """
jpayne@7:     if should_bypass_proxies(url, no_proxy=no_proxy):
jpayne@7:         return {}
jpayne@7:     else:
jpayne@7:         return getproxies()
jpayne@7: 
jpayne@7: 
jpayne@7: def select_proxy(url, proxies):
jpayne@7:     """Select a proxy for the url, if applicable.
jpayne@7: 
jpayne@7:     :param url: The url being for the request
jpayne@7:     :param proxies: A dictionary of schemes or schemes and hosts to proxy URLs
jpayne@7:     """
jpayne@7:     proxies = proxies or {}
jpayne@7:     urlparts = urlparse(url)
jpayne@7:     if urlparts.hostname is None:
jpayne@7:         return proxies.get(urlparts.scheme, proxies.get("all"))
jpayne@7: 
jpayne@7:     proxy_keys = [
jpayne@7:         urlparts.scheme + "://" + urlparts.hostname,
jpayne@7:         urlparts.scheme,
jpayne@7:         "all://" + urlparts.hostname,
jpayne@7:         "all",
jpayne@7:     ]
jpayne@7:     proxy = None
jpayne@7:     for proxy_key in proxy_keys:
jpayne@7:         if proxy_key in proxies:
jpayne@7:             proxy = proxies[proxy_key]
jpayne@7:             break
jpayne@7: 
jpayne@7:     return proxy
jpayne@7: 
jpayne@7: 
jpayne@7: def resolve_proxies(request, proxies, trust_env=True):
jpayne@7:     """This method takes proxy information from a request and configuration
jpayne@7:     input to resolve a mapping of target proxies. This will consider settings
jpayne@7:     such a NO_PROXY to strip proxy configurations.
jpayne@7: 
jpayne@7:     :param request: Request or PreparedRequest
jpayne@7:     :param proxies: A dictionary of schemes or schemes and hosts to proxy URLs
jpayne@7:     :param trust_env: Boolean declaring whether to trust environment configs
jpayne@7: 
jpayne@7:     :rtype: dict
jpayne@7:     """
jpayne@7:     proxies = proxies if proxies is not None else {}
jpayne@7:     url = request.url
jpayne@7:     scheme = urlparse(url).scheme
jpayne@7:     no_proxy = proxies.get("no_proxy")
jpayne@7:     new_proxies = proxies.copy()
jpayne@7: 
jpayne@7:     if trust_env and not should_bypass_proxies(url, no_proxy=no_proxy):
jpayne@7:         environ_proxies = get_environ_proxies(url, no_proxy=no_proxy)
jpayne@7: 
jpayne@7:         proxy = environ_proxies.get(scheme, environ_proxies.get("all"))
jpayne@7: 
jpayne@7:         if proxy:
jpayne@7:             new_proxies.setdefault(scheme, proxy)
jpayne@7:     return new_proxies
jpayne@7: 
jpayne@7: 
jpayne@7: def default_user_agent(name="python-requests"):
jpayne@7:     """
jpayne@7:     Return a string representing the default user agent.
jpayne@7: 
jpayne@7:     :rtype: str
jpayne@7:     """
jpayne@7:     return f"{name}/{__version__}"
jpayne@7: 
jpayne@7: 
jpayne@7: def default_headers():
jpayne@7:     """
jpayne@7:     :rtype: requests.structures.CaseInsensitiveDict
jpayne@7:     """
jpayne@7:     return CaseInsensitiveDict(
jpayne@7:         {
jpayne@7:             "User-Agent": default_user_agent(),
jpayne@7:             "Accept-Encoding": DEFAULT_ACCEPT_ENCODING,
jpayne@7:             "Accept": "*/*",
jpayne@7:             "Connection": "keep-alive",
jpayne@7:         }
jpayne@7:     )
jpayne@7: 
jpayne@7: 
jpayne@7: def parse_header_links(value):
jpayne@7:     """Return a list of parsed link headers proxies.
jpayne@7: 
jpayne@7:     i.e. Link: <http:/.../front.jpeg>; rel=front; type="image/jpeg",<http://.../back.jpeg>; rel=back;type="image/jpeg"
jpayne@7: 
jpayne@7:     :rtype: list
jpayne@7:     """
jpayne@7: 
jpayne@7:     links = []
jpayne@7: 
jpayne@7:     replace_chars = " '\""
jpayne@7: 
jpayne@7:     value = value.strip(replace_chars)
jpayne@7:     if not value:
jpayne@7:         return links
jpayne@7: 
jpayne@7:     for val in re.split(", *<", value):
jpayne@7:         try:
jpayne@7:             url, params = val.split(";", 1)
jpayne@7:         except ValueError:
jpayne@7:             url, params = val, ""
jpayne@7: 
jpayne@7:         link = {"url": url.strip("<> '\"")}
jpayne@7: 
jpayne@7:         for param in params.split(";"):
jpayne@7:             try:
jpayne@7:                 key, value = param.split("=")
jpayne@7:             except ValueError:
jpayne@7:                 break
jpayne@7: 
jpayne@7:             link[key.strip(replace_chars)] = value.strip(replace_chars)
jpayne@7: 
jpayne@7:         links.append(link)
jpayne@7: 
jpayne@7:     return links
jpayne@7: 
jpayne@7: 
jpayne@7: # Null bytes; no need to recreate these on each call to guess_json_utf
jpayne@7: _null = "\x00".encode("ascii")  # encoding to ASCII for Python 3
jpayne@7: _null2 = _null * 2
jpayne@7: _null3 = _null * 3
jpayne@7: 
jpayne@7: 
jpayne@7: def guess_json_utf(data):
jpayne@7:     """
jpayne@7:     :rtype: str
jpayne@7:     """
jpayne@7:     # JSON always starts with two ASCII characters, so detection is as
jpayne@7:     # easy as counting the nulls and from their location and count
jpayne@7:     # determine the encoding. Also detect a BOM, if present.
jpayne@7:     sample = data[:4]
jpayne@7:     if sample in (codecs.BOM_UTF32_LE, codecs.BOM_UTF32_BE):
jpayne@7:         return "utf-32"  # BOM included
jpayne@7:     if sample[:3] == codecs.BOM_UTF8:
jpayne@7:         return "utf-8-sig"  # BOM included, MS style (discouraged)
jpayne@7:     if sample[:2] in (codecs.BOM_UTF16_LE, codecs.BOM_UTF16_BE):
jpayne@7:         return "utf-16"  # BOM included
jpayne@7:     nullcount = sample.count(_null)
jpayne@7:     if nullcount == 0:
jpayne@7:         return "utf-8"
jpayne@7:     if nullcount == 2:
jpayne@7:         if sample[::2] == _null2:  # 1st and 3rd are null
jpayne@7:             return "utf-16-be"
jpayne@7:         if sample[1::2] == _null2:  # 2nd and 4th are null
jpayne@7:             return "utf-16-le"
jpayne@7:         # Did not detect 2 valid UTF-16 ascii-range characters
jpayne@7:     if nullcount == 3:
jpayne@7:         if sample[:3] == _null3:
jpayne@7:             return "utf-32-be"
jpayne@7:         if sample[1:] == _null3:
jpayne@7:             return "utf-32-le"
jpayne@7:         # Did not detect a valid UTF-32 ascii-range character
jpayne@7:     return None
jpayne@7: 
jpayne@7: 
jpayne@7: def prepend_scheme_if_needed(url, new_scheme):
jpayne@7:     """Given a URL that may or may not have a scheme, prepend the given scheme.
jpayne@7:     Does not replace a present scheme with the one provided as an argument.
jpayne@7: 
jpayne@7:     :rtype: str
jpayne@7:     """
jpayne@7:     parsed = parse_url(url)
jpayne@7:     scheme, auth, host, port, path, query, fragment = parsed
jpayne@7: 
jpayne@7:     # A defect in urlparse determines that there isn't a netloc present in some
jpayne@7:     # urls. We previously assumed parsing was overly cautious, and swapped the
jpayne@7:     # netloc and path. Due to a lack of tests on the original defect, this is
jpayne@7:     # maintained with parse_url for backwards compatibility.
jpayne@7:     netloc = parsed.netloc
jpayne@7:     if not netloc:
jpayne@7:         netloc, path = path, netloc
jpayne@7: 
jpayne@7:     if auth:
jpayne@7:         # parse_url doesn't provide the netloc with auth
jpayne@7:         # so we'll add it ourselves.
jpayne@7:         netloc = "@".join([auth, netloc])
jpayne@7:     if scheme is None:
jpayne@7:         scheme = new_scheme
jpayne@7:     if path is None:
jpayne@7:         path = ""
jpayne@7: 
jpayne@7:     return urlunparse((scheme, netloc, path, "", query, fragment))
jpayne@7: 
jpayne@7: 
jpayne@7: def get_auth_from_url(url):
jpayne@7:     """Given a url with authentication components, extract them into a tuple of
jpayne@7:     username,password.
jpayne@7: 
jpayne@7:     :rtype: (str,str)
jpayne@7:     """
jpayne@7:     parsed = urlparse(url)
jpayne@7: 
jpayne@7:     try:
jpayne@7:         auth = (unquote(parsed.username), unquote(parsed.password))
jpayne@7:     except (AttributeError, TypeError):
jpayne@7:         auth = ("", "")
jpayne@7: 
jpayne@7:     return auth
jpayne@7: 
jpayne@7: 
jpayne@7: def check_header_validity(header):
jpayne@7:     """Verifies that header parts don't contain leading whitespace
jpayne@7:     reserved characters, or return characters.
jpayne@7: 
jpayne@7:     :param header: tuple, in the format (name, value).
jpayne@7:     """
jpayne@7:     name, value = header
jpayne@7:     _validate_header_part(header, name, 0)
jpayne@7:     _validate_header_part(header, value, 1)
jpayne@7: 
jpayne@7: 
jpayne@7: def _validate_header_part(header, header_part, header_validator_index):
jpayne@7:     if isinstance(header_part, str):
jpayne@7:         validator = _HEADER_VALIDATORS_STR[header_validator_index]
jpayne@7:     elif isinstance(header_part, bytes):
jpayne@7:         validator = _HEADER_VALIDATORS_BYTE[header_validator_index]
jpayne@7:     else:
jpayne@7:         raise InvalidHeader(
jpayne@7:             f"Header part ({header_part!r}) from {header} "
jpayne@7:             f"must be of type str or bytes, not {type(header_part)}"
jpayne@7:         )
jpayne@7: 
jpayne@7:     if not validator.match(header_part):
jpayne@7:         header_kind = "name" if header_validator_index == 0 else "value"
jpayne@7:         raise InvalidHeader(
jpayne@7:             f"Invalid leading whitespace, reserved character(s), or return"
jpayne@7:             f"character(s) in header {header_kind}: {header_part!r}"
jpayne@7:         )
jpayne@7: 
jpayne@7: 
jpayne@7: def urldefragauth(url):
jpayne@7:     """
jpayne@7:     Given a url remove the fragment and the authentication part.
jpayne@7: 
jpayne@7:     :rtype: str
jpayne@7:     """
jpayne@7:     scheme, netloc, path, params, query, fragment = urlparse(url)
jpayne@7: 
jpayne@7:     # see func:`prepend_scheme_if_needed`
jpayne@7:     if not netloc:
jpayne@7:         netloc, path = path, netloc
jpayne@7: 
jpayne@7:     netloc = netloc.rsplit("@", 1)[-1]
jpayne@7: 
jpayne@7:     return urlunparse((scheme, netloc, path, params, query, ""))
jpayne@7: 
jpayne@7: 
jpayne@7: def rewind_body(prepared_request):
jpayne@7:     """Move file pointer back to its recorded starting position
jpayne@7:     so it can be read again on redirect.
jpayne@7:     """
jpayne@7:     body_seek = getattr(prepared_request.body, "seek", None)
jpayne@7:     if body_seek is not None and isinstance(
jpayne@7:         prepared_request._body_position, integer_types
jpayne@7:     ):
jpayne@7:         try:
jpayne@7:             body_seek(prepared_request._body_position)
jpayne@7:         except OSError:
jpayne@7:             raise UnrewindableBodyError(
jpayne@7:                 "An error occurred when rewinding request body for redirect."
jpayne@7:             )
jpayne@7:     else:
jpayne@7:         raise UnrewindableBodyError("Unable to rewind request body for redirect.")