#!/usr/bin/env python3 """ Phantom SOCKS5 Proxy Server Создает SOCKS5 прокси сервер, который маршрутизирует трафик через Phantom сеть Использование: python3 socks5-proxy.py --port 8080 --hops 3 Автор: Phantom Protocol Team 2025 """ import socket import threading import struct import select import sys import argparse import time import logging from typing import Optional, Tuple, List # Настройка логирования logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) # SOCKS5 константы SOCKS5_VERSION = 0x05 SOCKS5_CONNECT = 0x01 SOCKS5_IPV4 = 0x01 SOCKS5_DOMAIN = 0x03 SOCKS5_IPV6 = 0x04 # Коды ответов SOCKS5 SOCKS5_SUCCESS = 0x00 SOCKS5_GENERAL_FAILURE = 0x01 SOCKS5_CONNECTION_NOT_ALLOWED = 0x02 SOCKS5_NETWORK_UNREACHABLE = 0x03 SOCKS5_HOST_UNREACHABLE = 0x04 SOCKS5_CONNECTION_REFUSED = 0x05 SOCKS5_TTL_EXPIRED = 0x06 SOCKS5_COMMAND_NOT_SUPPORTED = 0x07 SOCKS5_ADDRESS_TYPE_NOT_SUPPORTED = 0x08 class PhantomRoute: """Представляет маршрут через Phantom сеть""" def __init__(self, hops: int = 3): self.hops = hops self.nodes: List[Tuple[str, int]] = [] self.build_route() def build_route(self): """Построение маршрута через Phantom узлы""" # В реальной реализации здесь будет запрос к Kademlia DHT # Сейчас используем симуляцию logger.info(f"Построение маршрута через {self.hops} хопов...") # Симуляция поиска узлов в Phantom сети available_nodes = [ ("10.0.1.100", 8050), ("10.0.1.101", 8050), ("10.0.1.102", 8050), ("10.0.1.103", 8050), ("10.0.1.104", 8050), ] # Выбор случайных узлов для маршрута import random self.nodes = random.sample(available_nodes, min(self.hops, len(available_nodes))) for i, (ip, port) in enumerate(self.nodes): logger.info(f" Хоп {i+1}: {ip}:{port}") class PhantomConnection: """Подключение через Phantom сеть""" def __init__(self, route: PhantomRoute): self.route = route self.socket: Optional[socket.socket] = None self.connected = False def connect(self, target_host: str, target_port: int) -> bool: """Подключение к целевому хосту через Phantom сеть""" try: logger.info(f"Подключение к {target_host}:{target_port} через Phantom сеть") # Подключение к первому узлу маршрута if not self.route.nodes: logger.error("Маршрут не построен") return False first_hop = self.route.nodes[0] logger.info(f"Подключение к первому хопу: {first_hop[0]}:{first_hop[1]}") self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.socket.settimeout(10) try: self.socket.connect(first_hop) except socket.error as e: logger.error(f"Не удалось подключиться к первому хопу: {e}") # Fallback: прямое подключение для демонстрации logger.info(f"Fallback: прямое подключение к {target_host}:{target_port}") self.socket.close() self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.socket.settimeout(10) self.socket.connect((target_host, target_port)) self.connected = True return True # Отправка команды построения туннеля tunnel_command = self._build_tunnel_command(target_host, target_port) self.socket.send(tunnel_command.encode('utf-8')) # Ожидание подтверждения response = self.socket.recv(1024).decode('utf-8') if response.startswith("PHANTOM_TUNNEL_OK"): logger.info("Туннель через Phantom сеть установлен") self.connected = True return True else: logger.error(f"Ошибка установки туннеля: {response}") # Fallback: прямое подключение logger.info(f"Fallback: прямое подключение к {target_host}:{target_port}") self.socket.close() self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.socket.settimeout(10) self.socket.connect((target_host, target_port)) self.connected = True return True except Exception as e: logger.error(f"Ошибка подключения: {e}") return False def _build_tunnel_command(self, target_host: str, target_port: int) -> str: """Построение команды для создания туннеля""" command = f"PHANTOM_BUILD_TUNNEL {target_host} {target_port}" # Добавление промежуточных хопов for i in range(1, len(self.route.nodes)): hop = self.route.nodes[i] command += f" {hop[0]}:{hop[1]}" return command + "\n" def send(self, data: bytes) -> int: """Отправка данных через Phantom соединение""" if not self.connected or not self.socket: raise ConnectionError("Соединение не установлено") return self.socket.send(data) def recv(self, size: int) -> bytes: """Получение данных через Phantom соединение""" if not self.connected or not self.socket: raise ConnectionError("Соединение не установлено") return self.socket.recv(size) def close(self): """Закрытие соединения""" if self.socket: self.socket.close() self.socket = None self.connected = False class SOCKS5Handler: """Обработчик SOCKS5 протокола""" def __init__(self, client_socket: socket.socket, hops: int = 3): self.client_socket = client_socket self.hops = hops self.phantom_connection: Optional[PhantomConnection] = None def handle(self): """Основная обработка SOCKS5 соединения""" try: # Этап 1: Аутентификация if not self._handle_authentication(): return # Этап 2: Обработка запроса подключения if not self._handle_connection_request(): return # Этап 3: Проксирование данных self._relay_data() except Exception as e: logger.error(f"Ошибка обработки SOCKS5: {e}") finally: self._cleanup() def _handle_authentication(self) -> bool: """Обработка этапа аутентификации SOCKS5""" try: # Получение запроса аутентификации data = self.client_socket.recv(256) if len(data) < 2: return False version, nmethods = struct.unpack('!BB', data[:2]) if version != SOCKS5_VERSION: logger.error(f"Неподдерживаемая версия SOCKS: {version}") return False if len(data) < 2 + nmethods: return False methods = struct.unpack('!' + 'B' * nmethods, data[2:2+nmethods]) # Поддерживаем только "без аутентификации" (метод 0) if 0 in methods: # Отправка ответа: версия + выбранный метод response = struct.pack('!BB', SOCKS5_VERSION, 0) self.client_socket.send(response) logger.debug("Аутентификация SOCKS5 пройдена") return True else: # Нет поддерживаемых методов response = struct.pack('!BB', SOCKS5_VERSION, 0xFF) self.client_socket.send(response) return False except Exception as e: logger.error(f"Ошибка аутентификации SOCKS5: {e}") return False def _handle_connection_request(self) -> bool: """Обработка запроса подключения SOCKS5""" try: # Получение запроса подключения data = self.client_socket.recv(4) if len(data) < 4: return False version, cmd, reserved, address_type = struct.unpack('!BBBB', data) if version != SOCKS5_VERSION: self._send_error_response(SOCKS5_GENERAL_FAILURE) return False if cmd != SOCKS5_CONNECT: self._send_error_response(SOCKS5_COMMAND_NOT_SUPPORTED) return False # Получение адреса назначения target_host, target_port = self._parse_target_address(address_type) if not target_host: self._send_error_response(SOCKS5_ADDRESS_TYPE_NOT_SUPPORTED) return False logger.info(f"SOCKS5 запрос подключения к {target_host}:{target_port}") # Подключение через Phantom сеть route = PhantomRoute(self.hops) self.phantom_connection = PhantomConnection(route) if self.phantom_connection.connect(target_host, target_port): # Отправка успешного ответа self._send_success_response() logger.info(f"SOCKS5 туннель к {target_host}:{target_port} установлен") return True else: self._send_error_response(SOCKS5_HOST_UNREACHABLE) return False except Exception as e: logger.error(f"Ошибка обработки запроса подключения: {e}") self._send_error_response(SOCKS5_GENERAL_FAILURE) return False def _parse_target_address(self, address_type: int) -> Tuple[Optional[str], Optional[int]]: """Парсинг адреса назначения""" try: if address_type == SOCKS5_IPV4: # IPv4 адрес (4 байта) + порт (2 байта) data = self.client_socket.recv(6) if len(data) < 6: return None, None ip_bytes = data[:4] port_bytes = data[4:6] ip = socket.inet_ntoa(ip_bytes) port = struct.unpack('!H', port_bytes)[0] return ip, port elif address_type == SOCKS5_DOMAIN: # Доменное имя: длина (1 байт) + имя + порт (2 байта) length_data = self.client_socket.recv(1) if len(length_data) < 1: return None, None domain_length = struct.unpack('!B', length_data)[0] domain_data = self.client_socket.recv(domain_length + 2) if len(domain_data) < domain_length + 2: return None, None domain = domain_data[:domain_length].decode('utf-8') port = struct.unpack('!H', domain_data[domain_length:])[0] return domain, port elif address_type == SOCKS5_IPV6: # IPv6 адрес (16 байт) + порт (2 байта) data = self.client_socket.recv(18) if len(data) < 18: return None, None ip_bytes = data[:16] port_bytes = data[16:18] ip = socket.inet_ntop(socket.AF_INET6, ip_bytes) port = struct.unpack('!H', port_bytes)[0] return ip, port return None, None except Exception as e: logger.error(f"Ошибка парсинга адреса: {e}") return None, None def _send_success_response(self): """Отправка успешного ответа SOCKS5""" # Версия + статус + зарезервировано + тип адреса response = struct.pack('!BBBB', SOCKS5_VERSION, SOCKS5_SUCCESS, 0, SOCKS5_IPV4) # Связанный адрес (0.0.0.0) + порт (0) response += struct.pack('!IH', 0, 0) self.client_socket.send(response) def _send_error_response(self, error_code: int): """Отправка ответа об ошибке SOCKS5""" response = struct.pack('!BBBB', SOCKS5_VERSION, error_code, 0, SOCKS5_IPV4) response += struct.pack('!IH', 0, 0) self.client_socket.send(response) def _relay_data(self): """Проксирование данных между клиентом и Phantom соединением""" if not self.phantom_connection: return logger.debug("Начало проксирования данных") try: while True: # Ожидание данных от клиента или сервера ready_sockets, _, error_sockets = select.select( [self.client_socket, self.phantom_connection.socket], [], [self.client_socket, self.phantom_connection.socket], 1.0 # Таймаут 1 секунда ) if error_sockets: logger.debug("Ошибка в сокетах, завершение проксирования") break if not ready_sockets: continue # Данные от клиента к серверу if self.client_socket in ready_sockets: data = self.client_socket.recv(4096) if not data: logger.debug("Клиент закрыл соединение") break self.phantom_connection.send(data) logger.debug(f"Отправлено {len(data)} байт через Phantom") # Данные от сервера к клиенту if self.phantom_connection.socket in ready_sockets: data = self.phantom_connection.recv(4096) if not data: logger.debug("Сервер закрыл соединение") break self.client_socket.send(data) logger.debug(f"Получено {len(data)} байт от Phantom") except Exception as e: logger.error(f"Ошибка проксирования данных: {e}") def _cleanup(self): """Очистка ресурсов""" try: if self.phantom_connection: self.phantom_connection.close() self.client_socket.close() except: pass class PhantomSOCKS5Server: """SOCKS5 прокси сервер через Phantom сеть""" def __init__(self, host: str = '127.0.0.1', port: int = 8080, hops: int = 3): self.host = host self.port = port self.hops = hops self.server_socket: Optional[socket.socket] = None self.running = False def start(self): """Запуск SOCKS5 сервера""" try: self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) self.server_socket.bind((self.host, self.port)) self.server_socket.listen(10) self.running = True logger.info(f"🚀 Phantom SOCKS5 прокси запущен на {self.host}:{self.port}") logger.info(f" Количество хопов через Phantom сеть: {self.hops}") logger.info(f" Настройте ваш браузер или приложение:") logger.info(f" - Тип прокси: SOCKS v5") logger.info(f" - Адрес: {self.host}") logger.info(f" - Порт: {self.port}") logger.info(f" - Аутентификация: Нет") while self.running: try: client_socket, client_address = self.server_socket.accept() logger.info(f"Новое SOCKS5 подключение от {client_address[0]}:{client_address[1]}") # Обработка клиента в отдельном потоке handler = SOCKS5Handler(client_socket, self.hops) client_thread = threading.Thread(target=handler.handle) client_thread.daemon = True client_thread.start() except socket.error as e: if self.running: logger.error(f"Ошибка принятия подключения: {e}") except Exception as e: logger.error(f"Ошибка запуска сервера: {e}") finally: self.stop() def stop(self): """Остановка SOCKS5 сервера""" self.running = False if self.server_socket: self.server_socket.close() logger.info("Phantom SOCKS5 прокси остановлен") def main(): """Главная функция""" parser = argparse.ArgumentParser( description='Phantom SOCKS5 Proxy - SOCKS5 прокси через Phantom сеть' ) parser.add_argument( '--host', default='127.0.0.1', help='IP адрес для прослушивания (по умолчанию: 127.0.0.1)' ) parser.add_argument( '--port', type=int, default=8080, help='Порт для прослушивания (по умолчанию: 8080)' ) parser.add_argument( '--hops', type=int, default=3, help='Количество хопов через Phantom сеть (по умолчанию: 3)' ) parser.add_argument( '--verbose', action='store_true', help='Подробный вывод' ) args = parser.parse_args() if args.verbose: logging.getLogger().setLevel(logging.DEBUG) # Создание и запуск SOCKS5 сервера server = PhantomSOCKS5Server(args.host, args.port, args.hops) try: server.start() except KeyboardInterrupt: logger.info("Получен сигнал прерывания, остановка сервера...") server.stop() if __name__ == '__main__': main()