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

Py3: pathoc

Browse Source
This commit is contained in:
Shadab Zafar 2016-06-15 10:33:56 +05:30 committed by Thomas Kriechbaumer
parent fa40531a80
commit f83433e674
3 changed files with 34 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
pathod/pathoc.py
View File

@ -42,7 +42,7 @@ class SSLInfo(object):
def __str__(self): def __str__(self):
parts = [ parts = [
"Application Layer Protocol: %s" % self.alp, "Application Layer Protocol: %s" % strutils.native(self.alp, "utf8"),
"Cipher: %s, %s bit, %s" % self.cipher, "Cipher: %s, %s bit, %s" % self.cipher,
"SSL certificate chain:" "SSL certificate chain:"
] ]
@ -50,18 +50,25 @@ class SSLInfo(object):
parts.append(" Certificate [%s]" % n) parts.append(" Certificate [%s]" % n)
parts.append("\tSubject: ") parts.append("\tSubject: ")
for cn in i.get_subject().get_components(): for cn in i.get_subject().get_components():
parts.append("\t\t%s=%s" % cn) parts.append("\t\t%s=%s" % (
strutils.native(cn[0], "utf8"),
strutils.native(cn[1], "utf8"))
)
parts.append("\tIssuer: ") parts.append("\tIssuer: ")
for cn in i.get_issuer().get_components(): for cn in i.get_issuer().get_components():
parts.append("\t\t%s=%s" % cn) parts.append("\t\t%s=%s" % (
strutils.native(cn[0], "utf8"),
strutils.native(cn[1], "utf8"))
)
parts.extend( parts.extend(
[ [
"\tVersion: %s" % i.get_version(), "\tVersion: %s" % i.get_version(),
"\tValidity: %s - %s" % ( "\tValidity: %s - %s" % (
i.get_notBefore(), i.get_notAfter() strutils.native(i.get_notBefore(), "utf8"),
strutils.native(i.get_notAfter(), "utf8")
), ),
"\tSerial: %s" % i.get_serial_number(), "\tSerial: %s" % i.get_serial_number(),
"\tAlgorithm: %s" % i.get_signature_algorithm() "\tAlgorithm: %s" % strutils.native(i.get_signature_algorithm(), "utf8")
] ]
) )
pk = i.get_pubkey() pk = i.get_pubkey()
@ -73,7 +80,7 @@ class SSLInfo(object):
parts.append("\tPubkey: %s bit %s" % (pk.bits(), t)) parts.append("\tPubkey: %s bit %s" % (pk.bits(), t))
s = certutils.SSLCert(i) s = certutils.SSLCert(i)
if s.altnames: if s.altnames:
parts.append("\tSANs: %s" % " ".join(s.altnames)) parts.append("\tSANs: %s" % " ".join(strutils.native(n, "utf8") for n in s.altnames))
return "\n".join(parts) return "\n".join(parts)
@ -239,7 +246,7 @@ class Pathoc(tcp.TCPClient):
) )
self.wfile.flush() self.wfile.flush()
try: try:
resp = self.protocol.read_response(self.rfile, treq(method="CONNECT")) resp = self.protocol.read_response(self.rfile, treq(method=b"CONNECT"))
if resp.status_code != 200: if resp.status_code != 200:
raise exceptions.HttpException("Unexpected status code: %s" % resp.status_code) raise exceptions.HttpException("Unexpected status code: %s" % resp.status_code)
except exceptions.HttpException as e: except exceptions.HttpException as e:
@ -437,7 +444,7 @@ class Pathoc(tcp.TCPClient):
finally: finally:
if resp: if resp:
lg("<< %s %s: %s bytes" % ( lg("<< %s %s: %s bytes" % (
resp.status_code, strutils.bytes_to_escaped_str(resp.reason), len(resp.content) resp.status_code, strutils.bytes_to_escaped_str(resp.reason.encode()), len(resp.content)
)) ))
if resp.status_code in self.ignorecodes: if resp.status_code in self.ignorecodes:
lg.suppress() lg.suppress()
@ -454,8 +461,8 @@ class Pathoc(tcp.TCPClient):
May raise a exceptions.NetlibException May raise a exceptions.NetlibException
""" """
if isinstance(r, basestring): if isinstance(r, six.string_types):
r = language.parse_pathoc(r, self.use_http2).next() r = next(language.parse_pathoc(r, self.use_http2))
if isinstance(r, language.http.Request): if isinstance(r, language.http.Request):
if r.ws: if r.ws:

31
test/pathod/test_pathoc.py
View File

