Disclaimer: Целью статьи является изучение протокола SOCKS4. Представленная реализация является исключительно учебной, не использует аутентификацию клиентов, шифрование или маскировку трафика, вследствие чего не может быть использована для обхода каких-либо блокировок.

Краткое описание протокола

Протокол SOCKS4 предназначен для проксирования TCP соединений.

Клиент устанавливает соединение с прокси и присылает один запрос с указанием команды и её параметров; прокси отвечает одним или двумя ответными сообщениями, и, если удалось выполнить команду, начинает пересылать данные между клиентом и целью.

Запрос клиента имеет вид: VERSION (1 байт), COMMAND (1 байт), PORT (2 байта), IPv4 ADDRESS (4 байта), USER_ID (нуль-терминированная строка).

Расширение протокола SOCKS4a позволяет клиенту делегировать прокси разрешение доменных имён; то есть, в конце запроса может присутствовать необязательный параметр DOMAIN (нуль-терминированная строка).

Поддерживаются две команды:

CONNECT - клиент соединяется с прокси и говорит, куда подключиться; прокси соединяется с целью, подтверждает клиенту успех соединения и начинает пересылать данные между клиентом и целью.

BIND - клиент соединяется с прокси; прокси готовится к приёму входящего соединения и сообщает клиенту адрес и порт для подключения; клиент передаёт эту информацию цели; цель подключается к прокси; прокси подтверждает клиенту успешное соединение и начинает пересылать данные между клиентом и целью.

Ответ прокси имеет вид: VERSION (1 байт), STATUS (1 байт), PORT (2 байта), IPv4 ADDRESS (4 байта).

Подробные описания протокола SOCKS4 и расширения SOCKS4a:

https://www.openssh.com/txt/socks4.protocol

https://www.openssh.com/txt/socks4a.protocol

Скучная часть реализации

Импортируем необходимые библиотеки:

import sys, os, argparse, traceback, logging
import enum, struct, socket, socketserver, select

Напишем стартовый код приложения, разбирающий аргументы командной строки (адрес и порт, по которым прокси будет принимать соединения клиентов, уровень логгирования, необязательное имя лог-файла) и настраивающий логгирование:

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)

Основой прокси выступит стандартная библиотека socketserver, а именно, класс ThreadingTCPServer. Он будет принимать входящие соединения от клиентов и запускать наш обработчик в отдельных потоках. Добавим заглушку обработчика клиентских соединений:

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)

В стартовый код, после инициализации логгирования, добавим запуск сервера с нашим обработчиком:

        with socketserver.ThreadingTCPServer((args.host, args.port), SOCKS4Handler) as server:
            server.serve_forever() # interrupt with Ctrl+C

Определим пару вспомогательных констант - таймаут для сокетов (в секундах) и кол-во байт, вычитываемых прокси (в рабочем режиме, когда трафик перенаправляется между клиентом и целью) из сокета за один раз:

SOCKET_TIMEOUT_SEC = 120
STREAM_BUFFER_SIZE = 4096

Нескучная часть реализации

В соответствии с полным описанием протокола определим наборы констант - версии запроса и ответа, коды команд, статусы ответов (в данной реализации будут использоваться только первые два):

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

socket.recv(bufsize [, flags]) может вернуть меньше байт, чем запрошено. Чтобы упростить код, разбирающий запрос клиента, напишем вспомогательные функции - вычитывающую строго n байт из сокета (или выбрасывающую исключение), и вычитывающую нуль-терминированную строку:

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)

Разбирать запрос клиента будем так: вычитываем часть запроса с фиксированной длиной (8 байт - VERSION, COMMAND, PORT, IPv4 ADDRESS), вычитываем обязательную часть запроса с переменной длиной (нуль-терминированная строка USER_ID). Если ADDRESS имеет вид 0.0.0.x (причём x != 0) - то имеем дело с SOCKS4a запросом; в таком случае вычитываем необязательную часть запроса с переменной длиной (нуль-терминированная строка DOMAIN). В случае заведомо ошибочных ситуаций (недопустимые VERSION, COMMAND, пустой DOMAIN для SOCKS4a запроса) будем кидать исключение. Добавим метод в обработчик запроса SOCKS4Handler:

    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)

Также добавим исключение, выбрасываемое в read_socks4_request():

class SOCKS4ProtocolError(Exception):
    pass

Ответ клиенту всегда имеет фиксированную длину 8 байт. Значения полей PORT и IPv4 ADDRESS не имеют смысла, если статус ответа отличен от SOCKS4_REPLY.GRANTED, т.е. можно задать им значения по умолчанию. Добавим в обработчик SOCKS4Handler метод для отправки ответов клиенту:

    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)

