Simple SOCKS4 proxy in Python
Disclaimer: The purpose of this article is to study the SOCKS4 protocol. The presented implementation is purely educational, does not use client authentication, encryption or traffic masking, and therefore cannot be used to bypass any blocking.
Brief description of the protocol
The SOCKS4 protocol is designed to proxy TCP connections.
The client establishes a connection with the proxy and sends one request specifying a command and its parameters; the proxy responds with one or two response messages and, if the command was successfully executed, begins sending data between the client and the target.
The client request has the form: VERSION (1 byte), COMMAND (1 byte), PORT (2 bytes), IPv4 ADDRESS (4 bytes), USER_ID (null-terminated string).
The SOCKS4a protocol extension allows the client to delegate domain name resolution to the proxy; that is, an optional DOMAIN parameter (a null-terminated string) may be present at the end of the request.
Two commands are supported:
CONNECT – the client connects to the proxy and tells it where to connect; the proxy connects to the target, confirms the connection success to the client, and starts sending data between the client and the target.
BIND – the client connects to the proxy; the proxy prepares to accept an incoming connection and tells the client the address and port to connect to; the client passes this information to the target; the target connects to the proxy; the proxy confirms a successful connection to the client and begins sending data between the client and the target.
The proxy response looks like: VERSION (1 byte), STATUS (1 byte), PORT (2 bytes), IPv4 ADDRESS (4 bytes).
Detailed descriptions of the SOCKS4 protocol and the SOCKS4a extension:
https://www.openssh.com/txt/socks4.protocol
https://www.openssh.com/txt/socks4a.protocol
The boring part of implementation
We import the necessary libraries:
import sys, os, argparse, traceback, logging
import enum, struct, socket, socketserver, selectLet's write the application's startup code that parses the command line arguments (the address and port at which the proxy will accept client connections, the logging level, the optional log file name) and configures logging:
def parse_args():
parser = argparse.ArgumentParser(description = 'Simple SOCKS4 server')
parser.add_argument('--log-level', action = 'store', type = str,
dest="log_level", default="DEBUG", help = 'Log level',
choices = ['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'])
parser.add_argument('--log-path', action = 'store', type = str,
dest="log_path", default = None, help = 'Log file path')
parser.add_argument('--host', action = 'store', type = str,
dest="host", default="127.0.0.1", help = 'Server IP or hostname')
parser.add_argument('--port', action = 'store', type = int,
dest="port", default = 1080, help = 'Port to listen')
return parser.parse_args()
if __name__ == '__main__':
try:
args = parse_args()
logging.basicConfig(format="%(asctime)s [%(levelname)s]: %(message)s",
level = logging.getLevelName(args.log_level.upper()), filename = args.log_path)
logging.info(f'Starting server at {args.host}:{args.port}')
# TODO
except KeyboardInterrupt as e:
logging.info('Shutting down')
sys.exit(0)
except Exception as e:
logging.error(f'{str(e)}\n{traceback.format_exc()}')
sys.exit(1)The proxy will be based on the standard library socketservernamely, the ThreadingTCPServer class. It will accept incoming connections from clients and run our handler in separate threads. Let's add a stub for the client connection handler:
class SOCKS4Handler(socketserver.StreamRequestHandler):
def handle(self):
logging.info('Connection accepted: %s:%s' % self.client_address)
try:
# TODO
pass
finally:
logging.info('Connection terminated: %s:%s' % self.client_address)
self.server.close_request(self.request)In the startup code, after initializing logging, we will add the launch of the server with our handler:
with socketserver.ThreadingTCPServer((args.host, args.port), SOCKS4Handler) as server:
server.serve_forever() # interrupt with Ctrl+CLet's define a couple of auxiliary constants – a timeout for sockets (in seconds) and the number of bytes read by the proxy (in working mode, when traffic is redirected between the client and the target) from the socket at a time:
SOCKET_TIMEOUT_SEC = 120
STREAM_BUFFER_SIZE = 4096The fun part of implementation
In accordance with the full description of the protocol, we define sets of constants – request and response versions, command codes, response statuses (in this implementation, only the first two will be used):
class SOCKS4_VER(enum.IntEnum):
REQUEST = 0x04
REPLY = 0x00
class SOCKS4_CMD(enum.IntEnum):
# Establish a TCP/IP stream connection
CONNECT = 0x01
# Establish a TCP/IP port binding
BIND = 0x02
class SOCKS4_REPLY(enum.IntEnum):
# Request granted
GRANTED = 0x5A
# Request rejected or failed
FAILED_OR_REJECTED = 0x5B
# Request failed because client is not running identd (or not reachable from server)
FAILED_NO_IDENTD = 0x5C
# Request failed because client's identd could not confirm the user ID in the request
FAILED_BAD_IDENTD = 0x5Dsocket.recv(bufsize [, flags]) may return fewer bytes than requested. To simplify the code parsing the client's request, we'll write helper functions – one that reads exactly n bytes from the socket (or throws an exception), and one that reads a null-terminated string:
def recvall(sock, n, /):
data, count = [], 0
while count < n:
packet = sock.recv(n - count)
if not packet:
raise EOFError(f'Read {count} bytes from socket, expected {n} bytes')
data.append(packet)
count += len(packet)
return b''.join(data)
def recv_null_terminated_string(sock, /):
data = []
while True:
char = recvall(sock, 1)
if char == b'\x00':
return b''.join(data).decode('utf-8')
data.append(char)We will analyze the client request as follows: we read the fixed-length part of the request (8 bytes – VERSION, COMMAND, PORT, IPv4 ADDRESS), we read the mandatory part of the request with variable length (null-terminated string USER_ID). If ADDRESS is 0.0.0.x (where x != 0) – then we are dealing with a SOCKS4a request; in this case, we read the optional part of the request with variable length (null-terminated string DOMAIN). In case of obviously erroneous situations (invalid VERSION, COMMAND, empty DOMAIN for SOCKS4a request), we will throw an exception. Let's add a method to the SOCKS4Handler request handler:
def read_socks4_request(self):
# fixed-length: VER(1), CMD(1), DSTPORT(2), DSTADDR(4)
header = recvall(self.connection, 8)
ver, cmd, dst_port = struct.unpack('!BBH', header[0 : 4])
dst_addr = socket.inet_ntop(socket.AF_INET, header[4 : 8])
if ver != SOCKS4_VER.REQUEST:
raise SOCKS4ProtocolError(f'Unknown version: {ver}')
if (cmd != SOCKS4_CMD.CONNECT) and (cmd != SOCKS4_CMD.BIND):
raise SOCKS4ProtocolError(f'Unknown command: {cmd}')
# variable-length: USERID, DOMAIN (optional)
user_id = recv_null_terminated_string(self.connection)
dst_domain = None # SOCKS 4A
if (header[4 : 7] == b'\x00\x00\x00') and (header[7 : 8] != b'\x00'):
dst_domain = recv_null_terminated_string(self.connection)
if len(dst_domain) == 0:
raise SOCKS4ProtocolError(f'Empty domain field in SOCKS 4A request: {dst_addr}')
return (ver, cmd, dst_port, dst_addr, user_id, dst_domain)Let's also add an exception thrown in read_socks4_request():
class SOCKS4ProtocolError(Exception):
passThe response to the client always has a fixed length of 8 bytes. The values of the PORT and IPv4 ADDRESS fields are meaningless if the response status is different from SOCKS4_REPLY.GRANTED, i.e. you can set their default values. Let's add a method to the SOCKS4Handler handler for sending responses to the client:
def write_socks4_reply(self, status, addr="0.0.0.0", port = 0, /):
logging.debug(f'Reply: {status};{addr};{port}')
reply = struct.pack('!BBH', SOCKS4_VER.REPLY, status, port)
reply += socket.inet_pton(socket.AF_INET, addr)
self.connection.sendall(reply)We will also add a helper method to set the timeout and keep alive flag for sockets:
def tune_socket_options(self, sock, /):
sock.settimeout(SOCKET_TIMEOUT_SEC)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)In the try block of the handle() method, we will add the basis of the future proxy – receiving a client request, resolving a domain name (if a SOCKS4a request has arrived), branching depending on the command:
self.tune_socket_options(self.connection)
(ver, cmd, dst_port, dst_addr, user_id, dst_domain) = self.read_socks4_request()
logging.debug(f'Request: {ver};{cmd};{dst_port};{dst_addr};{user_id};{dst_domain}')
if dst_domain:
dst_addr = socket.gethostbyname(dst_domain)
logging.debug(f'Resolved {dst_domain} into {dst_addr}')
if cmd == SOCKS4_CMD.CONNECT:
pass # TODO
elif cmd == SOCKS4_CMD.BIND:
pass # TODOThis code can throw a number of exceptions – EOFError (from the recvall() function), socket.gaierror (from socket.gethostbyname()), SOCKS4ProtocolError (from the read_socks4_request() method), etc. Depending on the exception, you can simply close the connection, or (in most cases) you need to send a response to the client with the SOCKS4_REPLY.FAILED_OR_REJECTED status, but if the proxy has entered working mode (transfers traffic between the client and the target), sending responses should be prohibited. Let's add a flag before the try block of the handle() method that allows sending responses:
reply_on_fail = TrueWe will supplement the try block of the handle() method with except blocks:
except EOFError as e:
logging.warning(f'{str(e)}\n{traceback.format_exc()}')
except socket.gaierror:
logging.warning(f'Unable to resolve domain: {dst_domain}')
self.write_socks4_reply(SOCKS4_REPLY.FAILED_OR_REJECTED)
except Exception as e:
logging.warning(f'{str(e)}\n{traceback.format_exc()}')
if reply_on_fail:
self.write_socks4_reply(SOCKS4_REPLY.FAILED_OR_REJECTED)We will add a stub to the SOCKS4Handler handler, which will then forward traffic between the client and the target:
def stream_tcp(self, socket_a, socket_b, /):
# TODO
passWe'll come back to this stub later, but for now let's finish working on the handle() method by adding processing for SOCKS4_CMD.CONNECT and SOCKS4_CMD.BIND.
Processing of the CONNECT command is reduced to
creating a socket
attempting to connect to the target (target address and port obtained from the client request) and sending a response with the SOCKS4_REPLY.GRANTED status
after which the proxy begins to forward traffic between the pair of sockets.
If the connection cannot be established, the exception handler will send a response to the client with the SOCKS4_REPLY.FAILED_OR_REJECTED status.
if cmd == SOCKS4_CMD.CONNECT:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as dst_socket:
self.tune_socket_options(dst_socket)
dst_socket.connect((dst_addr, dst_port))
(bind_addr, bind_port) = dst_socket.getsockname()
reply_on_fail = False
self.write_socks4_reply(SOCKS4_REPLY.GRANTED, bind_addr, bind_port)
self.stream_tcp(self.connection, dst_socket)Processing the BIND command (well, as far as I understand) comes down to
creating a socket listening on a random port
sending the address and port (on which the proxy is waiting for a connection) in response with the SOCKS4_REPLY.GRANTED status to the client
accepting an incoming connection from the target
checking whether the target address matches the address specified in the client's request
another response to the client with the SOCKS4_REPLY.GRANTED status, the address and port of the target connected to the proxy
after which the proxy begins to forward traffic between the pair of sockets.
If something goes wrong, the exception handler will send a response to the client with the SOCKS4_REPLY.FAILED_OR_REJECTED status.
elif cmd == SOCKS4_CMD.BIND:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as bind_socket:
self.tune_socket_options(bind_socket)
bind_addr = socket.gethostbyname(socket.gethostname())
bind_socket.bind((bind_addr, 0)) # any port
bind_socket.listen(1)
(bind_addr, bind_port) = bind_socket.getsockname()
self.write_socks4_reply(SOCKS4_REPLY.GRANTED, bind_addr, bind_port)
(peer_socket, (peer_addr, peer_port)) = bind_socket.accept()
with peer_socket:
self.tune_socket_options(peer_socket)
if dst_addr != peer_addr:
raise SOCKS4ProtocolError(f'Got connection from {peer_addr}; expected from {dst_addr}')
reply_on_fail = False
self.write_socks4_reply(SOCKS4_REPLY.GRANTED, peer_addr, peer_port)
self.stream_tcp(self.connection, peer_socket)This completes the main part of the proxy that directly concerns the SOCKS4 protocol.
Forwarding traffic between the client and the target
What do we need to do conceptually in the stream_tcp() method?
Find out if there is data to read in any of the sockets.
Read data, write it to buffers.
Send data from buffers to the corresponding sockets.
Remove the sent portion of data from the buffers.
If an exception occurs while reading or sending data, terminate the data transfer cycle.
Attempt to send the remainder of the data from the buffers, ignoring exceptions.
Let's add a method to SOCKS4Handler that generates a description of a pair of sockets that will be written to the log:
def get_stream_desc(self, socket_a, socket_b, /):
addr_a, port_a = socket_a.getpeername()
addr_b, port_b = socket_b.getpeername()
return f'{addr_a}:{port_a} <--> {addr_b}:{port_b}'Let's write helper functions for securely receiving and sending data to sockets:
safe_recv() takes a socket, a buffer of data, and a loop-complete flag; returns a buffer padded with new data and a potentially modified loop-complete flag.
safe_send() takes a socket, a buffer of data, and a send loop completion flag; returns a buffer with the sent data removed and the potentially modified send loop completion flag.
safe_sendfinal() takes a socket and a buffer from which to attempt to send data, ignoring errors.
def safe_recv(sock, buf, done, /):
try:
packet = sock.recv(STREAM_BUFFER_SIZE)
if len(packet) > 0:
buf += packet
else:
done = True
except Exception as e:
logging.warning(f'{str(e)}\n{traceback.format_exc()}')
done = True
return buf, done
def safe_send(sock, buf, done, /):
try:
bytes_sent = sock.send(buf)
buf = buf[bytes_sent : ]
except Exception as e:
logging.warning(f'{str(e)}\n{traceback.format_exc()}')
done = True
return buf, done
def safe_sendfinal(sock, buf):
try:
sock.sendall(buf)
except:
passUsing these functions, the stream_tcp() method of the SOCKS4Handler will look like this:
def stream_tcp(self, socket_a, socket_b, /):
stream_info = self.get_stream_desc(socket_a, socket_b)
logging.debug(f'Starting stream: {stream_info}')
sockets_list = [socket_a, socket_b]
buf_a2b, buf_b2a = b'', b''
done = False
while not done:
read_ready, write_ready, _ = select.select(sockets_list, sockets_list, [], 0.5)
if socket_a in read_ready:
buf_a2b, done = safe_recv(socket_a, buf_a2b, done)
if socket_b in read_ready:
buf_b2a, done = safe_recv(socket_b, buf_b2a, done)
if socket_a in write_ready:
buf_b2a, done = safe_send(socket_a, buf_b2a, done)
if socket_b in write_ready:
buf_a2b, done = safe_send(socket_b, buf_a2b, done)
logging.debug(f'Finalizing stream: {stream_info}')
safe_sendfinal(socket_a, buf_b2a)
safe_sendfinal(socket_b, buf_a2b)
logging.debug(f'Stopped stream: {stream_info}')This is not the best implementation, but even in this form uTorrent was able to push a data stream of 0.5 Gbps through the proxy (with a monstrous CPU load).
How to test?
The easiest way to check is the CONNECT command:
curl -v --socks4 127.0.0.1:1080 -U userid:ignored ya.ru
curl -v --socks4a 127.0.0.1:1080 -U userid:ignored ya.ruIn the proxy logs we will see information about the reception/breakdown of connections, the parsed client request and the proxy response, etc.:
2024-08-15 22:49:21,216 [INFO]: Starting server at 127.0.0.1:1080
2024-08-15 22:49:42,229 [INFO]: Connection accepted: 127.0.0.1:58551
2024-08-15 22:49:42,270 [DEBUG]: Request: 4;1;443;77.88.55.242;username;None
2024-08-15 22:49:42,280 [DEBUG]: Reply: 90;192.168.10.101;58554
2024-08-15 22:49:42,280 [DEBUG]: Starting stream: 127.0.0.1:58551 <--> 77.88.55.242:443
2024-08-15 22:49:42,630 [DEBUG]: Finalizing stream: 127.0.0.1:58551 <--> 77.88.55.242:443
2024-08-15 22:49:42,630 [DEBUG]: Stopped stream: 127.0.0.1:58551 <--> 77.88.55.242:443
2024-08-15 22:49:42,630 [INFO]: Connection terminated: 127.0.0.1:58551
2024-08-15 22:50:01,330 [INFO]: Connection accepted: 127.0.0.1:58558
2024-08-15 22:50:01,332 [DEBUG]: Request: 4;1;443;0.0.0.1;username;ya.ru
2024-08-15 22:50:01,339 [DEBUG]: Resolved ya.ru into 77.88.55.242
2024-08-15 22:50:01,350 [DEBUG]: Reply: 90;192.168.10.101;58559
2024-08-15 22:50:01,350 [DEBUG]: Starting stream: 127.0.0.1:58558 <--> 77.88.55.242:443
2024-08-15 22:50:01,466 [DEBUG]: Finalizing stream: 127.0.0.1:58558 <--> 77.88.55.242:443
2024-08-15 22:50:01,466 [DEBUG]: Stopped stream: 127.0.0.1:58558 <--> 77.88.55.242:443
2024-08-15 22:50:01,466 [INFO]: Connection terminated: 127.0.0.1:58558
(Ctrl+C)
2024-08-15 22:50:24,247 [INFO]: Shutting downFull proxy source code
###################################################################################################
import sys, os, argparse, traceback, logging
import enum, struct, socket, socketserver, select
###################################################################################################
SOCKET_TIMEOUT_SEC = 120
STREAM_BUFFER_SIZE = 4096
###################################################################################################
# SOCKS 4 / 4A (TCP)
# https://www.openssh.com/txt/socks4.protocol
# https://www.openssh.com/txt/socks4a.protocol
class SOCKS4_VER(enum.IntEnum):
REQUEST = 0x04
REPLY = 0x00
class SOCKS4_CMD(enum.IntEnum):
# Establish a TCP/IP stream connection
CONNECT = 0x01
# Establish a TCP/IP port binding
BIND = 0x02
class SOCKS4_REPLY(enum.IntEnum):
# Request granted
GRANTED = 0x5A
# Request rejected or failed
FAILED_OR_REJECTED = 0x5B
# Request failed because client is not running identd (or not reachable from server)
FAILED_NO_IDENTD = 0x5C
# Request failed because client's identd could not confirm the user ID in the request
FAILED_BAD_IDENTD = 0x5D
###################################################################################################
def recvall(sock, n, /):
data, count = [], 0
while count < n:
packet = sock.recv(n - count)
if not packet:
raise EOFError(f'Read {count} bytes from socket, expected {n} bytes')
data.append(packet)
count += len(packet)
return b''.join(data)
def recv_null_terminated_string(sock, /):
data = []
while True:
char = recvall(sock, 1)
if char == b'\x00':
return b''.join(data).decode('utf-8')
data.append(char)
###################################################################################################
def safe_recv(sock, buf, done, /):
try:
packet = sock.recv(STREAM_BUFFER_SIZE)
if len(packet) > 0:
buf += packet
else:
done = True
except Exception as e:
logging.warning(f'{str(e)}\n{traceback.format_exc()}')
done = True
return buf, done
def safe_send(sock, buf, done, /):
try:
bytes_sent = sock.send(buf)
buf = buf[bytes_sent : ]
except Exception as e:
logging.warning(f'{str(e)}\n{traceback.format_exc()}')
done = True
return buf, done
def safe_sendfinal(sock, buf):
try:
sock.sendall(buf)
except:
pass
###################################################################################################
class SOCKS4ProtocolError(Exception):
pass
class SOCKS4Handler(socketserver.StreamRequestHandler):
def tune_socket_options(self, sock, /):
sock.settimeout(SOCKET_TIMEOUT_SEC)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
def read_socks4_request(self):
# fixed-length: VER(1), CMD(1), DSTPORT(2), DSTADDR(4)
header = recvall(self.connection, 8)
ver, cmd, dst_port = struct.unpack('!BBH', header[0 : 4])
dst_addr = socket.inet_ntop(socket.AF_INET, header[4 : 8])
if ver != SOCKS4_VER.REQUEST:
raise SOCKS4ProtocolError(f'Unknown version: {ver}')
if (cmd != SOCKS4_CMD.CONNECT) and (cmd != SOCKS4_CMD.BIND):
raise SOCKS4ProtocolError(f'Unknown command: {cmd}')
# variable-length: USERID, DOMAIN (optional)
user_id = recv_null_terminated_string(self.connection)
dst_domain = None # SOCKS 4A
if (header[4 : 7] == b'\x00\x00\x00') and (header[7 : 8] != b'\x00'):
dst_domain = recv_null_terminated_string(self.connection)
if len(dst_domain) == 0:
raise SOCKS4ProtocolError(f'Empty domain field in SOCKS 4A request: {dst_addr}')
return (ver, cmd, dst_port, dst_addr, user_id, dst_domain)
def write_socks4_reply(self, status, addr="0.0.0.0", port = 0, /):
logging.debug(f'Reply: {status};{addr};{port}')
reply = struct.pack('!BBH', SOCKS4_VER.REPLY, status, port)
reply += socket.inet_pton(socket.AF_INET, addr)
self.connection.sendall(reply)
def get_stream_desc(self, socket_a, socket_b, /):
addr_a, port_a = socket_a.getpeername()
addr_b, port_b = socket_b.getpeername()
return f'{addr_a}:{port_a} <--> {addr_b}:{port_b}'
def stream_tcp(self, socket_a, socket_b, /):
stream_info = self.get_stream_desc(socket_a, socket_b)
logging.debug(f'Starting stream: {stream_info}')
sockets_list = [socket_a, socket_b]
buf_a2b, buf_b2a = b'', b''
done = False
while not done:
read_ready, write_ready, _ = select.select(sockets_list, sockets_list, [], 0.5)
if socket_a in read_ready:
buf_a2b, done = safe_recv(socket_a, buf_a2b, done)
if socket_b in read_ready:
buf_b2a, done = safe_recv(socket_b, buf_b2a, done)
if socket_a in write_ready:
buf_b2a, done = safe_send(socket_a, buf_b2a, done)
if socket_b in write_ready:
buf_a2b, done = safe_send(socket_b, buf_a2b, done)
logging.debug(f'Finalizing stream: {stream_info}')
safe_sendfinal(socket_a, buf_b2a)
safe_sendfinal(socket_b, buf_a2b)
logging.debug(f'Stopped stream: {stream_info}')
def handle(self):
logging.info('Connection accepted: %s:%s' % self.client_address)
reply_on_fail = True
try:
self.tune_socket_options(self.connection)
(ver, cmd, dst_port, dst_addr, user_id, dst_domain) = self.read_socks4_request()
logging.debug(f'Request: {ver};{cmd};{dst_port};{dst_addr};{user_id};{dst_domain}')
if dst_domain:
dst_addr = socket.gethostbyname(dst_domain)
logging.debug(f'Resolved {dst_domain} into {dst_addr}')
if cmd == SOCKS4_CMD.CONNECT:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as dst_socket:
self.tune_socket_options(dst_socket)
dst_socket.connect((dst_addr, dst_port))
(bind_addr, bind_port) = dst_socket.getsockname()
reply_on_fail = False
self.write_socks4_reply(SOCKS4_REPLY.GRANTED, bind_addr, bind_port)
self.stream_tcp(self.connection, dst_socket)
elif cmd == SOCKS4_CMD.BIND:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as bind_socket:
self.tune_socket_options(bind_socket)
bind_addr = socket.gethostbyname(socket.gethostname())
bind_socket.bind((bind_addr, 0)) # any port
bind_socket.listen(1)
(bind_addr, bind_port) = bind_socket.getsockname()
self.write_socks4_reply(SOCKS4_REPLY.GRANTED, bind_addr, bind_port)
(peer_socket, (peer_addr, peer_port)) = bind_socket.accept()
with peer_socket:
self.tune_socket_options(peer_socket)
if dst_addr != peer_addr:
raise SOCKS4ProtocolError(f'Got connection from {peer_addr}; expected from {dst_addr}')
reply_on_fail = False
self.write_socks4_reply(SOCKS4_REPLY.GRANTED, peer_addr, peer_port)
self.stream_tcp(self.connection, peer_socket)
except EOFError as e:
logging.warning(f'{str(e)}\n{traceback.format_exc()}')
except socket.gaierror:
logging.warning(f'Unable to resolve domain: {dst_domain}')
self.write_socks4_reply(SOCKS4_REPLY.FAILED_OR_REJECTED)
except Exception as e:
logging.warning(f'{str(e)}\n{traceback.format_exc()}')
if reply_on_fail:
self.write_socks4_reply(SOCKS4_REPLY.FAILED_OR_REJECTED)
finally:
logging.info('Connection terminated: %s:%s' % self.client_address)
self.server.close_request(self.request)
###################################################################################################
def parse_args():
parser = argparse.ArgumentParser(description = 'Simple SOCKS4 server')
parser.add_argument('--log-level', action = 'store', type = str,
dest="log_level", default="DEBUG", help = 'Log level',
choices = ['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'])
parser.add_argument('--log-path', action = 'store', type = str,
dest="log_path", default = None, help = 'Log file path')
parser.add_argument('--host', action = 'store', type = str,
dest="host", default="127.0.0.1", help = 'Server IP or hostname')
parser.add_argument('--port', action = 'store', type = int,
dest="port", default = 1080, help = 'Port to listen')
return parser.parse_args()
if __name__ == '__main__':
try:
args = parse_args()
logging.basicConfig(format="%(asctime)s [%(levelname)s]: %(message)s",
level = logging.getLevelName(args.log_level.upper()), filename = args.log_path)
logging.info(f'Starting server at {args.host}:{args.port}')
with socketserver.ThreadingTCPServer((args.host, args.port), SOCKS4Handler) as server:
server.serve_forever() # interrupt with Ctrl+C
except KeyboardInterrupt as e:
logging.info('Shutting down')
sys.exit(0)
except Exception as e:
logging.error(f'{str(e)}\n{traceback.format_exc()}')
sys.exit(1)
###################################################################################################
