fix sock5 connect

services/socks5_to_http_proxy.py

@@ -156,10 +156,9 @@ class Socks5ToHttpProxy:
         reader: asyncio.StreamReader,
         writer: asyncio.StreamWriter
     ):
-        """Обрабатывает CONNECT запрос."""
-        import socks
-        remote_socket = None
-        
+        """Обрабатывает CONNECT запрос через асинхронный SOCKS5."""
+        host = "unknown"
+        port = 443
         try:
             parts = request_line.split()
             if len(parts) < 2:
@@ -172,34 +171,38 @@ class Socks5ToHttpProxy:
                 port = int(port)
             else:
                 host = target
-                port = 443
 
-            # Пропускаем заголовки (с таймаутом)
+            # Пропускаем заголовки
             while True:
                 line = await asyncio.wait_for(reader.readline(), timeout=5)
                 if not line or line in [b'\r\n', b'\n']:
                     break
 
-            # Создаем SOCKS5 соединение
-            remote_socket = socks.socksocket()
-            remote_socket.settimeout(15)
+            # Асинхронное SOCKS5 соединение через python-socks
+            from python_socks.async_.asyncio import Proxy
+            from python_socks import ProxyType
 
             if self.username:
                 logger.info(f"SOCKS5 connect {host}:{port} via {self.socks5_host}:{self.socks5_port} auth=yes user={self.username[:4]}***")
-                remote_socket.set_proxy(
-                    socks.SOCKS5, self.socks5_host, self.socks5_port,
-                    username=self.username, password=self.password
+                proxy = Proxy.create(
+                    ProxyType.SOCKS5,
+                    self.socks5_host, self.socks5_port,
+                    username=self.username, password=self.password,
+                    rdns=True
                 )
             else:
-                logger.warning(f"SOCKS5 connect {host}:{port} via {self.socks5_host}:{self.socks5_port} auth=NO — credentials missing!")
-                remote_socket.set_proxy(
-                    socks.SOCKS5, self.socks5_host, self.socks5_port
+                logger.warning(f"SOCKS5 connect {host}:{port} via {self.socks5_host}:{self.socks5_port} auth=NO")
+                proxy = Proxy.create(
+                    ProxyType.SOCKS5,
+                    self.socks5_host, self.socks5_port,
+                    rdns=True
                 )
 
-            # Подключаемся (в отдельном потоке чтобы не блокировать)
-            loop = asyncio.get_running_loop()
             try:
-                await loop.run_in_executor(None, remote_socket.connect, (host, port))
+                sock = await asyncio.wait_for(
+                    proxy.connect(dest_host=host, dest_port=port),
+                    timeout=15
+                )
             except Exception as e:
                 logger.warning(f"SOCKS5 connect FAILED {host}:{port} via {self.socks5_host}: {e}")
                 writer.write(b'HTTP/1.1 502 Bad Gateway\r\n\r\n')
@@ -207,26 +210,50 @@ class Socks5ToHttpProxy:
                 return
 
             logger.info(f"SOCKS5 connected {host}:{port} OK")
-            # Отвечаем клиенту
             writer.write(b'HTTP/1.1 200 Connection Established\r\n\r\n')
             await writer.drain()
 
-            # Пересылка данных с таймаутом
+            # Полностью асинхронный relay через asyncio streams
+            remote_reader, remote_writer = await asyncio.open_connection(sock=sock)
             await asyncio.wait_for(
-                self._relay(reader, writer, remote_socket),
-                timeout=60
+                self._relay_streams(reader, writer, remote_reader, remote_writer),
+                timeout=120
             )
 
         except asyncio.TimeoutError:
             logger.warning(f"CONNECT relay timeout {host}:{port}")
         except Exception as e:
             logger.warning(f"CONNECT error {host}:{port}: {e}")
-        finally:
-            if remote_socket:
+    
+    async def _relay_streams(
+        self,
+        client_reader: asyncio.StreamReader,
+        client_writer: asyncio.StreamWriter,
+        remote_reader: asyncio.StreamReader,
+        remote_writer: asyncio.StreamWriter,
+    ):
+        """Асинхронная двунаправленная пересылка через asyncio streams."""
+        async def pipe(src: asyncio.StreamReader, dst: asyncio.StreamWriter):
             try:
-                    remote_socket.close()
+                while True:
+                    data = await src.read(65536)
+                    if not data:
+                        break
+                    dst.write(data)
+                    await dst.drain()
             except Exception:
                 pass
+            finally:
+                try:
+                    dst.close()
+                except Exception:
+                    pass
+
+        await asyncio.gather(
+            pipe(client_reader, remote_writer),
+            pipe(remote_reader, client_writer),
+            return_exceptions=True
+        )
 
     async def _relay(
         self,