Elgato_dark/python-tuf
1
0
Fork 0
mirror of https://github.com/theupdateframework/python-tuf synced 2026-05-24 10:08:28 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge branch 'trishankatdatadog/fix-for-https-proxies' into develop

Browse source 
Signed-off-by: Sebastien Awwad <sebastien.awwad@gmail.com>
This commit is contained in:
Sebastien Awwad 2018-10-02 17:24:06 -04:00
commit 9cd2d3a0ab
No known key found for this signature in database
GPG key ID: BC0C6DEDD5E5CC03
24 changed files with 1463 additions and 334 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
ci-requirements.txt
View file

@ -4,3 +4,4 @@ iso8601
coverage
coveralls
pylint
requests

1
dev-requirements.txt
View file

@ -31,6 +31,7 @@ pylint==2.1.1 ; python_version >= "3.0"
pylint==1.9.3 ; python_version < "3.0" # pyup: ignore
pynacl==1.2.1
pyyaml==3.13
requests==2.19.1
securesystemslib[crypto,pynacl]==0.11.2
singledispatch==3.4.0.3
six==1.11.0

1
requirements.in
View file

@ -4,5 +4,6 @@ securesystemslib
cryptography
colorama
pynacl
requests
six
iso8601

21
requirements.txt
View file

@ -8,6 +8,10 @@ asn1crypto==0.24.0 \
--hash=sha256:2f1adbb7546ed199e3c90ef23ec95c5cf3585bac7d11fb7eb562a3fe89c64e87 \
--hash=sha256:9d5c20441baf0cb60a4ac34cc447c6c189024b6b4c6cd7877034f4965c464e49 \
# via cryptography
certifi==2018.8.24 \
--hash=sha256:376690d6f16d32f9d1fe8932551d80b23e9d393a8578c5633a2ed39a64861638 \
--hash=sha256:456048c7e371c089d0a77a5212fb37a2c2dce1e24146e3b7e0261736aaeaa22a \
# via requests
cffi==1.11.5 \
--hash=sha256:151b7eefd035c56b2b2e1eb9963c90c6302dc15fbd8c1c0a83a163ff2c7d7743 \
--hash=sha256:1553d1e99f035ace1c0544050622b7bc963374a00c467edafac50ad7bd276aef \
@ -37,6 +41,10 @@ cffi==1.11.5 \
--hash=sha256:edabd457cd23a02965166026fd9bfd196f4324fe6032e866d0f3bd0301cd486f \
--hash=sha256:fdf1c1dc5bafc32bc5d08b054f94d659422b05aba244d6be4ddc1c72d9aa70fb \
# via cryptography, pynacl
chardet==3.0.4 \
--hash=sha256:84ab92ed1c4d4f16916e05906b6b75a6c0fb5db821cc65e70cbd64a3e2a5eaae \
--hash=sha256:fc323ffcaeaed0e0a02bf4d117757b98aed530d9ed4531e3e15460124c106691 \
# via requests
colorama==0.3.9 \
--hash=sha256:463f8483208e921368c9f306094eb6f725c6ca42b0f97e313cb5d5512459feda \
--hash=sha256:48eb22f4f8461b1df5734a074b57042430fb06e1d61bd1e11b078c0fe6d7a1f1
@ -69,11 +77,7 @@ enum34==1.1.6 \
idna==2.7 \
--hash=sha256:156a6814fb5ac1fc6850fb002e0852d56c0c8d2531923a51032d1b70760e186e \
--hash=sha256:684a38a6f903c1d71d6d5fac066b58d7768af4de2b832e426ec79c30daa94a16 \
# via cryptography
ipaddress==1.0.22 \
--hash=sha256:64b28eec5e78e7510698f6d4da08800a5c575caa4a286c93d651c5d3ff7b6794 \
--hash=sha256:b146c751ea45cad6188dd6cf2d9b757f6f4f8d6ffb96a023e6f2e26eea02a72c \
# via cryptography
# via cryptography, requests
iso8601==0.1.12 \
--hash=sha256:210e0134677cc0d02f6028087fee1df1e1d76d372ee1db0bf30bf66c5c1c89a3 \
--hash=sha256:49c4b20e1f38aa5cf109ddcd39647ac419f928512c869dc01d5c7098eddede82 \
@ -105,9 +109,16 @@ pynacl==1.2.1 \
--hash=sha256:eeee629828d0eb4f6d98ac41e9a3a6461d114d1d0aa111a8931c049359298da0 \
--hash=sha256:f5ce9e26d25eb0b2d96f3ef0ad70e1d3ae89b5d60255c462252a3e456a48c053 \
--hash=sha256:fabf73d5d0286f9e078774f3435601d2735c94ce9e514ac4fb945701edead7e4
requests==2.19.1 \
--hash=sha256:63b52e3c866428a224f97cab011de738c36aec0185aa91cfacd418b5d58911d1 \
--hash=sha256:ec22d826a36ed72a7358ff3fe56cbd4ba69dd7a6718ffd450ff0e9df7a47ce6a
securesystemslib==0.11.2 \
--hash=sha256:43554371feeef50196587aa066cffd6b9ceff6b484fa7b127e139fafb5c0e23e \
--hash=sha256:7fe1ed8a4139b12225986ff6f9ebab48c74eaa93265a73f988e8de10e6b237a8
six==1.11.0 \
--hash=sha256:70e8a77beed4562e7f14fe23a786b54f6296e34344c23bc42f07b15018ff98e9 \
--hash=sha256:832dc0e10feb1aa2c68dcc57dbb658f1c7e65b9b61af69048abc87a2db00a0eb
urllib3==1.23 \
--hash=sha256:a68ac5e15e76e7e5dd2b8f94007233e01effe3e50e8daddf69acfd81cb686baf \
--hash=sha256:b5725a0bd4ba422ab0e66e89e030c806576753ea3ee08554382c14e685d117b5 \
# via requests

10
setup.py
View file