Также добавим вспомогательный метод для установки таймаута и флага keep alive для сокетов:

    def tune_socket_options(self, sock, /):
        sock.settimeout(SOCKET_TIMEOUT_SEC)
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)

В блок try метода handle() добавим основу будущего прокси - приём запроса клиента, разрешение доменного имени (если пришел SOCKS4a запрос), ветвление в зависимости от команды:

            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 # TODO

Этот код может кидать ряд исключений - EOFError (из функции recvall()), socket.gaierror (из socket.gethostbyname()), SOCKS4ProtocolError (из метода read_socks4_request()) и т.д. В зависимости от исключения можно просто закрыть соединение, или (в большинстве случаев) нужно отправить клиенту ответ со статусом SOCKS4_REPLY.FAILED_OR_REJECTED, однако если прокси вышел в рабочий режим (перекидывает трафик между клиентом и целью) - отправку ответов следует запретить. Добавим перед блоком try метода handle() флажок, разрешающий отправку ответов:

        reply_on_fail = True

Блок try метода handle() дополним блоками except:

        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)

В обработчик SOCKS4Handler добавим заглушку, которая в дальнейшем будет пересылать трафик между клиентом и целью:

    def stream_tcp(self, socket_a, socket_b, /):
        # TODO
        pass

К этой заглушке мы ещё вернёмся, а пока что закончим работу над методом handle() добавив обработку SOCKS4_CMD.CONNECT и SOCKS4_CMD.BIND.

Обработка команды CONNECT сводится к

созданию сокета
попытке соединения с целью (адрес и порт цели получены из запроса клиента) и отправке ответа со статусом SOCKS4_REPLY.GRANTED
после чего прокси начинает пересылать трафик между парой сокетов.

Если соединение не получится установить - обработчик исключений отправит клиенту ответ со статусом SOCKS4_REPLY.FAILED_OR_REJECTED.

            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)

Обработка команды BIND (ну, насколько я понял) сводится к

созданию сокета, слушающего случайный порт
отправке адреса и порта (на которых прокси ожидает соединения) в ответе со статусом SOCKS4_REPLY.GRANTED клиенту
приеме входящего соединения от цели
проверке соответствия адреса цели адресу, указанному в запросе клиента
ещё одном ответе клиенту со статусом SOCKS4_REPLY.GRANTED, адресом и портом соединившейся с прокси цели
после чего прокси начинает пересылать трафик между парой сокетов.

Если что-то пойдет не так - обработчик исключений отправит клиенту ответ со статусом SOCKS4_REPLY.FAILED_OR_REJECTED.

            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)

На этом основная часть прокси, непосредственно касающаяся протокола SOCKS4, закончена.

Пересылка трафика между клиентом и целью

Что нам нужно сделать концептуально в методе stream_tcp()?

Узнать, есть ли данные для чтения в каком либо из сокетов.
Прочитать данные, записать их в буферы.
Отправить данные из буферов в соответствующие сокеты.
Удалить отправленную часть данных из буферов.
Если при чтении или отправке данных произошло исключение - завершить цикл пересылки данных.
Попытаться отправить остатки данных из буферов, игнорируя исключения.

Добавим в SOCKS4Handler метод, формирующий описание пары сокетов, которое будет записываться в лог:

    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}'

Напишем вспомогательные функции для безопасного приема и отправки данных в сокеты:

safe_recv() принимает сокет, буфер с данными и флаг завершения цикла пересылки; возвращает буфер, дополненный новыми данными и потенциально измененный флаг завершения цикла пересылки
safe_send() принимает сокет, буфер с данными и флаг завершения цикла пересылки; возвращает буфер, из которого удалены отправленные данные и потенциально измененный флаг завершения цикла пересылки
safe_sendfinal() принимает сокет и буфер, данные из которого надо попробовать отправить, игнорируя ошибки

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

С использованием этих функций метод stream_tcp() обработчика SOCKS4Handler будет иметь вид:

    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}')

Это не самая лучшая реализация, но даже в таком виде uTorrent смог пропихнуть через прокси поток данных в 0.5 Gbps (с чудовищной загрузкой CPU).

Как протестировать?

Проще всего проверить команду CONNECT:

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.ru

В логах прокси увидим информацию о приеме/разрыве соединений, разобранный запрос клиента и ответ прокси, и проч.:

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 down

Полный исходный код прокси


###################################################################################################

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)

###################################################################################################

Простой SOCKS4 прокси на Python