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

[sans-io] Add SOCKS5 proxy mode (#4482)

Browse Source 
* [sans-io] add SOCKS5 proxy mode

* fixup coverage
This commit is contained in:
Maximilian Hils 2021-03-07 22:37:50 +01:00 committed by GitHub
commit 7345daf3f4
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 374 additions and 113 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
mitmproxy/addons/next_layer.py
View File

@ -168,7 +168,7 @@ class NextLayer:
return layers.modes.ReverseProxy(context) return layers.modes.ReverseProxy(context)
elif ctx.options.mode == "socks5": elif ctx.options.mode == "socks5":
raise NotImplementedError("Mode not implemented.") return layers.modes.Socks5Proxy(context)
else: # pragma: no cover else: # pragma: no cover
raise AssertionError("Unknown mode.") raise AssertionError("Unknown mode.")

151
mitmproxy/proxy/layers/modes.py
View File

@ -1,4 +1,7 @@
import socket
import struct
from abc import ABCMeta from abc import ABCMeta
from typing import Optional
from mitmproxy import platform from mitmproxy import platform
from mitmproxy.net import server_spec from mitmproxy.net import server_spec
@ -16,18 +19,19 @@ class HttpProxy(layer.Layer):
class DestinationKnown(layer.Layer, metaclass=ABCMeta): class DestinationKnown(layer.Layer, metaclass=ABCMeta):
"""Base layer for layers that gather connection destination info and then delegate."""
child_layer: layer.Layer child_layer: layer.Layer
def finish_start(self): def finish_start(self) -> layer.CommandGenerator[Optional[str]]:
if self.context.options.connection_strategy == "eager": if self.context.options.connection_strategy == "eager":
err = yield commands.OpenConnection(self.context.server) err = yield commands.OpenConnection(self.context.server)
if err: if err:
yield commands.CloseConnection(self.context.client) self._handle_event = self.done # type: ignore
self._handle_event = self.done return err
return
self._handle_event = self.child_layer.handle_event self._handle_event = self.child_layer.handle_event # type: ignore
yield from self.child_layer.handle_event(events.Start()) yield from self.child_layer.handle_event(events.Start())
return None
@expect(events.DataReceived, events.ConnectionClosed) @expect(events.DataReceived, events.ConnectionClosed)
def done(self, _) -> layer.CommandGenerator[None]: def done(self, _) -> layer.CommandGenerator[None]:
@ -47,7 +51,9 @@ class ReverseProxy(DestinationKnown):
else: else:
self.child_layer = layer.NextLayer(self.context) self.child_layer = layer.NextLayer(self.context)
yield from self.finish_start() err = yield from self.finish_start()
if err:
yield commands.CloseConnection(self.context.client)
class TransparentProxy(DestinationKnown): class TransparentProxy(DestinationKnown):
@ -62,4 +68,135 @@ class TransparentProxy(DestinationKnown):
self.child_layer = layer.NextLayer(self.context) self.child_layer = layer.NextLayer(self.context)
yield from self.finish_start() err = yield from self.finish_start()
if err:
yield commands.CloseConnection(self.context.client)
SOCKS5_VERSION = 0x05
SOCKS5_METHOD_NO_AUTHENTICATION_REQUIRED = 0x00
SOCKS5_METHOD_NO_ACCEPTABLE_METHODS = 0xFF
SOCKS5_ATYP_IPV4_ADDRESS = 0x01
SOCKS5_ATYP_DOMAINNAME = 0x03
SOCKS5_ATYP_IPV6_ADDRESS = 0x04
SOCKS5_REP_HOST_UNREACHABLE = 0x04
SOCKS5_REP_COMMAND_NOT_SUPPORTED = 0x07
SOCKS5_REP_ADDRESS_TYPE_NOT_SUPPORTED = 0x08
class Socks5Proxy(DestinationKnown):
buf: bytes = b""
greeted: bool = False
def socks_err(
self,
message: str,
reply_code: Optional[int] = None,
) -> layer.CommandGenerator[None]:
if reply_code is not None:
yield commands.SendData(
self.context.client,
bytes([SOCKS5_VERSION, reply_code]) + b"\x00\x01\x00\x00\x00\x00\x00\x00"
)
yield commands.CloseConnection(self.context.client)
yield commands.Log(message)
self._handle_event = self.done
@expect(events.Start, events.DataReceived, events.ConnectionClosed)
def _handle_event(self, event: events.Event) -> layer.CommandGenerator[None]:
if isinstance(event, events.Start):
pass
elif isinstance(event, events.DataReceived):
self.buf += event.data
if not self.greeted:
# Parse Client Greeting
if len(self.buf) < 2:
return
if self.buf[0] != SOCKS5_VERSION:
if self.buf[:3].isupper():
guess = "Probably not a SOCKS request but a regular HTTP request. "
else:
guess = ""
yield from self.socks_err(guess + "Invalid SOCKS version. Expected 0x05, got 0x%x" % self.buf[0])
return
n_methods = self.buf[1]
if len(self.buf) < 2 + n_methods:
return
if SOCKS5_METHOD_NO_AUTHENTICATION_REQUIRED not in self.buf[2:2 + n_methods]:
yield from self.socks_err("mitmproxy only supports SOCKS without authentication",
SOCKS5_METHOD_NO_ACCEPTABLE_METHODS)
return
# Send Server Greeting
# Ver = SOCKS5, Auth = NO_AUTH
yield commands.SendData(self.context.client, b"\x05\x00")
self.buf = self.buf[2 + n_methods:]
self.greeted = True
# Parse Connect Request
if len(self.buf) < 4:
return
if self.buf[:3] != b"\x05\x01\x00":
yield from self.socks_err(f"Unsupported SOCKS5 request: {self.buf!r}", SOCKS5_REP_COMMAND_NOT_SUPPORTED)
return
# Determine message length
atyp = self.buf[3]
message_len: int
if atyp == SOCKS5_ATYP_IPV4_ADDRESS:
message_len = 4 + 4 + 2
elif atyp == SOCKS5_ATYP_IPV6_ADDRESS:
message_len = 4 + 16 + 2
elif atyp == SOCKS5_ATYP_DOMAINNAME:
message_len = 4 + 1 + self.buf[4] + 2
else:
yield from self.socks_err(f"Unknown address type: {atyp}", SOCKS5_REP_ADDRESS_TYPE_NOT_SUPPORTED)
return
# Do we have enough bytes yet?
if len(self.buf) < message_len:
return
# Parse host and port
msg, self.buf = self.buf[:message_len], self.buf[message_len:]
host: str
if atyp == SOCKS5_ATYP_IPV4_ADDRESS:
host = socket.inet_ntop(socket.AF_INET, msg[4:-2])
elif atyp == SOCKS5_ATYP_IPV6_ADDRESS:
host = socket.inet_ntop(socket.AF_INET6, msg[4:-2])
else:
host_bytes = msg[5:-2]
host = host_bytes.decode("ascii", "replace")
port, = struct.unpack("!H", msg[-2:])
# We now have all we need, let's get going.
self.context.server.address = (host, port)
self.child_layer = layer.NextLayer(self.context)
# this already triggers the child layer's Start event,
# but that's not a problem in practice...
err = yield from self.finish_start()
if err:
yield commands.SendData(self.context.client, b"\x05\x04\x00\x01\x00\x00\x00\x00\x00\x00")
yield commands.CloseConnection(self.context.client)
else:
yield commands.SendData(self.context.client, b"\x05\x00\x00\x01\x00\x00\x00\x00\x00\x00")
if self.buf:
yield from self.child_layer.handle_event(events.DataReceived(self.context.client, self.buf))
del self.buf
elif isinstance(event, events.ConnectionClosed):
if self.buf:
yield commands.Log(f"Client closed connection before completing SOCKS5 handshake: {self.buf!r}")
yield commands.CloseConnection(event.connection)
else:
raise AssertionError(f"Unknown event: {event}")

3
test/mitmproxy/addons/test_next_layer.py
View File

@ -78,8 +78,7 @@ class TestNextLayer:
assert isinstance(nl.make_top_layer(ctx), layers.modes.ReverseProxy) assert isinstance(nl.make_top_layer(ctx), layers.modes.ReverseProxy)
tctx.configure(nl, mode="socks5") tctx.configure(nl, mode="socks5")
with pytest.raises(NotImplementedError): assert isinstance(nl.make_top_layer(ctx), layers.modes.Socks5Proxy)
nl.make_top_layer(ctx)
def test_next_layer(self): def test_next_layer(self):
nl = NextLayer() nl = NextLayer()

314
test/mitmproxy/proxy/layers/test_modes.py
View File

@ -3,14 +3,14 @@ import copy
import pytest import pytest
from mitmproxy import platform from mitmproxy import platform
from mitmproxy.proxy.context import Context
from mitmproxy.proxy.layers.http import HTTPMode
from mitmproxy.proxy.commands import CloseConnection, OpenConnection, SendData, GetSocket, Log
from mitmproxy.connection import Client, Server from mitmproxy.connection import Client, Server
from mitmproxy.proxy.events import DataReceived, ConnectionClosed from mitmproxy.proxy.commands import CloseConnection, GetSocket, Log, OpenConnection, SendData
from mitmproxy.proxy.context import Context
from mitmproxy.proxy.events import ConnectionClosed, DataReceived
from mitmproxy.proxy.layer import NextLayer, NextLayerHook from mitmproxy.proxy.layer import NextLayer, NextLayerHook
from mitmproxy.proxy.layers import http, modes, tcp, tls from mitmproxy.proxy.layers import http, modes, tcp, tls
from mitmproxy.proxy.layers.tcp import TcpStartHook, TcpMessageHook from mitmproxy.proxy.layers.http import HTTPMode
from mitmproxy.proxy.layers.tcp import TcpMessageHook, TcpStartHook
from mitmproxy.proxy.layers.tls import ClientTLSLayer, TlsStartHook from mitmproxy.proxy.layers.tls import ClientTLSLayer, TlsStartHook
from mitmproxy.tcp import TCPFlow from mitmproxy.tcp import TCPFlow
from test.mitmproxy.proxy.layers.test_tls import reply_tls_start from test.mitmproxy.proxy.layers.test_tls import reply_tls_start
@ -60,52 +60,52 @@ def test_upstream_https(tctx):
h2_server_settings_ack = Placeholder(bytes) h2_server_settings_ack = Placeholder(bytes)
assert ( assert (
proxy1 proxy1
>> DataReceived(tctx1.client, b"GET http://example.com/ HTTP/1.1\r\nHost: example.com\r\n\r\n") >> DataReceived(tctx1.client, b"GET http://example.com/ HTTP/1.1\r\nHost: example.com\r\n\r\n")
<< NextLayerHook(Placeholder(NextLayer)) << NextLayerHook(Placeholder(NextLayer))
>> reply_next_layer(lambda ctx: http.HttpLayer(ctx, HTTPMode.upstream)) >> reply_next_layer(lambda ctx: http.HttpLayer(ctx, HTTPMode.upstream))
<< OpenConnection(upstream) << OpenConnection(upstream)
>> reply(None) >> reply(None)
<< TlsStartHook(Placeholder()) << TlsStartHook(Placeholder())
>> reply_tls_start(alpn=b"h2") >> reply_tls_start(alpn=b"h2")
<< SendData(upstream, clienthello) << SendData(upstream, clienthello)
) )
assert upstream().address == ("example.mitmproxy.org", 8081) assert upstream().address == ("example.mitmproxy.org", 8081)
assert ( assert (
proxy2 proxy2
>> DataReceived(tctx2.client, clienthello()) >> DataReceived(tctx2.client, clienthello())
<< NextLayerHook(Placeholder(NextLayer)) << NextLayerHook(Placeholder(NextLayer))
>> reply_next_layer(ClientTLSLayer) >> reply_next_layer(ClientTLSLayer)
<< TlsStartHook(Placeholder()) << TlsStartHook(Placeholder())
>> reply_tls_start(alpn=b"h2") >> reply_tls_start(alpn=b"h2")
<< SendData(tctx2.client, serverhello) << SendData(tctx2.client, serverhello)
) )
assert ( assert (
proxy1 proxy1
>> DataReceived(upstream, serverhello()) >> DataReceived(upstream, serverhello())
<< SendData(upstream, request) << SendData(upstream, request)
) )
assert ( assert (
proxy2 proxy2
>> DataReceived(tctx2.client, request()) >> DataReceived(tctx2.client, request())
<< SendData(tctx2.client, tls_finished) << SendData(tctx2.client, tls_finished)
<< NextLayerHook(Placeholder(NextLayer)) << NextLayerHook(Placeholder(NextLayer))
>> reply_next_layer(lambda ctx: http.HttpLayer(ctx, HTTPMode.regular)) >> reply_next_layer(lambda ctx: http.HttpLayer(ctx, HTTPMode.regular))
<< SendData(tctx2.client, h2_client_settings_ack) << SendData(tctx2.client, h2_client_settings_ack)
<< OpenConnection(server) << OpenConnection(server)
>> reply(None) >> reply(None)
<< SendData(server, b'GET / HTTP/1.1\r\nhost: example.com\r\n\r\n') << SendData(server, b'GET / HTTP/1.1\r\nhost: example.com\r\n\r\n')
>> DataReceived(server, b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n") >> DataReceived(server, b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n")
<< CloseConnection(server) << CloseConnection(server)
<< SendData(tctx2.client, response) << SendData(tctx2.client, response)
) )
assert server().address == ("example.com", 80) assert server().address == ("example.com", 80)
assert ( assert (
proxy1 proxy1
>> DataReceived(upstream, tls_finished() + h2_client_settings_ack() + response()) >> DataReceived(upstream, tls_finished() + h2_client_settings_ack() + response())
<< SendData(upstream, h2_server_settings_ack) << SendData(upstream, h2_server_settings_ack)
<< SendData(tctx1.client, b"HTTP/1.1 200 OK\r\ncontent-length: 0\r\n\r\n") << SendData(tctx1.client, b"HTTP/1.1 200 OK\r\ncontent-length: 0\r\n\r\n")
) )
@ -121,17 +121,17 @@ def test_reverse_proxy(tctx, keep_host_header):
tctx.options.mode = "reverse:http://localhost:8000" tctx.options.mode = "reverse:http://localhost:8000"
tctx.options.keep_host_header = keep_host_header tctx.options.keep_host_header = keep_host_header
assert ( assert (
Playbook(modes.ReverseProxy(tctx), hooks=False) Playbook(modes.ReverseProxy(tctx), hooks=False)
>> DataReceived(tctx.client, b"GET /foo HTTP/1.1\r\n" >> DataReceived(tctx.client, b"GET /foo HTTP/1.1\r\n"
b"Host: example.com\r\n\r\n") b"Host: example.com\r\n\r\n")
<< NextLayerHook(Placeholder(NextLayer)) << NextLayerHook(Placeholder(NextLayer))
>> reply_next_layer(lambda ctx: http.HttpLayer(ctx, HTTPMode.transparent)) >> reply_next_layer(lambda ctx: http.HttpLayer(ctx, HTTPMode.transparent))
<< OpenConnection(server) << OpenConnection(server)
>> reply(None) >> reply(None)
<< SendData(server, b"GET /foo HTTP/1.1\r\n" << SendData(server, b"GET /foo HTTP/1.1\r\n"
b"Host: " + (b"example.com" if keep_host_header else b"localhost:8000") + b"\r\n\r\n") b"Host: " + (b"example.com" if keep_host_header else b"localhost:8000") + b"\r\n\r\n")
>> DataReceived(server, b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n") >> DataReceived(server, b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n")
<< SendData(tctx.client, b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n") << SendData(tctx.client, b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n")
) )
assert server().address == ("localhost", 8000) assert server().address == ("localhost", 8000)
@ -155,53 +155,53 @@ def test_reverse_proxy_tcp_over_tls(tctx: Context, monkeypatch, patch, connectio
playbook = Playbook(modes.ReverseProxy(tctx)) playbook = Playbook(modes.ReverseProxy(tctx))
if connection_strategy == "eager": if connection_strategy == "eager":
( (
playbook playbook
<< OpenConnection(tctx.server) << OpenConnection(tctx.server)
>> DataReceived(tctx.client, b"\x01\x02\x03") >> DataReceived(tctx.client, b"\x01\x02\x03")
>> reply(None, to=OpenConnection(tctx.server)) >> reply(None, to=OpenConnection(tctx.server))
) )
else: else:
( (
playbook playbook
>> DataReceived(tctx.client, b"\x01\x02\x03") >> DataReceived(tctx.client, b"\x01\x02\x03")
) )
if patch: if patch:
( (
playbook playbook
<< NextLayerHook(Placeholder(NextLayer)) << NextLayerHook(Placeholder(NextLayer))
>> reply_next_layer(tcp.TCPLayer) >> reply_next_layer(tcp.TCPLayer)
<< TcpStartHook(flow) << TcpStartHook(flow)
>> reply() >> reply()
) )
if connection_strategy == "lazy": if connection_strategy == "lazy":
( (
playbook playbook
<< OpenConnection(tctx.server) << OpenConnection(tctx.server)
>> reply(None) >> reply(None)
) )
assert ( assert (
playbook playbook
<< TcpMessageHook(flow) << TcpMessageHook(flow)
>> reply() >> reply()
<< SendData(tctx.server, data) << SendData(tctx.server, data)
) )
assert data() == b"\x01\x02\x03" assert data() == b"\x01\x02\x03"
else: else:
if connection_strategy == "lazy": if connection_strategy == "lazy":
( (
playbook playbook
<< NextLayerHook(Placeholder(NextLayer)) << NextLayerHook(Placeholder(NextLayer))
>> reply_next_layer(tcp.TCPLayer) >> reply_next_layer(tcp.TCPLayer)
<< TcpStartHook(flow) << TcpStartHook(flow)
>> reply() >> reply()
<< OpenConnection(tctx.server) << OpenConnection(tctx.server)
>> reply(None) >> reply(None)
) )
assert ( assert (
playbook playbook
<< TlsStartHook(Placeholder()) << TlsStartHook(Placeholder())
>> reply_tls_start() >> reply_tls_start()
<< SendData(tctx.server, data) << SendData(tctx.server, data)
) )
assert tls.parse_client_hello(data()).sni == "localhost" assert tls.parse_client_hello(data()).sni == "localhost"
@ -216,25 +216,25 @@ def test_transparent_tcp(tctx: Context, monkeypatch, connection_strategy):
sock = object() sock = object()
playbook = Playbook(modes.TransparentProxy(tctx)) playbook = Playbook(modes.TransparentProxy(tctx))
( (
playbook playbook
<< GetSocket(tctx.client) << GetSocket(tctx.client)
>> reply(sock) >> reply(sock)
) )
if connection_strategy == "lazy": if connection_strategy == "lazy":
assert playbook assert playbook
else: else:
assert ( assert (
playbook playbook
<< OpenConnection(tctx.server) << OpenConnection(tctx.server)
>> reply(None) >> reply(None)
>> DataReceived(tctx.server, b"hello") >> DataReceived(tctx.server, b"hello")
<< NextLayerHook(Placeholder(NextLayer)) << NextLayerHook(Placeholder(NextLayer))
>> reply_next_layer(tcp.TCPLayer) >> reply_next_layer(tcp.TCPLayer)
<< TcpStartHook(flow) << TcpStartHook(flow)
>> reply() >> reply()
<< TcpMessageHook(flow) << TcpMessageHook(flow)
>> reply() >> reply()
<< SendData(tctx.client, b"hello") << SendData(tctx.client, b"hello")
) )
assert flow().messages[0].content == b"hello" assert flow().messages[0].content == b"hello"
assert not flow().messages[0].from_client assert not flow().messages[0].from_client
@ -250,10 +250,10 @@ def test_transparent_failure(tctx: Context, monkeypatch):
monkeypatch.setattr(platform, "original_addr", raise_err) monkeypatch.setattr(platform, "original_addr", raise_err)
assert ( assert (
Playbook(modes.TransparentProxy(tctx), logs=True) Playbook(modes.TransparentProxy(tctx), logs=True)
<< GetSocket(tctx.client) << GetSocket(tctx.client)
>> reply(object()) >> reply(object())
<< Log("Transparent mode failure: RuntimeError('platform-specific error')", "info") << Log("Transparent mode failure: RuntimeError('platform-specific error')", "info")
) )
@ -268,9 +268,117 @@ def test_reverse_eager_connect_failure(tctx: Context):
tctx.options.connection_strategy = "eager" tctx.options.connection_strategy = "eager"
playbook = Playbook(modes.ReverseProxy(tctx)) playbook = Playbook(modes.ReverseProxy(tctx))
assert ( assert (
playbook playbook
<< OpenConnection(tctx.server) << OpenConnection(tctx.server)
>> reply("IPoAC unstable") >> reply("IPoAC unstable")
<< CloseConnection(tctx.client) << CloseConnection(tctx.client)
>> ConnectionClosed(tctx.client) >> ConnectionClosed(tctx.client)
)
def test_transparent_eager_connect_failure(tctx: Context, monkeypatch):
"""Test that we recover from a transparent mode resolve error."""
tctx.options.connection_strategy = "eager"
monkeypatch.setattr(platform, "original_addr", lambda sock: ("address", 22))
assert (
Playbook(modes.TransparentProxy(tctx), logs=True)
<< GetSocket(tctx.client)
>> reply(object())
<< OpenConnection(tctx.server)
>> reply("something something")
<< CloseConnection(tctx.client)
>> ConnectionClosed(tctx.client)
)
CLIENT_HELLO = b"\x05\x01\x00"
SERVER_HELLO = b"\x05\x00"
@pytest.mark.parametrize("address,packed", [
("127.0.0.1", b"\x01\x7f\x00\x00\x01"),
("::1", b"\x04\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01"),
("example.com", b"\x03\x0bexample.com"),
])
def test_socks5_success(address: str, packed: bytes, tctx: Context):
tctx.options.connection_strategy = "eager"
playbook = Playbook(modes.Socks5Proxy(tctx))
server = Placeholder(Server)
nextlayer = Placeholder(NextLayer)
assert (
playbook
>> DataReceived(tctx.client, CLIENT_HELLO)
<< SendData(tctx.client, SERVER_HELLO)
>> DataReceived(tctx.client, b"\x05\x01\x00" + packed + b"\x12\x34applicationdata")
<< OpenConnection(server)
>> reply(None)
<< SendData(tctx.client, b"\x05\x00\x00\x01\x00\x00\x00\x00\x00\x00")
<< NextLayerHook(nextlayer)
)
assert server().address == (address, 0x1234)
assert nextlayer().data_client() == b"applicationdata"
def test_socks5_trickle(tctx: Context):
playbook = Playbook(modes.Socks5Proxy(tctx))
for x in CLIENT_HELLO:
playbook >> DataReceived(tctx.client, bytes([x]))
playbook << SendData(tctx.client, b"\x05\x00")
for x in b"\x05\x01\x00\x01\x7f\x00\x00\x01\x12\x34":
playbook >> DataReceived(tctx.client, bytes([x]))
assert playbook << SendData(tctx.client, b"\x05\x00\x00\x01\x00\x00\x00\x00\x00\x00")
@pytest.mark.parametrize("data,err,msg", [
(b"GET / HTTP/1.1",
None,
"Probably not a SOCKS request but a regular HTTP request. Invalid SOCKS version. Expected 0x05, got 0x47"),
(b"abcd",
None,
"Invalid SOCKS version. Expected 0x05, got 0x61"),
(b"\x05\x01\x02",
b"\x05\xFF\x00\x01\x00\x00\x00\x00\x00\x00",
"mitmproxy only supports SOCKS without authentication"),
(CLIENT_HELLO + b"\x05\x02\x00\x01\x7f\x00\x00\x01\x12\x34",
SERVER_HELLO + b"\x05\x07\x00\x01\x00\x00\x00\x00\x00\x00",
r"Unsupported SOCKS5 request: b'\x05\x02\x00\x01\x7f\x00\x00\x01\x124'"),
(CLIENT_HELLO + b"\x05\x01\x00\xFF\x00\x00",
SERVER_HELLO + b"\x05\x08\x00\x01\x00\x00\x00\x00\x00\x00",
r"Unknown address type: 255"),
])
def test_socks5_err(data: bytes, err: bytes, msg: str, tctx: Context):
playbook = (
Playbook(modes.Socks5Proxy(tctx), logs=True)
>> DataReceived(tctx.client, data)
)
if err:
playbook << SendData(tctx.client, err)
playbook << CloseConnection(tctx.client)
playbook << Log(msg)
assert playbook
def test_socks5_eager_err(tctx: Context):
tctx.options.connection_strategy = "eager"
server = Placeholder(Server)
assert (
Playbook(modes.Socks5Proxy(tctx))
>> DataReceived(tctx.client, CLIENT_HELLO)
<< SendData(tctx.client, SERVER_HELLO)
>> DataReceived(tctx.client, b"\x05\x01\x00\x01\x7f\x00\x00\x01\x12\x34")
<< OpenConnection(server)
>> reply("out of socks")
<< SendData(tctx.client, b"\x05\x04\x00\x01\x00\x00\x00\x00\x00\x00")
<< CloseConnection(tctx.client)
)
def test_socks5_premature_close(tctx: Context):
assert (
Playbook(modes.Socks5Proxy(tctx), logs=True)
>> DataReceived(tctx.client, b"\x05")
>> ConnectionClosed(tctx.client)
<< Log(r"Client closed connection before completing SOCKS5 handshake: b'\x05'")
<< CloseConnection(tctx.client)
) )

17
test/mitmproxy/proxy/layers/test_socks5_fuzz.py Normal file
View File

@ -0,0 +1,17 @@
from hypothesis import given
from hypothesis.strategies import binary
from mitmproxy import options
from mitmproxy.connection import Client
from mitmproxy.proxy.context import Context
from mitmproxy.proxy.events import DataReceived
from mitmproxy.proxy.layers.modes import Socks5Proxy
opts = options.Options()
tctx = Context(Client(("client", 1234), ("127.0.0.1", 8080), 1605699329), opts)
@given(binary())
def test_socks5_fuzz(data):
layer = Socks5Proxy(tctx)
list(layer.handle_event(DataReceived(tctx.client, data)))