@ -1,4 +1,5 @@
from six.moves import cStringIO as StringIO from six.moves import cStringIO as StringIO
from six import BytesIO
from mock import Mock from mock import Mock
from netlib import http from netlib import http
@ -12,7 +13,7 @@ import tutils
def test_response(): def test_response():
r = http.Response("HTTP/1.1", 200, "Message", {}, None, None) r = http.Response(b"HTTP/1.1", 200, b"Message", {}, None, None)
assert repr(r) assert repr(r)
@ -29,7 +30,7 @@ class PathocTestDaemon(tutils.DaemonTests):
if timeout: if timeout:
c.settimeout(timeout) c.settimeout(timeout)
for i in requests: for i in requests:
r = language.parse_pathoc(i).next() r = next(language.parse_pathoc(i))
if kwargs.get("explain"): if kwargs.get("explain"):
r = r.freeze(language.Settings()) r = r.freeze(language.Settings())
try: try:
@ -44,17 +45,17 @@ class TestDaemonSSL(PathocTestDaemon):
ssl = True ssl = True
ssloptions = dict( ssloptions = dict(
request_client_cert=True, request_client_cert=True,
sans=["test1.com", "test2.com"], sans=[b"test1.com", b"test2.com"],
alpn_select=b'h2', alpn_select=b'h2',
) )
def test_sni(self): def test_sni(self):
self.tval( self.tval(
["get:/p/200"], ["get:/p/200"],
sni="foobar.com" sni=b"foobar.com"
) )
log = self.d.log() log = self.d.log()
assert log[0]["request"]["sni"] == "foobar.com" assert log[0]["request"]["sni"] == b"foobar.com"
def test_showssl(self): def test_showssl(self):
assert "certificate chain" in self.tval(["get:/p/200"], showssl=True) assert "certificate chain" in self.tval(["get:/p/200"], showssl=True)
@ -171,36 +172,36 @@ class TestDaemon(PathocTestDaemon):
c.rfile, c.wfile = StringIO(), StringIO() c.rfile, c.wfile = StringIO(), StringIO()
with raises("connect failed"): with raises("connect failed"):
c.http_connect(to) c.http_connect(to)
c.rfile = StringIO( c.rfile = BytesIO(
"HTTP/1.1 500 OK\r\n" b"HTTP/1.1 500 OK\r\n"
) )
with raises("connect failed"): with raises("connect failed"):
c.http_connect(to) c.http_connect(to)
c.rfile = StringIO( c.rfile = BytesIO(
"HTTP/1.1 200 OK\r\n" b"HTTP/1.1 200 OK\r\n"
) )
c.http_connect(to) c.http_connect(to)
def test_socks_connect(self): def test_socks_connect(self):
to = ("foobar", 80) to = ("foobar", 80)
c = pathoc.Pathoc(("127.0.0.1", self.d.port), fp=None) c = pathoc.Pathoc(("127.0.0.1", self.d.port), fp=None)
c.rfile, c.wfile = tutils.treader(""), StringIO() c.rfile, c.wfile = tutils.treader(b""), BytesIO()
tutils.raises(pathoc.PathocError, c.socks_connect, to) tutils.raises(pathoc.PathocError, c.socks_connect, to)
c.rfile = tutils.treader( c.rfile = tutils.treader(
"\x05\xEE" b"\x05\xEE"
) )
tutils.raises("SOCKS without authentication", c.socks_connect, ("example.com", 0xDEAD)) tutils.raises("SOCKS without authentication", c.socks_connect, ("example.com", 0xDEAD))
c.rfile = tutils.treader( c.rfile = tutils.treader(
"\x05\x00" + b"\x05\x00" +
"\x05\xEE\x00\x03\x0bexample.com\xDE\xAD" b"\x05\xEE\x00\x03\x0bexample.com\xDE\xAD"
) )
tutils.raises("SOCKS server error", c.socks_connect, ("example.com", 0xDEAD)) tutils.raises("SOCKS server error", c.socks_connect, ("example.com", 0xDEAD))
c.rfile = tutils.treader( c.rfile = tutils.treader(
"\x05\x00" + b"\x05\x00" +
"\x05\x00\x00\x03\x0bexample.com\xDE\xAD" b"\x05\x00\x00\x03\x0bexample.com\xDE\xAD"
) )
c.socks_connect(("example.com", 0xDEAD)) c.socks_connect(("example.com", 0xDEAD))

2
test/pathod/tutils.py
View File

@ -20,7 +20,7 @@ def treader(bytes):
""" """
Construct a tcp.Read object from bytes. Construct a tcp.Read object from bytes.
""" """
fp = StringIO(bytes) fp = BytesIO(bytes)
return tcp.Reader(fp) return tcp.Reader(fp)