@ -78,9 +78,10 @@
with open('README.md') as file_object:
long_description = file_object.read()
setup(
name = 'tuf',
version = '0.11.1',
version = '0.11.1', # If updating version, also update it in tuf/__init__.py
description = 'A secure updater framework for Python',
long_description = long_description,
long_description_content_type='text/markdown',
@ -109,7 +110,12 @@
'Topic :: Security',
'Topic :: Software Development'
],
install_requires = ['iso8601>=0.1.12', 'six>=1.11.0', 'securesystemslib>=0.11.2'],
install_requires = [
'iso8601>=0.1.12',
'requests>=2.19.1',
'six>=1.11.0',
'securesystemslib>=0.11.2'
],
packages = find_packages(exclude=['tests']),
scripts = [
'tuf/scripts/repo.py',

48
tests/aggregate_tests.py
View file

@ -43,22 +43,54 @@
# 'test_' and end with '.py'. A shell-style wildcard is used with glob() to
# match desired filenames. All the tests matching the pattern will be loaded
# and run in a test suite.
tests_list = glob.glob('test_*.py')
available_tests = glob.glob('test_*.py')
# Remove '.py' from each filename to allow loadTestsFromNames() (called below)
# to properly load the file as a module.
tests_without_extension = []
for test in tests_list:
# A dictionary of test modules that should only run in certain python versions.
# Carefully consider the impact of only testing these in a given version.
# test_proxy_use.py: uses a proxy that only runs in Python2.7. TUF's
# compatibility with proxies is not likely to vary based on the Python version
# in use, so this is OK for now. See comments in that module.
# The semantics here are: only add to this list the particular tests that are
# to be run in a single major version or a single minor version. An entry must
# include major version, and may include minor version.
# Skip the test if any such listed constraints don't match the python version
# currently running.
# Note that aggregate_tests.py is run for each version of Python that tox is
# configured to use. Note also that this TUF implementation does not support
# any Python versions <2.7 or any Python3 versions <3.4.
VERSION_SPECIFIC_TESTS = {
'test_proxy_use': {'major': 2, 'minor': 7}} # Run test only if Python2.7
# Further example:
# 'test_abc': {'major': 2} # Run test only if Python2
# Determine which tests should be run.
test_modules_to_run = []
for test in available_tests:
# Remove '.py' from each filename to allow loadTestsFromNames() (called below)
# to properly load the file as a module.
assert test[-3:] == '.py', 'aggregate_tests.py is inconsistent; fix.'
test = test[:-3]
tests_without_extension.append(test)
if test in VERSION_SPECIFIC_TESTS:
# Consistency checks.
assert 'major' in VERSION_SPECIFIC_TESTS[test], 'Empty/illogical constraint'
for keyword in VERSION_SPECIFIC_TESTS[test]:
assert keyword in ['major', 'minor'], 'Unrecognized test constraint'
if sys.version_info.major != VERSION_SPECIFIC_TESTS[test]['major']:
continue
if 'minor' in VERSION_SPECIFIC_TESTS[test] \
and sys.version_info.minor != VERSION_SPECIFIC_TESTS[test]['minor']:
continue
test_modules_to_run.append(test)
# Randomize the order in which the tests run. Randomization might catch errors
# with unit tests that do not properly clean up or restore monkey-patched
# modules.
random.shuffle(tests_without_extension)
random.shuffle(test_modules_to_run)
if __name__ == '__main__':
suite = unittest.TestLoader().loadTestsFromNames(tests_without_extension)
suite = unittest.TestLoader().loadTestsFromNames(test_modules_to_run)
all_tests_passed = unittest.TextTestRunner(verbosity=1).run(suite).wasSuccessful()
if not all_tests_passed:
sys.exit(1)

500
tests/proxy_server.py Normal file
View file

@ -0,0 +1,500 @@
#!/usr/bin/env python
# This code is taken from: github.com/inaz2/proxy2
# Credit goes to the author. It has been very slightly modified here to use
# IPv4 instead of IPv6, and to only attempt interception of HTTPS traffic
# (instead of relaying via HTTP CONNECT) if new global variable INTERCEPT is
# set to True. (Modified sections are marked '# MODIFIED'.)
#
# Because this is a helper module for a test, the style is less important, and
# so to minimize changes from the source, it has NOT been changed to match the
# TUF project's code style outside of rewritten sections.
"""
<Program>
proxy_server.py
<Copyright>
Taken from a repository set to BSD 3-Clause "New" or "Revised" License. See:
https://github.com/inaz2/proxy2/blob/b2bab648173ac69f0a10421750125517accdfe26/LICENSE
<Purpose>
Serves as an HTTP, HTTP CONNECT (TCP), and HTTPS proxy, for testing purposes.
This is used by test_proxy_use.py.
In Python versions < 2.7.9, this proxy does not perform certificate
validation of the target server. As that is not part of what the current
tests using this script require, that is currently OK. In Python
versions > 2.7.9 (SSLContext was added in 2.7.9), the same code actually does
check the certificate, using the system's trusted CAs. As a result, since we
are using custom certificates, we need to either disable certificate
checking in 2.7.9 or load the specific CA for target test server, using the
SSLContext and create_default_context functionality also added in 2.7.9. It
is easier to do the latter, so the behavior in 2.7.9+ is to check the cert
and below 2.7.9 is not to. Note that we do not support Python < 2.7.
SSLContext is also available in all Python3 versions that we support.
This module requires Python2.7 and does not support Python3.
Note that this is not thread-safe, in part due to its use of globals.
"""
import sys
import os
import socket
import ssl
import select
import httplib
import urlparse
import threading
import gzip
import zlib
import time
import json
import re
from BaseHTTPServer import HTTPServer, BaseHTTPRequestHandler
from SocketServer import ThreadingMixIn
from cStringIO import StringIO
from subprocess import Popen, PIPE
from HTMLParser import HTMLParser
# MODIFIED: (added) three globals
# INTERCEPT: A boolean:
# False: normal HTTP proxy. Support HTTP & HTTPS connections to target server
# True: intercepting MITM transparent HTTPS proxy. Makes own TLS connections
# and has its own cert; must be trusted by the client and is able to
# modify requests.
# TARGET_SERVER_CA_FILEPATH: location of certificate to use as CA for
# connections to target servers (to constrain certs to trust from target
# servers).
# The remaining globals define the certs and keys to be used in communications
# with the client, with the proxy's CA signing new certs for individual hosts
# the client wishes to connect to, and placing them in dir PROXY_CERTS_DIR.
INTERCEPT = False
CERTS_DIR = os.path.join(
os.path.dirname(os.path.abspath(__file__)), 'ssl_certs')
TARGET_SERVER_CA_FILEPATH = os.path.join(CERTS_DIR, 'ssl_cert.crt')
PROXY_CA_KEY = os.path.join(CERTS_DIR, 'proxy_ca.key') # was cakey
PROXY_CA_CERT = os.path.join(CERTS_DIR, 'proxy_ca.crt') # was cacert
PROXY_CERTS_KEY = os.path.join(CERTS_DIR, 'proxy_cert.key') # was certkey
PROXY_CERTS_DIR = os.path.join(CERTS_DIR, 'proxy_certs') # was certdir
def with_color(c, s):
return "\x1b[%dm%s\x1b[0m" % (c, s)
# MODIFIED: removed join_with_script_dir
# def get_cert_filepath(path):
# return os.path.join(CERTS_DIR, path)
class ThreadingHTTPServer(ThreadingMixIn, HTTPServer):
address_family = socket.AF_INET # MODIFIED to use IPv4 instead of IPv6
daemon_threads = True
def handle_error(self, request, client_address):
# surpress socket/ssl related errors
cls, e = sys.exc_info()[:2]
if cls is socket.error or cls is ssl.SSLError:
pass
else:
return HTTPServer.handle_error(self, request, client_address)
class ProxyRequestHandler(BaseHTTPRequestHandler):
# MODIFIED: Variables here made into globals.
#Calls below modified: filenames changed, function changed to
# include ssl_certs directory.
timeout = 5
lock = threading.Lock()
def __init__(self, *args, **kwargs):
self.tls = threading.local()
self.tls.conns = {}
BaseHTTPRequestHandler.__init__(self, *args, **kwargs)
def log_error(self, format, *args):
# surpress "Request timed out: timeout('timed out',)"
if isinstance(args[0], socket.timeout):
return
self.log_message(format, *args)
def do_CONNECT(self):
# MODIFIED: This function has been modified to use new global INTERCEPT
# and to issue an error if the necessary certificate/key files are
# missing for interception attempts.
if not INTERCEPT:
print('\n\nRELAYING\n\n')
self.connect_relay()
else:
assert os.path.isfile(PROXY_CA_KEY) \
and os.path.isfile(PROXY_CA_CERT) \
and os.path.isfile(PROXY_CERTS_KEY) \
and os.path.isdir(PROXY_CERTS_DIR), \
'\nMissing key or certificate files; unable to perform TLS ' \
'handshake with client to intercept traffic.\n'
print('\n\nINTERCEPTING\n\n')
self.connect_intercept()
def connect_intercept(self):
hostname = self.path.split(':')[0]
certpath = os.path.join(PROXY_CERTS_DIR, hostname + '.crt') # MODIFIED for Windows compatibility and to use new globals
with self.lock:
if not os.path.isfile(certpath):
epoch = "%d" % (time.time() * 1000)
p1 = Popen(["openssl", "req", "-new", "-key", PROXY_CERTS_KEY, "-subj", "/CN=%s" % hostname], stdout=PIPE)
p2 = Popen(["openssl", "x509", "-req", "-days", "3650", "-CA", PROXY_CA_CERT, "-CAkey", PROXY_CA_KEY, "-set_serial", epoch, "-out", certpath], stdin=p1.stdout, stderr=PIPE) # MODIFIED to use the new globals
p2.communicate()
self.wfile.write("%s %d %s\r\n" % (self.protocol_version, 200, 'Connection Established'))
self.end_headers()
self.connection = ssl.wrap_socket(self.connection, keyfile=PROXY_CERTS_KEY, certfile=certpath, server_side=True) # MODIFIED: Updated to use new globals
self.rfile = self.connection.makefile("rb", self.rbufsize)
self.wfile = self.connection.makefile("wb", self.wbufsize)
conntype = self.headers.get('Proxy-Connection', '')
if self.protocol_version == "HTTP/1.1" and conntype.lower() != 'close':
self.close_connection = 0
else:
self.close_connection = 1
def connect_relay(self):
address = self.path.split(':', 1)
address[1] = int(address[1]) or 443
try:
s = socket.create_connection(address, timeout=self.timeout)
except Exception as e:
self.send_error(502)
return
self.send_response(200, 'Connection Established')
self.end_headers()
conns = [self.connection, s]
self.close_connection = 0
while not self.close_connection:
rlist, wlist, xlist = select.select(conns, [], conns, self.timeout)
if xlist or not rlist:
break
for r in rlist:
other = conns[1] if r is conns[0] else conns[0]
data = r.recv(8192)
if not data:
self.close_connection = 1
break
other.sendall(data)
def do_GET(self):
if self.path == 'http://proxy2.test/':
self.send_cacert()
return
req = self
content_length = int(req.headers.get('Content-Length', 0))
req_body = self.rfile.read(content_length) if content_length else None
if req.path[0] == '/':
if isinstance(self.connection, ssl.SSLSocket):
req.path = "https://%s%s" % (req.headers['Host'], req.path)
else:
req.path = "http://%s%s" % (req.headers['Host'], req.path)
req_body_modified = self.request_handler(req, req_body)
if req_body_modified is False:
self.send_error(403)
return
elif req_body_modified is not None:
req_body = req_body_modified
req.headers['Content-length'] = str(len(req_body))
u = urlparse.urlsplit(req.path)
scheme, netloc, path = u.scheme, u.netloc, (u.path + '?' + u.query if u.query else u.path)
assert scheme in ('http', 'https')
if netloc:
req.headers['Host'] = netloc
setattr(req, 'headers', self.filter_headers(req.headers))
try:
origin = (scheme, netloc)
if not origin in self.tls.conns:
if scheme == 'https':
# MODIFIED: Added Python version checking and changed behavior
# in Python2.7.9+ to use custom certificate for target server
# inherited from command line argument.
# In Python versions < 2.7.9, there is no certificate
# validation through this method of the target server.
# In supported Python versions > 2.7.9, we check the target
# server's certificate against our expected custom cert.
# See this script's docstring.
if sys.version_info.major == 2 \
and sys.version_info.minor == 7 \
and sys.version_info.micro < 9:
self.tls.conns[origin] = httplib.HTTPSConnection(
netloc, timeout=self.timeout)
else:
self.tls.conns[origin] = httplib.HTTPSConnection(
netloc, timeout=self.timeout,
context=ssl.create_default_context( # reqs Python2.7.9+
cafile=TARGET_SERVER_CA_FILEPATH))
else:
self.tls.conns[origin] = httplib.HTTPConnection(netloc, timeout=self.timeout)
conn = self.tls.conns[origin]
conn.request(self.command, path, req_body, dict(req.headers))
res = conn.getresponse()
version_table = {10: 'HTTP/1.0', 11: 'HTTP/1.1'}
setattr(res, 'headers', res.msg)
setattr(res, 'response_version', version_table[res.version])
# support streaming
if not 'Content-Length' in res.headers and 'no-store' in res.headers.get('Cache-Control', ''):
self.response_handler(req, req_body, res, '')
setattr(res, 'headers', self.filter_headers(res.headers))
self.relay_streaming(res)
with self.lock:
self.save_handler(req, req_body, res, '')
return
res_body = res.read()
except Exception as e:
if origin in self.tls.conns:
del self.tls.conns[origin]
self.send_error(502)
return
content_encoding = res.headers.get('Content-Encoding', 'identity')
res_body_plain = self.decode_content_body(res_body, content_encoding)
res_body_modified = self.response_handler(req, req_body, res, res_body_plain)
if res_body_modified is False:
self.send_error(403)
return
elif res_body_modified is not None:
res_body_plain = res_body_modified
res_body = self.encode_content_body(res_body_plain, content_encoding)
res.headers['Content-Length'] = str(len(res_body))
setattr(res, 'headers', self.filter_headers(res.headers))
self.wfile.write("%s %d %s\r\n" % (self.protocol_version, res.status, res.reason))
for line in res.headers.headers:
self.wfile.write(line)
self.end_headers()
self.wfile.write(res_body)
self.wfile.flush()
with self.lock:
self.save_handler(req, req_body, res, res_body_plain)
def relay_streaming(self, res):
self.wfile.write("%s %d %s\r\n" % (self.protocol_version, res.status, res.reason))
for line in res.headers.headers:
self.wfile.write(line)
self.end_headers()
try:
while True:
chunk = res.read(8192)
if not chunk:
break
self.wfile.write(chunk)
self.wfile.flush()
except socket.error:
# connection closed by client
pass
do_HEAD = do_GET
do_POST = do_GET
do_PUT = do_GET
do_DELETE = do_GET
do_OPTIONS = do_GET
def filter_headers(self, headers):
# http://tools.ietf.org/html/rfc2616#section-13.5.1
hop_by_hop = ('connection', 'keep-alive', 'proxy-authenticate', 'proxy-authorization', 'te', 'trailers', 'transfer-encoding', 'upgrade')
for k in hop_by_hop:
del headers[k]
# accept only supported encodings
if 'Accept-Encoding' in headers:
ae = headers['Accept-Encoding']
filtered_encodings = [x for x in re.split(r',\s*', ae) if x in ('identity', 'gzip', 'x-gzip', 'deflate')]
headers['Accept-Encoding'] = ', '.join(filtered_encodings)
return headers
def encode_content_body(self, text, encoding):
if encoding == 'identity':
data = text
elif encoding in ('gzip', 'x-gzip'):
io = StringIO()
with gzip.GzipFile(fileobj=io, mode='wb') as f:
f.write(text)
data = io.getvalue()
elif encoding == 'deflate':
data = zlib.compress(text)
else:
raise Exception("Unknown Content-Encoding: %s" % encoding)
return data
def decode_content_body(self, data, encoding):
if encoding == 'identity':
text = data
elif encoding in ('gzip', 'x-gzip'):
io = StringIO(data)
with gzip.GzipFile(fileobj=io) as f:
text = f.read()
elif encoding == 'deflate':
try:
text = zlib.decompress(data)
except zlib.error:
text = zlib.decompress(data, -zlib.MAX_WBITS)
else:
raise Exception("Unknown Content-Encoding: %s" % encoding)
return text
def send_cacert(self):
with open(PROXY_CA_CERT, 'rb') as f: # MODIFIED to use new globals
data = f.read()
self.wfile.write("%s %d %s\r\n" % (self.protocol_version, 200, 'OK'))
self.send_header('Content-Type', 'application/x-x509-ca-cert')
self.send_header('Content-Length', len(data))
self.send_header('Connection', 'close')
self.end_headers()
self.wfile.write(data)
def print_info(self, req, req_body, res, res_body):
def parse_qsl(s):
return '\n'.join("%-20s %s" % (k, v) for k, v in urlparse.parse_qsl(s, keep_blank_values=True))
req_header_text = "%s %s %s\n%s" % (req.command, req.path, req.request_version, req.headers)
res_header_text = "%s %d %s\n%s" % (res.response_version, res.status, res.reason, res.headers)
print with_color(33, req_header_text)
u = urlparse.urlsplit(req.path)
if u.query:
query_text = parse_qsl(u.query)
print with_color(32, "==== QUERY PARAMETERS ====\n%s\n" % query_text)
cookie = req.headers.get('Cookie', '')
if cookie:
cookie = parse_qsl(re.sub(r';\s*', '&', cookie))
print with_color(32, "==== COOKIE ====\n%s\n" % cookie)
auth = req.headers.get('Authorization', '')
if auth.lower().startswith('basic'):
token = auth.split()[1].decode('base64')
print with_color(31, "==== BASIC AUTH ====\n%s\n" % token)
if req_body is not None:
req_body_text = None
content_type = req.headers.get('Content-Type', '')
if content_type.startswith('application/x-www-form-urlencoded'):
req_body_text = parse_qsl(req_body)
elif content_type.startswith('application/json'):
try:
json_obj = json.loads(req_body)
json_str = json.dumps(json_obj, indent=2)
if json_str.count('\n') < 50:
req_body_text = json_str
else:
lines = json_str.splitlines()
req_body_text = "%s\n(%d lines)" % ('\n'.join(lines[:50]), len(lines))
except ValueError:
req_body_text = req_body
elif len(req_body) < 1024:
req_body_text = req_body
if req_body_text:
print with_color(32, "==== REQUEST BODY ====\n%s\n" % req_body_text)
print with_color(36, res_header_text)
cookies = res.headers.getheaders('Set-Cookie')
if cookies:
cookies = '\n'.join(cookies)
print with_color(31, "==== SET-COOKIE ====\n%s\n" % cookies)
if res_body is not None:
res_body_text = None
content_type = res.headers.get('Content-Type', '')
if content_type.startswith('application/json'):
try:
json_obj = json.loads(res_body)
json_str = json.dumps(json_obj, indent=2)
if json_str.count('\n') < 50:
res_body_text = json_str
else:
lines = json_str.splitlines()
res_body_text = "%s\n(%d lines)" % ('\n'.join(lines[:50]), len(lines))
except ValueError:
res_body_text = res_body
elif content_type.startswith('text/html'):
m = re.search(r'<title[^>]*>\s*([^<]+?)\s*</title>', res_body, re.I)
if m:
h = HTMLParser()
print with_color(32, "==== HTML TITLE ====\n%s\n" % h.unescape(m.group(1).decode('utf-8')))
elif content_type.startswith('text/') and len(res_body) < 1024:
res_body_text = res_body
if res_body_text:
print with_color(32, "==== RESPONSE BODY ====\n%s\n" % res_body_text)
def request_handler(self, req, req_body):
pass
def response_handler(self, req, req_body, res, res_body):
pass
def save_handler(self, req, req_body, res, res_body):
self.print_info(req, req_body, res, res_body)
def test(HandlerClass=ProxyRequestHandler, ServerClass=ThreadingHTTPServer, protocol="HTTP/1.1"):
# MODIFIED: Added these globals.
global INTERCEPT
global TARGET_SERVER_CA_FILEPATH
if sys.argv[1:]:
port = int(sys.argv[1])
else:
port = 8080
server_address = ('127.0.0.1', port) # MODIFIED: changed from '::1'
# MODIFIED: Argument added, conditional below added to control INTERCEPT
# setting.
if len(sys.argv) > 2:
if sys.argv[2].lower() == 'intercept':
INTERCEPT = True
# MODIFIED: Argument added to control certificate(s) the proxy expects of
# the target server(s), and added default value.
if len(sys.argv) > 3:
if os.path.exists(sys.argv[3]):
TARGET_SERVER_CA_FILEPATH = sys.argv[3]
else:
raise Exception('Target server cert file not found: ' + sys.argv[3])
# MODIFIED: Create the target-host-specific proxy certificates directory if
# it doesn't already exist.
if not os.path.exists(PROXY_CERTS_DIR):
os.mkdir(PROXY_CERTS_DIR)
HandlerClass.protocol_version = protocol
httpd = ServerClass(server_address, HandlerClass)
sa = httpd.socket.getsockname()
print "Serving HTTP Proxy on", sa[0], "port", sa[1], "..."
httpd.serve_forever()
if __name__ == '__main__':
test()

18
tests/simple_https_server.py
View file

@ -40,11 +40,14 @@
import sys
import random
import ssl
import os
import six
PORT = 0
keyfile = os.path.join('ssl_certs', 'ssl_cert.key')
certfile = os.path.join('ssl_certs', 'ssl_cert.crt')
def _generate_random_port():
return random.randint(30000, 45000)
@ -60,12 +63,19 @@ def _generate_random_port():
else:
PORT = _generate_random_port()
if len(sys.argv) > 2:
if os.path.exists(sys.argv[2]):
certfile = sys.argv[2]
else:
print('simple_https_server: cert file not found: ' + sys.argv[2] +
'; using default: ' + certfile)
httpd = six.moves.BaseHTTPServer.HTTPServer(('localhost', PORT),
six.moves.SimpleHTTPServer.SimpleHTTPRequestHandler)
httpd.socket = ssl.wrap_socket(httpd.socket, keyfile='ssl_cert.key',
certfile='ssl_cert.crt',
server_side=True)
httpd.socket = ssl.wrap_socket(
httpd.socket, keyfile=keyfile, certfile=certfile, server_side=True)
#print('Starting https server on port: ' + str(PORT))
httpd.serve_forever()

6
tests/slow_retrieval_server.py
View file

@ -73,11 +73,7 @@ def do_GET(self):
# 'mode_2'
else:
DELAY = 1
# Throttle the file by sending a character every few seconds.
# NOTE: The for-loop below completes early if the download file
# (len(data)) is small. 'download.py' waits at least
# 'settings.SLOW_START_GRACE_PERIOD' seconds before triggering a
# potential slow retrieval error.
# Throttle the file by sending a character every DELAY seconds.
for i in range(len(data)):
self.wfile.write(data[i].encode('utf-8'))
time.sleep(DELAY)

17
tests/ssl_certs/proxy_ca.crt Normal file
View file

@ -0,0 +1,17 @@
-----BEGIN CERTIFICATE-----
MIICpDCCAYwCCQCFr/EhHmzVajANBgkqhkiG9w0BAQsFADAUMRIwEAYDVQQDDAlw
cm94eTIgQ0EwHhcNMTgwOTIwMTkyOTQ2WhcNMjgwOTE3MTkyOTQ2WjAUMRIwEAYD
VQQDDAlwcm94eTIgQ0EwggEiMA0GCSqGSIb3DQEBAQUAA4IBDwAwggEKAoIBAQC/
rVOeqSzJb01Vyliw3dnfLJsWfDfs/Lq5HLn+Xqnzl6MqnYirDqHzTErD3vl8lo/o
OJrziO0vYCWGXEylRQlZp+P37bLToSWiVqWZ8pH6CAh+AhA3WtegN5JwTgIUSP7A
aDlxuZrXlJM50QVlXJIPkc74M8ALz0nu5zmyWkGFvmTYS8503T8cXs9Alr4Bo++9
Ilixv6lW4QS7FKTeQXlI49K4TeGGGsfmEO6Uj4WTUkwMZym9wfiqtaWc6I9ZMese
WmU3LuufY+pFCdjsdMWDJpYc+HabTSrbgXSF5Iq9a84Xuum39qhVpYhBwBtLk3ye
cxZmIxde1vnkWAitJFETAgMBAAEwDQYJKoZIhvcNAQELBQADggEBAKV09r/x3WyO
McH0RU4WRVzvQN5F0e7swpDlLUX7YnfvpPEkavqQfmrL1cYyEDgsm/347Gvcs1Aa
iaT77axYroXOvCEJ3DxZdzUErKH6Jr3MmHKcZ/L35u6ZXKnmx/edFjdWr6ENkjuZ
NVvKbTrm4cl6Wy4bXkp6b24rBa9IFJncOouSkIvHENEcH//OD4xeTK8vSJTJ9nmw
TiJ0TjCRujtJWC6yb03ZV32VbeiHa1zLlZhcyKqUtt81dLti5t5+L2hAAVCcnEgI
DBWQdlRs/wilHGWVBo/9srOoMNsmvecTBpLH2JyC5VZ1+faYLPrNlgkWgHIFOTTi
h4ByR95Wbi8=
-----END CERTIFICATE-----

27
tests/ssl_certs/proxy_ca.key Normal file
View file

@ -0,0 +1,27 @@
-----BEGIN RSA PRIVATE KEY-----
MIIEpQIBAAKCAQEAv61TnqksyW9NVcpYsN3Z3yybFnw37Py6uRy5/l6p85ejKp2I
qw6h80xKw975fJaP6Dia84jtL2AlhlxMpUUJWafj9+2y06ElolalmfKR+ggIfgIQ
N1rXoDeScE4CFEj+wGg5cbma15STOdEFZVySD5HO+DPAC89J7uc5slpBhb5k2EvO
dN0/HF7PQJa+AaPvvSJYsb+pVuEEuxSk3kF5SOPSuE3hhhrH5hDulI+Fk1JMDGcp
vcH4qrWlnOiPWTHrHlplNy7rn2PqRQnY7HTFgyaWHPh2m00q24F0heSKvWvOF7rp
t/aoVaWIQcAbS5N8nnMWZiMXXtb55FgIrSRREwIDAQABAoIBACxJObbA064+3xlh
RRioSXx86+BIFwvUYLgAYSDacl3rvTFNcJRFLznteKDE1dPpXZqD6Zk3G8YEauce
UD8nMj/awJs5+kVXSEC30E8/cmbYkE284E5J2OQVsunrvCM/skx2SD90aMhCdbm4
B40h1EVwpOdH3alc3XIrTnNc0yK5MWAu41qwkxYxXHmW9Y0L8AjZve9JBrnKsJMB
ETEZFhHgi/IWtfh5PLbJO2dbSe7Nqo4ikyWo3r5b3yvuphFz1il88ZLjJ5nDmtlH
is7sk7pd0tYNsK1Di5G1ku50XvcbOE4F7mOVCxICTwjN+sdyG8o+AVlgbTKBo/JF
uEhthCECgYEA/3YXS9mAEujlstrV4VOksYWtySSrLHC56tLjj8cHVPJ1qkzT4OOC
X9TsWReDG4J8/t0DOHn+5dnhnqGcYjMMAQx095KHU1bQGrcRdmi6cjnNLTvfEbge
IcJTYG5P7NpLfLjB3DOGqFR4o0iz4K9ZLTYJc+BaCB9qJBEw6nuoP+sCgYEAwBTN
WpRDrmch0+LFPQwboLwtEPiFscTj8SInV0KsI/MK8+5Sm+tXS8PQHYJYcECEQxQM
2gfyM8vy33UP4yn4edJGWlaz7a4hyDxn944vv2fBQ3vjJTNz3X3skkhZ2/F+ZW9e
SFxPj+Vbif8VTEU+wK0f5SUmpRec4E7y3fq+kXkCgYEAib8ZbLLI1mlygfBx51/8
rCRSwuTcz8ew2CgCwGInV+ys+bkXfmnuwNHE531AGrNPxvVRaUCO602C1NB7zI+N
53raDyyZf5yN9fnElr592l3EfqGL9Lf8t2NbJeIVgrdqgMP29E9sSpPRwOnQ5FRo
l3JNwoe0xDB8QRpr7+PhoyUCgYEAp+GGmmR7wzLgnhDV00WB4DqYKP0N3RH5KAhx
2hKr4b/LEuh5y00mP1Il06TZJ0M8VmRv1yCa0CqxXB00hZdpVRAz7UFagaJwZFJn
jDb6BJDqmdDt9tXBrxUgb7pMz6+CiaWNAjGsWFheaX5JXyAmeMDX369Y13KL6oEW
RG2jogECgYEA/1vLZcWNK/0yd4ClU+Xbu8xC29q82JUMsaazHtbgSNlOfo9LMQlH
z6xBiMYfHZ/SiHCy9RsO8GD4caXiF0RsTVnhqjSRJf3EARamufelNsu2ApLclkSN
fzSoB7ZHddGaYKYpXkGzcwFcKd/QjAlHm1yIsZu4B52AhCxC/WS2X54=
-----END RSA PRIVATE KEY-----

27
tests/ssl_certs/proxy_cert.key Normal file
View file

@ -0,0 +1,27 @@
-----BEGIN RSA PRIVATE KEY-----
MIIEpAIBAAKCAQEAzZO36nZvb9wLxBNB2cZyHqcX5poChJd1YnFBtxbtQwiISxid
eGdiWImQE80vpUyTQbI7TxM+w1xZeEeu4PXuYrOgdTDRFEnjM2mteG+3WpHQBN4H
xoah0msp3046fMkYqcEvhvHbsc5DAWgLK4JFHQPtG/+CIH0ZY+lBBPQhFIhBLYkt
YxNVqwpsXOGreASSw6mO6cVehCuVFJQO5NnI1sCAvp3SeosMKeIcDZxpZWmZhSwH
n3Rj6RMNM66C8zG4YlpvIniGzgV4UiW8XrTUG8HmzQ2295IcfB4No2DZeJDSR9oq
jOkyqJXll+tSiAMuzBRtTQKvGZ5bpZWW4XELEQIDAQABAoIBAQCAfW2cjD4GimCI
QwkLlq9JXWLg7S3ZtdjWmLdcOmY9WZ3mYhI6aVPcxs5Ysgyvonb/vui2+e5mqNf7
B8LUNKK06lTGKqbjqXLqdYjJF/pgD3cXM7dkbE3EeNqJChogWIijwW11SMHqFmNn
A6LHpPqRshyHPWIV8FroSagr8nKio5BjUEuUiQUUAmSJPGN5qUhdIWXcQu8R1JB8
9qqqtwPR4FELbFVGI2vYHaSWGnf9V0boPOsfFXWbSq/Ksj3Lm3gAqMtlAeOFu84l
fhP9RkgeXfaCXq0VaOM83UDgLqXm4Ni4wAMKRLwNs4LzumqMM/dfUTn+mGncj33q
idp5qnDhAoGBAOXkwuf60F7aBbo98A0vWZli2CbkspsJz2J573pf+lVWI+ZHBZLI
MOM2DgCOEIUfa2TIMkwFr2t9x6uXlACEwFbEtEBpM4J5qUHgGtXZIsnTsv3qUg/C
L89cNrMddOuuRkxQbyK1QMYZZmZQjSKG2jW6m1KING+shtkOzQ/P9ildAoGBAOTs
DLyyPeEZPj1UMqxVNmeYYRfWnt+YyTPulOIbSuFN0DhZPNLsjrhSxvDwe/3sYH/p
nKdjnlFlx8frz9wtkCt0hWvY0pG2Zam4IBCvreFN7rSvpzHwUAK3oXic2TRKKu1m
xUPZqMJwnWAPX+XxGFn0m7UJj+95VTEOJ2d12ClFAoGAdexXMgmM8uqg/3yf8xNz
wWNbfu/W0gJBN8FWXw52aWmrNob9y+IWeaYTnqNAxBhuzR6H9kkAR4IYduNkzrNJ
ufhigZu1CVuAv8LF4SXlW2PVL7wPZff08Efb4xrcC7y0YJbtuv8Af90tkpQFIU3N
Brx2yeoGA7aa4SJfe5nwKh0CgYAo1yP+lh4MBqDf+CGCNUGbgcfwpM17PprGtQ3C
uPPG9kbrhqAfUSy1Ha94VK8KQh2FNHxKMK+R/gKCXEOdGFPcLNGQyAHpFQ1WFg9C
atUumOS5P40oj6L2mSQpjHIDrieyat9Ol4pQBh9Nf/Cv6S9a/RS6W5ZeNttIASpu
fsutsQKBgQCq+BFeDYJH4f+C1233W3PXM0P1ivj+9TJMRUP63RRay6rv2ZTZXyPc
Rx6Lv4OVWh9VMfv1kHRloJ1GKEBo/uD3nid1WqoNxpXv1iwxeGtjXkFHfvCB7Ruu
vTyQhJQQ7WSCJJOfarstusIn0udOG3MLRgG4X1pPQghyS1AT8NUglw==
-----END RSA PRIVATE KEY-----

0
tests/ssl_cert.crt → tests/ssl_certs/ssl_cert.crt
View file

0
tests/ssl_cert.key → tests/ssl_certs/ssl_cert.key
View file

30
tests/ssl_certs/ssl_cert_2.crt Normal file
View file

@ -0,0 +1,30 @@
-----BEGIN CERTIFICATE-----
MIIFOTCCA6GgAwIBAgIJAO+bbero+zKtMA0GCSqGSIb3DQEBCwUAMIGAMQswCQYD
VQQGEwJVUzERMA8GA1UECAwITmV3IFlvcmsxETAPBgNVBAcMCEJyb29rbHluMQww
CgYDVQQKDANOWVUxKTAnBgNVBAsMIENvbXB1dGVyIFNjaWVuY2UgYW5kIEVuZ2lu
ZWVyaW5nMRIwEAYDVQQDDAlsb2NhbGhvc3QwHhcNMTgwOTI2MTgwMDAzWhcNMzgw
OTIxMTgwMDAzWjCBgDELMAkGA1UEBhMCVVMxETAPBgNVBAgMCE5ldyBZb3JrMREw
DwYDVQQHDAhCcm9va2x5bjEMMAoGA1UECgwDTllVMSkwJwYDVQQLDCBDb21wdXRl
ciBTY2llbmNlIGFuZCBFbmdpbmVlcmluZzESMBAGA1UEAwwJbG9jYWxob3N0MIIB
ojANBgkqhkiG9w0BAQEFAAOCAY8AMIIBigKCAYEAxyFVeRsWnb1UlCKBks2azM9W
9K+J/ZkzdSb6eCxOIxv79M/Ug54CfWqkySSaQejsu0U/gJxkFYRvwQAy5lATrspY
2kyiWYiggWXFDWz+i8ETPkL9zn59v13sNIpT/IXQj0S3Mr9ZnsUn1qCyEOOIxJxZ
lyuV/M/XP1DP4tArhEvrex12V6MQIK+8fYzEjHG/W7vIIet+wTStIR8ArvVQi0Kv
PbbGCfrZ+e+gq+UpBLBuAfMzM95TW+YJ5duMchie2n6LDmOeegA4jMEv2ppeOr8Q
JJtZuKpXWVbJvLg81yrDjr1rAwJR/WQrnk8GQWPCyPLneAA4mJbi75LqjLxn0AoJ
b3kzLfGEMJJEWXspxNg06bLQU948hB4L7nKARq6s7KoESjEV+/L4koMPWJoNq6fx
OUVw2+S3ITNrDctecRQ1j3RGVPaj5l6bn03C7KV9uRrfqFY3OUjn7A0kDczvRnmr
e1BZIpe+mfGFB+Uu7JiQoBv6I6fqyrdH9rX1LUKlAgMBAAGjgbMwgbAwgZ8GA1Ud
IwSBlzCBlKGBhqSBgzCBgDELMAkGA1UEBhMCVVMxETAPBgNVBAgMCE5ldyBZb3Jr
MREwDwYDVQQHDAhCcm9va2x5bjEMMAoGA1UECgwDTllVMSkwJwYDVQQLDCBDb21w
dXRlciBTY2llbmNlIGFuZCBFbmdpbmVlcmluZzESMBAGA1UEAwwJbG9jYWxob3N0
ggkA75tt6uj7Mq0wDAYDVR0TBAUwAwEB/zANBgkqhkiG9w0BAQsFAAOCAYEAFWcl
1tAmt/3DJDjk0ppF62jbwcEOu1N9Nono9a70ojAQYYuMC7Ditw6rLbeXS8tP8ae/
drlci3VxlE5PpmAjuP67Uv2CuGu/2iMqa99AWZ4mVN+x4YL6awvYs8ea6I1Xe8tQ
5+RqvNA+QtnjtfOeb6yWQBAGrc2eTX87IzqvV/EewkdKAs4GZUWG1Zjv3effqjTO
qRX94ltW1GWud7fVcqpZLOaK9U+4IaI2nNHuCtWODoyQmMoVApXyig/YQqFe0eyj
76m1T+2SZLRtn0xn1fTHuLZ2bdtTMZ7k5PTAKnBNEn1Rr9MAS+WEASN1ZyoQ3reL
VYrgkMTrrXPO8bdDTvP7z1Jzv5Cq9WMHFvOLfnj/vN9ZPH6w4QT3Zb97SAAOSPK/
gzOzRtIe+hqCYBh/cwMoeeoAzes/nJgorj3IOTu8JXmtZrZGrdLIhu2Q8U+yKasf
+TUrr6xdcJI/fyVM5BVelpGhqHzzOQe1tO4VYQlAVaaVvFidDPHqTI2/S272
-----END CERTIFICATE-----

30
tests/ssl_certs/ssl_cert_expired.crt Normal file
View file

@ -0,0 +1,30 @@
-----BEGIN CERTIFICATE-----
MIIFOTCCA6GgAwIBAgIJALtyUsChEIJpMA0GCSqGSIb3DQEBCwUAMIGAMQswCQYD
VQQGEwJVUzERMA8GA1UECAwITmV3IFlvcmsxETAPBgNVBAcMCEJyb29rbHluMQww
CgYDVQQKDANOWVUxKTAnBgNVBAsMIENvbXB1dGVyIFNjaWVuY2UgYW5kIEVuZ2lu
ZWVyaW5nMRIwEAYDVQQDDAlsb2NhbGhvc3QwHhcNMTgwOTI2MTc0NTM2WhcNMTgw
OTI1MTc0NTM2WjCBgDELMAkGA1UEBhMCVVMxETAPBgNVBAgMCE5ldyBZb3JrMREw
DwYDVQQHDAhCcm9va2x5bjEMMAoGA1UECgwDTllVMSkwJwYDVQQLDCBDb21wdXRl
ciBTY2llbmNlIGFuZCBFbmdpbmVlcmluZzESMBAGA1UEAwwJbG9jYWxob3N0MIIB
ojANBgkqhkiG9w0BAQEFAAOCAY8AMIIBigKCAYEAxyFVeRsWnb1UlCKBks2azM9W
9K+J/ZkzdSb6eCxOIxv79M/Ug54CfWqkySSaQejsu0U/gJxkFYRvwQAy5lATrspY
2kyiWYiggWXFDWz+i8ETPkL9zn59v13sNIpT/IXQj0S3Mr9ZnsUn1qCyEOOIxJxZ
lyuV/M/XP1DP4tArhEvrex12V6MQIK+8fYzEjHG/W7vIIet+wTStIR8ArvVQi0Kv
PbbGCfrZ+e+gq+UpBLBuAfMzM95TW+YJ5duMchie2n6LDmOeegA4jMEv2ppeOr8Q
JJtZuKpXWVbJvLg81yrDjr1rAwJR/WQrnk8GQWPCyPLneAA4mJbi75LqjLxn0AoJ
b3kzLfGEMJJEWXspxNg06bLQU948hB4L7nKARq6s7KoESjEV+/L4koMPWJoNq6fx
OUVw2+S3ITNrDctecRQ1j3RGVPaj5l6bn03C7KV9uRrfqFY3OUjn7A0kDczvRnmr
e1BZIpe+mfGFB+Uu7JiQoBv6I6fqyrdH9rX1LUKlAgMBAAGjgbMwgbAwgZ8GA1Ud
IwSBlzCBlKGBhqSBgzCBgDELMAkGA1UEBhMCVVMxETAPBgNVBAgMCE5ldyBZb3Jr
MREwDwYDVQQHDAhCcm9va2x5bjEMMAoGA1UECgwDTllVMSkwJwYDVQQLDCBDb21w
dXRlciBTY2llbmNlIGFuZCBFbmdpbmVlcmluZzESMBAGA1UEAwwJbG9jYWxob3N0
ggkAu3JSwKEQgmkwDAYDVR0TBAUwAwEB/zANBgkqhkiG9w0BAQsFAAOCAYEAW4I1
TacdFv3L9ENFkSLciPb7zFMckLUZfk/P+4VjdapWrfuydO4W/ogMxA4DK09thTsK
N/BgcExyKjDldGUfUv57Tqv3v2E5kbygNcNtP53fwMz3y+7QourzkDE5HWciw1Lb
hmbnCBTzt/UioSBdJnAH29GWpSS+Jzu745sRaI48AS/J5ApH2aVEnNQTCE7v1LNH
2bTTPYl3eDXiD8yOhvyiW1F4y2BSFbQRH/3aE6Goe4A75m8sX50+JlOgjyyQnAMf
vbfvZsjGfqdXv9Qpci50qKCFxHJLXXNAUbX3fDgKE+RoZUNZnmn2VDgJYnToz6on
RcVnppV09kmSjHXZBT04XXUA0vG3p+oU0TO4puJlePVf4Oz23/DRCPHSfVWgMeB2
c1PpKit4+Bz7mypnsWVw8kk//l0GJ1cHnkkZElKJtPEB7I587jgTCDcN811TGNBc
rLLd/JwtYAvi1CPFt2ICGDvA4AKLY3rBNg5z1DrSE/iom1NTC00SFZJztYiX
-----END CERTIFICATE-----

31
tests/ssl_certs/ssl_cert_wronghost.crt Normal file
View file

@ -0,0 +1,31 @@
-----BEGIN CERTIFICATE-----
MIIFRTCCA62gAwIBAgIJAKY6b706lpuDMA0GCSqGSIb3DQEBCwUAMIGEMQswCQYD
VQQGEwJVUzERMA8GA1UECAwITmV3IFlvcmsxETAPBgNVBAcMCEJyb29rbHluMQww
CgYDVQQKDANOWVUxKTAnBgNVBAsMIENvbXB1dGVyIFNjaWVuY2UgYW5kIEVuZ2lu
ZWVyaW5nMRYwFAYDVQQDDA1ub3RteWhvc3RuYW1lMB4XDTE4MDkxMjE2NTkxN1oX
DTM4MDkwNzE2NTkxN1owgYQxCzAJBgNVBAYTAlVTMREwDwYDVQQIDAhOZXcgWW9y
azERMA8GA1UEBwwIQnJvb2tseW4xDDAKBgNVBAoMA05ZVTEpMCcGA1UECwwgQ29t
cHV0ZXIgU2NpZW5jZSBhbmQgRW5naW5lZXJpbmcxFjAUBgNVBAMMDW5vdG15aG9z
dG5hbWUwggGiMA0GCSqGSIb3DQEBAQUAA4IBjwAwggGKAoIBgQDHIVV5GxadvVSU
IoGSzZrMz1b0r4n9mTN1Jvp4LE4jG/v0z9SDngJ9aqTJJJpB6Oy7RT+AnGQVhG/B
ADLmUBOuyljaTKJZiKCBZcUNbP6LwRM+Qv3Ofn2/Xew0ilP8hdCPRLcyv1mexSfW
oLIQ44jEnFmXK5X8z9c/UM/i0CuES+t7HXZXoxAgr7x9jMSMcb9bu8gh637BNK0h
HwCu9VCLQq89tsYJ+tn576Cr5SkEsG4B8zMz3lNb5gnl24xyGJ7afosOY556ADiM
wS/aml46vxAkm1m4qldZVsm8uDzXKsOOvWsDAlH9ZCueTwZBY8LI8ud4ADiYluLv
kuqMvGfQCglveTMt8YQwkkRZeynE2DTpstBT3jyEHgvucoBGrqzsqgRKMRX78viS
gw9Ymg2rp/E5RXDb5LchM2sNy15xFDWPdEZU9qPmXpufTcLspX25Gt+oVjc5SOfs
DSQNzO9Geat7UFkil76Z8YUH5S7smJCgG/ojp+rKt0f2tfUtQqUCAwEAAaOBtzCB
tDCBowYDVR0jBIGbMIGYoYGKpIGHMIGEMQswCQYDVQQGEwJVUzERMA8GA1UECAwI
TmV3IFlvcmsxETAPBgNVBAcMCEJyb29rbHluMQwwCgYDVQQKDANOWVUxKTAnBgNV
BAsMIENvbXB1dGVyIFNjaWVuY2UgYW5kIEVuZ2luZWVyaW5nMRYwFAYDVQQDDA1u
b3RteWhvc3RuYW1lggkApjpvvTqWm4MwDAYDVR0TBAUwAwEB/zANBgkqhkiG9w0B
AQsFAAOCAYEAvpBMce3kxwo9W0o4RqezkSxnNyax0ezbUNodIkx5kbzX09qQLqhK
SkhQY3CNmtrpsczUg1W2nldxioEouwfTlhi15H98E/8XytpGaHO7Rnbtq8nkOp3E
N1+DMfFR95OynbHSd7bfK9UEmH1CmCnttvCuQkLTxDCpEsQNAxvmU/yDONoDr+cu
jGo80XTnYTqHl5/UtGbCS4SAIdWgrXTIqVvY/eF+mR+3nQEYjBuqW0cNfXLyYLXH
XMc6qtfGX1P+NRWtlrWgGQmc0fry+GczRHMJuKtJMV2xZzPJAJqwwvj3Fjz8HNGu
ZX3kVdbkDjf8is2cWgyZqDecqPHDBW4Ey539s/5eurgOkEvhriS4/9RnVhgdzduj
nRdXkD10ficrFcBQO0KaTWT+iFBc9duuYPuLRyRTye5p3t0liOikH2XrRXs4IBfz
2mT4npXQl1liNixcCf/yUEUOSQAJDG6aRjDjD4SZBUPDLjfqKLid8M0BpLQrks9L
5hAg1WZXorY6
-----END CERTIFICATE-----

227
tests/test_download.py
View file

@ -21,6 +21,8 @@
NOTE: Make sure test_download.py is ran in 'tuf/tests/' directory.
Otherwise, module that launches simple server would not be found.
TODO: Adopt the environment variable management from test_proxy_use.py here.
"""
# Help with Python 3 compatibility, where the print statement is a function, an
@ -36,6 +38,7 @@
import os
import random
import subprocess
import sys
import time
import unittest
@ -45,6 +48,8 @@
import tuf.unittest_toolbox as unittest_toolbox
import tuf.exceptions
import requests.exceptions
import securesystemslib
import six
@ -69,8 +74,7 @@ def setUp(self):
# Launch a SimpleHTTPServer (serves files in the current dir).
self.PORT = random.randint(30000, 45000)
command = ['python', 'simple_server.py', str(self.PORT)]
self.server_proc = subprocess.Popen(command, stderr=subprocess.PIPE)
self.server_proc = popen_python(['simple_server.py', str(self.PORT)])
logger.info('\n\tServer process started.')
logger.info('\tServer process id: '+str(self.server_proc.pid))
logger.info('\tServing on port: '+str(self.PORT))
@ -168,81 +172,201 @@ def test_download_url_to_tempfileobj_and_urls(self):
self.assertRaises(securesystemslib.exceptions.FormatError,
download_file, None, self.target_data_length)
self.assertRaises(securesystemslib.exceptions.FormatError,
self.assertRaises(tuf.exceptions.URLParsingError,
download_file,
self.random_string(), self.target_data_length)
self.assertRaises(six.moves.urllib.error.HTTPError,
self.assertRaises(requests.exceptions.HTTPError,
download_file,
'http://localhost:' + str(self.PORT) + '/' + self.random_string(),
self.target_data_length)
self.assertRaises(six.moves.urllib.error.URLError,
self.assertRaises(requests.exceptions.ConnectionError,
download_file,
'http://localhost:' + str(self.PORT+1) + '/' + self.random_string(),
self.target_data_length)
# Specify an unsupported URI scheme.
url_with_unsupported_uri = self.url.replace('http', 'file')
self.assertRaises(securesystemslib.exceptions.FormatError, download_file, url_with_unsupported_uri,
self.assertRaises(requests.exceptions.InvalidSchema, download_file, url_with_unsupported_uri,
self.target_data_length)
self.assertRaises(securesystemslib.exceptions.FormatError, unsafe_download_file,
self.assertRaises(requests.exceptions.InvalidSchema, unsafe_download_file,
url_with_unsupported_uri, self.target_data_length)
def test__get_opener(self):
# Test normal case.
# A simple https server should be used to test the rest of the optional
# ssl-related functions of 'tuf.download.py'.
fake_cacert = self.make_temp_data_file()
with open(fake_cacert, 'wt') as file_object:
file_object.write('fake cacert')
tuf.settings.ssl_certificates = fake_cacert
tuf.download._get_opener('https')
tuf.settings.ssl_certificates = None
'''
# This test uses sites on the internet, requiring a net connection to succeed.
# Since this is the only such test in TUF, I'm not going to enable it... but
# it's here in case it's useful for diagnosis.
def test_https_validation(self):
"""
Use some known URLs on the net to ensure that TUF download checks SSL
certificates appropriately.
"""
# We should never get as far as the target file download itself, so the
# length we pass to safe_download and unsafe_download shouldn't matter.
irrelevant_length = 10
for bad_url in [
'https://expired.badssl.com/', # expired certificate
'https://wrong.host.badssl.com/', ]: # hostname verification fail
with self.assertRaises(requests.exceptions.SSLError):
download.safe_download(bad_url, irrelevant_length)
with self.assertRaises(requests.exceptions.SSLError):
download.unsafe_download(bad_url, irrelevant_length)
'''
def test_https_connection(self):
"""
Try various HTTPS downloads using trusted and untrusted certificates with
and without the correct hostname listed in the SSL certificate.
"""
# Make a temporary file to be served to the client.
current_directory = os.getcwd()
target_filepath = self.make_temp_data_file(directory=current_directory)
target_data = None
target_data_length = 0
with open(target_filepath, 'r') as target_file_object:
target_data = target_file_object.read()
target_data_length = len(target_data)
target_data_length = len(target_file_object.read())
# Launch an https server (serves files in the current dir).
port = random.randint(30000, 45000)
command = ['python', 'simple_https_server.py', str(port)]
https_server_process = subprocess.Popen(command, stderr=subprocess.PIPE)
# These cert files provide various test cases:
# good: A valid cert from an older generation of test_download.py tests.
# good2: A valid cert made simultaneous to the bad certs below, with the
# same settings otherwise, tested here in case the difference
# between the way the new bad certs and the old good cert were
# generated turns out to matter at some point.
# bad: An otherwise-valid cert with the wrong hostname. The good certs
# list "localhost", but this lists "notmyhostname".
# expired: An otherwise-valid cert but which is expired (no valid dates
# exist, fwiw: startdate > enddate).
good_cert_fname = os.path.join('ssl_certs', 'ssl_cert.crt')
good2_cert_fname = os.path.join('ssl_certs', 'ssl_cert_2.crt')
bad_cert_fname = os.path.join('ssl_certs', 'ssl_cert_wronghost.crt')
expired_cert_fname = os.path.join('ssl_certs', 'ssl_cert_expired.crt')
# NOTE: Following error is raised if delay is not applied:
# <urlopen error [Errno 111] Connection refused>
time.sleep(1)
# Launch four HTTPS servers (serve files in the current dir).
# 1: we expect to operate correctly
# 2: also good; uses a slightly different cert (controls for the cert
# generation method used for the next two, in case it comes to matter)
# 3: run with an HTTPS certificate with an unexpected hostname
# 4: run with an HTTPS certificate that is expired
port1 = str(random.randint(30000, 45000))
port2 = str(int(port1) + 1)
port3 = str(int(port1) + 2)
port4 = str(int(port1) + 3)
good_https_server_proc = popen_python(
['simple_https_server.py', port1, good_cert_fname])
good2_https_server_proc = popen_python(
['simple_https_server.py', port2, good2_cert_fname])
bad_https_server_proc = popen_python(
['simple_https_server.py', port3, bad_cert_fname])
expd_https_server_proc = popen_python(
['simple_https_server.py', port4, expired_cert_fname])
junk, relative_target_filepath = os.path.split(target_filepath)
https_url = 'https://localhost:' + str(port) + '/' + relative_target_filepath
# Provide a delay long enough to allow the HTTPS servers to start.
# Encountered an error on one test system at delay value of 0.2s, so
# increasing to 0.5s.
# Expect to see "Connection refused" if this delay is not long enough
# (though other issues could cause that).
time.sleep(0.5)
# Download the target file using an https connection.
tuf.settings.ssl_certificates = 'ssl_cert.crt'
message = 'Downloading target file from https server: ' + https_url
logger.info(message)
relative_target_fpath = os.path.basename(target_filepath)
good_https_url = 'https://localhost:' + port1 + '/' + relative_target_fpath
good2_https_url = good_https_url.replace(':' + port1, ':' + port2)
bad_https_url = good_https_url.replace(':' + port1, ':' + port3)
expired_https_url = good_https_url.replace(':' + port1, ':' + port4)
# Download the target file using an HTTPS connection.
# Use try-finally solely to ensure that the server processes are killed.
try:
download.safe_download(https_url, target_data_length)
download.unsafe_download(https_url, target_data_length)
# Trust the certfile that happens to use a different hostname than we
# will expect.
os.environ['REQUESTS_CA_BUNDLE'] = bad_cert_fname
# Clear sessions to ensure that the certificate we just specified is used.
# TODO: Confirm necessity of this session clearing and lay out mechanics.
tuf.download._sessions = {}
# Try connecting to the server process with the bad cert while trusting
# the bad cert. Expect failure because even though we trust it, the
# hostname we're connecting to does not match the hostname in the cert.
logger.info('Trying HTTPS download of target file: ' + bad_https_url)
with self.assertRaises(requests.exceptions.SSLError):
download.safe_download(bad_https_url, target_data_length)
with self.assertRaises(requests.exceptions.SSLError):
download.unsafe_download(bad_https_url, target_data_length)
# Try connecting to the server processes with the good certs while not
# trusting the good certs (trusting the bad cert instead). Expect failure
# because even though the server's cert file is otherwise OK, we don't
# trust it.
print('Trying HTTPS download of target file: ' + good_https_url)
with self.assertRaises(requests.exceptions.SSLError):
download.safe_download(good_https_url, target_data_length)
with self.assertRaises(requests.exceptions.SSLError):
download.unsafe_download(good_https_url, target_data_length)
print('Trying HTTPS download of target file: ' + good2_https_url)
with self.assertRaises(requests.exceptions.SSLError):
download.safe_download(good2_https_url, target_data_length)
with self.assertRaises(requests.exceptions.SSLError):
download.unsafe_download(good2_https_url, target_data_length)
# Configure environment to now trust the certfile that is expired.
os.environ['REQUESTS_CA_BUNDLE'] = expired_cert_fname
# Clear sessions to ensure that the certificate we just specified is used.
# TODO: Confirm necessity of this session clearing and lay out mechanics.
tuf.download._sessions = {}
# Try connecting to the server process with the expired cert while
# trusting the expired cert. Expect failure because even though we trust
# it, it is expired.
logger.info('Trying HTTPS download of target file: ' + expired_https_url)
with self.assertRaises(requests.exceptions.SSLError):
download.safe_download(expired_https_url, target_data_length)
with self.assertRaises(requests.exceptions.SSLError):
download.unsafe_download(expired_https_url, target_data_length)
# Try connecting to the server processes with the good certs while
# trusting the appropriate good certs. Expect success.
# TODO: expand testing to switch expected certificates back and forth a
# bit more while clearing / not clearing sessions.
os.environ['REQUESTS_CA_BUNDLE'] = good_cert_fname
# Clear sessions to ensure that the certificate we just specified is used.
# TODO: Confirm necessity of this session clearing and lay out mechanics.
tuf.download._sessions = {}
logger.info('Trying HTTPS download of target file: ' + good_https_url)
download.safe_download(good_https_url, target_data_length)
download.unsafe_download(good_https_url, target_data_length)
os.environ['REQUESTS_CA_BUNDLE'] = good2_cert_fname
# Clear sessions to ensure that the certificate we just specified is used.
# TODO: Confirm necessity of this session clearing and lay out mechanics.
tuf.download._sessions = {}
logger.info('Trying HTTPS download of target file: ' + good2_https_url)
download.safe_download(good2_https_url, target_data_length)
download.unsafe_download(good2_https_url, target_data_length)
finally:
if https_server_process.returncode is None:
message = \
'Server process ' + str(https_server_process.pid) + ' terminated.'
logger.info(message)
https_server_process.kill()
for proc in [
good_https_server_proc,
good2_https_server_proc,
bad_https_server_proc,
expd_https_server_proc]:
if proc.returncode is None:
logger.info('Terminating server process ' + str(proc.pid))
proc.kill()
@ -252,6 +376,27 @@ def test__get_content_length(self):
self.assertEqual(content_length, None)
# TODO: Move this to a common test module (tests/common.py?)
# and strip it test_proxy_use.py and test_download.py.
def popen_python(command_arg_list):
"""
Run subprocess.Popen() to produce a process running a Python interpreter.
Uses the same Python interpreter that the current process is using, via
sys.executable.
"""
assert sys.executable, 'Test cannot function: unable to determine ' \
'current Python interpreter via sys.executable.'
return subprocess.Popen(
[sys.executable] + command_arg_list, stderr=subprocess.PIPE)
# Run unit test.
if __name__ == '__main__':
unittest.main()

385
tests/test_proxy_use.py Normal file
View file

@ -0,0 +1,385 @@
#!/usr/bin/env python
# Copyright 2018, New York University and the TUF contributors
# SPDX-License-Identifier: MIT OR Apache-2.0
"""
<Program>
test_proxy_use.py
<Copyright>
See LICENSE-MIT OR LICENSE for licensing information.
<Purpose>
Integration/regression test of TUF downloads through proxies.
NOTE: Make sure test_proxy_use.py is run in 'tuf/tests/' directory.
Otherwise, test data or scripts may not be found.
THIS module requires Python3.6+, as it uses mitmproxy, which only supports
Python3.6+.
So long as the tests succeed in Python 3.6+, it is unlikely that TUF
behaves differently with respect to proxies when it runs in other Python
versions.
As a result of this dependency on a modern Python version, this test does
not run with the other unit tests, being specifically excluded from
aggregate_tests.py.
"""
# Help with Python 3 compatibility, where the print statement is a function, an
# implicit relative import is invalid, and the '/' operator performs true
# division. Example: print 'hello world' raises a 'SyntaxError' exception.
from __future__ import print_function
from __future__ import absolute_import
from __future__ import division
from __future__ import unicode_literals
import hashlib
import logging
import os
import random
import subprocess
import sys
import time
import unittest
import tuf
import tuf.download as download
import tuf.log
import tuf.unittest_toolbox as unittest_toolbox
import tuf.exceptions
import requests.exceptions
import securesystemslib
import six
logger = logging.getLogger('tuf.test_download')
class TestWithProxies(unittest_toolbox.Modified_TestCase):
@classmethod
def setUpClass(cls):
"""
Setup performed before the first test function (TestWithProxies class
method) runs.
Launch HTTP, HTTPS, and proxy servers in the current working directory.
We'll set up four servers:
- HTTP server (simple_server.py)
- HTTPS server (simple_https_server.py)
- HTTP proxy server (proxy_server.py)
(that supports HTTP CONNECT to funnel HTTPS connections)
- HTTPS proxy server (proxy_server.py)
(trusted by the client to intercept and resign connections)
"""
unittest_toolbox.Modified_TestCase.setUpClass()
# Launch a simple HTTP server (serves files in the current dir).
cls.http_port = random.randint(30000, 45000)
cls.http_server_proc = popen_python(
['simple_server.py', str(cls.http_port)])
# Launch an HTTPS server (serves files in the current dir).
cls.https_port = cls.http_port + 1
cls.https_server_proc = popen_python(
['simple_https_server.py', str(cls.https_port)])
# Launch an HTTP proxy server derived from inaz2/proxy2.
# This one is able to handle HTTP CONNECT requests, and so can pass HTTPS
# requests on to the target server.
cls.http_proxy_port = cls.http_port + 2
cls.http_proxy_proc = popen_python(
['proxy_server.py', str(cls.http_proxy_port)])
# Note that the HTTP proxy server's address uses http://, regardless of the
# type of connection used with the target server.
cls.http_proxy_addr = 'http://127.0.0.1:' + str(cls.http_proxy_port)
# Launch an HTTPS proxy server, also derived from inaz2/proxy2.
# (An HTTPS proxy performs its own TLS connection with the client and must
# be trusted by it, and is capable of tampering.)
# We instruct the proxy server to expect certain certificates from the
# target server.
# 1st arg: port
# 2nd arg: whether to intercept (HTTPS proxy) or relay (TCP tunnel using
# HTTP CONNECT verb, to facilitate an HTTPS connection between the client
# and server which the proxy cannot inspect)
# 3rd arg: (optional) certificate file for telling the proxy what target
# server certs to accept in its HTTPS connection to the target server.
# This is only relevant if the proxy is in intercept mode.
cls.https_proxy_port = cls.http_port + 3
cls.https_proxy_proc = popen_python(
['proxy_server.py', str(cls.https_proxy_port), 'intercept',
os.path.join('ssl_certs', 'ssl_cert.crt')])
# Note that the HTTPS proxy server's address uses https://, regardless of
# the type of connection used with the target server.
cls.https_proxy_addr = 'https://127.0.0.1:' + str(cls.https_proxy_port)
# Give the HTTP server and proxy server processes a little bit of time to
# start listening before allowing tests to begin, lest we get "Connection
# refused" errors. On the first test system. 0.1s was too short and 0.15s
# was long enough. Use 0.5s to be safe, and if issues arise, increase it.
# Observed some occasional AppVeyor failures, so increasing this to 1s.
time.sleep(1)
@classmethod
def tearDownClass(cls):
"""
Cleanup performed after the last of the tests (TestWithProxies methods)
has been run.
Stop server process and perform clean up.
"""
unittest_toolbox.Modified_TestCase.tearDownClass()
for proc in [
cls.http_server_proc,
cls.https_server_proc,
cls.http_proxy_proc,
cls.https_proxy_proc,
]:
if proc.returncode is None:
logger.info('\tTerminating process ' + str(proc.pid) + ' in cleanup.')
proc.kill()
def setUp(self):
"""
Setup performed before EACH test function (TestWithProxies class method)
runs.
"""
unittest_toolbox.Modified_TestCase.setUp(self)
# Dictionary for saving environment values to restore.
self.old_env_values = {}
# Make a temporary file to serve on the server, and determine its length,
# and its url on the server.
current_dir = os.getcwd()
target_filepath = self.make_temp_data_file(directory=current_dir)
rel_target_filepath = os.path.basename(target_filepath)
with open(target_filepath, 'r') as target_file_object:
self.target_data_length = len(target_file_object.read())
self.url = \
'http://localhost:' + str(self.http_port) + '/' + rel_target_filepath
self.url_https = \
'https://localhost:' + str(self.https_port) + '/' + rel_target_filepath
def tearDown(self):
"""
Cleanup performed after each test (each TestWithProxies method).
Reset environment variables (for next test, etc.).
"""
unittest_toolbox.Modified_TestCase.tearDown(self)
self.restore_all_modified_env_values()
def test_baseline_no_proxy(self):
"""
Test a length-validating TUF download of a file through a proxy. Use an
HTTP proxy, and perform an HTTP connection with the final server.
"""
logger.info('Trying HTTP download with no proxy: ' + self.url)
download.safe_download(self.url, self.target_data_length)
download.unsafe_download(self.url, self.target_data_length)
def test_http_dl_via_smart_http_proxy(self):
"""
Test a length-validating TUF download of a file through a proxy. Use an
HTTP proxy normally, and make an HTTP connection with the final server.
"""
self.set_env_value('HTTP_PROXY', self.http_proxy_addr)
logger.info('Trying HTTP download via HTTP proxy: ' + self.url)
download.safe_download(self.url, self.target_data_length)
download.unsafe_download(self.url, self.target_data_length)
def test_https_dl_via_smart_http_proxy(self):
"""
Test a length-validating TUF download of a file through a proxy. Use an
HTTP proxy that supports HTTP CONNECT (which essentially causes it to act
as a TCP proxy), and perform an HTTPS connection through with the final
server.
Note that the proxy address is still http://... even though the connection
with the target server is an HTTPS connection. The proxy itself will act as
a TCP proxy via HTTP CONNECT.
"""
self.set_env_value('HTTP_PROXY', self.http_proxy_addr) # http as intended
self.set_env_value('HTTPS_PROXY', self.http_proxy_addr) # http as intended
self.set_env_value('REQUESTS_CA_BUNDLE',
os.path.join('ssl_certs', 'ssl_cert.crt'))
# Clear sessions to ensure that the certificate we just specified is used.
# TODO: Confirm necessity of this session clearing and lay out mechanics.
tuf.download._sessions = {}
logger.info('Trying HTTPS download via HTTP proxy: ' + self.url_https)
download.safe_download(self.url_https, self.target_data_length)
download.unsafe_download(self.url_https, self.target_data_length)
def test_http_dl_via_https_proxy(self):
"""
Test a length-validating TUF download of a file through a proxy. Use an
HTTPS proxy, and perform an HTTP connection with the final server.
"""
self.set_env_value('HTTP_PROXY', self.https_proxy_addr)
self.set_env_value('HTTPS_PROXY', self.https_proxy_addr) # unnecessary
# We're making an HTTPS connection with the proxy. The proxy will make a
# plain HTTP connection to the target server.
self.set_env_value('REQUESTS_CA_BUNDLE',
os.path.join('ssl_certs', 'proxy_ca.crt'))
# Clear sessions to ensure that the certificate we just specified is used.
# TODO: Confirm necessity of this session clearing and lay out mechanics.
tuf.download._sessions = {}
logger.info('Trying HTTP download via HTTPS proxy: ' + self.url_https)
download.safe_download(self.url, self.target_data_length)
download.unsafe_download(self.url, self.target_data_length)
def test_https_dl_via_https_proxy(self):
"""
Test a length-validating TUF download of a file through a proxy. Use an
HTTPS proxy, and perform an HTTPS connection with the final server.
"""
self.set_env_value('HTTP_PROXY', self.https_proxy_addr) # unnecessary
self.set_env_value('HTTPS_PROXY', self.https_proxy_addr)
# We're making an HTTPS connection with the proxy. The proxy will make its
# own HTTPS connection with the target server, and will have to know what
# certificate to trust. It was told what certs to trust when it was
# started in setUpClass().
self.set_env_value('REQUESTS_CA_BUNDLE',
os.path.join('ssl_certs', 'proxy_ca.crt'))
# Clear sessions to ensure that the certificate we just specified is used.
# TODO: Confirm necessity of this session clearing and lay out mechanics.
tuf.download._sessions = {}
logger.info('Trying HTTPS download via HTTPS proxy: ' + self.url_https)
download.safe_download(self.url_https, self.target_data_length)
download.unsafe_download(self.url_https, self.target_data_length)
def set_env_value(self, key, value):
"""
Set an environment variable after noting what the original value was, if it
was set, and add it to the queue for restoring to its original value / lack
of a value after the test finishes.
Safe for multiple uses in one test: does not overwrite original saved value
with new saved values.
"""
# Only save the current value if we have not previously saved an older
# value. The original one is the one we'll restore to, not whatever we
# most recently overwrote.
if key not in self.old_env_values:
# If the value was previously unset in os.environ, save the old value
# as None so that we know to unset it.
self.old_env_values[key] = os.environ.get(key, None)
# Actually set the new value.
os.environ[key] = value
def restore_env_value(self, key):
# Save old values for environment variables for restoration after the test.
# Save the pre-existing value of the environment variables HTTP_PROXY and
# HTTPS_PROXY so that we can restore them in tearDown() after the test.
# If the value was not originally set at all, we'll try to unset it again,
# too.
assert key in self.old_env_values, 'Test coding mistake: something is ' \
'trying to restore environment variable ' + key + ', but that ' \
'variable does not appear in the list of values to restore. ' \
'Please make sure to use set_env_value().'
if self.old_env_values[key] is None:
# If it was not previously set, try to unset it.
# If the platform provides a way to unset environment variables,
# del os.environ[key] should unset the variable. Otherwise, we'll just
# have to settle for setting it to an empty string.
# See os.environ in:
# https://docs.python.org/2/library/os.html#process-parameters
os.environ[key] = ''
del os.environ[key]
else:
# If it was previously set, restore the original value from when the
# test was being set up.
os.environ[key] = self.old_env_values[key]
def restore_all_modified_env_values(self):
for key in self.old_env_values:
self.restore_env_value(key)
# TODO: Move this to a common test module (tests/common.py?)
# and strip it test_proxy_use.py and test_download.py.
def popen_python(command_arg_list):
"""
Run subprocess.Popen() to produce a process running a Python interpreter.
Uses the same Python interpreter that the current process is using, via
sys.executable.
"""
assert sys.executable, 'Test cannot function: unable to determine ' \
'current Python interpreter via sys.executable.'
return subprocess.Popen(
[sys.executable] + command_arg_list, stderr=subprocess.PIPE)
# Run unit test.
if __name__ == '__main__':
unittest.main()

38
tests/test_slow_retrieval_attack.py
View file

@ -31,6 +31,9 @@
does not fall below a required rate (tuf.settings.MIN_AVERAGE_DOWNLOAD_SPEED).
Note: There is no difference between 'updates' and 'target' files.
# TODO: Consider additional tests for slow metadata download. Tests here only
use slow target download.
"""
# Help with Python 3 compatibility, where the print statement is a function, an
@ -154,13 +157,20 @@ def setUp(self):
shutil.copytree(original_client, self.client_directory)
shutil.copytree(original_keystore, self.keystore_directory)
# Produce a longer target file than exists in the other test repository
# data, to provide for a long-duration slow attack. Then we'll write new
# top-level metadata that includes a hash over that file, and provide that
# metadata to the client as well.
# The slow retrieval server, in mode 2 (1 byte per second), will only
# sleep for a total of (target file size) seconds. Add a target file
# that contains sufficient number of bytes to trigger a slow retrieval
# error. "sufficient number of bytes" assumed to be
# >> 'tuf.settings.SLOW_START_GRACE_PERIOD' bytes.
extra_bytes = 8
total_bytes = tuf.settings.SLOW_START_GRACE_PERIOD + extra_bytes
# error. A transfer should not be permitted to take 1 second per byte
# transferred. Because this test is currently expected to fail, I'm
# limiting the size to 10 bytes (10 seconds) to avoid expected testing
# delays.... Consider increasing again after fix, to, e.g. 400.
total_bytes = 10
repository = repo_tool.load_repository(self.repository_directory)
file1_filepath = os.path.join(self.repository_directory, 'targets',
@ -190,6 +200,19 @@ def setUp(self):
shutil.copytree(os.path.join(self.repository_directory, 'metadata.staged'),
os.path.join(self.repository_directory, 'metadata'))
# Since we've changed the repository metadata in this setup (by lengthening
# a target file and then writing new metadata), we also have to update the
# client metadata to get to the expected initial state, where the client
# knows the right target info (and so expects the right, longer target
# length.
# We'll skip using updater.refresh since we don't have a server running,
# and we'll update the metadata locally, manually.
shutil.rmtree(os.path.join(
self.client_directory, self.repository_name, 'metadata', 'current'))
shutil.copytree(os.path.join(self.repository_directory, 'metadata'),
os.path.join(self.client_directory, self.repository_name, 'metadata',
'current'))
# Set the url prefix required by the 'tuf/client/updater.py' updater.
# 'path/to/tmp/repository' -> 'localhost:8001/tmp/repository'.
repository_basepath = self.repository_directory[len(os.getcwd()):]
@ -252,6 +275,13 @@ def test_with_tuf_mode_1(self):
# The following test fails as a result of a change to TUF's download code.
# Rather than constructing urllib2 requests, we now use the requests library.
# This solves an HTTPS proxy issue, but has for the moment deprived us of a
# way to prevent certain this kind of slow retrieval attack.
# See conversation in PR: https://github.com/theupdateframework/tuf/pull/781
# TODO: Update download code to resolve the slow retrieval vulnerability.
@unittest.expectedFailure
def test_with_tuf_mode_2(self):
# Simulate a slow retrieval attack.
# 'mode_2': During the download process, the server blocks the download

5
tuf/__init__.py
View file

@ -0,0 +1,5 @@
# This value is used in the requests user agent.
# setup.py has it hard-coded separately.
# Currently, when the version is changed, it must be set in both locations.
# TODO: Single-source the version number.
__version__ = "0.11.1"

351
tuf/download.py
View file

@ -34,30 +34,37 @@
from __future__ import division
from __future__ import unicode_literals
import os
import socket
import logging
import time
import timeit
import ssl
import tuf
import requests
import securesystemslib
import securesystemslib.util
import six
import tuf.exceptions
# 'ssl.match_hostname' was added in Python 3.2. The vendored version is needed
# for Python 2.7.
try:
from ssl import match_hostname
except ImportError: # pragma: no cover
from securesystemslib._vendor.ssl_match_hostname import match_hostname
import urllib3.exceptions
# See 'log.py' to learn how logging is handled in TUF.
logger = logging.getLogger('tuf.download')
# From http://docs.python-requests.org/en/master/user/advanced/#session-objects:
#
# "The Session object allows you to persist certain parameters across requests.
# It also persists cookies across all requests made from the Session instance,
# and will use urllib3's connection pooling. So if you're making several
# requests to the same host, the underlying TCP connection will be reused,
# which can result in a significant performance increase (see HTTP persistent
# connection)."
#
# NOTE: We use a separate requests.Session per scheme+hostname combination, in
# order to reuse connections to the same hostname to improve efficiency, but
# avoiding sharing state between different hosts-scheme combinations to
# minimize subtle security issues. Some cookies may not be HTTP-safe.
_sessions = {}
def safe_download(url, required_length):
@ -76,8 +83,7 @@ def safe_download(url, required_length):
<Arguments>
url:
A URL string that represents the location of the file. The URI scheme
component must be one of 'tuf.settings.SUPPORTED_URI_SCHEMES'.
A URL string that represents the location of the file.
required_length:
An integer value representing the length of the file. This is an exact
@ -106,20 +112,6 @@ def safe_download(url, required_length):
securesystemslib.formats.URL_SCHEMA.check_match(url)
securesystemslib.formats.LENGTH_SCHEMA.check_match(required_length)
# Ensure 'url' specifies one of the URI schemes in
# 'tuf.settings.SUPPORTED_URI_SCHEMES'. Be default, ['http', 'https'] is
# supported. If the URI scheme of 'url' is empty or "file", files on the
# local system can be accessed. Unexpected files may be accessed by
# compromised metadata (unlikely to happen if targets.json metadata is signed
# with offline keys).
parsed_url = six.moves.urllib.parse.urlparse(url)
if parsed_url.scheme not in tuf.settings.SUPPORTED_URI_SCHEMES:
message = \
repr(url) + ' specifies an unsupported URI scheme. Supported ' + \
' URI Schemes: ' + repr(tuf.settings.SUPPORTED_URI_SCHEMES)
raise securesystemslib.exceptions.FormatError(message)
return _download_file(url, required_length, STRICT_REQUIRED_LENGTH=True)
@ -142,8 +134,7 @@ def unsafe_download(url, required_length):
<Arguments>
url:
A URL string that represents the location of the file. The URI scheme
component must be one of 'tuf.settings.SUPPORTED_URI_SCHEMES'.
A URL string that represents the location of the file.
required_length:
An integer value representing the length of the file. This is an upper
@ -172,20 +163,6 @@ def unsafe_download(url, required_length):
securesystemslib.formats.URL_SCHEMA.check_match(url)
securesystemslib.formats.LENGTH_SCHEMA.check_match(required_length)
# Ensure 'url' specifies one of the URI schemes in
# 'tuf.settings.SUPPORTED_URI_SCHEMES'. Be default, ['http', 'https'] is
# supported. If the URI scheme of 'url' is empty or "file", files on the
# local system can be accessed. Unexpected files may be accessed by
# compromised metadata (unlikely to happen if targets.json metadata is signed
# with offline keys).
parsed_url = six.moves.urllib.parse.urlparse(url)
if parsed_url.scheme not in tuf.settings.SUPPORTED_URI_SCHEMES:
message = \
repr(url) + ' specifies an unsupported URI scheme. Supported ' + \
' URI Schemes: ' + repr(tuf.settings.SUPPORTED_URI_SCHEMES)
raise securesystemslib.exceptions.FormatError(message)
return _download_file(url, required_length, STRICT_REQUIRED_LENGTH=False)
@ -229,9 +206,6 @@ def _download_file(url, required_length, STRICT_REQUIRED_LENGTH=True):
securesystemslib.exceptions.FormatError, if any of the arguments are
improperly formatted.
tuf.exceptions.Error, if the certificates pointed to by
tuf.settings.ssl_certificates cannot be loaded.
Any other unforeseen runtime exception.
<Returns>
@ -257,14 +231,62 @@ def _download_file(url, required_length, STRICT_REQUIRED_LENGTH=True):
temp_file = securesystemslib.util.TempFile()
try:
# Open the connection to the remote file. _open_connection() can raise
# socket connection exceptions (such as SSLError). Connection errors of
# this kind can be minimized by adjusting the socket timeout in
# settings.py.
connection = _open_connection(url)
# Use a different requests.Session per schema+hostname combination, to
# reuse connections while minimizing subtle security issues.
parsed_url = six.moves.urllib.parse.urlparse(url)
if not parsed_url.scheme or not parsed_url.hostname:
raise tuf.exceptions.URLParsingError(
'Could not get scheme and hostname from URL: ' + url)
session_index = parsed_url.scheme + '+' + parsed_url.hostname
logger.debug('url: ' + url)
logger.debug('session index: ' + session_index)
session = _sessions.get(session_index)
if not session:
session = requests.Session()
_sessions[session_index] = session
# Attach some default headers to every Session.
requests_user_agent = session.headers['User-Agent']
# Follows the RFC: https://tools.ietf.org/html/rfc7231#section-5.5.3
tuf_user_agent = 'tuf/' + tuf.__version__ + ' ' + requests_user_agent
session.headers.update({
# Tell the server not to compress or modify anything.
# https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Accept-Encoding#Directives
'Accept-Encoding': 'identity',
# The TUF user agent.
'User-Agent': tuf_user_agent})
logger.debug('Made new session for ' + session_index)
else:
logger.debug('Reusing session for ' + session_index)
# Get the requests.Response object for this URL.
#
# Always stream to control how requests are downloaded:
# http://docs.python-requests.org/en/master/user/advanced/#body-content-workflow
#
# We will always manually close Responses, so no need for a context
# manager.
#
# Always set the timeout. This timeout value is interpreted by requests as:
# - connect timeout (max delay before first byte is received)
# - read (gap) timeout (max delay between bytes received)
# These are NOT overall/total, wall-clock timeouts for any single read.
# http://docs.python-requests.org/en/master/user/advanced/#timeouts
response = session.get(
url, stream=True, timeout=tuf.settings.SOCKET_TIMEOUT)
# Check response status.
response.raise_for_status()
# We ask the server about how big it thinks this file should be.
reported_length = _get_content_length(connection)
reported_length = _get_content_length(response)
# Then, we check whether the required length matches the reported length.
_check_content_length(reported_length, required_length,
@ -273,7 +295,7 @@ def _download_file(url, required_length, STRICT_REQUIRED_LENGTH=True):
# Download the contents of the URL, up to the required length, to a
# temporary file, and get the total number of downloaded bytes.
total_downloaded, average_download_speed = \
_download_fixed_amount_of_data(connection, temp_file, required_length)
_download_fixed_amount_of_data(response, temp_file, required_length)
# Does the total number of downloaded bytes match the required length?
_check_downloaded_length(total_downloaded, required_length,
@ -293,21 +315,20 @@ def _download_file(url, required_length, STRICT_REQUIRED_LENGTH=True):
def _download_fixed_amount_of_data(connection, temp_file, required_length):
def _download_fixed_amount_of_data(response, temp_file, required_length):
"""
<Purpose>
This is a helper function, where the download really happens. While-block
reads data from connection a fixed chunk of data at a time, or less, until
reads data from response a fixed chunk of data at a time, or less, until
'required_length' is reached.
<Arguments>
connection:
The object that the _open_connection returns for communicating with the
server about the contents of a URL.
response:
The object for communicating with the server about the contents of a URL.
temp_file:
A temporary file where the contents at the URL specified by the
'connection' object will be stored.
'response' object will be stored.
required_length:
The number of bytes that we must download for the file. This is almost
@ -319,7 +340,11 @@ def _download_fixed_amount_of_data(connection, temp_file, required_length):
Data from the server will be written to 'temp_file'.
<Exceptions>
Runtime or network exceptions will be raised without question.
tuf.exceptions.SlowRetrievalError
will be raised if urllib3.exceptions.ReadTimeoutError is caught (if the
download times out).
Otherwise, runtime or network exceptions will be raised without question.
<Returns>
A (total_downloaded, average_download_speed) tuple, where
@ -328,12 +353,6 @@ def _download_fixed_amount_of_data(connection, temp_file, required_length):
attempt.
"""
# Tolerate servers with a slow start by ignoring their delivery speed for
# 'tuf.settings.SLOW_START_GRACE_PERIOD' seconds. Set 'seconds_spent_receiving'
# to negative SLOW_START_GRACE_PERIOD seconds, and begin checking the average
# download speed once it is positive.
grace_period = -tuf.settings.SLOW_START_GRACE_PERIOD
# Keep track of total bytes downloaded.
number_of_bytes_received = 0
average_download_speed = 0
@ -350,20 +369,17 @@ def _download_fixed_amount_of_data(connection, temp_file, required_length):
if tuf.settings.SLEEP_BEFORE_ROUND:
time.sleep(tuf.settings.SLEEP_BEFORE_ROUND)
data = b''
read_amount = min(tuf.settings.CHUNK_SIZE,
required_length - number_of_bytes_received)
read_amount = min(
tuf.settings.CHUNK_SIZE, required_length - number_of_bytes_received)
try:
data = connection.read(read_amount)
# Python 3.2 returns 'IOError' if the remote file object has timed out.
except (socket.error, IOError):
pass
# NOTE: This may not handle some servers adding a Content-Encoding
# header, which may cause urllib3 to misbehave:
# https://github.com/pypa/pip/blob/404838abcca467648180b358598c597b74d568c9/src/pip/_internal/download.py#L547-L582
data = response.raw.read(read_amount)
number_of_bytes_received = number_of_bytes_received + len(data)
# Data successfully read from the connection. Store it.
# Data successfully read from the response. Store it.
temp_file.write(data)
if number_of_bytes_received == required_length:
@ -372,9 +388,6 @@ def _download_fixed_amount_of_data(connection, temp_file, required_length):
stop_time = timeit.default_timer()
seconds_spent_receiving = stop_time - start_time
if (seconds_spent_receiving + grace_period) < 0:
continue
# Measure the average download speed.
average_download_speed = number_of_bytes_received / seconds_spent_receiving
@ -395,121 +408,31 @@ def _download_fixed_amount_of_data(connection, temp_file, required_length):
# Finally, we signal that the download is complete.
break
except urllib3.exceptions.ReadTimeoutError as e:
# Whatever happens, make sure that we always close the connection.
response.close()
raise tuf.exceptions.SlowRetrievalError(str(e))
except:
# Whatever happens, make sure that we always close the connection.
connection.close()
response.close()
raise
connection.close()
response.close()
return number_of_bytes_received, average_download_speed
def _get_request(url):
"""
Wraps the URL to retrieve to protects against "creative"
interpretation of the RFC: http://bugs.python.org/issue8732
https://github.com/pypa/pip/blob/d0fa66ecc03ab20b7411b35f7c7b423f31f77761/pip/download.py#L147
"""
return six.moves.urllib.request.Request(url, headers={'Accept-encoding': 'identity'})
def _get_opener(scheme=None):
"""
Build a urllib2 opener based on whether the user now wants SSL.
https://github.com/pypa/pip/blob/d0fa66ecc03ab20b7411b35f7c7b423f31f77761/pip/download.py#L178
Raises tuf.exceptions.Error if tuf.settings.ssl_certificates is not a valid
file.
"""
if scheme == "https":
if not os.path.isfile(tuf.settings.ssl_certificates):
raise tuf.exceptions.Error('The SSL certificate specified in'
' tuf.settings.ssl_certificates is not a valid file.'
' ssl_certificates set to: ' + repr(tuf.settings.ssl_certificates))
else:
# If we are going over https, use an opener which will provide SSL
# certificate verification.
https_handler = VerifiedHTTPSHandler()
opener = six.moves.urllib.request.build_opener(https_handler)
# Strip out HTTPHandler to prevent MITM spoof.
for handler in opener.handlers:
if isinstance(handler, six.moves.urllib.request.HTTPHandler):
opener.handlers.remove(handler)
else:
# Otherwise, use the default opener.
opener = six.moves.urllib.request.build_opener()
return opener
def _open_connection(url):
"""
<Purpose>
Helper function that opens a connection to the url. urllib2 supports http,
ftp, and file. In python (2.6+) where the ssl module is available, urllib2
also supports https.
TODO: Determine whether this follows http redirects and decide if we like
that. For example, would we not want to allow redirection from ssl to
non-ssl urls?
<Arguments>
url:
URL string (e.g., 'http://...' or 'ftp://...' or 'file://...')
<Exceptions>
tuf.exceptions.Error, if the certificates pointed by
tuf.settings.ssl_certificates cannot be loaded.
<Side Effects>
Opens a connection to a remote server.
<Returns>
File-like object.
"""
# urllib2.Request produces a Request object that allows for a finer control
# of the requesting process. Request object allows to add headers or data to
# the HTTP request. For instance, request method add_header(key, val) can be
# used to change/spoof 'User-Agent' from default Python-urllib/x.y to
# 'Mozilla/4.0 (compatible; MSIE 7.0; Windows NT 5.1)' this can be useful if
# servers do not recognize connections that originates from
# Python-urllib/x.y.
parsed_url = six.moves.urllib.parse.urlparse(url)
opener = _get_opener(scheme=parsed_url.scheme)
request = _get_request(url)
return opener.open(request, timeout = tuf.settings.SOCKET_TIMEOUT)
def _get_content_length(connection):
def _get_content_length(response):
"""
<Purpose>
A helper function that gets the purported file length from server.
<Arguments>
connection:
The object that the _open_connection function returns for communicating
with the server about the contents of a URL.
response:
The object for communicating with the server about the contents of a URL.
<Side Effects>
No known side effects.
@ -525,7 +448,7 @@ def _get_content_length(connection):
try:
# What is the length of this document according to the HTTP spec?
reported_length = connection.info().get('Content-Length')
reported_length = response.headers.get('Content-Length')
# Try casting it as a decimal number.
reported_length = int(reported_length, 10)
@ -536,7 +459,7 @@ def _get_content_length(connection):
except Exception as e:
logger.exception('Could not get content length'
' about ' + str(connection) + ' from server: ' + str(e))
' about ' + str(response) + ' from server: ' + str(e))
return None
return reported_length
@ -682,71 +605,3 @@ def _check_downloaded_length(total_downloaded, required_length,
logger.info('Downloaded ' + str(total_downloaded) + ' bytes out of an'
' upper limit of ' + str(required_length) + ' bytes.')
class VerifiedHTTPSConnection(six.moves.http_client.HTTPSConnection):
"""
A connection that wraps connections with ssl certificate verification.
https://github.com/pypa/pip/blob/d0fa66ecc03ab20b7411b35f7c7b423f31f77761/pip/download.py#L72
Raise tuf.exeptions.Error if the certificates specified in
tuf.settings.ssl_certificates cannot be loaded.
"""
def connect(self):
connection_kwargs = {}
connection_kwargs.update(timeout = self.timeout)
# for >= py2.7
if hasattr(self, 'source_address'):
connection_kwargs.update(source_address = self.source_address)
sock = socket.create_connection((self.host, self.port), **connection_kwargs)
# for >= py2.7
if getattr(self, '_tunnel_host', None):
self.sock = sock
self._tunnel()
# set location of certificate authorities
if not os.path.isfile(tuf.settings.ssl_certificates):
raise tuf.exceptions.Error('The SSL certificate specified in'
' tuf.settings.ssl_certificates is not a valid file.'
' ssl_certificates set to: ' + repr(tuf.settings.ssl_certificates))
else:
cert_path = tuf.settings.ssl_certificates
# TODO: Disallow SSLv2.
# http://docs.python.org/dev/library/ssl.html#protocol-versions
# TODO: Select the right ciphers.
# http://docs.python.org/dev/library/ssl.html#cipher-selection
# ssl.PROTOCOL_SSLv23, the default value for 'ssl_version', is deprecated in
# Python 2.7.13, but becomes an alias for ssl.PROTOCOL_TLS.
self.sock = ssl.wrap_socket(sock, self.key_file, self.cert_file,
cert_reqs=ssl.CERT_REQUIRED, ca_certs=cert_path,
ssl_version=ssl.PROTOCOL_SSLv23)
match_hostname(self.sock.getpeercert(), self.host)
class VerifiedHTTPSHandler(six.moves.urllib.request.HTTPSHandler):
"""
A HTTPSHandler that uses our own VerifiedHTTPSConnection.
https://github.com/pypa/pip/blob/d0fa66ecc03ab20b7411b35f7c7b423f31f77761/pip/download.py#L109
"""
def __init__(self, connection_class = VerifiedHTTPSConnection):
self.specialized_conn_class = connection_class
six.moves.urllib.request.HTTPSHandler.__init__(self)
def https_open(self, req):
return self.do_open(self.specialized_conn_class, req)

8
tuf/exceptions.py
View file

@ -172,8 +172,8 @@ def __init__(self, expected_length, observed_length):
self.observed_length = observed_length #bytes
def __str__(self):
return 'Observed length (' + repr(self.observed_length)+\
') <= expected length (' + repr(self.expected_length) + ').'
return 'Observed length (' + repr(self.observed_length) + \
') < expected length (' + repr(self.expected_length) + ').'
class SlowRetrievalError(DownloadError):
@ -270,6 +270,10 @@ class URLMatchesNoPatternError(Error):
"""If a URL does not match a user-specified regular expression."""
pass
class URLParsingError(Error):
"""If we are unable to parse a URL -- for example, if a hostname element
cannot be isoalted."""
pass
class InvalidConfigurationError(Error):
"""If a configuration object does not match the expected format."""

15
tuf/settings.py
View file

@ -46,11 +46,6 @@
# /tmp/repositories/django_repo/metadata/current/root.METADATA_EXTENSION
repositories_directory = None
# A PEM (RFC 1422) file where you may find SSL certificate authorities
# https://en.wikipedia.org/wiki/Certificate_authority
# http://docs.python.org/2/library/ssl.html#certificates
ssl_certificates = None
# The 'log.py' module manages TUF's logging system. Users have the option to
# enable/disable logging to a file via 'ENABLE_FILE_LOGGING', or
# tuf.log.enable_file_logging() and tuf.log.disable_file_logging().
@ -89,16 +84,6 @@
# avoid being considered as a slow retrieval attack.
MIN_AVERAGE_DOWNLOAD_SPEED = 50 #bytes/second
# The time (in seconds) we ignore a server with a slow initial retrieval speed.
SLOW_START_GRACE_PERIOD = 0.1 #seconds
# Software updaters that integrate the framework are required to specify
# the URL prefix for the mirrors that clients can contact to download updates.
# The following URI schemes are those that download.py support. By default,
# the ['http', 'https'] URI schemes are supported, but may be modified by
# integrators to schemes that they wish to support for their integration.
SUPPORTED_URI_SCHEMES = ['http', 'https']
# By default, limit number of delegatees we visit for any target.
MAX_NUMBER_OF_DELEGATIONS = 2**5