Grasscutters/mitmproxy
3
0
Fork 0
mirror of https://github.com/Grasscutters/mitmproxy.git synced 2024-11-26 18:18:25 +00:00
Code Issues Projects Releases Wiki Activity

python3++

Browse Source
This commit is contained in:
Maximilian Hils 2015-09-20 19:56:45 +02:00
parent 693cdfc6d7
commit 0ad5cbc6bf
3 changed files with 16 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
netlib/tcp.py
View File

@ -7,6 +7,8 @@ import threading
import time
import traceback
from six.moves import range
import certifi
import six
import OpenSSL
@ -227,7 +229,7 @@ class Reader(_FileLike):
return result
def readline(self, size=None):
result = ''
result = b''
bytes_read = 0
while True:
if size is not None and bytes_read >= size:
@ -399,7 +401,7 @@ def close_socket(sock):
sock.settimeout(sock.gettimeout() or 20)
# limit at a megabyte so that we don't read infinitely
for _ in xrange(1024 ** 3 // 4096):
for _ in range(1024 ** 3 // 4096):
# may raise a timeout/disconnect exception.
if not sock.recv(4096):
break
@ -649,7 +651,7 @@ class TCPClient(_Connection):
if OpenSSL._util.lib.Cryptography_HAS_ALPN and self.ssl_established:
return self.connection.get_alpn_proto_negotiated()
else:
return ""
return b""
class BaseHandler(_Connection):

2
test/test_socks.py
View File

@ -121,7 +121,7 @@ def test_message_ipv4():
def test_message_ipv6():
# Test ATYP=0x04 (IPV6)
ipv6_addr = "2001:db8:85a3:8d3:1319:8a2e:370:7344"
ipv6_addr = u"2001:db8:85a3:8d3:1319:8a2e:370:7344"
raw = tutils.treader(
b"\x05\x01\x00\x04" +

20
test/test_tcp.py
View File

@ -49,9 +49,9 @@ class ALPNHandler(tcp.BaseHandler):
def handle(self):
alp = self.get_alpn_proto_negotiated()
if alp:
self.wfile.write("%s" % alp)
self.wfile.write(("%s" % alp).encode("ascii"))
else:
self.wfile.write("NONE")
self.wfile.write(b"NONE")
self.wfile.flush()
@ -503,24 +503,24 @@ class TestALPNClient(tservers.ServerTestBase):
def test_alpn(self):
c = tcp.TCPClient(("127.0.0.1", self.port))
c.connect()
c.convert_to_ssl(alpn_protos=["foo", "bar", "fasel"])
assert c.get_alpn_proto_negotiated() == "bar"
assert c.rfile.readline().strip() == "bar"
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() == ""
assert c.rfile.readline().strip() == "NONE"
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=["foo", "bar", "fasel"])
assert c.get_alpn_proto_negotiated() == ""
assert c.rfile.readline() == "NONE"
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):