mirror of
https://github.com/Grasscutters/mitmproxy.git
synced 2024-11-23 16:17:49 +00:00
5af9df326a
This commit fixes netlib's optional (turned off by default)
certificate verification, which previously did not validate the
cert's host name. As it turns out, verifying the connection's host
name on an intercepting proxy is not really straightforward - if
we receive a connection in transparent mode without SNI, we have no
clue which hosts the client intends to connect to. There are two
basic approaches to solve this problem:
1. Exactly mirror the host names presented by the server in the
spoofed certificate presented to the client.
2. Require the client to send the TLS Server Name Indication
extension. While this does not work with older clients,
we can validate the hostname on the proxy.
Approach 1 is problematic in mitmproxy's use case, as we may want
to deliberately divert connections without the client's knowledge.
As a consequence, we opt for approach 2. While mitmproxy does now
require a SNI value to be sent by the client if certificate
verification is turned on, we retain our ability to present
certificates to the client which are accepted with a maximum
likelihood.
764 lines
21 KiB
Python
764 lines
21 KiB
Python
from io import BytesIO
|
|
from six.moves import queue
|
|
import time
|
|
import socket
|
|
import random
|
|
import os
|
|
import threading
|
|
import mock
|
|
|
|
from OpenSSL import SSL
|
|
import OpenSSL
|
|
|
|
from netlib import tcp, certutils, tutils
|
|
from . import tservers
|
|
from netlib.exceptions import InvalidCertificateException, TcpReadIncomplete, TlsException, \
|
|
TcpTimeout, TcpDisconnect, TcpException
|
|
|
|
|
|
class EchoHandler(tcp.BaseHandler):
|
|
sni = None
|
|
|
|
def handle_sni(self, connection):
|
|
self.sni = connection.get_servername()
|
|
|
|
def handle(self):
|
|
v = self.rfile.readline()
|
|
self.wfile.write(v)
|
|
self.wfile.flush()
|
|
|
|
|
|
class ClientCipherListHandler(tcp.BaseHandler):
|
|
sni = None
|
|
|
|
def handle(self):
|
|
self.wfile.write("%s" % self.connection.get_cipher_list())
|
|
self.wfile.flush()
|
|
|
|
|
|
class HangHandler(tcp.BaseHandler):
|
|
|
|
def handle(self):
|
|
while True:
|
|
time.sleep(1)
|
|
|
|
|
|
class ALPNHandler(tcp.BaseHandler):
|
|
sni = None
|
|
|
|
def handle(self):
|
|
alp = self.get_alpn_proto_negotiated()
|
|
if alp:
|
|
self.wfile.write(alp)
|
|
else:
|
|
self.wfile.write(b"NONE")
|
|
self.wfile.flush()
|
|
|
|
|
|
class TestServer(tservers.ServerTestBase):
|
|
handler = EchoHandler
|
|
|
|
def test_echo(self):
|
|
testval = b"echo!\n"
|
|
c = tcp.TCPClient(("127.0.0.1", self.port))
|
|
c.connect()
|
|
c.wfile.write(testval)
|
|
c.wfile.flush()
|
|
assert c.rfile.readline() == testval
|
|
|
|
def test_thread_start_error(self):
|
|
with mock.patch.object(threading.Thread, "start", side_effect=threading.ThreadError("nonewthread")) as m:
|
|
c = tcp.TCPClient(("127.0.0.1", self.port))
|
|
c.connect()
|
|
assert not c.rfile.read(1)
|
|
assert m.called
|
|
assert "nonewthread" in self.q.get_nowait()
|
|
self.test_echo()
|
|
|
|
|
|
class TestServerBind(tservers.ServerTestBase):
|
|
|
|
class handler(tcp.BaseHandler):
|
|
|
|
def handle(self):
|
|
self.wfile.write(str(self.connection.getpeername()).encode())
|
|
self.wfile.flush()
|
|
|
|
def test_bind(self):
|
|
""" Test to bind to a given random port. Try again if the random port turned out to be blocked. """
|
|
for i in range(20):
|
|
random_port = random.randrange(1024, 65535)
|
|
try:
|
|
c = tcp.TCPClient(
|
|
("127.0.0.1", self.port), source_address=(
|
|
"127.0.0.1", random_port))
|
|
c.connect()
|
|
assert c.rfile.readline() == str(("127.0.0.1", random_port)).encode()
|
|
return
|
|
except TcpException: # port probably already in use
|
|
pass
|
|
|
|
|
|
class TestServerIPv6(tservers.ServerTestBase):
|
|
handler = EchoHandler
|
|
addr = tcp.Address(("localhost", 0), use_ipv6=True)
|
|
|
|
def test_echo(self):
|
|
testval = b"echo!\n"
|
|
c = tcp.TCPClient(tcp.Address(("::1", self.port), use_ipv6=True))
|
|
c.connect()
|
|
c.wfile.write(testval)
|
|
c.wfile.flush()
|
|
assert c.rfile.readline() == testval
|
|
|
|
|
|
class TestEcho(tservers.ServerTestBase):
|
|
handler = EchoHandler
|
|
|
|
def test_echo(self):
|
|
testval = b"echo!\n"
|
|
c = tcp.TCPClient(("127.0.0.1", self.port))
|
|
c.connect()
|
|
c.wfile.write(testval)
|
|
c.wfile.flush()
|
|
assert c.rfile.readline() == testval
|
|
|
|
|
|
class HardDisconnectHandler(tcp.BaseHandler):
|
|
|
|
def handle(self):
|
|
self.connection.close()
|
|
|
|
|
|
class TestFinishFail(tservers.ServerTestBase):
|
|
|
|
"""
|
|
This tests a difficult-to-trigger exception in the .finish() method of
|
|
the handler.
|
|
"""
|
|
handler = EchoHandler
|
|
|
|
def test_disconnect_in_finish(self):
|
|
c = tcp.TCPClient(("127.0.0.1", self.port))
|
|
c.connect()
|
|
c.wfile.write(b"foo\n")
|
|
c.wfile.flush = mock.Mock(side_effect=TcpDisconnect)
|
|
c.finish()
|
|
|
|
|
|
class TestServerSSL(tservers.ServerTestBase):
|
|
handler = EchoHandler
|
|
ssl = dict(
|
|
cipher_list="AES256-SHA",
|
|
chain_file=tutils.test_data.path("data/server.crt")
|
|
)
|
|
|
|
def test_echo(self):
|
|
c = tcp.TCPClient(("127.0.0.1", self.port))
|
|
c.connect()
|
|
c.convert_to_ssl(sni=b"foo.com", options=SSL.OP_ALL)
|
|
testval = b"echo!\n"
|
|
c.wfile.write(testval)
|
|
c.wfile.flush()
|
|
assert c.rfile.readline() == testval
|
|
|
|
def test_get_current_cipher(self):
|
|
c = tcp.TCPClient(("127.0.0.1", self.port))
|
|
c.connect()
|
|
assert not c.get_current_cipher()
|
|
c.convert_to_ssl(sni=b"foo.com")
|
|
ret = c.get_current_cipher()
|
|
assert ret
|
|
assert "AES" in ret[0]
|
|
|
|
|
|
class TestSSLv3Only(tservers.ServerTestBase):
|
|
handler = EchoHandler
|
|
ssl = dict(
|
|
request_client_cert=False,
|
|
v3_only=True
|
|
)
|
|
|
|
def test_failure(self):
|
|
c = tcp.TCPClient(("127.0.0.1", self.port))
|
|
c.connect()
|
|
tutils.raises(TlsException, c.convert_to_ssl, sni=b"foo.com")
|
|
|
|
|
|
class TestSSLUpstreamCertVerificationWBadServerCert(tservers.ServerTestBase):
|
|
handler = EchoHandler
|
|
|
|
ssl = dict(
|
|
cert=tutils.test_data.path("data/verificationcerts/self-signed.crt"),
|
|
key=tutils.test_data.path("data/verificationcerts/self-signed.key")
|
|
)
|
|
|
|
def test_mode_default_should_pass(self):
|
|
c = tcp.TCPClient(("127.0.0.1", self.port))
|
|
c.connect()
|
|
|
|
c.convert_to_ssl()
|
|
|
|
# Verification errors should be saved even if connection isn't aborted
|
|
# aborted
|
|
assert c.ssl_verification_error is not None
|
|
|
|
testval = b"echo!\n"
|
|
c.wfile.write(testval)
|
|
c.wfile.flush()
|
|
assert c.rfile.readline() == testval
|
|
|
|
def test_mode_none_should_pass(self):
|
|
c = tcp.TCPClient(("127.0.0.1", self.port))
|
|
c.connect()
|
|
|
|
c.convert_to_ssl(verify_options=SSL.VERIFY_NONE)
|
|
|
|
# Verification errors should be saved even if connection isn't aborted
|
|
assert c.ssl_verification_error is not None
|
|
|
|
testval = b"echo!\n"
|
|
c.wfile.write(testval)
|
|
c.wfile.flush()
|
|
assert c.rfile.readline() == testval
|
|
|
|
def test_mode_strict_should_fail(self):
|
|
c = tcp.TCPClient(("127.0.0.1", self.port))
|
|
c.connect()
|
|
|
|
with tutils.raises(InvalidCertificateException):
|
|
c.convert_to_ssl(
|
|
sni=b"example.mitmproxy.org",
|
|
verify_options=SSL.VERIFY_PEER,
|
|
ca_pemfile=tutils.test_data.path("data/verificationcerts/trusted-root.crt")
|
|
)
|
|
|
|
assert c.ssl_verification_error is not None
|
|
|
|
# Unknown issuing certificate authority for first certificate
|
|
assert c.ssl_verification_error['errno'] == 18
|
|
assert c.ssl_verification_error['depth'] == 0
|
|
|
|
|
|
class TestSSLUpstreamCertVerificationWBadHostname(tservers.ServerTestBase):
|
|
handler = EchoHandler
|
|
|
|
ssl = dict(
|
|
cert=tutils.test_data.path("data/verificationcerts/trusted-leaf.crt"),
|
|
key=tutils.test_data.path("data/verificationcerts/trusted-leaf.key")
|
|
)
|
|
|
|
def test_should_fail_without_sni(self):
|
|
c = tcp.TCPClient(("127.0.0.1", self.port))
|
|
c.connect()
|
|
|
|
with tutils.raises(TlsException):
|
|
c.convert_to_ssl(
|
|
verify_options=SSL.VERIFY_PEER,
|
|
ca_pemfile=tutils.test_data.path("data/verificationcerts/trusted-root.crt")
|
|
)
|
|
|
|
def test_should_fail(self):
|
|
c = tcp.TCPClient(("127.0.0.1", self.port))
|
|
c.connect()
|
|
|
|
with tutils.raises(InvalidCertificateException):
|
|
c.convert_to_ssl(
|
|
sni=b"mitmproxy.org",
|
|
verify_options=SSL.VERIFY_PEER,
|
|
ca_pemfile=tutils.test_data.path("data/verificationcerts/trusted-root.crt")
|
|
)
|
|
|
|
assert c.ssl_verification_error is not None
|
|
|
|
|
|
class TestSSLUpstreamCertVerificationWValidCertChain(tservers.ServerTestBase):
|
|
handler = EchoHandler
|
|
|
|
ssl = dict(
|
|
cert=tutils.test_data.path("data/verificationcerts/trusted-leaf.crt"),
|
|
key=tutils.test_data.path("data/verificationcerts/trusted-leaf.key")
|
|
)
|
|
|
|
def test_mode_strict_w_pemfile_should_pass(self):
|
|
c = tcp.TCPClient(("127.0.0.1", self.port))
|
|
c.connect()
|
|
|
|
c.convert_to_ssl(
|
|
sni=b"example.mitmproxy.org",
|
|
verify_options=SSL.VERIFY_PEER,
|
|
ca_pemfile=tutils.test_data.path("data/verificationcerts/trusted-root.crt")
|
|
)
|
|
|
|
assert c.ssl_verification_error is None
|
|
|
|
testval = b"echo!\n"
|
|
c.wfile.write(testval)
|
|
c.wfile.flush()
|
|
assert c.rfile.readline() == testval
|
|
|
|
def test_mode_strict_w_cadir_should_pass(self):
|
|
c = tcp.TCPClient(("127.0.0.1", self.port))
|
|
c.connect()
|
|
|
|
c.convert_to_ssl(
|
|
sni=b"example.mitmproxy.org",
|
|
verify_options=SSL.VERIFY_PEER,
|
|
ca_path=tutils.test_data.path("data/verificationcerts/")
|
|
)
|
|
|
|
assert c.ssl_verification_error is None
|
|
|
|
testval = b"echo!\n"
|
|
c.wfile.write(testval)
|
|
c.wfile.flush()
|
|
assert c.rfile.readline() == testval
|
|
|
|
|
|
class TestSSLClientCert(tservers.ServerTestBase):
|
|
|
|
class handler(tcp.BaseHandler):
|
|
sni = None
|
|
|
|
def handle_sni(self, connection):
|
|
self.sni = connection.get_servername()
|
|
|
|
def handle(self):
|
|
self.wfile.write(b"%d\n" % self.clientcert.serial)
|
|
self.wfile.flush()
|
|
|
|
ssl = dict(
|
|
request_client_cert=True,
|
|
v3_only=False
|
|
)
|
|
|
|
def test_clientcert(self):
|
|
c = tcp.TCPClient(("127.0.0.1", self.port))
|
|
c.connect()
|
|
c.convert_to_ssl(
|
|
cert=tutils.test_data.path("data/clientcert/client.pem"))
|
|
assert c.rfile.readline().strip() == b"1"
|
|
|
|
def test_clientcert_err(self):
|
|
c = tcp.TCPClient(("127.0.0.1", self.port))
|
|
c.connect()
|
|
tutils.raises(
|
|
TlsException,
|
|
c.convert_to_ssl,
|
|
cert=tutils.test_data.path("data/clientcert/make")
|
|
)
|
|
|
|
|
|
class TestSNI(tservers.ServerTestBase):
|
|
|
|
class handler(tcp.BaseHandler):
|
|
sni = None
|
|
|
|
def handle_sni(self, connection):
|
|
self.sni = connection.get_servername()
|
|
|
|
def handle(self):
|
|
self.wfile.write(self.sni)
|
|
self.wfile.flush()
|
|
|
|
ssl = True
|
|
|
|
def test_echo(self):
|
|
c = tcp.TCPClient(("127.0.0.1", self.port))
|
|
c.connect()
|
|
c.convert_to_ssl(sni=b"foo.com")
|
|
assert c.sni == b"foo.com"
|
|
assert c.rfile.readline() == b"foo.com"
|
|
|
|
|
|
class TestServerCipherList(tservers.ServerTestBase):
|
|
handler = ClientCipherListHandler
|
|
ssl = dict(
|
|
cipher_list='RC4-SHA'
|
|
)
|
|
|
|
def test_echo(self):
|
|
c = tcp.TCPClient(("127.0.0.1", self.port))
|
|
c.connect()
|
|
c.convert_to_ssl(sni=b"foo.com")
|
|
assert c.rfile.readline() == b"['RC4-SHA']"
|
|
|
|
|
|
class TestServerCurrentCipher(tservers.ServerTestBase):
|
|
|
|
class handler(tcp.BaseHandler):
|
|
sni = None
|
|
|
|
def handle(self):
|
|
self.wfile.write(str(self.get_current_cipher()).encode())
|
|
self.wfile.flush()
|
|
|
|
ssl = dict(
|
|
cipher_list='RC4-SHA'
|
|
)
|
|
|
|
def test_echo(self):
|
|
c = tcp.TCPClient(("127.0.0.1", self.port))
|
|
c.connect()
|
|
c.convert_to_ssl(sni=b"foo.com")
|
|
assert b"RC4-SHA" in c.rfile.readline()
|
|
|
|
|
|
class TestServerCipherListError(tservers.ServerTestBase):
|
|
handler = ClientCipherListHandler
|
|
ssl = dict(
|
|
cipher_list='bogus'
|
|
)
|
|
|
|
def test_echo(self):
|
|
c = tcp.TCPClient(("127.0.0.1", self.port))
|
|
c.connect()
|
|
tutils.raises("handshake error", c.convert_to_ssl, sni=b"foo.com")
|
|
|
|
|
|
class TestClientCipherListError(tservers.ServerTestBase):
|
|
handler = ClientCipherListHandler
|
|
ssl = dict(
|
|
cipher_list='RC4-SHA'
|
|
)
|
|
|
|
def test_echo(self):
|
|
c = tcp.TCPClient(("127.0.0.1", self.port))
|
|
c.connect()
|
|
tutils.raises(
|
|
"cipher specification",
|
|
c.convert_to_ssl,
|
|
sni=b"foo.com",
|
|
cipher_list="bogus")
|
|
|
|
|
|
class TestSSLDisconnect(tservers.ServerTestBase):
|
|
|
|
class handler(tcp.BaseHandler):
|
|
|
|
def handle(self):
|
|
self.finish()
|
|
|
|
ssl = True
|
|
|
|
def test_echo(self):
|
|
c = tcp.TCPClient(("127.0.0.1", self.port))
|
|
c.connect()
|
|
c.convert_to_ssl()
|
|
# Excercise SSL.ZeroReturnError
|
|
c.rfile.read(10)
|
|
c.close()
|
|
tutils.raises(TcpDisconnect, c.wfile.write, b"foo")
|
|
tutils.raises(queue.Empty, self.q.get_nowait)
|
|
|
|
|
|
class TestSSLHardDisconnect(tservers.ServerTestBase):
|
|
handler = HardDisconnectHandler
|
|
ssl = True
|
|
|
|
def test_echo(self):
|
|
c = tcp.TCPClient(("127.0.0.1", self.port))
|
|
c.connect()
|
|
c.convert_to_ssl()
|
|
# Exercise SSL.SysCallError
|
|
c.rfile.read(10)
|
|
c.close()
|
|
tutils.raises(TcpDisconnect, c.wfile.write, b"foo")
|
|
|
|
|
|
class TestDisconnect(tservers.ServerTestBase):
|
|
|
|
def test_echo(self):
|
|
c = tcp.TCPClient(("127.0.0.1", self.port))
|
|
c.connect()
|
|
c.rfile.read(10)
|
|
c.wfile.write(b"foo")
|
|
c.close()
|
|
c.close()
|
|
|
|
|
|
class TestServerTimeOut(tservers.ServerTestBase):
|
|
|
|
class handler(tcp.BaseHandler):
|
|
|
|
def handle(self):
|
|
self.timeout = False
|
|
self.settimeout(0.01)
|
|
try:
|
|
self.rfile.read(10)
|
|
except TcpTimeout:
|
|
self.timeout = True
|
|
|
|
def test_timeout(self):
|
|
c = tcp.TCPClient(("127.0.0.1", self.port))
|
|
c.connect()
|
|
time.sleep(0.3)
|
|
assert self.last_handler.timeout
|
|
|
|
|
|
class TestTimeOut(tservers.ServerTestBase):
|
|
handler = HangHandler
|
|
|
|
def test_timeout(self):
|
|
c = tcp.TCPClient(("127.0.0.1", self.port))
|
|
c.connect()
|
|
c.settimeout(0.1)
|
|
assert c.gettimeout() == 0.1
|
|
tutils.raises(TcpTimeout, c.rfile.read, 10)
|
|
|
|
|
|
class TestALPNClient(tservers.ServerTestBase):
|
|
handler = ALPNHandler
|
|
ssl = dict(
|
|
alpn_select=b"bar"
|
|
)
|
|
|
|
if OpenSSL._util.lib.Cryptography_HAS_ALPN:
|
|
def test_alpn(self):
|
|
c = tcp.TCPClient(("127.0.0.1", self.port))
|
|
c.connect()
|
|
c.convert_to_ssl(alpn_protos=[b"foo", b"bar", b"fasel"])
|
|
assert c.get_alpn_proto_negotiated() == b"bar"
|
|
assert c.rfile.readline().strip() == b"bar"
|
|
|
|
def test_no_alpn(self):
|
|
c = tcp.TCPClient(("127.0.0.1", self.port))
|
|
c.connect()
|
|
c.convert_to_ssl()
|
|
assert c.get_alpn_proto_negotiated() == b""
|
|
assert c.rfile.readline().strip() == b"NONE"
|
|
|
|
else:
|
|
def test_none_alpn(self):
|
|
c = tcp.TCPClient(("127.0.0.1", self.port))
|
|
c.connect()
|
|
c.convert_to_ssl(alpn_protos=[b"foo", b"bar", b"fasel"])
|
|
assert c.get_alpn_proto_negotiated() == b""
|
|
assert c.rfile.readline() == b"NONE"
|
|
|
|
|
|
class TestNoSSLNoALPNClient(tservers.ServerTestBase):
|
|
handler = ALPNHandler
|
|
|
|
def test_no_ssl_no_alpn(self):
|
|
c = tcp.TCPClient(("127.0.0.1", self.port))
|
|
c.connect()
|
|
assert c.get_alpn_proto_negotiated() == b""
|
|
assert c.rfile.readline().strip() == b"NONE"
|
|
|
|
|
|
class TestSSLTimeOut(tservers.ServerTestBase):
|
|
handler = HangHandler
|
|
ssl = True
|
|
|
|
def test_timeout_client(self):
|
|
c = tcp.TCPClient(("127.0.0.1", self.port))
|
|
c.connect()
|
|
c.convert_to_ssl()
|
|
c.settimeout(0.1)
|
|
tutils.raises(TcpTimeout, c.rfile.read, 10)
|
|
|
|
|
|
class TestDHParams(tservers.ServerTestBase):
|
|
handler = HangHandler
|
|
ssl = dict(
|
|
dhparams=certutils.CertStore.load_dhparam(
|
|
tutils.test_data.path("data/dhparam.pem"),
|
|
),
|
|
cipher_list="DHE-RSA-AES256-SHA"
|
|
)
|
|
|
|
def test_dhparams(self):
|
|
c = tcp.TCPClient(("127.0.0.1", self.port))
|
|
c.connect()
|
|
c.convert_to_ssl()
|
|
ret = c.get_current_cipher()
|
|
assert ret[0] == "DHE-RSA-AES256-SHA"
|
|
|
|
def test_create_dhparams(self):
|
|
with tutils.tmpdir() as d:
|
|
filename = os.path.join(d, "dhparam.pem")
|
|
certutils.CertStore.load_dhparam(filename)
|
|
assert os.path.exists(filename)
|
|
|
|
|
|
class TestTCPClient:
|
|
|
|
def test_conerr(self):
|
|
c = tcp.TCPClient(("127.0.0.1", 0))
|
|
tutils.raises(TcpException, c.connect)
|
|
|
|
|
|
class TestFileLike:
|
|
|
|
def test_blocksize(self):
|
|
s = BytesIO(b"1234567890abcdefghijklmnopqrstuvwxyz")
|
|
s = tcp.Reader(s)
|
|
s.BLOCKSIZE = 2
|
|
assert s.read(1) == b"1"
|
|
assert s.read(2) == b"23"
|
|
assert s.read(3) == b"456"
|
|
assert s.read(4) == b"7890"
|
|
d = s.read(-1)
|
|
assert d.startswith(b"abc") and d.endswith(b"xyz")
|
|
|
|
def test_wrap(self):
|
|
s = BytesIO(b"foobar\nfoobar")
|
|
s.flush()
|
|
s = tcp.Reader(s)
|
|
assert s.readline() == b"foobar\n"
|
|
assert s.readline() == b"foobar"
|
|
# Test __getattr__
|
|
assert s.isatty
|
|
|
|
def test_limit(self):
|
|
s = BytesIO(b"foobar\nfoobar")
|
|
s = tcp.Reader(s)
|
|
assert s.readline(3) == b"foo"
|
|
|
|
def test_limitless(self):
|
|
s = BytesIO(b"f" * (50 * 1024))
|
|
s = tcp.Reader(s)
|
|
ret = s.read(-1)
|
|
assert len(ret) == 50 * 1024
|
|
|
|
def test_readlog(self):
|
|
s = BytesIO(b"foobar\nfoobar")
|
|
s = tcp.Reader(s)
|
|
assert not s.is_logging()
|
|
s.start_log()
|
|
assert s.is_logging()
|
|
s.readline()
|
|
assert s.get_log() == b"foobar\n"
|
|
s.read(1)
|
|
assert s.get_log() == b"foobar\nf"
|
|
s.start_log()
|
|
assert s.get_log() == b""
|
|
s.read(1)
|
|
assert s.get_log() == b"o"
|
|
s.stop_log()
|
|
tutils.raises(ValueError, s.get_log)
|
|
|
|
def test_writelog(self):
|
|
s = BytesIO()
|
|
s = tcp.Writer(s)
|
|
s.start_log()
|
|
assert s.is_logging()
|
|
s.write(b"x")
|
|
assert s.get_log() == b"x"
|
|
s.write(b"x")
|
|
assert s.get_log() == b"xx"
|
|
|
|
def test_writer_flush_error(self):
|
|
s = BytesIO()
|
|
s = tcp.Writer(s)
|
|
o = mock.MagicMock()
|
|
o.flush = mock.MagicMock(side_effect=socket.error)
|
|
s.o = o
|
|
tutils.raises(TcpDisconnect, s.flush)
|
|
|
|
def test_reader_read_error(self):
|
|
s = BytesIO(b"foobar\nfoobar")
|
|
s = tcp.Reader(s)
|
|
o = mock.MagicMock()
|
|
o.read = mock.MagicMock(side_effect=socket.error)
|
|
s.o = o
|
|
tutils.raises(TcpDisconnect, s.read, 10)
|
|
|
|
def test_reset_timestamps(self):
|
|
s = BytesIO(b"foobar\nfoobar")
|
|
s = tcp.Reader(s)
|
|
s.first_byte_timestamp = 500
|
|
s.reset_timestamps()
|
|
assert not s.first_byte_timestamp
|
|
|
|
def test_first_byte_timestamp_updated_on_read(self):
|
|
s = BytesIO(b"foobar\nfoobar")
|
|
s = tcp.Reader(s)
|
|
s.read(1)
|
|
assert s.first_byte_timestamp
|
|
expected = s.first_byte_timestamp
|
|
s.read(5)
|
|
assert s.first_byte_timestamp == expected
|
|
|
|
def test_first_byte_timestamp_updated_on_readline(self):
|
|
s = BytesIO(b"foobar\nfoobar\nfoobar")
|
|
s = tcp.Reader(s)
|
|
s.readline()
|
|
assert s.first_byte_timestamp
|
|
expected = s.first_byte_timestamp
|
|
s.readline()
|
|
assert s.first_byte_timestamp == expected
|
|
|
|
def test_read_ssl_error(self):
|
|
s = mock.MagicMock()
|
|
s.read = mock.MagicMock(side_effect=SSL.Error())
|
|
s = tcp.Reader(s)
|
|
tutils.raises(TlsException, s.read, 1)
|
|
|
|
def test_read_syscall_ssl_error(self):
|
|
s = mock.MagicMock()
|
|
s.read = mock.MagicMock(side_effect=SSL.SysCallError())
|
|
s = tcp.Reader(s)
|
|
tutils.raises(TlsException, s.read, 1)
|
|
|
|
def test_reader_readline_disconnect(self):
|
|
o = mock.MagicMock()
|
|
o.read = mock.MagicMock(side_effect=socket.error)
|
|
s = tcp.Reader(o)
|
|
tutils.raises(TcpDisconnect, s.readline, 10)
|
|
|
|
def test_reader_incomplete_error(self):
|
|
s = BytesIO(b"foobar")
|
|
s = tcp.Reader(s)
|
|
tutils.raises(TcpReadIncomplete, s.safe_read, 10)
|
|
|
|
|
|
class TestAddress:
|
|
|
|
def test_simple(self):
|
|
a = tcp.Address("localhost", True)
|
|
assert a.use_ipv6
|
|
b = tcp.Address("foo.com", True)
|
|
assert not a == b
|
|
assert str(b) == str(tuple("foo.com"))
|
|
c = tcp.Address("localhost", True)
|
|
assert a == c
|
|
assert not a != c
|
|
assert repr(a)
|
|
|
|
|
|
class TestSSLKeyLogger(tservers.ServerTestBase):
|
|
handler = EchoHandler
|
|
ssl = dict(
|
|
cipher_list="AES256-SHA"
|
|
)
|
|
|
|
def test_log(self):
|
|
testval = b"echo!\n"
|
|
_logfun = tcp.log_ssl_key
|
|
|
|
with tutils.tmpdir() as d:
|
|
logfile = os.path.join(d, "foo", "bar", "logfile")
|
|
tcp.log_ssl_key = tcp.SSLKeyLogger(logfile)
|
|
|
|
c = tcp.TCPClient(("127.0.0.1", self.port))
|
|
c.connect()
|
|
c.convert_to_ssl()
|
|
c.wfile.write(testval)
|
|
c.wfile.flush()
|
|
assert c.rfile.readline() == testval
|
|
c.finish()
|
|
|
|
tcp.log_ssl_key.close()
|
|
with open(logfile, "rb") as f:
|
|
assert f.read().count(b"CLIENT_RANDOM") == 2
|
|
|
|
tcp.log_ssl_key = _logfun
|
|
|
|
def test_create_logfun(self):
|
|
assert isinstance(
|
|
tcp.SSLKeyLogger.create_logfun("test"),
|
|
tcp.SSLKeyLogger)
|
|
assert not tcp.SSLKeyLogger.create_logfun(False)
|