codekingpro/portable-devtools

1import argparse2import http.client3import json4import sys5from contextlib import contextmanager6from time import monotonic7from typing import Any, Dict, Callable, Iterable8from urllib.parse import urlparse, urlunparse9 10import requests11# noinspection PyPackageRequirements12import urllib313from urllib3.util import SKIP_HEADER, SKIPPABLE_HEADERS14 15from . import __version__16from .adapters import HTTPieHTTPAdapter17from .cli.constants import HTTP_OPTIONS18from .cli.dicts import HTTPHeadersDict19from .cli.nested_json import unwrap_top_level_list_if_needed20from .context import Environment21from .encoding import UTF822from .models import RequestsMessage23from .plugins.registry import plugin_manager24from .sessions import get_httpie_session25from .ssl_ import AVAILABLE_SSL_VERSION_ARG_MAPPING, HTTPieCertificate, HTTPieHTTPSAdapter26from .uploads import (27    compress_request, prepare_request_body,28    get_multipart_data_and_content_type,29)30from .utils import get_expired_cookies, repr_dict31 32 33urllib3.disable_warnings()34 35FORM_CONTENT_TYPE = f'application/x-www-form-urlencoded; charset={UTF8}'36JSON_CONTENT_TYPE = 'application/json'37JSON_ACCEPT = f'{JSON_CONTENT_TYPE}, */*;q=0.5'38DEFAULT_UA = f'HTTPie/{__version__}'39 40IGNORE_CONTENT_LENGTH_METHODS = frozenset([HTTP_OPTIONS])41 42 43def collect_messages(44    env: Environment,45    args: argparse.Namespace,46    request_body_read_callback: Callable[[bytes], None] = None,47) -> Iterable[RequestsMessage]:48    httpie_session = None49    httpie_session_headers = None50    if args.session or args.session_read_only:51        httpie_session = get_httpie_session(52            env=env,53            config_dir=env.config.directory,54            session_name=args.session or args.session_read_only,55            host=args.headers.get('Host'),56            url=args.url,57        )58        httpie_session_headers = httpie_session.headers59 60    request_kwargs = make_request_kwargs(61        env,62        args=args,63        base_headers=httpie_session_headers,64        request_body_read_callback=request_body_read_callback65    )66    send_kwargs = make_send_kwargs(args)67    send_kwargs_mergeable_from_env = make_send_kwargs_mergeable_from_env(args)68    requests_session = build_requests_session(69        ssl_version=args.ssl_version,70        ciphers=args.ciphers,71        verify=bool(send_kwargs_mergeable_from_env['verify'])72    )73 74    if httpie_session:75        httpie_session.update_headers(request_kwargs['headers'])76        requests_session.cookies = httpie_session.cookies77        if args.auth_plugin:78            # Save auth from CLI to HTTPie session.79            httpie_session.auth = {80                'type': args.auth_plugin.auth_type,81                'raw_auth': args.auth_plugin.raw_auth,82            }83        elif httpie_session.auth:84            # Apply auth from HTTPie session85            request_kwargs['auth'] = httpie_session.auth86 87    if args.debug:88        # TODO: reflect the split between request and send kwargs.89        dump_request(request_kwargs)90 91    request = requests.Request(**request_kwargs)92    prepared_request = requests_session.prepare_request(request)93    transform_headers(request, prepared_request)94    if args.path_as_is:95        prepared_request.url = ensure_path_as_is(96            orig_url=args.url,97            prepped_url=prepared_request.url,98        )99    if args.compress and prepared_request.body:100        compress_request(101            request=prepared_request,102            always=args.compress > 1,103        )104    response_count = 0105    expired_cookies = []106    while prepared_request:107        yield prepared_request108        if not args.offline:109            send_kwargs_merged = requests_session.merge_environment_settings(110                url=prepared_request.url,111                **send_kwargs_mergeable_from_env,112            )113            with max_headers(args.max_headers):114                response = requests_session.send(115                    request=prepared_request,116                    **send_kwargs_merged,117                    **send_kwargs,118                )119            response._httpie_headers_parsed_at = monotonic()120            expired_cookies += get_expired_cookies(121                response.headers.get('Set-Cookie', '')122            )123 124            response_count += 1125            if response.next:126                if args.max_redirects and response_count == args.max_redirects:127                    raise requests.TooManyRedirects128                if args.follow:129                    prepared_request = response.next130                    if args.all:131                        yield response132                    continue133            yield response134        break135 136    if httpie_session:137        if httpie_session.is_new() or not args.session_read_only:138            httpie_session.cookies = requests_session.cookies139            httpie_session.remove_cookies(expired_cookies)140            httpie_session.save()141 142 143# noinspection PyProtectedMember144@contextmanager145def max_headers(limit):146    # <https://github.com/httpie/cli/issues/802>147    # noinspection PyUnresolvedReferences148    orig = http.client._MAXHEADERS149    http.client._MAXHEADERS = limit or float('Inf')150    try:151        yield152    finally:153        http.client._MAXHEADERS = orig154 155 156def build_requests_session(157    verify: bool,158    ssl_version: str = None,159    ciphers: str = None,160) -> requests.Session:161    requests_session = requests.Session()162 163    # Install our adapter.164    http_adapter = HTTPieHTTPAdapter()165    https_adapter = HTTPieHTTPSAdapter(166        ciphers=ciphers,167        verify=verify,168        ssl_version=(169            AVAILABLE_SSL_VERSION_ARG_MAPPING[ssl_version]170            if ssl_version else None171        ),172    )173    requests_session.mount('http://', http_adapter)174    requests_session.mount('https://', https_adapter)175 176    # Install adapters from plugins.177    for plugin_cls in plugin_manager.get_transport_plugins():178        transport_plugin = plugin_cls()179        requests_session.mount(180            prefix=transport_plugin.prefix,181            adapter=transport_plugin.get_adapter(),182        )183 184    return requests_session185 186 187def dump_request(kwargs: dict):188    sys.stderr.write(189        f'\n>>> requests.request(**{repr_dict(kwargs)})\n\n')190 191 192def finalize_headers(headers: HTTPHeadersDict) -> HTTPHeadersDict:193    final_headers = HTTPHeadersDict()194    for name, value in headers.items():195        if value is not None:196            # “leading or trailing LWS MAY be removed without197            # changing the semantics of the field value”198            # <https://www.w3.org/Protocols/rfc2616/rfc2616-sec4.html>199            # Also, requests raises `InvalidHeader` for leading spaces.200            value = value.strip()201            if isinstance(value, str):202                # See <https://github.com/httpie/cli/issues/212>203                value = value.encode()204        elif name.lower() in SKIPPABLE_HEADERS:205            # Some headers get overwritten by urllib3 when set to `None`206            # and should be replaced with the `SKIP_HEADER` constant.207            value = SKIP_HEADER208        final_headers.add(name, value)209    return final_headers210 211 212def transform_headers(213    request: requests.Request,214    prepared_request: requests.PreparedRequest215) -> None:216    """Apply various transformations on top of the `prepared_requests`'s217    headers to change the request prepreation behavior."""218 219    # Remove 'Content-Length' when it is misplaced by requests.220    if (221        prepared_request.method in IGNORE_CONTENT_LENGTH_METHODS222        and prepared_request.headers.get('Content-Length') == '0'223        and request.headers.get('Content-Length') != '0'224    ):225        prepared_request.headers.pop('Content-Length')226 227    apply_missing_repeated_headers(228        request.headers,229        prepared_request230    )231 232 233def apply_missing_repeated_headers(234    original_headers: HTTPHeadersDict,235    prepared_request: requests.PreparedRequest236) -> None:237    """Update the given `prepared_request`'s headers with the original238    ones. This allows the requests to be prepared as usual, and then later239    merged with headers that are specified multiple times."""240 241    new_headers = HTTPHeadersDict(prepared_request.headers)242    for prepared_name, prepared_value in prepared_request.headers.items():243        if prepared_name not in original_headers:244            continue245 246        original_keys, original_values = zip(*filter(247            lambda item: item[0].casefold() == prepared_name.casefold(),248            original_headers.items()249        ))250 251        if prepared_value not in original_values:252            # If the current value is not among the initial values253            # set for this field, then it means that this field got254            # overridden on the way, and we should preserve it.255            continue256 257        new_headers.popone(prepared_name)258        new_headers.update(zip(original_keys, original_values))259 260    prepared_request.headers = new_headers261 262 263def make_default_headers(args: argparse.Namespace) -> HTTPHeadersDict:264    default_headers = HTTPHeadersDict({265        'User-Agent': DEFAULT_UA266    })267 268    auto_json = args.data and not args.form269    if args.json or auto_json:270        default_headers['Accept'] = JSON_ACCEPT271        if args.json or (auto_json and args.data):272            default_headers['Content-Type'] = JSON_CONTENT_TYPE273 274    elif args.form and not args.files:275        # If sending files, `requests` will set276        # the `Content-Type` for us.277        default_headers['Content-Type'] = FORM_CONTENT_TYPE278    return default_headers279 280 281def make_send_kwargs(args: argparse.Namespace) -> dict:282    return {283        'timeout': args.timeout or None,284        'allow_redirects': False,285    }286 287 288def make_send_kwargs_mergeable_from_env(args: argparse.Namespace) -> dict:289    cert = None290    if args.cert:291        cert = args.cert292        if args.cert_key:293            # Having a client certificate key passphrase is not supported294            # by requests. So we are using our own transportation structure295            # which is compatible with their format (a tuple of minimum two296            # items).297            #298            # See: https://github.com/psf/requests/issues/2519299            cert = HTTPieCertificate(cert, args.cert_key, args.cert_key_pass.value)300 301    return {302        'proxies': {p.key: p.value for p in args.proxy},303        'stream': True,304        'verify': {305            'yes': True,306            'true': True,307            'no': False,308            'false': False,309        }.get(args.verify.lower(), args.verify),310        'cert': cert,311    }312 313 314def json_dict_to_request_body(data: Dict[str, Any]) -> str:315    data = unwrap_top_level_list_if_needed(data)316    if data:317        data = json.dumps(data)318    else:319        # We need to set data to an empty string to prevent requests320        # from assigning an empty list to `response.request.data`.321        data = ''322    return data323 324 325def make_request_kwargs(326    env: Environment,327    args: argparse.Namespace,328    base_headers: HTTPHeadersDict = None,329    request_body_read_callback=lambda chunk: chunk330) -> dict:331    """332    Translate our `args` into `requests.Request` keyword arguments.333 334    """335    files = args.files336    # Serialize JSON data, if needed.337    data = args.data338    auto_json = data and not args.form339    if (args.json or auto_json) and isinstance(data, dict):340        data = json_dict_to_request_body(data)341 342    # Finalize headers.343    headers = make_default_headers(args)344    if base_headers:345        headers.update(base_headers)346    headers.update(args.headers)347    if args.offline and args.chunked and 'Transfer-Encoding' not in headers:348        # When online, we let requests set the header instead to be able more349        # easily verify chunking is taking place.350        headers['Transfer-Encoding'] = 'chunked'351    headers = finalize_headers(headers)352 353    if (args.form and files) or args.multipart:354        data, headers['Content-Type'] = get_multipart_data_and_content_type(355            data=args.multipart_data,356            boundary=args.boundary,357            content_type=args.headers.get('Content-Type'),358        )359 360    return {361        'method': args.method.lower(),362        'url': args.url,363        'headers': headers,364        'data': prepare_request_body(365            env,366            data,367            body_read_callback=request_body_read_callback,368            chunked=args.chunked,369            offline=args.offline,370            content_length_header_value=headers.get('Content-Length'),371        ),372        'auth': args.auth,373        'params': args.params.items(),374    }375 376 377def ensure_path_as_is(orig_url: str, prepped_url: str) -> str:378    """379    Handle `--path-as-is` by replacing the path component of the prepared380    URL with the path component from the original URL. Other parts stay381    untouched because other (welcome) processing on the URL might have382    taken place.383 384    <https://github.com/httpie/cli/issues/895>385 386 387    <https://ec.haxx.se/http/http-basics#path-as-is>388    <https://curl.haxx.se/libcurl/c/CURLOPT_PATH_AS_IS.html>389 390    >>> ensure_path_as_is('http://foo/../', 'http://foo/?foo=bar')391    'http://foo/../?foo=bar'392 393    """394    parsed_orig, parsed_prepped = urlparse(orig_url), urlparse(prepped_url)395    final_dict = {396        # noinspection PyProtectedMember397        **parsed_prepped._asdict(),398        'path': parsed_orig.path,399    }400    return urlunparse(tuple(final_dict.values()))401