Spaces:
Paused
Paused
File size: 2,657 Bytes
a5784e9 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 | import asyncio
import ssl as ssl_module
import urllib.parse
from typing import Any, Optional, Tuple
from aiohttp import TCPConnector
from python_socks.async_.asyncio import Proxy
class ProxyConnector:
"""
Class to handle connections through different types of proxies
"""
def __init__(self, proxy_url: Optional[str] = None):
self.proxy_url = proxy_url
self.connector = None
if proxy_url:
self._setup_connector()
def _setup_connector(self):
"""Set up the appropriate connector based on the proxy URL"""
if not self.proxy_url:
self.connector = TCPConnector()
return
# Parse the proxy URL
parsed = urllib.parse.urlparse(self.proxy_url)
proxy_type = parsed.scheme.lower()
if proxy_type in ("http", "https", "socks4", "socks5"):
self.connector = "SocksConnector"
else:
raise ValueError(f"Unsupported proxy type: {proxy_type}")
async def create_connection(
self, host: str, port: int, ssl: Optional[Any] = None
) -> Tuple[Any, Any]:
"""Create a connection to the target host through the proxy"""
if not self.connector:
# Direct connection without proxy
reader, writer = await asyncio.open_connection(host, port, ssl=ssl)
return reader, writer
# SOCKS proxy connection
assert self.proxy_url is not None # Type guard for pyright
proxy = Proxy.from_url(self.proxy_url) # type: ignore[misc]
sock = await proxy.connect(dest_host=host, dest_port=port)
if ssl is None:
reader, writer = await asyncio.open_connection(
host=None,
port=None,
sock=sock,
ssl=None,
)
return reader, writer
else:
ssl_context = ssl_module.SSLContext(ssl_module.PROTOCOL_TLS_CLIENT)
ssl_context.check_hostname = False
ssl_context.verify_mode = ssl_module.CERT_NONE
ssl_context.minimum_version = (
ssl_module.TLSVersion.TLSv1_2
) # Force TLS 1.2 or higher
ssl_context.maximum_version = (
ssl_module.TLSVersion.TLSv1_3
) # Allow TLS 1.3 if supported
ssl_context.set_ciphers("DEFAULT@SECLEVEL=2") # Use secure ciphers
reader, writer = await asyncio.open_connection(
host=None,
port=None,
sock=sock,
ssl=ssl_context,
server_hostname=host,
)
return reader, writer
|