File manager - Edit - /home/u478019808/domains/bestandroidphones.store/public_html/static/img/logo/utils.py.tar
Back
lib/python3.9/site-packages/dateutil/utils.py 0000644 00000003647 15025037654 0015153 0 ustar 00 # -*- coding: utf-8 -*- """ This module offers general convenience and utility functions for dealing with datetimes. .. versionadded:: 2.7.0 """ from __future__ import unicode_literals from datetime import datetime, time def today(tzinfo=None): """ Returns a :py:class:`datetime` representing the current day at midnight :param tzinfo: The time zone to attach (also used to determine the current day). :return: A :py:class:`datetime.datetime` object representing the current day at midnight. """ dt = datetime.now(tzinfo) return datetime.combine(dt.date(), time(0, tzinfo=tzinfo)) def default_tzinfo(dt, tzinfo): """ Sets the ``tzinfo`` parameter on naive datetimes only This is useful for example when you are provided a datetime that may have either an implicit or explicit time zone, such as when parsing a time zone string. .. doctest:: >>> from dateutil.tz import tzoffset >>> from dateutil.parser import parse >>> from dateutil.utils import default_tzinfo >>> dflt_tz = tzoffset("EST", -18000) >>> print(default_tzinfo(parse('2014-01-01 12:30 UTC'), dflt_tz)) 2014-01-01 12:30:00+00:00 >>> print(default_tzinfo(parse('2014-01-01 12:30'), dflt_tz)) 2014-01-01 12:30:00-05:00 :param dt: The datetime on which to replace the time zone :param tzinfo: The :py:class:`datetime.tzinfo` subclass instance to assign to ``dt`` if (and only if) it is naive. :return: Returns an aware :py:class:`datetime.datetime`. """ if dt.tzinfo is not None: return dt else: return dt.replace(tzinfo=tzinfo) def within_delta(dt1, dt2, delta): """ Useful for comparing two datetimes that may a negilible difference to be considered equal. """ delta = abs(delta) difference = dt1 - dt2 return -delta <= difference <= delta lib/python3.9/site-packages/requests/utils.py 0000644 00000076634 15025043533 0015212 0 ustar 00 # -*- coding: utf-8 -*- """ requests.utils ~~~~~~~~~~~~~~ This module provides utility functions that are used within Requests that are also useful for external consumption. """ import codecs import contextlib import io import os import re import socket import struct import sys import tempfile import warnings import zipfile from collections import OrderedDict from .__version__ import __version__ from . import certs # to_native_string is unused here, but imported here for backwards compatibility from ._internal_utils import to_native_string from .compat import parse_http_list as _parse_list_header from .compat import ( quote, urlparse, bytes, str, unquote, getproxies, proxy_bypass, urlunparse, basestring, integer_types, is_py3, proxy_bypass_environment, getproxies_environment, Mapping) from .cookies import cookiejar_from_dict from .structures import CaseInsensitiveDict from .exceptions import ( InvalidURL, InvalidHeader, FileModeWarning, UnrewindableBodyError) NETRC_FILES = ('.netrc', '_netrc') DEFAULT_CA_BUNDLE_PATH = certs.where() DEFAULT_PORTS = {'http': 80, 'https': 443} if sys.platform == 'win32': # provide a proxy_bypass version on Windows without DNS lookups def proxy_bypass_registry(host): try: if is_py3: import winreg else: import _winreg as winreg except ImportError: return False try: internetSettings = winreg.OpenKey(winreg.HKEY_CURRENT_USER, r'Software\Microsoft\Windows\CurrentVersion\Internet Settings') # ProxyEnable could be REG_SZ or REG_DWORD, normalizing it proxyEnable = int(winreg.QueryValueEx(internetSettings, 'ProxyEnable')[0]) # ProxyOverride is almost always a string proxyOverride = winreg.QueryValueEx(internetSettings, 'ProxyOverride')[0] except OSError: return False if not proxyEnable or not proxyOverride: return False # make a check value list from the registry entry: replace the # '<local>' string by the localhost entry and the corresponding # canonical entry. proxyOverride = proxyOverride.split(';') # now check if we match one of the registry values. for test in proxyOverride: if test == '<local>': if '.' not in host: return True test = test.replace(".", r"\.") # mask dots test = test.replace("*", r".*") # change glob sequence test = test.replace("?", r".") # change glob char if re.match(test, host, re.I): return True return False def proxy_bypass(host): # noqa """Return True, if the host should be bypassed. Checks proxy settings gathered from the environment, if specified, or the registry. """ if getproxies_environment(): return proxy_bypass_environment(host) else: return proxy_bypass_registry(host) def dict_to_sequence(d): """Returns an internal sequence dictionary update.""" if hasattr(d, 'items'): d = d.items() return d def super_len(o): total_length = None current_position = 0 if hasattr(o, '__len__'): total_length = len(o) elif hasattr(o, 'len'): total_length = o.len elif hasattr(o, 'fileno'): try: fileno = o.fileno() except io.UnsupportedOperation: pass else: total_length = os.fstat(fileno).st_size # Having used fstat to determine the file length, we need to # confirm that this file was opened up in binary mode. if 'b' not in o.mode: warnings.warn(( "Requests has determined the content-length for this " "request using the binary size of the file: however, the " "file has been opened in text mode (i.e. without the 'b' " "flag in the mode). This may lead to an incorrect " "content-length. In Requests 3.0, support will be removed " "for files in text mode."), FileModeWarning ) if hasattr(o, 'tell'): try: current_position = o.tell() except (OSError, IOError): # This can happen in some weird situations, such as when the file # is actually a special file descriptor like stdin. In this # instance, we don't know what the length is, so set it to zero and # let requests chunk it instead. if total_length is not None: current_position = total_length else: if hasattr(o, 'seek') and total_length is None: # StringIO and BytesIO have seek but no useable fileno try: # seek to end of file o.seek(0, 2) total_length = o.tell() # seek back to current position to support # partially read file-like objects o.seek(current_position or 0) except (OSError, IOError): total_length = 0 if total_length is None: total_length = 0 return max(0, total_length - current_position) def get_netrc_auth(url, raise_errors=False): """Returns the Requests tuple auth for a given url from netrc.""" netrc_file = os.environ.get('NETRC') if netrc_file is not None: netrc_locations = (netrc_file,) else: netrc_locations = ('~/{}'.format(f) for f in NETRC_FILES) try: from netrc import netrc, NetrcParseError netrc_path = None for f in netrc_locations: try: loc = os.path.expanduser(f) except KeyError: # os.path.expanduser can fail when $HOME is undefined and # getpwuid fails. See https://bugs.python.org/issue20164 & # https://github.com/psf/requests/issues/1846 return if os.path.exists(loc): netrc_path = loc break # Abort early if there isn't one. if netrc_path is None: return ri = urlparse(url) # Strip port numbers from netloc. This weird `if...encode`` dance is # used for Python 3.2, which doesn't support unicode literals. splitstr = b':' if isinstance(url, str): splitstr = splitstr.decode('ascii') host = ri.netloc.split(splitstr)[0] try: _netrc = netrc(netrc_path).authenticators(host) if _netrc: # Return with login / password login_i = (0 if _netrc[0] else 1) return (_netrc[login_i], _netrc[2]) except (NetrcParseError, IOError): # If there was a parsing error or a permissions issue reading the file, # we'll just skip netrc auth unless explicitly asked to raise errors. if raise_errors: raise # App Engine hackiness. except (ImportError, AttributeError): pass def guess_filename(obj): """Tries to guess the filename of the given object.""" name = getattr(obj, 'name', None) if (name and isinstance(name, basestring) and name[0] != '<' and name[-1] != '>'): return os.path.basename(name) def extract_zipped_paths(path): """Replace nonexistent paths that look like they refer to a member of a zip archive with the location of an extracted copy of the target, or else just return the provided path unchanged. """ if os.path.exists(path): # this is already a valid path, no need to do anything further return path # find the first valid part of the provided path and treat that as a zip archive # assume the rest of the path is the name of a member in the archive archive, member = os.path.split(path) while archive and not os.path.exists(archive): archive, prefix = os.path.split(archive) member = '/'.join([prefix, member]) if not zipfile.is_zipfile(archive): return path zip_file = zipfile.ZipFile(archive) if member not in zip_file.namelist(): return path # we have a valid zip archive and a valid member of that archive tmp = tempfile.gettempdir() extracted_path = os.path.join(tmp, *member.split('/')) if not os.path.exists(extracted_path): extracted_path = zip_file.extract(member, path=tmp) return extracted_path def from_key_val_list(value): """Take an object and test to see if it can be represented as a dictionary. Unless it can not be represented as such, return an OrderedDict, e.g., :: >>> from_key_val_list([('key', 'val')]) OrderedDict([('key', 'val')]) >>> from_key_val_list('string') Traceback (most recent call last): ... ValueError: cannot encode objects that are not 2-tuples >>> from_key_val_list({'key': 'val'}) OrderedDict([('key', 'val')]) :rtype: OrderedDict """ if value is None: return None if isinstance(value, (str, bytes, bool, int)): raise ValueError('cannot encode objects that are not 2-tuples') return OrderedDict(value) def to_key_val_list(value): """Take an object and test to see if it can be represented as a dictionary. If it can be, return a list of tuples, e.g., :: >>> to_key_val_list([('key', 'val')]) [('key', 'val')] >>> to_key_val_list({'key': 'val'}) [('key', 'val')] >>> to_key_val_list('string') Traceback (most recent call last): ... ValueError: cannot encode objects that are not 2-tuples :rtype: list """ if value is None: return None if isinstance(value, (str, bytes, bool, int)): raise ValueError('cannot encode objects that are not 2-tuples') if isinstance(value, Mapping): value = value.items() return list(value) # From mitsuhiko/werkzeug (used with permission). def parse_list_header(value): """Parse lists as described by RFC 2068 Section 2. In particular, parse comma-separated lists where the elements of the list may include quoted-strings. A quoted-string could contain a comma. A non-quoted string could have quotes in the middle. Quotes are removed automatically after parsing. It basically works like :func:`parse_set_header` just that items may appear multiple times and case sensitivity is preserved. The return value is a standard :class:`list`: >>> parse_list_header('token, "quoted value"') ['token', 'quoted value'] To create a header from the :class:`list` again, use the :func:`dump_header` function. :param value: a string with a list header. :return: :class:`list` :rtype: list """ result = [] for item in _parse_list_header(value): if item[:1] == item[-1:] == '"': item = unquote_header_value(item[1:-1]) result.append(item) return result # From mitsuhiko/werkzeug (used with permission). def parse_dict_header(value): """Parse lists of key, value pairs as described by RFC 2068 Section 2 and convert them into a python dict: >>> d = parse_dict_header('foo="is a fish", bar="as well"') >>> type(d) is dict True >>> sorted(d.items()) [('bar', 'as well'), ('foo', 'is a fish')] If there is no value for a key it will be `None`: >>> parse_dict_header('key_without_value') {'key_without_value': None} To create a header from the :class:`dict` again, use the :func:`dump_header` function. :param value: a string with a dict header. :return: :class:`dict` :rtype: dict """ result = {} for item in _parse_list_header(value): if '=' not in item: result[item] = None continue name, value = item.split('=', 1) if value[:1] == value[-1:] == '"': value = unquote_header_value(value[1:-1]) result[name] = value return result # From mitsuhiko/werkzeug (used with permission). def unquote_header_value(value, is_filename=False): r"""Unquotes a header value. (Reversal of :func:`quote_header_value`). This does not use the real unquoting but what browsers are actually using for quoting. :param value: the header value to unquote. :rtype: str """ if value and value[0] == value[-1] == '"': # this is not the real unquoting, but fixing this so that the # RFC is met will result in bugs with internet explorer and # probably some other browsers as well. IE for example is # uploading files with "C:\foo\bar.txt" as filename value = value[1:-1] # if this is a filename and the starting characters look like # a UNC path, then just return the value without quotes. Using the # replace sequence below on a UNC path has the effect of turning # the leading double slash into a single slash and then # _fix_ie_filename() doesn't work correctly. See #458. if not is_filename or value[:2] != '\\\\': return value.replace('\\\\', '\\').replace('\\"', '"') return value def dict_from_cookiejar(cj): """Returns a key/value dictionary from a CookieJar. :param cj: CookieJar object to extract cookies from. :rtype: dict """ cookie_dict = {} for cookie in cj: cookie_dict[cookie.name] = cookie.value return cookie_dict def add_dict_to_cookiejar(cj, cookie_dict): """Returns a CookieJar from a key/value dictionary. :param cj: CookieJar to insert cookies into. :param cookie_dict: Dict of key/values to insert into CookieJar. :rtype: CookieJar """ return cookiejar_from_dict(cookie_dict, cj) def get_encodings_from_content(content): """Returns encodings from given content string. :param content: bytestring to extract encodings from. """ warnings.warn(( 'In requests 3.0, get_encodings_from_content will be removed. For ' 'more information, please see the discussion on issue #2266. (This' ' warning should only appear once.)'), DeprecationWarning) charset_re = re.compile(r'<meta.*?charset=["\']*(.+?)["\'>]', flags=re.I) pragma_re = re.compile(r'<meta.*?content=["\']*;?charset=(.+?)["\'>]', flags=re.I) xml_re = re.compile(r'^<\?xml.*?encoding=["\']*(.+?)["\'>]') return (charset_re.findall(content) + pragma_re.findall(content) + xml_re.findall(content)) def _parse_content_type_header(header): """Returns content type and parameters from given header :param header: string :return: tuple containing content type and dictionary of parameters """ tokens = header.split(';') content_type, params = tokens[0].strip(), tokens[1:] params_dict = {} items_to_strip = "\"' " for param in params: param = param.strip() if param: key, value = param, True index_of_equals = param.find("=") if index_of_equals != -1: key = param[:index_of_equals].strip(items_to_strip) value = param[index_of_equals + 1:].strip(items_to_strip) params_dict[key.lower()] = value return content_type, params_dict def get_encoding_from_headers(headers): """Returns encodings from given HTTP Header Dict. :param headers: dictionary to extract encoding from. :rtype: str """ content_type = headers.get('content-type') if not content_type: return None content_type, params = _parse_content_type_header(content_type) if 'charset' in params: return params['charset'].strip("'\"") if 'text' in content_type: return 'ISO-8859-1' if 'application/json' in content_type: # Assume UTF-8 based on RFC 4627: https://www.ietf.org/rfc/rfc4627.txt since the charset was unset return 'utf-8' def stream_decode_response_unicode(iterator, r): """Stream decodes a iterator.""" if r.encoding is None: for item in iterator: yield item return decoder = codecs.getincrementaldecoder(r.encoding)(errors='replace') for chunk in iterator: rv = decoder.decode(chunk) if rv: yield rv rv = decoder.decode(b'', final=True) if rv: yield rv def iter_slices(string, slice_length): """Iterate over slices of a string.""" pos = 0 if slice_length is None or slice_length <= 0: slice_length = len(string) while pos < len(string): yield string[pos:pos + slice_length] pos += slice_length def get_unicode_from_response(r): """Returns the requested content back in unicode. :param r: Response object to get unicode content from. Tried: 1. charset from content-type 2. fall back and replace all unicode characters :rtype: str """ warnings.warn(( 'In requests 3.0, get_unicode_from_response will be removed. For ' 'more information, please see the discussion on issue #2266. (This' ' warning should only appear once.)'), DeprecationWarning) tried_encodings = [] # Try charset from content-type encoding = get_encoding_from_headers(r.headers) if encoding: try: return str(r.content, encoding) except UnicodeError: tried_encodings.append(encoding) # Fall back: try: return str(r.content, encoding, errors='replace') except TypeError: return r.content # The unreserved URI characters (RFC 3986) UNRESERVED_SET = frozenset( "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz" + "0123456789-._~") def unquote_unreserved(uri): """Un-escape any percent-escape sequences in a URI that are unreserved characters. This leaves all reserved, illegal and non-ASCII bytes encoded. :rtype: str """ parts = uri.split('%') for i in range(1, len(parts)): h = parts[i][0:2] if len(h) == 2 and h.isalnum(): try: c = chr(int(h, 16)) except ValueError: raise InvalidURL("Invalid percent-escape sequence: '%s'" % h) if c in UNRESERVED_SET: parts[i] = c + parts[i][2:] else: parts[i] = '%' + parts[i] else: parts[i] = '%' + parts[i] return ''.join(parts) def requote_uri(uri): """Re-quote the given URI. This function passes the given URI through an unquote/quote cycle to ensure that it is fully and consistently quoted. :rtype: str """ safe_with_percent = "!#$%&'()*+,/:;=?@[]~" safe_without_percent = "!#$&'()*+,/:;=?@[]~" try: # Unquote only the unreserved characters # Then quote only illegal characters (do not quote reserved, # unreserved, or '%') return quote(unquote_unreserved(uri), safe=safe_with_percent) except InvalidURL: # We couldn't unquote the given URI, so let's try quoting it, but # there may be unquoted '%'s in the URI. We need to make sure they're # properly quoted so they do not cause issues elsewhere. return quote(uri, safe=safe_without_percent) def _get_mask_bits(mask, totalbits=32): """Converts a mask from /xx format to a int to be used as a mask for IP's in int format Example: if mask is 24 function returns 0xFFFFFF00 if mask is 24 and totalbits=128 function returns 0xFFFFFF00000000000000000000000000 :rtype: int """ bits = ((1 << mask) - 1) << (totalbits - mask) return bits def address_in_network(ip, net): """This function allows you to check if an IP belongs to a network subnet Example: returns True if ip = 192.168.1.1 and net = 192.168.1.0/24 returns False if ip = 192.168.1.1 and net = 192.168.100.0/24 returns True if ip = 1:2:3:4::1 and net = 1:2:3:4::/64 :rtype: bool """ netaddr, bits = net.split('/') if is_ipv4_address(ip) and is_ipv4_address(netaddr): ipaddr = struct.unpack(">L", socket.inet_aton(ip))[0] netmask = _get_mask_bits(int(bits)) network = struct.unpack(">L", socket.inet_aton(netaddr))[0] elif is_ipv6_address(ip) and is_ipv6_address(netaddr): ipaddr_msb, ipaddr_lsb = struct.unpack( ">QQ", socket.inet_pton(socket.AF_INET6, ip) ) ipaddr = (ipaddr_msb << 64) ^ ipaddr_lsb netmask = _get_mask_bits(int(bits), 128) network_msb, network_lsb = struct.unpack( ">QQ", socket.inet_pton(socket.AF_INET6, netaddr) ) network = (network_msb << 64) ^ network_lsb else: return False return (ipaddr & netmask) == (network & netmask) def dotted_netmask(mask): """Converts mask from /xx format to xxx.xxx.xxx.xxx Example: if mask is 24 function returns 255.255.255.0 :rtype: str """ bits = 0xffffffff ^ (1 << 32 - mask) - 1 return socket.inet_ntoa(struct.pack('>I', bits)) def is_ipv4_address(string_ip): """ :rtype: bool """ try: socket.inet_pton(socket.AF_INET, string_ip) except socket.error: return False return True def is_ipv6_address(string_ip): """ :rtype: bool """ try: socket.inet_pton(socket.AF_INET6, string_ip) except socket.error: return False return True def compare_ips(a, b): """ Compare 2 IP's, uses socket.inet_pton to normalize IPv6 IPs :rtype: bool """ if a == b: return True try: return socket.inet_pton(socket.AF_INET6, a) == socket.inet_pton( socket.AF_INET6, b ) except OSError: return False def is_valid_cidr(string_network): """ Very simple check of the cidr format in no_proxy variable. :rtype: bool """ if string_network.count('/') == 1: address, mask = string_network.split("/") try: mask = int(mask) except ValueError: return False if is_ipv4_address(address): if mask < 1 or mask > 32: return False elif is_ipv6_address(address): if mask < 1 or mask > 128: return False else: return False else: return False return True @contextlib.contextmanager def set_environ(env_name, value): """Set the environment variable 'env_name' to 'value' Save previous value, yield, and then restore the previous value stored in the environment variable 'env_name'. If 'value' is None, do nothing""" value_changed = value is not None if value_changed: old_value = os.environ.get(env_name) os.environ[env_name] = value try: yield finally: if value_changed: if old_value is None: del os.environ[env_name] else: os.environ[env_name] = old_value def should_bypass_proxies(url, no_proxy): """ Returns whether we should bypass proxies or not. :rtype: bool """ # Prioritize lowercase environment variables over uppercase # to keep a consistent behaviour with other http projects (curl, wget). get_proxy = lambda k: os.environ.get(k) or os.environ.get(k.upper()) # First check whether no_proxy is defined. If it is, check that the URL # we're getting isn't in the no_proxy list. no_proxy_arg = no_proxy if no_proxy is None: no_proxy = get_proxy('no_proxy') parsed = urlparse(url) if parsed.hostname is None: # URLs don't always have hostnames, e.g. file:/// urls. return True if no_proxy: # We need to check whether we match here. We need to see if we match # the end of the hostname, both with and without the port. no_proxy = ( host for host in no_proxy.replace(' ', '').split(',') if host ) if is_ipv4_address(parsed.hostname) or is_ipv6_address(parsed.hostname): for proxy_ip in no_proxy: if is_valid_cidr(proxy_ip): if address_in_network(parsed.hostname, proxy_ip): return True elif compare_ips(parsed.hostname, proxy_ip): # If no_proxy ip was defined in plain IP notation instead of cidr notation & # matches the IP of the index return True else: host_with_port = parsed.hostname if parsed.port: host_with_port += ':{}'.format(parsed.port) for host in no_proxy: if parsed.hostname.endswith(host) or host_with_port.endswith(host): # The URL does match something in no_proxy, so we don't want # to apply the proxies on this URL. return True with set_environ('no_proxy', no_proxy_arg): # parsed.hostname can be `None` in cases such as a file URI. try: bypass = proxy_bypass(parsed.hostname) except (TypeError, socket.gaierror): bypass = False if bypass: return True return False def get_environ_proxies(url, no_proxy=None): """ Return a dict of environment proxies. :rtype: dict """ if should_bypass_proxies(url, no_proxy=no_proxy): return {} else: return getproxies() def select_proxy(url, proxies): """Select a proxy for the url, if applicable. :param url: The url being for the request :param proxies: A dictionary of schemes or schemes and hosts to proxy URLs """ proxies = proxies or {} urlparts = urlparse(url) if urlparts.hostname is None: return proxies.get(urlparts.scheme, proxies.get('all')) proxy_keys = [ urlparts.scheme + '://' + urlparts.hostname, urlparts.scheme, 'all://' + urlparts.hostname, 'all', ] proxy = None for proxy_key in proxy_keys: if proxy_key in proxies: proxy = proxies[proxy_key] break return proxy def default_user_agent(name="python-requests"): """ Return a string representing the default user agent. :rtype: str """ return '%s/%s' % (name, __version__) def default_headers(): """ :rtype: requests.structures.CaseInsensitiveDict """ return CaseInsensitiveDict({ 'User-Agent': default_user_agent(), 'Accept-Encoding': ', '.join(('gzip', 'deflate')), 'Accept': '*/*', 'Connection': 'keep-alive', }) def parse_header_links(value): """Return a list of parsed link headers proxies. i.e. Link: <http:/.../front.jpeg>; rel=front; type="image/jpeg",<http://.../back.jpeg>; rel=back;type="image/jpeg" :rtype: list """ links = [] replace_chars = ' \'"' value = value.strip(replace_chars) if not value: return links for val in re.split(', *<', value): try: url, params = val.split(';', 1) except ValueError: url, params = val, '' link = {'url': url.strip('<> \'"')} for param in params.split(';'): try: key, value = param.split('=') except ValueError: break link[key.strip(replace_chars)] = value.strip(replace_chars) links.append(link) return links # Null bytes; no need to recreate these on each call to guess_json_utf _null = '\x00'.encode('ascii') # encoding to ASCII for Python 3 _null2 = _null * 2 _null3 = _null * 3 def guess_json_utf(data): """ :rtype: str """ # JSON always starts with two ASCII characters, so detection is as # easy as counting the nulls and from their location and count # determine the encoding. Also detect a BOM, if present. sample = data[:4] if sample in (codecs.BOM_UTF32_LE, codecs.BOM_UTF32_BE): return 'utf-32' # BOM included if sample[:3] == codecs.BOM_UTF8: return 'utf-8-sig' # BOM included, MS style (discouraged) if sample[:2] in (codecs.BOM_UTF16_LE, codecs.BOM_UTF16_BE): return 'utf-16' # BOM included nullcount = sample.count(_null) if nullcount == 0: return 'utf-8' if nullcount == 2: if sample[::2] == _null2: # 1st and 3rd are null return 'utf-16-be' if sample[1::2] == _null2: # 2nd and 4th are null return 'utf-16-le' # Did not detect 2 valid UTF-16 ascii-range characters if nullcount == 3: if sample[:3] == _null3: return 'utf-32-be' if sample[1:] == _null3: return 'utf-32-le' # Did not detect a valid UTF-32 ascii-range character return None def prepend_scheme_if_needed(url, new_scheme): """Given a URL that may or may not have a scheme, prepend the given scheme. Does not replace a present scheme with the one provided as an argument. :rtype: str """ scheme, netloc, path, params, query, fragment = urlparse(url, new_scheme) # urlparse is a finicky beast, and sometimes decides that there isn't a # netloc present. Assume that it's being over-cautious, and switch netloc # and path if urlparse decided there was no netloc. if not netloc: netloc, path = path, netloc return urlunparse((scheme, netloc, path, params, query, fragment)) def get_auth_from_url(url): """Given a url with authentication components, extract them into a tuple of username,password. :rtype: (str,str) """ parsed = urlparse(url) try: auth = (unquote(parsed.username), unquote(parsed.password)) except (AttributeError, TypeError): auth = ('', '') return auth # Moved outside of function to avoid recompile every call _CLEAN_HEADER_REGEX_BYTE = re.compile(b'^\\S[^\\r\\n]*$|^$') _CLEAN_HEADER_REGEX_STR = re.compile(r'^\S[^\r\n]*$|^$') def check_header_validity(header): """Verifies that header value is a string which doesn't contain leading whitespace or return characters. This prevents unintended header injection. :param header: tuple, in the format (name, value). """ name, value = header if isinstance(value, bytes): pat = _CLEAN_HEADER_REGEX_BYTE else: pat = _CLEAN_HEADER_REGEX_STR try: if not pat.match(value): raise InvalidHeader("Invalid return character or leading space in header: %s" % name) except TypeError: raise InvalidHeader("Value for header {%s: %s} must be of type str or " "bytes, not %s" % (name, value, type(value))) def urldefragauth(url): """ Given a url remove the fragment and the authentication part. :rtype: str """ scheme, netloc, path, params, query, fragment = urlparse(url) # see func:`prepend_scheme_if_needed` if not netloc: netloc, path = path, netloc netloc = netloc.rsplit('@', 1)[-1] return urlunparse((scheme, netloc, path, params, query, '')) def rewind_body(prepared_request): """Move file pointer back to its recorded starting position so it can be read again on redirect. """ body_seek = getattr(prepared_request.body, 'seek', None) if body_seek is not None and isinstance(prepared_request._body_position, integer_types): try: body_seek(prepared_request._body_position) except (IOError, OSError): raise UnrewindableBodyError("An error occurred when rewinding request " "body for redirect.") else: raise UnrewindableBodyError("Unable to rewind request body for redirect.") lib/python3.9/site-packages/dnf/cli/utils.py 0000644 00000010650 15025166323 0014642 0 ustar 00 # Copyright (C) 2016 Red Hat, Inc. # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation; either version 2 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Library General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program; if not, write to the Free Software # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. """Various utility functions, and a utility class.""" from __future__ import absolute_import from __future__ import unicode_literals from dnf.cli.format import format_number from dnf.i18n import _ import dnf.util import logging import os import time _USER_HZ = os.sysconf(os.sysconf_names['SC_CLK_TCK']) logger = logging.getLogger('dnf') def jiffies_to_seconds(jiffies): """Convert a number of jiffies to seconds. How many jiffies are in a second is system-dependent, e.g. 100 jiffies = 1 second is common. :param jiffies: a number of jiffies :return: the equivalent number of seconds """ return int(jiffies) / _USER_HZ def seconds_to_ui_time(seconds): """Return a human-readable string representation of the length of a time interval given in seconds. :param seconds: the length of the time interval in seconds :return: a human-readable string representation of the length of the time interval """ if seconds >= 60 * 60 * 24: return "%d day(s) %d:%02d:%02d" % (seconds // (60 * 60 * 24), (seconds // (60 * 60)) % 24, (seconds // 60) % 60, seconds % 60) if seconds >= 60 * 60: return "%d:%02d:%02d" % (seconds // (60 * 60), (seconds // 60) % 60, (seconds % 60)) return "%02d:%02d" % ((seconds // 60), seconds % 60) def get_process_info(pid): """Return info dict about a process.""" pid = int(pid) # Maybe true if /proc isn't mounted, or not Linux ... or something. if (not os.path.exists("/proc/%d/status" % pid) or not os.path.exists("/proc/stat") or not os.path.exists("/proc/%d/stat" % pid)): return ps = {} with open("/proc/%d/status" % pid) as status_file: for line in status_file: if line[-1] != '\n': continue data = line[:-1].split(':\t', 1) if len(data) < 2: continue data[1] = dnf.util.rtrim(data[1], ' kB') ps[data[0].strip().lower()] = data[1].strip() if 'vmrss' not in ps: return if 'vmsize' not in ps: return boot_time = None with open("/proc/stat") as stat_file: for line in stat_file: if line.startswith("btime "): boot_time = int(line[len("btime "):-1]) break if boot_time is None: return with open('/proc/%d/stat' % pid) as stat_file: ps_stat = stat_file.read().split() ps['start_time'] = boot_time + jiffies_to_seconds(ps_stat[21]) ps['state'] = {'R' : _('Running'), 'S' : _('Sleeping'), 'D' : _('Uninterruptible'), 'Z' : _('Zombie'), 'T' : _('Traced/Stopped') }.get(ps_stat[2], _('Unknown')) return ps def show_lock_owner(pid): """Output information about process holding a lock.""" ps = get_process_info(pid) if not ps: msg = _('Unable to find information about the locking process (PID %d)') logger.critical(msg, pid) return msg = _(' The application with PID %d is: %s') % (pid, ps['name']) logger.critical("%s", msg) logger.critical(_(" Memory : %5s RSS (%5sB VSZ)"), format_number(int(ps['vmrss']) * 1024), format_number(int(ps['vmsize']) * 1024)) ago = seconds_to_ui_time(int(time.time()) - ps['start_time']) logger.critical(_(' Started: %s - %s ago'), dnf.util.normalize_time(ps['start_time']), ago) logger.critical(_(' State : %s'), ps['state']) return lib64/python3.9/email/utils.py 0000644 00000044131 15025175045 0012047 0 ustar 00 # Copyright (C) 2001-2010 Python Software Foundation # Author: Barry Warsaw # Contact: email-sig@python.org """Miscellaneous utilities.""" __all__ = [ 'collapse_rfc2231_value', 'decode_params', 'decode_rfc2231', 'encode_rfc2231', 'formataddr', 'formatdate', 'format_datetime', 'getaddresses', 'make_msgid', 'mktime_tz', 'parseaddr', 'parsedate', 'parsedate_tz', 'parsedate_to_datetime', 'unquote', ] import os import re import time import random import socket import datetime import urllib.parse from email._parseaddr import quote from email._parseaddr import AddressList as _AddressList from email._parseaddr import mktime_tz from email._parseaddr import parsedate, parsedate_tz, _parsedate_tz # Intrapackage imports from email.charset import Charset COMMASPACE = ', ' EMPTYSTRING = '' UEMPTYSTRING = '' CRLF = '\r\n' TICK = "'" specialsre = re.compile(r'[][\\()<>@,:;".]') escapesre = re.compile(r'[\\"]') _EMAIL_CONFIG_FILE = "/etc/python/email.cfg" _cached_strict_addr_parsing = None def _use_strict_email_parsing(): """"Cache implementation for _cached_strict_addr_parsing""" global _cached_strict_addr_parsing if _cached_strict_addr_parsing is None: _cached_strict_addr_parsing = _use_strict_email_parsing_impl() return _cached_strict_addr_parsing def _use_strict_email_parsing_impl(): """Returns True if strict email parsing is not disabled by config file or env variable. """ disabled = bool(os.environ.get("PYTHON_EMAIL_DISABLE_STRICT_ADDR_PARSING")) if disabled: return False try: file = open(_EMAIL_CONFIG_FILE) except FileNotFoundError: pass else: with file: import configparser config = configparser.ConfigParser( interpolation=None, comment_prefixes=('#', ), ) config.read_file(file) disabled = config.getboolean('email_addr_parsing', "PYTHON_EMAIL_DISABLE_STRICT_ADDR_PARSING", fallback=None) if disabled: return False return True def _has_surrogates(s): """Return True if s contains surrogate-escaped binary data.""" # This check is based on the fact that unless there are surrogates, utf8 # (Python's default encoding) can encode any string. This is the fastest # way to check for surrogates, see issue 11454 for timings. try: s.encode() return False except UnicodeEncodeError: return True # How to deal with a string containing bytes before handing it to the # application through the 'normal' interface. def _sanitize(string): # Turn any escaped bytes into unicode 'unknown' char. If the escaped # bytes happen to be utf-8 they will instead get decoded, even if they # were invalid in the charset the source was supposed to be in. This # seems like it is not a bad thing; a defect was still registered. original_bytes = string.encode('utf-8', 'surrogateescape') return original_bytes.decode('utf-8', 'replace') # Helpers def formataddr(pair, charset='utf-8'): """The inverse of parseaddr(), this takes a 2-tuple of the form (realname, email_address) and returns the string value suitable for an RFC 2822 From, To or Cc header. If the first element of pair is false, then the second element is returned unmodified. The optional charset is the character set that is used to encode realname in case realname is not ASCII safe. Can be an instance of str or a Charset-like object which has a header_encode method. Default is 'utf-8'. """ name, address = pair # The address MUST (per RFC) be ascii, so raise a UnicodeError if it isn't. address.encode('ascii') if name: try: name.encode('ascii') except UnicodeEncodeError: if isinstance(charset, str): charset = Charset(charset) encoded_name = charset.header_encode(name) return "%s <%s>" % (encoded_name, address) else: quotes = '' if specialsre.search(name): quotes = '"' name = escapesre.sub(r'\\\g<0>', name) return '%s%s%s <%s>' % (quotes, name, quotes, address) return address def _iter_escaped_chars(addr): pos = 0 escape = False for pos, ch in enumerate(addr): if escape: yield (pos, '\\' + ch) escape = False elif ch == '\\': escape = True else: yield (pos, ch) if escape: yield (pos, '\\') def _strip_quoted_realnames(addr): """Strip real names between quotes.""" if '"' not in addr: # Fast path return addr start = 0 open_pos = None result = [] for pos, ch in _iter_escaped_chars(addr): if ch == '"': if open_pos is None: open_pos = pos else: if start != open_pos: result.append(addr[start:open_pos]) start = pos + 1 open_pos = None if start < len(addr): result.append(addr[start:]) return ''.join(result) supports_strict_parsing = True def getaddresses(fieldvalues, *, strict=None): """Return a list of (REALNAME, EMAIL) or ('','') for each fieldvalue. When parsing fails for a fieldvalue, a 2-tuple of ('', '') is returned in its place. If strict is true, use a strict parser which rejects malformed inputs. """ # If default is used, it's True unless disabled # by env variable or config file. if strict == None: strict = _use_strict_email_parsing() # If strict is true, if the resulting list of parsed addresses is greater # than the number of fieldvalues in the input list, a parsing error has # occurred and consequently a list containing a single empty 2-tuple [('', # '')] is returned in its place. This is done to avoid invalid output. # # Malformed input: getaddresses(['alice@example.com <bob@example.com>']) # Invalid output: [('', 'alice@example.com'), ('', 'bob@example.com')] # Safe output: [('', '')] if not strict: all = COMMASPACE.join(str(v) for v in fieldvalues) a = _AddressList(all) return a.addresslist fieldvalues = [str(v) for v in fieldvalues] fieldvalues = _pre_parse_validation(fieldvalues) addr = COMMASPACE.join(fieldvalues) a = _AddressList(addr) result = _post_parse_validation(a.addresslist) # Treat output as invalid if the number of addresses is not equal to the # expected number of addresses. n = 0 for v in fieldvalues: # When a comma is used in the Real Name part it is not a deliminator. # So strip those out before counting the commas. v = _strip_quoted_realnames(v) # Expected number of addresses: 1 + number of commas n += 1 + v.count(',') if len(result) != n: return [('', '')] return result def _check_parenthesis(addr): # Ignore parenthesis in quoted real names. addr = _strip_quoted_realnames(addr) opens = 0 for pos, ch in _iter_escaped_chars(addr): if ch == '(': opens += 1 elif ch == ')': opens -= 1 if opens < 0: return False return (opens == 0) def _pre_parse_validation(email_header_fields): accepted_values = [] for v in email_header_fields: if not _check_parenthesis(v): v = "('', '')" accepted_values.append(v) return accepted_values def _post_parse_validation(parsed_email_header_tuples): accepted_values = [] # The parser would have parsed a correctly formatted domain-literal # The existence of an [ after parsing indicates a parsing failure for v in parsed_email_header_tuples: if '[' in v[1]: v = ('', '') accepted_values.append(v) return accepted_values def _format_timetuple_and_zone(timetuple, zone): return '%s, %02d %s %04d %02d:%02d:%02d %s' % ( ['Mon', 'Tue', 'Wed', 'Thu', 'Fri', 'Sat', 'Sun'][timetuple[6]], timetuple[2], ['Jan', 'Feb', 'Mar', 'Apr', 'May', 'Jun', 'Jul', 'Aug', 'Sep', 'Oct', 'Nov', 'Dec'][timetuple[1] - 1], timetuple[0], timetuple[3], timetuple[4], timetuple[5], zone) def formatdate(timeval=None, localtime=False, usegmt=False): """Returns a date string as specified by RFC 2822, e.g.: Fri, 09 Nov 2001 01:08:47 -0000 Optional timeval if given is a floating point time value as accepted by gmtime() and localtime(), otherwise the current time is used. Optional localtime is a flag that when True, interprets timeval, and returns a date relative to the local timezone instead of UTC, properly taking daylight savings time into account. Optional argument usegmt means that the timezone is written out as an ascii string, not numeric one (so "GMT" instead of "+0000"). This is needed for HTTP, and is only used when localtime==False. """ # Note: we cannot use strftime() because that honors the locale and RFC # 2822 requires that day and month names be the English abbreviations. if timeval is None: timeval = time.time() if localtime or usegmt: dt = datetime.datetime.fromtimestamp(timeval, datetime.timezone.utc) else: dt = datetime.datetime.utcfromtimestamp(timeval) if localtime: dt = dt.astimezone() usegmt = False return format_datetime(dt, usegmt) def format_datetime(dt, usegmt=False): """Turn a datetime into a date string as specified in RFC 2822. If usegmt is True, dt must be an aware datetime with an offset of zero. In this case 'GMT' will be rendered instead of the normal +0000 required by RFC2822. This is to support HTTP headers involving date stamps. """ now = dt.timetuple() if usegmt: if dt.tzinfo is None or dt.tzinfo != datetime.timezone.utc: raise ValueError("usegmt option requires a UTC datetime") zone = 'GMT' elif dt.tzinfo is None: zone = '-0000' else: zone = dt.strftime("%z") return _format_timetuple_and_zone(now, zone) def make_msgid(idstring=None, domain=None): """Returns a string suitable for RFC 2822 compliant Message-ID, e.g: <142480216486.20800.16526388040877946887@nightshade.la.mastaler.com> Optional idstring if given is a string used to strengthen the uniqueness of the message id. Optional domain if given provides the portion of the message id after the '@'. It defaults to the locally defined hostname. """ timeval = int(time.time()*100) pid = os.getpid() randint = random.getrandbits(64) if idstring is None: idstring = '' else: idstring = '.' + idstring if domain is None: domain = socket.getfqdn() msgid = '<%d.%d.%d%s@%s>' % (timeval, pid, randint, idstring, domain) return msgid def parsedate_to_datetime(data): *dtuple, tz = _parsedate_tz(data) if tz is None: return datetime.datetime(*dtuple[:6]) return datetime.datetime(*dtuple[:6], tzinfo=datetime.timezone(datetime.timedelta(seconds=tz))) def parseaddr(addr, *, strict=None): """ Parse addr into its constituent realname and email address parts. Return a tuple of realname and email address, unless the parse fails, in which case return a 2-tuple of ('', ''). If strict is True, use a strict parser which rejects malformed inputs. """ # If default is used, it's True unless disabled # by env variable or config file. if strict == None: strict = _use_strict_email_parsing() if not strict: addrs = _AddressList(addr).addresslist if not addrs: return ('', '') return addrs[0] if isinstance(addr, list): addr = addr[0] if not isinstance(addr, str): return ('', '') addr = _pre_parse_validation([addr])[0] addrs = _post_parse_validation(_AddressList(addr).addresslist) if not addrs or len(addrs) > 1: return ('', '') return addrs[0] # rfc822.unquote() doesn't properly de-backslash-ify in Python pre-2.3. def unquote(str): """Remove quotes from a string.""" if len(str) > 1: if str.startswith('"') and str.endswith('"'): return str[1:-1].replace('\\\\', '\\').replace('\\"', '"') if str.startswith('<') and str.endswith('>'): return str[1:-1] return str # RFC2231-related functions - parameter encoding and decoding def decode_rfc2231(s): """Decode string according to RFC 2231""" parts = s.split(TICK, 2) if len(parts) <= 2: return None, None, s return parts def encode_rfc2231(s, charset=None, language=None): """Encode string according to RFC 2231. If neither charset nor language is given, then s is returned as-is. If charset is given but not language, the string is encoded using the empty string for language. """ s = urllib.parse.quote(s, safe='', encoding=charset or 'ascii') if charset is None and language is None: return s if language is None: language = '' return "%s'%s'%s" % (charset, language, s) rfc2231_continuation = re.compile(r'^(?P<name>\w+)\*((?P<num>[0-9]+)\*?)?$', re.ASCII) def decode_params(params): """Decode parameters list according to RFC 2231. params is a sequence of 2-tuples containing (param name, string value). """ new_params = [params[0]] # Map parameter's name to a list of continuations. The values are a # 3-tuple of the continuation number, the string value, and a flag # specifying whether a particular segment is %-encoded. rfc2231_params = {} for name, value in params[1:]: encoded = name.endswith('*') value = unquote(value) mo = rfc2231_continuation.match(name) if mo: name, num = mo.group('name', 'num') if num is not None: num = int(num) rfc2231_params.setdefault(name, []).append((num, value, encoded)) else: new_params.append((name, '"%s"' % quote(value))) if rfc2231_params: for name, continuations in rfc2231_params.items(): value = [] extended = False # Sort by number continuations.sort() # And now append all values in numerical order, converting # %-encodings for the encoded segments. If any of the # continuation names ends in a *, then the entire string, after # decoding segments and concatenating, must have the charset and # language specifiers at the beginning of the string. for num, s, encoded in continuations: if encoded: # Decode as "latin-1", so the characters in s directly # represent the percent-encoded octet values. # collapse_rfc2231_value treats this as an octet sequence. s = urllib.parse.unquote(s, encoding="latin-1") extended = True value.append(s) value = quote(EMPTYSTRING.join(value)) if extended: charset, language, value = decode_rfc2231(value) new_params.append((name, (charset, language, '"%s"' % value))) else: new_params.append((name, '"%s"' % value)) return new_params def collapse_rfc2231_value(value, errors='replace', fallback_charset='us-ascii'): if not isinstance(value, tuple) or len(value) != 3: return unquote(value) # While value comes to us as a unicode string, we need it to be a bytes # object. We do not want bytes() normal utf-8 decoder, we want a straight # interpretation of the string as character bytes. charset, language, text = value if charset is None: # Issue 17369: if charset/lang is None, decode_rfc2231 couldn't parse # the value, so use the fallback_charset. charset = fallback_charset rawbytes = bytes(text, 'raw-unicode-escape') try: return str(rawbytes, charset, errors) except LookupError: # charset is not a known codec. return unquote(text) # # datetime doesn't provide a localtime function yet, so provide one. Code # adapted from the patch in issue 9527. This may not be perfect, but it is # better than not having it. # def localtime(dt=None, isdst=-1): """Return local time as an aware datetime object. If called without arguments, return current time. Otherwise *dt* argument should be a datetime instance, and it is converted to the local time zone according to the system time zone database. If *dt* is naive (that is, dt.tzinfo is None), it is assumed to be in local time. In this case, a positive or zero value for *isdst* causes localtime to presume initially that summer time (for example, Daylight Saving Time) is or is not (respectively) in effect for the specified time. A negative value for *isdst* causes the localtime() function to attempt to divine whether summer time is in effect for the specified time. """ if dt is None: return datetime.datetime.now(datetime.timezone.utc).astimezone() if dt.tzinfo is not None: return dt.astimezone() # We have a naive datetime. Convert to a (localtime) timetuple and pass to # system mktime together with the isdst hint. System mktime will return # seconds since epoch. tm = dt.timetuple()[:-1] + (isdst,) seconds = time.mktime(tm) localtm = time.localtime(seconds) try: delta = datetime.timedelta(seconds=localtm.tm_gmtoff) tz = datetime.timezone(delta, localtm.tm_zone) except AttributeError: # Compute UTC offset and compare with the value implied by tm_isdst. # If the values match, use the zone name implied by tm_isdst. delta = dt - datetime.datetime(*time.gmtime(seconds)[:6]) dst = time.daylight and localtm.tm_isdst > 0 gmtoff = -(time.altzone if dst else time.timezone) if delta == datetime.timedelta(seconds=gmtoff): tz = datetime.timezone(delta, time.tzname[dst]) else: tz = datetime.timezone(delta) return dt.replace(tzinfo=tz) opt/gsutil/third_party/requests/tests/utils.py 0000644 00000000556 15025261016 0015741 0 ustar 00 import contextlib import os @contextlib.contextmanager def override_environ(**kwargs): save_env = dict(os.environ) for key, value in kwargs.items(): if value is None: del os.environ[key] else: os.environ[key] = value try: yield finally: os.environ.clear() os.environ.update(save_env) opt/gsutil/third_party/requests/src/requests/utils.py 0000644 00000101523 15025314105 0017233 0 ustar 00 """ requests.utils ~~~~~~~~~~~~~~ This module provides utility functions that are used within Requests that are also useful for external consumption. """ import codecs import contextlib import io import os import re import socket import struct import sys import tempfile import warnings import zipfile from collections import OrderedDict from urllib3.util import make_headers, parse_url from . import certs from .__version__ import __version__ # to_native_string is unused here, but imported here for backwards compatibility from ._internal_utils import ( # noqa: F401 _HEADER_VALIDATORS_BYTE, _HEADER_VALIDATORS_STR, HEADER_VALIDATORS, to_native_string, ) from .compat import ( Mapping, basestring, bytes, getproxies, getproxies_environment, integer_types, ) from .compat import parse_http_list as _parse_list_header from .compat import ( proxy_bypass, proxy_bypass_environment, quote, str, unquote, urlparse, urlunparse, ) from .cookies import cookiejar_from_dict from .exceptions import ( FileModeWarning, InvalidHeader, InvalidURL, UnrewindableBodyError, ) from .structures import CaseInsensitiveDict NETRC_FILES = (".netrc", "_netrc") DEFAULT_CA_BUNDLE_PATH = certs.where() DEFAULT_PORTS = {"http": 80, "https": 443} # Ensure that ', ' is used to preserve previous delimiter behavior. DEFAULT_ACCEPT_ENCODING = ", ".join( re.split(r",\s*", make_headers(accept_encoding=True)["accept-encoding"]) ) if sys.platform == "win32": # provide a proxy_bypass version on Windows without DNS lookups def proxy_bypass_registry(host): try: import winreg except ImportError: return False try: internetSettings = winreg.OpenKey( winreg.HKEY_CURRENT_USER, r"Software\Microsoft\Windows\CurrentVersion\Internet Settings", ) # ProxyEnable could be REG_SZ or REG_DWORD, normalizing it proxyEnable = int(winreg.QueryValueEx(internetSettings, "ProxyEnable")[0]) # ProxyOverride is almost always a string proxyOverride = winreg.QueryValueEx(internetSettings, "ProxyOverride")[0] except (OSError, ValueError): return False if not proxyEnable or not proxyOverride: return False # make a check value list from the registry entry: replace the # '<local>' string by the localhost entry and the corresponding # canonical entry. proxyOverride = proxyOverride.split(";") # filter out empty strings to avoid re.match return true in the following code. proxyOverride = filter(None, proxyOverride) # now check if we match one of the registry values. for test in proxyOverride: if test == "<local>": if "." not in host: return True test = test.replace(".", r"\.") # mask dots test = test.replace("*", r".*") # change glob sequence test = test.replace("?", r".") # change glob char if re.match(test, host, re.I): return True return False def proxy_bypass(host): # noqa """Return True, if the host should be bypassed. Checks proxy settings gathered from the environment, if specified, or the registry. """ if getproxies_environment(): return proxy_bypass_environment(host) else: return proxy_bypass_registry(host) def dict_to_sequence(d): """Returns an internal sequence dictionary update.""" if hasattr(d, "items"): d = d.items() return d def super_len(o): total_length = None current_position = 0 if isinstance(o, str): o = o.encode("utf-8") if hasattr(o, "__len__"): total_length = len(o) elif hasattr(o, "len"): total_length = o.len elif hasattr(o, "fileno"): try: fileno = o.fileno() except (io.UnsupportedOperation, AttributeError): # AttributeError is a surprising exception, seeing as how we've just checked # that `hasattr(o, 'fileno')`. It happens for objects obtained via # `Tarfile.extractfile()`, per issue 5229. pass else: total_length = os.fstat(fileno).st_size # Having used fstat to determine the file length, we need to # confirm that this file was opened up in binary mode. if "b" not in o.mode: warnings.warn( ( "Requests has determined the content-length for this " "request using the binary size of the file: however, the " "file has been opened in text mode (i.e. without the 'b' " "flag in the mode). This may lead to an incorrect " "content-length. In Requests 3.0, support will be removed " "for files in text mode." ), FileModeWarning, ) if hasattr(o, "tell"): try: current_position = o.tell() except OSError: # This can happen in some weird situations, such as when the file # is actually a special file descriptor like stdin. In this # instance, we don't know what the length is, so set it to zero and # let requests chunk it instead. if total_length is not None: current_position = total_length else: if hasattr(o, "seek") and total_length is None: # StringIO and BytesIO have seek but no usable fileno try: # seek to end of file o.seek(0, 2) total_length = o.tell() # seek back to current position to support # partially read file-like objects o.seek(current_position or 0) except OSError: total_length = 0 if total_length is None: total_length = 0 return max(0, total_length - current_position) def get_netrc_auth(url, raise_errors=False): """Returns the Requests tuple auth for a given url from netrc.""" netrc_file = os.environ.get("NETRC") if netrc_file is not None: netrc_locations = (netrc_file,) else: netrc_locations = (f"~/{f}" for f in NETRC_FILES) try: from netrc import NetrcParseError, netrc netrc_path = None for f in netrc_locations: try: loc = os.path.expanduser(f) except KeyError: # os.path.expanduser can fail when $HOME is undefined and # getpwuid fails. See https://bugs.python.org/issue20164 & # https://github.com/psf/requests/issues/1846 return if os.path.exists(loc): netrc_path = loc break # Abort early if there isn't one. if netrc_path is None: return ri = urlparse(url) # Strip port numbers from netloc. This weird `if...encode`` dance is # used for Python 3.2, which doesn't support unicode literals. splitstr = b":" if isinstance(url, str): splitstr = splitstr.decode("ascii") host = ri.netloc.split(splitstr)[0] try: _netrc = netrc(netrc_path).authenticators(host) if _netrc: # Return with login / password login_i = 0 if _netrc[0] else 1 return (_netrc[login_i], _netrc[2]) except (NetrcParseError, OSError): # If there was a parsing error or a permissions issue reading the file, # we'll just skip netrc auth unless explicitly asked to raise errors. if raise_errors: raise # App Engine hackiness. except (ImportError, AttributeError): pass def guess_filename(obj): """Tries to guess the filename of the given object.""" name = getattr(obj, "name", None) if name and isinstance(name, basestring) and name[0] != "<" and name[-1] != ">": return os.path.basename(name) def extract_zipped_paths(path): """Replace nonexistent paths that look like they refer to a member of a zip archive with the location of an extracted copy of the target, or else just return the provided path unchanged. """ if os.path.exists(path): # this is already a valid path, no need to do anything further return path # find the first valid part of the provided path and treat that as a zip archive # assume the rest of the path is the name of a member in the archive archive, member = os.path.split(path) while archive and not os.path.exists(archive): archive, prefix = os.path.split(archive) if not prefix: # If we don't check for an empty prefix after the split (in other words, archive remains unchanged after the split), # we _can_ end up in an infinite loop on a rare corner case affecting a small number of users break member = "/".join([prefix, member]) if not zipfile.is_zipfile(archive): return path zip_file = zipfile.ZipFile(archive) if member not in zip_file.namelist(): return path # we have a valid zip archive and a valid member of that archive tmp = tempfile.gettempdir() extracted_path = os.path.join(tmp, member.split("/")[-1]) if not os.path.exists(extracted_path): # use read + write to avoid the creating nested folders, we only want the file, avoids mkdir racing condition with atomic_open(extracted_path) as file_handler: file_handler.write(zip_file.read(member)) return extracted_path @contextlib.contextmanager def atomic_open(filename): """Write a file to the disk in an atomic fashion""" tmp_descriptor, tmp_name = tempfile.mkstemp(dir=os.path.dirname(filename)) try: with os.fdopen(tmp_descriptor, "wb") as tmp_handler: yield tmp_handler os.replace(tmp_name, filename) except BaseException: os.remove(tmp_name) raise def from_key_val_list(value): """Take an object and test to see if it can be represented as a dictionary. Unless it can not be represented as such, return an OrderedDict, e.g., :: >>> from_key_val_list([('key', 'val')]) OrderedDict([('key', 'val')]) >>> from_key_val_list('string') Traceback (most recent call last): ... ValueError: cannot encode objects that are not 2-tuples >>> from_key_val_list({'key': 'val'}) OrderedDict([('key', 'val')]) :rtype: OrderedDict """ if value is None: return None if isinstance(value, (str, bytes, bool, int)): raise ValueError("cannot encode objects that are not 2-tuples") return OrderedDict(value) def to_key_val_list(value): """Take an object and test to see if it can be represented as a dictionary. If it can be, return a list of tuples, e.g., :: >>> to_key_val_list([('key', 'val')]) [('key', 'val')] >>> to_key_val_list({'key': 'val'}) [('key', 'val')] >>> to_key_val_list('string') Traceback (most recent call last): ... ValueError: cannot encode objects that are not 2-tuples :rtype: list """ if value is None: return None if isinstance(value, (str, bytes, bool, int)): raise ValueError("cannot encode objects that are not 2-tuples") if isinstance(value, Mapping): value = value.items() return list(value) # From mitsuhiko/werkzeug (used with permission). def parse_list_header(value): """Parse lists as described by RFC 2068 Section 2. In particular, parse comma-separated lists where the elements of the list may include quoted-strings. A quoted-string could contain a comma. A non-quoted string could have quotes in the middle. Quotes are removed automatically after parsing. It basically works like :func:`parse_set_header` just that items may appear multiple times and case sensitivity is preserved. The return value is a standard :class:`list`: >>> parse_list_header('token, "quoted value"') ['token', 'quoted value'] To create a header from the :class:`list` again, use the :func:`dump_header` function. :param value: a string with a list header. :return: :class:`list` :rtype: list """ result = [] for item in _parse_list_header(value): if item[:1] == item[-1:] == '"': item = unquote_header_value(item[1:-1]) result.append(item) return result # From mitsuhiko/werkzeug (used with permission). def parse_dict_header(value): """Parse lists of key, value pairs as described by RFC 2068 Section 2 and convert them into a python dict: >>> d = parse_dict_header('foo="is a fish", bar="as well"') >>> type(d) is dict True >>> sorted(d.items()) [('bar', 'as well'), ('foo', 'is a fish')] If there is no value for a key it will be `None`: >>> parse_dict_header('key_without_value') {'key_without_value': None} To create a header from the :class:`dict` again, use the :func:`dump_header` function. :param value: a string with a dict header. :return: :class:`dict` :rtype: dict """ result = {} for item in _parse_list_header(value): if "=" not in item: result[item] = None continue name, value = item.split("=", 1) if value[:1] == value[-1:] == '"': value = unquote_header_value(value[1:-1]) result[name] = value return result # From mitsuhiko/werkzeug (used with permission). def unquote_header_value(value, is_filename=False): r"""Unquotes a header value. (Reversal of :func:`quote_header_value`). This does not use the real unquoting but what browsers are actually using for quoting. :param value: the header value to unquote. :rtype: str """ if value and value[0] == value[-1] == '"': # this is not the real unquoting, but fixing this so that the # RFC is met will result in bugs with internet explorer and # probably some other browsers as well. IE for example is # uploading files with "C:\foo\bar.txt" as filename value = value[1:-1] # if this is a filename and the starting characters look like # a UNC path, then just return the value without quotes. Using the # replace sequence below on a UNC path has the effect of turning # the leading double slash into a single slash and then # _fix_ie_filename() doesn't work correctly. See #458. if not is_filename or value[:2] != "\\\\": return value.replace("\\\\", "\\").replace('\\"', '"') return value def dict_from_cookiejar(cj): """Returns a key/value dictionary from a CookieJar. :param cj: CookieJar object to extract cookies from. :rtype: dict """ cookie_dict = {cookie.name: cookie.value for cookie in cj} return cookie_dict def add_dict_to_cookiejar(cj, cookie_dict): """Returns a CookieJar from a key/value dictionary. :param cj: CookieJar to insert cookies into. :param cookie_dict: Dict of key/values to insert into CookieJar. :rtype: CookieJar """ return cookiejar_from_dict(cookie_dict, cj) def get_encodings_from_content(content): """Returns encodings from given content string. :param content: bytestring to extract encodings from. """ warnings.warn( ( "In requests 3.0, get_encodings_from_content will be removed. For " "more information, please see the discussion on issue #2266. (This" " warning should only appear once.)" ), DeprecationWarning, ) charset_re = re.compile(r'<meta.*?charset=["\']*(.+?)["\'>]', flags=re.I) pragma_re = re.compile(r'<meta.*?content=["\']*;?charset=(.+?)["\'>]', flags=re.I) xml_re = re.compile(r'^<\?xml.*?encoding=["\']*(.+?)["\'>]') return ( charset_re.findall(content) + pragma_re.findall(content) + xml_re.findall(content) ) def _parse_content_type_header(header): """Returns content type and parameters from given header :param header: string :return: tuple containing content type and dictionary of parameters """ tokens = header.split(";") content_type, params = tokens[0].strip(), tokens[1:] params_dict = {} items_to_strip = "\"' " for param in params: param = param.strip() if param: key, value = param, True index_of_equals = param.find("=") if index_of_equals != -1: key = param[:index_of_equals].strip(items_to_strip) value = param[index_of_equals + 1 :].strip(items_to_strip) params_dict[key.lower()] = value return content_type, params_dict def get_encoding_from_headers(headers): """Returns encodings from given HTTP Header Dict. :param headers: dictionary to extract encoding from. :rtype: str """ content_type = headers.get("content-type") if not content_type: return None content_type, params = _parse_content_type_header(content_type) if "charset" in params: return params["charset"].strip("'\"") if "text" in content_type: return "ISO-8859-1" if "application/json" in content_type: # Assume UTF-8 based on RFC 4627: https://www.ietf.org/rfc/rfc4627.txt since the charset was unset return "utf-8" def stream_decode_response_unicode(iterator, r): """Stream decodes an iterator.""" if r.encoding is None: yield from iterator return decoder = codecs.getincrementaldecoder(r.encoding)(errors="replace") for chunk in iterator: rv = decoder.decode(chunk) if rv: yield rv rv = decoder.decode(b"", final=True) if rv: yield rv def iter_slices(string, slice_length): """Iterate over slices of a string.""" pos = 0 if slice_length is None or slice_length <= 0: slice_length = len(string) while pos < len(string): yield string[pos : pos + slice_length] pos += slice_length def get_unicode_from_response(r): """Returns the requested content back in unicode. :param r: Response object to get unicode content from. Tried: 1. charset from content-type 2. fall back and replace all unicode characters :rtype: str """ warnings.warn( ( "In requests 3.0, get_unicode_from_response will be removed. For " "more information, please see the discussion on issue #2266. (This" " warning should only appear once.)" ), DeprecationWarning, ) tried_encodings = [] # Try charset from content-type encoding = get_encoding_from_headers(r.headers) if encoding: try: return str(r.content, encoding) except UnicodeError: tried_encodings.append(encoding) # Fall back: try: return str(r.content, encoding, errors="replace") except TypeError: return r.content # The unreserved URI characters (RFC 3986) UNRESERVED_SET = frozenset( "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz" + "0123456789-._~" ) def unquote_unreserved(uri): """Un-escape any percent-escape sequences in a URI that are unreserved characters. This leaves all reserved, illegal and non-ASCII bytes encoded. :rtype: str """ parts = uri.split("%") for i in range(1, len(parts)): h = parts[i][0:2] if len(h) == 2 and h.isalnum(): try: c = chr(int(h, 16)) except ValueError: raise InvalidURL(f"Invalid percent-escape sequence: '{h}'") if c in UNRESERVED_SET: parts[i] = c + parts[i][2:] else: parts[i] = f"%{parts[i]}" else: parts[i] = f"%{parts[i]}" return "".join(parts) def requote_uri(uri): """Re-quote the given URI. This function passes the given URI through an unquote/quote cycle to ensure that it is fully and consistently quoted. :rtype: str """ safe_with_percent = "!#$%&'()*+,/:;=?@[]~" safe_without_percent = "!#$&'()*+,/:;=?@[]~" try: # Unquote only the unreserved characters # Then quote only illegal characters (do not quote reserved, # unreserved, or '%') return quote(unquote_unreserved(uri), safe=safe_with_percent) except InvalidURL: # We couldn't unquote the given URI, so let's try quoting it, but # there may be unquoted '%'s in the URI. We need to make sure they're # properly quoted so they do not cause issues elsewhere. return quote(uri, safe=safe_without_percent) def address_in_network(ip, net): """This function allows you to check if an IP belongs to a network subnet Example: returns True if ip = 192.168.1.1 and net = 192.168.1.0/24 returns False if ip = 192.168.1.1 and net = 192.168.100.0/24 :rtype: bool """ ipaddr = struct.unpack("=L", socket.inet_aton(ip))[0] netaddr, bits = net.split("/") netmask = struct.unpack("=L", socket.inet_aton(dotted_netmask(int(bits))))[0] network = struct.unpack("=L", socket.inet_aton(netaddr))[0] & netmask return (ipaddr & netmask) == (network & netmask) def dotted_netmask(mask): """Converts mask from /xx format to xxx.xxx.xxx.xxx Example: if mask is 24 function returns 255.255.255.0 :rtype: str """ bits = 0xFFFFFFFF ^ (1 << 32 - mask) - 1 return socket.inet_ntoa(struct.pack(">I", bits)) def is_ipv4_address(string_ip): """ :rtype: bool """ try: socket.inet_aton(string_ip) except OSError: return False return True def is_valid_cidr(string_network): """ Very simple check of the cidr format in no_proxy variable. :rtype: bool """ if string_network.count("/") == 1: try: mask = int(string_network.split("/")[1]) except ValueError: return False if mask < 1 or mask > 32: return False try: socket.inet_aton(string_network.split("/")[0]) except OSError: return False else: return False return True @contextlib.contextmanager def set_environ(env_name, value): """Set the environment variable 'env_name' to 'value' Save previous value, yield, and then restore the previous value stored in the environment variable 'env_name'. If 'value' is None, do nothing""" value_changed = value is not None if value_changed: old_value = os.environ.get(env_name) os.environ[env_name] = value try: yield finally: if value_changed: if old_value is None: del os.environ[env_name] else: os.environ[env_name] = old_value def should_bypass_proxies(url, no_proxy): """ Returns whether we should bypass proxies or not. :rtype: bool """ # Prioritize lowercase environment variables over uppercase # to keep a consistent behaviour with other http projects (curl, wget). def get_proxy(key): return os.environ.get(key) or os.environ.get(key.upper()) # First check whether no_proxy is defined. If it is, check that the URL # we're getting isn't in the no_proxy list. no_proxy_arg = no_proxy if no_proxy is None: no_proxy = get_proxy("no_proxy") parsed = urlparse(url) if parsed.hostname is None: # URLs don't always have hostnames, e.g. file:/// urls. return True if no_proxy: # We need to check whether we match here. We need to see if we match # the end of the hostname, both with and without the port. no_proxy = (host for host in no_proxy.replace(" ", "").split(",") if host) if is_ipv4_address(parsed.hostname): for proxy_ip in no_proxy: if is_valid_cidr(proxy_ip): if address_in_network(parsed.hostname, proxy_ip): return True elif parsed.hostname == proxy_ip: # If no_proxy ip was defined in plain IP notation instead of cidr notation & # matches the IP of the index return True else: host_with_port = parsed.hostname if parsed.port: host_with_port += f":{parsed.port}" for host in no_proxy: if parsed.hostname.endswith(host) or host_with_port.endswith(host): # The URL does match something in no_proxy, so we don't want # to apply the proxies on this URL. return True with set_environ("no_proxy", no_proxy_arg): # parsed.hostname can be `None` in cases such as a file URI. try: bypass = proxy_bypass(parsed.hostname) except (TypeError, socket.gaierror): bypass = False if bypass: return True return False def get_environ_proxies(url, no_proxy=None): """ Return a dict of environment proxies. :rtype: dict """ if should_bypass_proxies(url, no_proxy=no_proxy): return {} else: return getproxies() def select_proxy(url, proxies): """Select a proxy for the url, if applicable. :param url: The url being for the request :param proxies: A dictionary of schemes or schemes and hosts to proxy URLs """ proxies = proxies or {} urlparts = urlparse(url) if urlparts.hostname is None: return proxies.get(urlparts.scheme, proxies.get("all")) proxy_keys = [ urlparts.scheme + "://" + urlparts.hostname, urlparts.scheme, "all://" + urlparts.hostname, "all", ] proxy = None for proxy_key in proxy_keys: if proxy_key in proxies: proxy = proxies[proxy_key] break return proxy def resolve_proxies(request, proxies, trust_env=True): """This method takes proxy information from a request and configuration input to resolve a mapping of target proxies. This will consider settings such as NO_PROXY to strip proxy configurations. :param request: Request or PreparedRequest :param proxies: A dictionary of schemes or schemes and hosts to proxy URLs :param trust_env: Boolean declaring whether to trust environment configs :rtype: dict """ proxies = proxies if proxies is not None else {} url = request.url scheme = urlparse(url).scheme no_proxy = proxies.get("no_proxy") new_proxies = proxies.copy() if trust_env and not should_bypass_proxies(url, no_proxy=no_proxy): environ_proxies = get_environ_proxies(url, no_proxy=no_proxy) proxy = environ_proxies.get(scheme, environ_proxies.get("all")) if proxy: new_proxies.setdefault(scheme, proxy) return new_proxies def default_user_agent(name="python-requests"): """ Return a string representing the default user agent. :rtype: str """ return f"{name}/{__version__}" def default_headers(): """ :rtype: requests.structures.CaseInsensitiveDict """ return CaseInsensitiveDict( { "User-Agent": default_user_agent(), "Accept-Encoding": DEFAULT_ACCEPT_ENCODING, "Accept": "*/*", "Connection": "keep-alive", } ) def parse_header_links(value): """Return a list of parsed link headers proxies. i.e. Link: <http:/.../front.jpeg>; rel=front; type="image/jpeg",<http://.../back.jpeg>; rel=back;type="image/jpeg" :rtype: list """ links = [] replace_chars = " '\"" value = value.strip(replace_chars) if not value: return links for val in re.split(", *<", value): try: url, params = val.split(";", 1) except ValueError: url, params = val, "" link = {"url": url.strip("<> '\"")} for param in params.split(";"): try: key, value = param.split("=") except ValueError: break link[key.strip(replace_chars)] = value.strip(replace_chars) links.append(link) return links # Null bytes; no need to recreate these on each call to guess_json_utf _null = "\x00".encode("ascii") # encoding to ASCII for Python 3 _null2 = _null * 2 _null3 = _null * 3 def guess_json_utf(data): """ :rtype: str """ # JSON always starts with two ASCII characters, so detection is as # easy as counting the nulls and from their location and count # determine the encoding. Also detect a BOM, if present. sample = data[:4] if sample in (codecs.BOM_UTF32_LE, codecs.BOM_UTF32_BE): return "utf-32" # BOM included if sample[:3] == codecs.BOM_UTF8: return "utf-8-sig" # BOM included, MS style (discouraged) if sample[:2] in (codecs.BOM_UTF16_LE, codecs.BOM_UTF16_BE): return "utf-16" # BOM included nullcount = sample.count(_null) if nullcount == 0: return "utf-8" if nullcount == 2: if sample[::2] == _null2: # 1st and 3rd are null return "utf-16-be" if sample[1::2] == _null2: # 2nd and 4th are null return "utf-16-le" # Did not detect 2 valid UTF-16 ascii-range characters if nullcount == 3: if sample[:3] == _null3: return "utf-32-be" if sample[1:] == _null3: return "utf-32-le" # Did not detect a valid UTF-32 ascii-range character return None def prepend_scheme_if_needed(url, new_scheme): """Given a URL that may or may not have a scheme, prepend the given scheme. Does not replace a present scheme with the one provided as an argument. :rtype: str """ parsed = parse_url(url) scheme, auth, host, port, path, query, fragment = parsed # A defect in urlparse determines that there isn't a netloc present in some # urls. We previously assumed parsing was overly cautious, and swapped the # netloc and path. Due to a lack of tests on the original defect, this is # maintained with parse_url for backwards compatibility. netloc = parsed.netloc if not netloc: netloc, path = path, netloc if auth: # parse_url doesn't provide the netloc with auth # so we'll add it ourselves. netloc = "@".join([auth, netloc]) if scheme is None: scheme = new_scheme if path is None: path = "" return urlunparse((scheme, netloc, path, "", query, fragment)) def get_auth_from_url(url): """Given a url with authentication components, extract them into a tuple of username,password. :rtype: (str,str) """ parsed = urlparse(url) try: auth = (unquote(parsed.username), unquote(parsed.password)) except (AttributeError, TypeError): auth = ("", "") return auth def check_header_validity(header): """Verifies that header parts don't contain leading whitespace reserved characters, or return characters. :param header: tuple, in the format (name, value). """ name, value = header _validate_header_part(header, name, 0) _validate_header_part(header, value, 1) def _validate_header_part(header, header_part, header_validator_index): if isinstance(header_part, str): validator = _HEADER_VALIDATORS_STR[header_validator_index] elif isinstance(header_part, bytes): validator = _HEADER_VALIDATORS_BYTE[header_validator_index] else: raise InvalidHeader( f"Header part ({header_part!r}) from {header} " f"must be of type str or bytes, not {type(header_part)}" ) if not validator.match(header_part): header_kind = "name" if header_validator_index == 0 else "value" raise InvalidHeader( f"Invalid leading whitespace, reserved character(s), or return " f"character(s) in header {header_kind}: {header_part!r}" ) def urldefragauth(url): """ Given a url remove the fragment and the authentication part. :rtype: str """ scheme, netloc, path, params, query, fragment = urlparse(url) # see func:`prepend_scheme_if_needed` if not netloc: netloc, path = path, netloc netloc = netloc.rsplit("@", 1)[-1] return urlunparse((scheme, netloc, path, params, query, "")) def rewind_body(prepared_request): """Move file pointer back to its recorded starting position so it can be read again on redirect. """ body_seek = getattr(prepared_request.body, "seek", None) if body_seek is not None and isinstance( prepared_request._body_position, integer_types ): try: body_seek(prepared_request._body_position) except OSError: raise UnrewindableBodyError( "An error occurred when rewinding request body for redirect." ) else: raise UnrewindableBodyError("Unable to rewind request body for redirect.")
| ver. 1.4 |
Github
|
.
| PHP 8.2.28 | Generation time: 0.03 |
proxy
|
phpinfo
|
Settings