Create a connection to the target host through the proxy
(
self, host: str, port: int, ssl: Optional[Any] = None
)
| 35 | raise ValueError(f"Unsupported proxy type: {proxy_type}") |
| 36 | |
| 37 | async def create_connection( |
| 38 | self, host: str, port: int, ssl: Optional[Any] = None |
| 39 | ) -> Tuple[Any, Any]: |
| 40 | """Create a connection to the target host through the proxy""" |
| 41 | if not self.connector: |
| 42 | # Direct connection without proxy |
| 43 | reader, writer = await asyncio.open_connection(host, port, ssl=ssl) |
| 44 | return reader, writer |
| 45 | |
| 46 | # SOCKS proxy connection |
| 47 | assert self.proxy_url is not None # Type guard for pyright |
| 48 | proxy = Proxy.from_url(self.proxy_url) # type: ignore[misc] |
| 49 | sock = await proxy.connect(dest_host=host, dest_port=port) |
| 50 | if ssl is None: |
| 51 | reader, writer = await asyncio.open_connection( |
| 52 | host=None, |
| 53 | port=None, |
| 54 | sock=sock, |
| 55 | ssl=None, |
| 56 | ) |
| 57 | return reader, writer |
| 58 | else: |
| 59 | ssl_context = ssl_module.SSLContext(ssl_module.PROTOCOL_TLS_CLIENT) |
| 60 | ssl_context.check_hostname = False |
| 61 | ssl_context.verify_mode = ssl_module.CERT_NONE |
| 62 | ssl_context.minimum_version = ( |
| 63 | ssl_module.TLSVersion.TLSv1_2 |
| 64 | ) # Force TLS 1.2 or higher |
| 65 | ssl_context.maximum_version = ( |
| 66 | ssl_module.TLSVersion.TLSv1_3 |
| 67 | ) # Allow TLS 1.3 if supported |
| 68 | ssl_context.set_ciphers("DEFAULT@SECLEVEL=2") # Use secure ciphers |
| 69 | |
| 70 | reader, writer = await asyncio.open_connection( |
| 71 | host=None, |
| 72 | port=None, |
| 73 | sock=sock, |
| 74 | ssl=ssl_context, |
| 75 | server_hostname=host, |
| 76 | ) |
| 77 | return reader, writer |