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

100% test coverage, though still need plenty more

Browse Source
This commit is contained in:
Chandler Abraham 2015-04-13 13:36:09 -07:00
parent 0ed2a29063
commit 2d72a1b6b5
4 changed files with 18 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
netlib/http.py
View File

@ -29,20 +29,6 @@ def _is_valid_host(host):
return None return None
return True return True
def is_successful_upgrade(request, response):
"""
determines if a client and server successfully agreed to an HTTP protocol upgrade
https://developer.mozilla.org/en-US/docs/Web/HTTP/Protocol_upgrade_mechanism
"""
http_switching_protocols_code = 101
if request and response:
responseUpgrade = request.headers.get("Upgrade")
requestUpgrade = response.headers.get("Upgrade")
if response.code == http_switching_protocols_code and responseUpgrade == requestUpgrade:
return requestUpgrade[0] if len(requestUpgrade) > 0 else None
return None
def parse_url(url): def parse_url(url):
""" """

6
netlib/websockets/implementations.py
View File

@ -66,19 +66,13 @@ class WebSocketsClient(tcp.TCPClient):
server_handshake = ws.read_handshake(self.rfile.read, 1) server_handshake = ws.read_handshake(self.rfile.read, 1)
if not server_handshake:
self.close()
server_nounce = ws.process_handshake_from_server(server_handshake, self.client_nounce) server_nounce = ws.process_handshake_from_server(server_handshake, self.client_nounce)
if not server_nounce == ws.create_server_nounce(self.client_nounce): if not server_nounce == ws.create_server_nounce(self.client_nounce):
self.close() self.close()
def read_next_message(self): def read_next_message(self):
try:
return ws.WebSocketsFrame.from_byte_stream(self.rfile.read).payload return ws.WebSocketsFrame.from_byte_stream(self.rfile.read).payload
except IndexError:
self.close()
def send_message(self, message): def send_message(self, message):
frame = ws.WebSocketsFrame.default(message, from_client = True) frame = ws.WebSocketsFrame.default(message, from_client = True)

5
netlib/websockets/websockets.py
View File

@ -158,10 +158,9 @@ class WebSocketsFrame(object):
("actual_payload_length - " + str(self.actual_payload_length))]) ("actual_payload_length - " + str(self.actual_payload_length))])
def safe_to_bytes(self): def safe_to_bytes(self):
try: if self.is_valid():
assert self.is_valid()
return self.to_bytes() return self.to_bytes()
except: else:
raise WebSocketFrameValidationException() raise WebSocketFrameValidationException()
def to_bytes(self): def to_bytes(self):

10
test/test_websockets.py
View File

@ -42,6 +42,10 @@ class TestWebSockets(test.ServerTestBase):
assert server_frame.is_valid() assert server_frame.is_valid()
def test_serialization_bijection(self): def test_serialization_bijection(self):
"""
Ensure that various frame types can be serialized/deserialized back and forth
between to_bytes() and from_bytes()
"""
for is_client in [True, False]: for is_client in [True, False]:
for num_bytes in [100, 50000, 150000]: for num_bytes in [100, 50000, 150000]:
frame = ws.WebSocketsFrame.default(self.random_bytes(num_bytes), is_client) frame = ws.WebSocketsFrame.default(self.random_bytes(num_bytes), is_client)
@ -50,6 +54,12 @@ class TestWebSockets(test.ServerTestBase):
bytes = b'\x81\x11cba' bytes = b'\x81\x11cba'
assert ws.WebSocketsFrame.from_bytes(bytes).to_bytes() == bytes assert ws.WebSocketsFrame.from_bytes(bytes).to_bytes() == bytes
@raises(ws.WebSocketFrameValidationException)
def test_safe_to_bytes(self):
frame = ws.WebSocketsFrame.default(self.random_bytes(8))
frame.actual_payload_length = 1 #corrupt the frame
frame.safe_to_bytes()
class BadHandshakeHandler(impl.WebSocketsEchoHandler): class BadHandshakeHandler(impl.WebSocketsEchoHandler):
def handshake(self): def handshake(self):