/usr/lib/python2.7/dist-packages/httpie/client.py is in httpie 0.9.8-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 | import json
import sys
import requests
from requests.adapters import HTTPAdapter
from requests.packages import urllib3
from httpie import sessions
from httpie import __version__
from httpie.compat import str
from httpie.input import SSL_VERSION_ARG_MAPPING
from httpie.plugins import plugin_manager
from httpie.utils import repr_dict_nice
try:
# https://urllib3.readthedocs.io/en/latest/security.html
urllib3.disable_warnings()
except AttributeError:
# In some rare cases, the user may have an old version of the requests
# or urllib3, and there is no method called "disable_warnings." In these
# cases, we don't need to call the method.
# They may get some noisy output but execution shouldn't die. Move on.
pass
FORM_CONTENT_TYPE = 'application/x-www-form-urlencoded; charset=utf-8'
JSON_CONTENT_TYPE = 'application/json'
JSON_ACCEPT = '{0}, */*'.format(JSON_CONTENT_TYPE)
DEFAULT_UA = 'HTTPie/%s' % __version__
class HTTPieHTTPAdapter(HTTPAdapter):
def __init__(self, ssl_version=None, **kwargs):
self._ssl_version = ssl_version
super(HTTPieHTTPAdapter, self).__init__(**kwargs)
def init_poolmanager(self, *args, **kwargs):
kwargs['ssl_version'] = self._ssl_version
super(HTTPieHTTPAdapter, self).init_poolmanager(*args, **kwargs)
def get_requests_session(ssl_version):
requests_session = requests.Session()
requests_session.mount(
'https://',
HTTPieHTTPAdapter(ssl_version=ssl_version)
)
for cls in plugin_manager.get_transport_plugins():
transport_plugin = cls()
requests_session.mount(prefix=transport_plugin.prefix,
adapter=transport_plugin.get_adapter())
return requests_session
def get_response(args, config_dir):
"""Send the request and return a `request.Response`."""
ssl_version = None
if args.ssl_version:
ssl_version = SSL_VERSION_ARG_MAPPING[args.ssl_version]
requests_session = get_requests_session(ssl_version)
requests_session.max_redirects = args.max_redirects
if not args.session and not args.session_read_only:
kwargs = get_requests_kwargs(args)
if args.debug:
dump_request(kwargs)
response = requests_session.request(**kwargs)
else:
response = sessions.get_response(
requests_session=requests_session,
args=args,
config_dir=config_dir,
session_name=args.session or args.session_read_only,
read_only=bool(args.session_read_only),
)
return response
def dump_request(kwargs):
sys.stderr.write('\n>>> requests.request(**%s)\n\n'
% repr_dict_nice(kwargs))
def finalize_headers(headers):
final_headers = {}
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/jkbrzt/httpie/issues/212
value = value.encode('utf8')
final_headers[name] = value
return final_headers
def get_default_headers(args):
default_headers = {
'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 get_requests_kwargs(args, base_headers=None):
"""
Translate our `args` into `requests.request` keyword arguments.
"""
# 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 = get_default_headers(args)
if base_headers:
headers.update(base_headers)
headers.update(args.headers)
headers = finalize_headers(headers)
cert = None
if args.cert:
cert = args.cert
if args.cert_key:
cert = cert, args.cert_key
kwargs = {
'stream': True,
'method': args.method.lower(),
'url': args.url,
'headers': headers,
'data': data,
'verify': {
'yes': True,
'no': False
}.get(args.verify, args.verify),
'cert': cert,
'timeout': args.timeout,
'auth': args.auth,
'proxies': dict((p.key, p.value) for p in args.proxy),
'files': args.files,
'allow_redirects': args.follow,
'params': args.params,
}
return kwargs
|