### Curio Proxy Connection Example Source: https://github.com/romis2012/python-socks/blob/master/_autodocs/other-async-frameworks.md Demonstrates how to establish an SSL connection to 'example.com' on port 443 using a SOCKS5 proxy with the Curio framework. It shows socket wrapping, SSL handshake, and basic HTTP GET request. ```python import curio import curio.ssl as curiossl from python_socks.async_.curio import Proxy from python_socks import ProxyType async def main(): proxy = Proxy( proxy_type=ProxyType.SOCKS5, host='127.0.0.1', port=1080 ) sock = await proxy.connect('example.com', 443, timeout=10) ssl_context = curiossl.create_default_context() ssl_sock = await ssl_context.wrap_socket( sock, do_handshake_on_connect=False, server_hostname='example.com' ) await ssl_sock.do_handshake() stream = ssl_sock.as_stream() await stream.write(b'GET / HTTP/1.1\r\nHost: example.com\r\n\r\n') response = await stream.read(1024) await stream.close() curio.run(main) ``` -------------------------------- ### Authenticated Proxy Examples Source: https://github.com/romis2012/python-socks/blob/master/_autodocs/COMPLETION_REPORT.txt Provides examples of establishing connections through proxies that require authentication (username and password). This covers both synchronous and asynchronous scenarios. ```python # Synchronous Example from python_socks.sync_proxy import SyncProxy proxy_sync = SyncProxy(proxy_host='127.0.0.1', proxy_port=1080, username='user', password='password') reader, writer = proxy_sync.connect('example.com', 80) # ... use reader/writer ... writer.close() # Asynchronous Example (asyncio) from python_socks.async_.asyncio.proxy import AsyncioProxy import asyncio async def async_example(): proxy_async = AsyncioProxy(proxy_host='127.0.0.1', proxy_port=1080, username='user', password='password') reader, writer = await proxy_async.connect('example.com', 80) # ... use reader/writer ... writer.close() asyncio.run(async_example()) ``` -------------------------------- ### Install python-socks with AnyIO Support Source: https://github.com/romis2012/python-socks/blob/master/README.md Install the package including optional support for AnyIO. This enables asynchronous proxy operations compatible with the AnyIO interface. ```bash pip install python-socks[anyio] ``` -------------------------------- ### Configuration Options Source: https://github.com/romis2012/python-socks/blob/master/_autodocs/COMPLETION_REPORT.txt Guide to common and framework-specific constructor parameters, connection settings, and timeout configurations. ```APIDOC ## Configuration Options ### Common Constructor Parameters Table of parameters common to all proxy constructors. ### Framework-Specific Parameters Details on parameters unique to specific frameworks (e.g., `loop`, `proxy_ssl`). ### Connection Method Parameters Parameters available for the `connect` method. ### Timeout Configuration Options for configuring connection and operation timeouts. ### URL Format Specification Specification of the supported URL formats for proxy configuration, with examples. ### Environment Variables Guidance on using environment variables for proxy configuration. ### Credentials and Timeouts Best practices for managing credentials and timeout settings. ``` -------------------------------- ### Install python-socks (Sync Only) Source: https://github.com/romis2012/python-socks/blob/master/README.md Install the package with only synchronous proxy support. This is the minimal installation. ```bash pip install python-socks ``` -------------------------------- ### Integrate with aiohttp using SocksConnector Source: https://github.com/romis2012/python-socks/blob/master/_autodocs/asyncio-api.md This example demonstrates integrating `python-socks` with `aiohttp` by creating a `SocksConnector` from a URL. It then uses this connector to establish an `aiohttp.ClientSession` and make a GET request. ```python import aiohttp import aiohttp_socks async def main(): connector = aiohttp_socks.SocksConnector.from_url('socks5://127.0.0.1:1080') async with aiohttp.ClientSession(connector=connector) as session: async with session.get('https://example.com') as response: data = await response.text() ``` -------------------------------- ### Example Usage of ProxyChain with Anyio Source: https://github.com/romis2012/python-socks/blob/master/_autodocs/other-async-frameworks.md Demonstrates how to chain two SOCKS5 proxies using ProxyChain and connect to a destination. This example uses the deprecated ProxyChain class. ```python import asyncio import ssl from python_socks.async_.anyio import Proxy, ProxyChain from python_socks import ProxyType import warnings async def main(): warnings.filterwarnings('ignore', category=DeprecationWarning) proxy1 = Proxy(ProxyType.SOCKS5, 'proxy1.example.com', 1080) proxy2 = Proxy(ProxyType.SOCKS5, 'proxy2.example.com', 1080) chain = ProxyChain([proxy1, proxy2]) stream = await chain.connect('example.com', 443, dest_ssl=ssl.create_default_context()) await stream.close() asyncio.run(main()) ``` -------------------------------- ### Install python-socks with Asyncio Support Source: https://github.com/romis2012/python-socks/blob/master/README.md Install the package including optional support for asyncio. This enables asynchronous proxy operations using the asyncio library. ```bash pip install python-socks[asyncio] ``` -------------------------------- ### Quick Reference Commands Source: https://github.com/romis2012/python-socks/blob/master/_autodocs/COMPLETION_REPORT.txt Provides a list of quick reference commands for common tasks related to the python-socks library, such as installation and basic usage examples. ```bash # Install the library pip install python-socks # Basic sync usage python -c "from python_socks.sync_proxy import SyncProxy; proxy = SyncProxy(proxy_host='127.0.0.1', proxy_port=1080); reader, writer = proxy.connect('example.com', 80); print(reader.read(100))" # Basic async usage (asyncio) python -c "import asyncio; from python_socks.async_.asyncio.proxy import AsyncioProxy; async def main(): proxy = AsyncioProxy(proxy_host='127.0.0.1', proxy_port=1080); reader, writer = await proxy.connect('example.com', 80); print(await reader.read(100)); writer.close(); asyncio.run(main())" ``` -------------------------------- ### Install python-socks with Trio Support Source: https://github.com/romis2012/python-socks/blob/master/README.md Install the package including optional support for Trio. This enables asynchronous proxy operations using the Trio framework. ```bash pip install python-socks[trio] ``` -------------------------------- ### SOCKS4 Connection Setup Source: https://github.com/romis2012/python-socks/blob/master/_autodocs/protocol-details.md Example of setting up a SOCKS4 proxy connection using python-socks. The username parameter is used as the user_id for SOCKS4. ```python from python_socks.sync import Proxy from python_socks import ProxyType proxy = Proxy( proxy_type=ProxyType.SOCKS4, host='proxy.example.com', port=1080, username='myuser', # Becomes user_id in SOCKS4 ) sock = proxy.connect('example.com', 80) ``` -------------------------------- ### Trio Proxy from_url Example Source: https://github.com/romis2012/python-socks/blob/master/_autodocs/other-async-frameworks.md Shows how to initialize a Trio Proxy object using a SOCKS5 URL and then establish a connection to a destination. ```python import trio from python_socks.async_.trio import Proxy async def main(): proxy = Proxy.from_url('socks5://user:pass@127.0.0.1:1080') sock = await proxy.connect('example.com', 80) sock.close() trio.run(main) ``` -------------------------------- ### Install python-socks with Curio Support Source: https://github.com/romis2012/python-socks/blob/master/README.md Install the package including optional support for Curio. This enables asynchronous proxy operations using the Curio concurrency library. ```bash pip install python-socks[curio] ``` -------------------------------- ### Trio Proxy Connection Example Source: https://github.com/romis2012/python-socks/blob/master/_autodocs/other-async-frameworks.md Demonstrates establishing a connection to 'example.com' on port 443 via a SOCKS5 proxy using Trio. Includes SSL wrapping and basic HTTP request/response. ```python import trio import ssl from python_socks.async_.trio import Proxy from python_socks import ProxyType async def main(): proxy = Proxy( proxy_type=ProxyType.SOCKS5, host='127.0.0.1', port=1080, username='user', password='pass' ) sock = await proxy.connect('example.com', 443, timeout=10) # Use with trio.SocketStream stream = trio.SocketStream(sock) # Wrap with SSL ssl_stream = trio.SSLStream( stream, ssl.create_default_context(), server_hostname='example.com' ) await ssl_stream.do_handshake() # Send request await ssl_stream.send_all(b'GET / HTTP/1.1\r\nHost: example.com\r\n\r\n') # Read response response = await ssl_stream.receive_some(4096) await ssl_stream.aclose() trio.run(main) ``` -------------------------------- ### Create Proxy from URL for httpx Integration Source: https://github.com/romis2012/python-socks/blob/master/_autodocs/asyncio-api.md This example shows how to create a `Proxy` instance using `Proxy.from_url()` for integration with the `httpx` library. It demonstrates connecting to a proxy and preparing for a custom stream wrapper. ```python import asyncio import httpx from python_socks.async_.asyncio import Proxy async def main(): proxy = Proxy.from_url('socks5://127.0.0.1:1080') sock = await proxy.connect('httpbin.org', 443) # Requires custom stream wrapper async with httpx.AsyncClient() as client: response = await client.get('https://httpbin.org/ip') ``` -------------------------------- ### ProxyChain Example Usage Source: https://github.com/romis2012/python-socks/blob/master/_autodocs/sync-api.md Demonstrates how to create and use a ProxyChain with two SOCKS5 proxies. Note the use of warnings.filterwarnings to ignore DeprecationWarning. ```python from python_socks.sync import Proxy, ProxyChain from python_socks import ProxyType import warnings warnings.filterwarnings('ignore', category=DeprecationWarning) proxy1 = Proxy(ProxyType.SOCKS5, 'proxy1.example.com', 1080) proxy2 = Proxy(ProxyType.SOCKS5, 'proxy2.example.com', 1080) chain = ProxyChain([proxy1, proxy2]) sock = chain.connect('example.com', 80) ``` -------------------------------- ### Async with Context Manager (Custom) Source: https://github.com/romis2012/python-socks/blob/master/_autodocs/INDEX.md An example of creating a custom asynchronous context manager for managing proxy connections. ```APIDOC ## Async with Context Manager (Custom) ### Description An example of creating a custom asynchronous context manager for managing proxy connections. This allows for cleaner handling of resource acquisition and release in asynchronous code. ### Usage This pattern is useful when you need to manage the lifecycle of a proxy-connected socket within an `async with` block. ### Example ```python from python_socks import Proxy from contextlib import asynccontextmanager @asynccontextmanager async def proxy_conn(url, host, port): proxy = Proxy.from_url(url) sock = await proxy.connect(host, port) try: yield sock finally: sock.close() # Example usage within an async function: # async def main(): # async with proxy_conn('socks5://host:port', 'example.com', 80) as sock: # # Use the socket here # pass ``` ``` -------------------------------- ### URL Format Specification with Examples Source: https://github.com/romis2012/python-socks/blob/master/_autodocs/COMPLETION_REPORT.txt Specifies the supported URL formats for configuring proxies, including examples for SOCKS4, SOCKS5, and HTTP proxies with and without authentication. ```python # SOCKS4 socks4://host:port/ socks4://user@host:port/ # SOCKS5 socks5://host:port/ socks5://user:password@host:port/ # HTTP http://host:port/ http://user:password@host:port/ ``` -------------------------------- ### Usage Patterns Source: https://github.com/romis2012/python-socks/blob/master/_autodocs/COMPLETION_REPORT.txt Examples and patterns for using the library in various scenarios, including synchronous, asynchronous, and authenticated proxy usage. ```APIDOC ## Usage Patterns ### Basic Synchronous HTTP Request Example of making a basic synchronous HTTP request through a proxy. ### Synchronous HTTPS with SSL Example for synchronous HTTPS connections with SSL/TLS. ### Asyncio HTTP Request Example of an asynchronous HTTP request using asyncio. ### Trio HTTP Request Example of an HTTP request using the Trio framework. ### Anyio HTTP Request Example of an HTTP request using the Anyio framework. ### Error Handling Patterns Comprehensive examples of error handling strategies. ### Authenticated Proxy Examples Demonstrates how to use proxies requiring authentication. ### Multiple Proxies with Fallback Pattern for configuring and using multiple proxies with fallback mechanisms. ### Requests Library Integration Example of integrating with the popular `requests` library. ### Async Context Manager Pattern Usage of async context managers for proxy connections. ``` -------------------------------- ### SOCKS4 URL Example Source: https://github.com/romis2012/python-socks/blob/master/_autodocs/configuration.md Use this format for SOCKS4 proxies. ```python 'socks4://proxy.example.com:1080' ``` -------------------------------- ### Parse Proxy URL with Credentials Example Source: https://github.com/romis2012/python-socks/blob/master/_autodocs/types.md Demonstrates parsing a proxy URL that includes username and password credentials. ```python from python_socks import parse_proxy_url, ProxyType # With credentials ptype, host, port, user, pwd = parse_proxy_url('socks5://user:pass@proxy.example.com:1080') # Result: (ProxyType.SOCKS5, 'proxy.example.com', 1080, 'user', 'pass') ``` -------------------------------- ### ProxyType Usage Example Source: https://github.com/romis2012/python-socks/blob/master/_autodocs/types.md Demonstrates how to use the ProxyType enumeration when creating a Proxy instance and checking the proxy protocol. ```python from python_socks import ProxyType from python_socks.sync import Proxy # Using enum value proxy = Proxy( proxy_type=ProxyType.SOCKS5, host='127.0.0.1', port=1080 ) # Check proxy type if proxy._proxy_type == ProxyType.SOCKS5: print("Using SOCKS5 protocol") # Iterate all types for ptype in ProxyType: print(f"Protocol: {ptype.name} (value: {ptype.value})") ``` -------------------------------- ### Connect to Destination via SOCKS5 Proxy with SSL Source: https://github.com/romis2012/python-socks/blob/master/_autodocs/other-async-frameworks.md Example demonstrating how to use the AnyIO Proxy to connect to a destination host and port through a SOCKS5 proxy. This snippet shows automatic SSL wrapping for the destination connection and includes a timeout. The returned stream object provides methods for network operations. ```python import asyncio import ssl from python_socks.async_.anyio import Proxy from python_socks import ProxyType async def main(): proxy = Proxy( proxy_type=ProxyType.SOCKS5, host='127.0.0.1', port=1080, username='user', password='pass' ) # Connect with automatic SSL wrapping stream = await proxy.connect( dest_host='example.com', dest_port=443, dest_ssl=ssl.create_default_context(), timeout=10 ) # Use stream interface await stream.write_all(b'GET / HTTP/1.1\r\nHost: example.com\r\n\r\n') response = await stream.read(4096) await stream.close() asyncio.run(main()) ``` -------------------------------- ### HTTP CONNECT Response Example Source: https://github.com/romis2012/python-socks/blob/master/_autodocs/protocol-details.md Shows the expected HTTP/1.1 200 OK response from an HTTP proxy upon successful establishment of a CONNECT tunnel. May include optional headers. ```http HTTP/1.1 200 Connection Established [optional headers] [blank line] ``` -------------------------------- ### Proxy URL Format Example Source: https://github.com/romis2012/python-socks/blob/master/_autodocs/overview.md Illustrates the standard URL format for defining proxy connections, including scheme, credentials, host, and port. ```text socks5://user:pass@127.0.0.1:1080 ``` -------------------------------- ### Proxy Connection from URL Source: https://github.com/romis2012/python-socks/blob/master/_autodocs/INDEX.md Configure and connect through a proxy using a connection URL. This simplifies proxy setup by embedding all necessary details in a single string. ```python from python_socks.sync import Proxy proxy = Proxy.from_url('socks5://user:pass@127.0.0.1:1080') sock = proxy.connect('example.com', 80) sock.close() ``` -------------------------------- ### Proxy Configuration from URL with Credentials Source: https://github.com/romis2012/python-socks/blob/master/_autodocs/usage-patterns.md Configure a proxy connection by providing a URL that includes username and password. This simplifies setup when credentials are part of the proxy address. ```python from python_socks.sync import Proxy proxy = Proxy.from_url('socks5://user:password@proxy.example.com:1080') sock = proxy.connect('example.com', 80) sock.close() ``` -------------------------------- ### Parse HTTP CONNECT Proxy URL Example Source: https://github.com/romis2012/python-socks/blob/master/_autodocs/types.md Demonstrates parsing a URL for the HTTP CONNECT tunneling protocol, including optional credentials. ```python from python_socks import parse_proxy_url, ProxyType # HTTP CONNECT ptype, host, port, user, pwd = parse_proxy_url('http://user:pass@proxy.example.com:8080') # Result: (ProxyType.HTTP, 'proxy.example.com', 8080, 'user', 'pass') ``` -------------------------------- ### Parse SOCKS4 Proxy URL Example Source: https://github.com/romis2012/python-socks/blob/master/_autodocs/types.md Shows how to parse a URL specifically for the SOCKS4 proxy protocol. ```python from python_socks import parse_proxy_url, ProxyType # SOCKS4 ptype, host, port, user, pwd = parse_proxy_url('socks4://proxy.example.com:1080') # Result: (ProxyType.SOCKS4, 'proxy.example.com', 1080, '', '') ``` -------------------------------- ### Asyncio Integration with aiohttp Source: https://github.com/romis2012/python-socks/blob/master/_autodocs/COMPLETION_REPORT.txt Integrate AsyncioProxy with the aiohttp library for asynchronous web requests. This example shows how to configure aiohttp to use the proxy. ```python import asyncio import aiohttp proxy = AsyncioProxy.from_url('socks5://127.0.0.1:1080/') async def main(): async with aiohttp.ClientSession() as session: async with session.get('http://example.com', proxy=proxy.as_url()) as response: print(await response.text()) asyncio.run(main()) ``` -------------------------------- ### Basic Synchronous HTTP Request Source: https://github.com/romis2012/python-socks/blob/master/_autodocs/COMPLETION_REPORT.txt Demonstrates a basic synchronous HTTP request using the library's SyncProxy. This is a fundamental example for making simple web requests through a proxy. ```python from python_socks.sync_proxy import SyncProxy import socket proxy = SyncProxy(proxy_host='127.0.0.1', proxy_port=1080) reader, writer = proxy.connect('example.com', 80) request = b"GET / HTTP/1.1\r\nHost: example.com\r\nConnection: close\r\n\r\n" writer.write(request) writer.flush() response = reader.read(4096) print(response.decode()) writer.close() ``` -------------------------------- ### AsyncioProxy from_url() Class Method Source: https://github.com/romis2012/python-socks/blob/master/_autodocs/COMPLETION_REPORT.txt Create an AsyncioProxy instance from a proxy URL string. This method simplifies the setup process for asynchronous proxy connections. ```python proxy = AsyncioProxy.from_url('socks5://user:password@127.0.0.1:1080/') ``` -------------------------------- ### Asyncio Integration with httpx Source: https://github.com/romis2012/python-socks/blob/master/_autodocs/COMPLETION_REPORT.txt Use the AsyncioProxy with the httpx library for making asynchronous HTTP requests through a proxy. This example demonstrates setting up the proxy for httpx client. ```python import httpx proxy = AsyncioProxy.from_url('socks5://127.0.0.1:1080/') async def main(): async with httpx.AsyncClient(proxies=proxy.as_dict()) as client: response = await client.get('http://example.com') print(response.text) asyncio.run(main()) ``` -------------------------------- ### HTTP CONNECT Request Example Source: https://github.com/romis2012/python-socks/blob/master/_autodocs/protocol-details.md Illustrates the format of an HTTP CONNECT request sent by a client to establish a tunnel through an HTTP proxy. Includes Host header and optional Proxy-Authorization. ```http CONNECT destination:port HTTP/1.1 Host: destination:port [Proxy-Authorization: Basic base64(username:password)] [other headers] [blank line] ``` -------------------------------- ### Trio HTTP Request with SSL Source: https://github.com/romis2012/python-socks/blob/master/_autodocs/COMPLETION_REPORT.txt Demonstrates making an HTTPS request through a proxy using the Trio framework. This example highlights Trio's concurrency model and TrioProxy usage. ```python import trio from python_socks.sync_proxy import SyncProxy async def main(): proxy = SyncProxy(proxy_host='127.0.0.1', proxy_port=1080) # Trio uses sync sockets, so SyncProxy is appropriate reader, writer = proxy.connect('example.com', 443) # SSL wrapping would typically be handled by the underlying socket library or trio-tls # For simplicity, this example focuses on the proxy connection itself. request = b"GET / HTTP/1.1\r\nHost: example.com\r\nConnection: close\r\n\r\n" writer.write(request) writer.flush() response = reader.read(4096) print(response.decode()) writer.close() trio.run(main) ``` -------------------------------- ### Python Synchronous HTTP Proxy Connection Source: https://github.com/romis2012/python-socks/blob/master/_autodocs/protocol-details.md Demonstrates how to establish a synchronous socket connection through an HTTP proxy using the python-socks library. This example includes basic authentication and upgrades the socket to SSL/TLS for HTTPS traffic. ```python from python_socks.sync import Proxy from python_socks import ProxyType proxy = Proxy( proxy_type=ProxyType.HTTP, host='proxy.example.com', port=8080, username='user', password='pass', ) sock = proxy.connect('example.com', 443) sock = ssl.wrap_socket(sock) # Upgrade to HTTPS ``` -------------------------------- ### AnyIO HTTP Request with Automatic SSL Source: https://github.com/romis2012/python-socks/blob/master/_autodocs/usage-patterns.md Connect to a SOCKS5 proxy and perform an HTTPS request using AnyIO with automatic SSL wrapping. Simplifies SSL setup. ```python import asyncio import ssl from python_socks.async_.anyio import Proxy async def fetch(): proxy = Proxy.from_url('socks5://127.0.0.1:1080') # Connect with automatic SSL wrapping stream = await proxy.connect( 'example.com', 443, dest_ssl=ssl.create_default_context(), timeout=10 ) await stream.write_all(b'GET / HTTP/1.1\r\nHost: example.com\r\n\r\n') response = await stream.read(4096) await stream.close() return response asyncio.run(fetch()) ``` -------------------------------- ### Asynchronous HTTP Request via SOCKS5 Proxy with asyncio Source: https://github.com/romis2012/python-socks/blob/master/_autodocs/usage-patterns.md Perform an asynchronous HTTP GET request using `asyncio` and `python-socks`. This example demonstrates how to integrate proxy connections within an asyncio event loop for non-blocking I/O. ```python import asyncio from python_socks.async_.asyncio import Proxy async def fetch(): proxy = Proxy.from_url('socks5://127.0.0.1:1080') sock = await proxy.connect('example.com', 80, timeout=10) # Use with asyncio loop = asyncio.get_event_loop() request = ( b'GET / HTTP/1.1\r\n' b'Host: example.com\r\n' b'Connection: close\r\n\r\n' ) await loop.sock_sendall(sock, request) response = await loop.sock_recv(sock, 4096) sock.close() return response asyncio.run(fetch()) ``` -------------------------------- ### Steps to Add New Async Framework Support Source: https://github.com/romis2012/python-socks/blob/master/_autodocs/architecture.md Details the process for integrating support for a new asynchronous framework, involving the creation of specific directories and files for proxy implementation, stream wrapping, TCP connection, and DNS resolution. ```text 1. Create `python_socks/async_/newlib/` directory 2. Create `_proxy.py` implementing Proxy with framework's event loop 3. Create `_stream.py` implementing `AsyncSocketStream` wrapper 4. Create `_connect.py` for framework-specific TCP connection 5. Create `_resolver.py` for framework-specific DNS resolution 6. Export Proxy from `__init__.py` ``` -------------------------------- ### SyncProxy Constructor with All Parameters Source: https://github.com/romis2012/python-socks/blob/master/_autodocs/COMPLETION_REPORT.txt Instantiate the SyncProxy class with all available configuration options. This is useful for setting up a proxy connection with specific host, port, username, password, and timeout settings. ```python proxy = SyncProxy( proxy_host='127.0.0.1', proxy_port=1080, proxy_type=ProxyType.SOCKS5, username='user', password='password', timeout=10, rdns=True, ) ``` -------------------------------- ### ProxyConnectionError Example Source: https://github.com/romis2012/python-socks/blob/master/_autodocs/COMPLETION_REPORT.txt Example of catching a ProxyConnectionError, which indicates a failure to establish a connection with the proxy server. This error often includes an errno mapping. ```python from python_socks.errors import ProxyConnectionError try: # ... proxy connection attempt ... pass except ProxyConnectionError as e: print(f'Proxy connection failed: {e}') ``` -------------------------------- ### Proxy.__init__() Source: https://github.com/romis2012/python-socks/blob/master/_autodocs/asyncio-api.md Initializes a Proxy instance for establishing asynchronous proxy connections. It configures the proxy type, host, port, and optional authentication credentials. ```APIDOC ## Proxy.__init__() ### Description Initializes a Proxy instance for establishing asynchronous proxy connections. It configures the proxy type, host, port, and optional authentication credentials. ### Method __init__ ### Parameters #### Path Parameters None #### Query Parameters None #### Request Body None ### Parameters - **proxy_type** (ProxyType) - Required - Proxy protocol (SOCKS4, SOCKS5, or HTTP) - **host** (str) - Required - Proxy server hostname or IP address - **port** (int) - Required - Proxy server port number - **username** (str | None) - Optional - Username for authentication (SOCKS5/HTTP). For SOCKS4, this is treated as user_id. - **password** (str | None) - Optional - Password for authentication (SOCKS5/HTTP). - **rdns** (bool | None) - Optional - Remote DNS resolution flag; None uses protocol default. For SOCKS5, True resolves hostname on the proxy, False resolves locally first. - **loop** (asyncio.AbstractEventLoop | None) - Optional - Event loop to use; defaults to `asyncio.get_event_loop()`. Deprecated in Python 3.10+. ### Notes - The loop parameter is deprecated in newer Python versions (3.10+) and defaults to the running event loop. - For SOCKS4, `username` is treated as the user_id. - For HTTP, `username` and `password` are used for basic authentication. - For SOCKS5, `rdns=True` (default) sends hostname to proxy for resolution; `rdns=False` resolves locally first. ``` -------------------------------- ### parse_proxy_url() Function Example Source: https://github.com/romis2012/python-socks/blob/master/_autodocs/COMPLETION_REPORT.txt Example usage of the parse_proxy_url() function, which parses a proxy URL string into its components. This is useful for dynamic proxy configuration. ```python from python_socks.utils import parse_proxy_url url = 'socks5://user:password@127.0.0.1:1080/' parsed_url = parse_proxy_url(url) print(parsed_url) ``` -------------------------------- ### Proxy.connect() Source: https://github.com/romis2012/python-socks/blob/master/_autodocs/other-async-frameworks.md Establishes a connection to a destination host and port through the configured proxy. It returns an Anyio compatible stream object. ```APIDOC ## Proxy.connect() ### Description Establishes a connection to a destination host and port through the configured proxy. It returns an Anyio compatible stream object. ### Method async def connect ### Parameters #### Path Parameters - **dest_host** (str) - Required - Destination hostname or IP address - **dest_port** (int) - Required - Destination port number - **dest_ssl** (ssl.SSLContext | None) - Optional - SSL context for destination connection - **timeout** (float | None) - Optional - Connection timeout in seconds (default: 60) - **local_host** (str | None) - Optional - Local address to bind to (keyword-only) ### Return Type `AnyioSocketStream` ### Notable Features - Automatically wraps with TLS if `dest_ssl` is provided - Automatically wraps proxy connection with TLS if `proxy_ssl` was set at construction - Returns a stream-like object instead of raw socket (more intuitive for async operations) ### Timeout Behavior Uses `anyio.fail_after()` context manager. Raises `TimeoutError` (wrapped as `ProxyTimeoutError`). ### Request Example ```python import asyncio import ssl from python_socks.async_.anyio import Proxy from python_socks import ProxyType async def main(): proxy = Proxy( proxy_type=ProxyType.SOCKS5, host='127.0.0.1', port=1080, username='user', password='pass' ) # Connect with automatic SSL wrapping stream = await proxy.connect( dest_host='example.com', dest_port=443, dest_ssl=ssl.create_default_context(), timeout=10 ) # Use stream interface await stream.write_all(b'GET / HTTP/1.1\r\nHost: example.com\r\n\r\n') response = await stream.read(4096) await stream.close() asyncio.run(main()) ``` ``` -------------------------------- ### Steps to Add a New Protocol Source: https://github.com/romis2012/python-socks/blob/master/_autodocs/architecture.md Outlines the necessary steps to extend the library with support for a new proxy protocol, such as SOCKS6, by creating new files and updating enums and factory modules. ```text 1. Create `python_socks/_protocols/socks6.py` with protocol logic 2. Create `python_socks/_connectors/socks6_sync.py` implementing `SyncConnector` 3. Create `python_socks/_connectors/socks6_async.py` implementing `AsyncConnector` 4. Add `ProxyType.SOCKS6` to enum in `_types.py` 5. Update `factory_sync.py` and `factory_async.py` to handle new type ``` -------------------------------- ### SOCKS5 Anonymous URL Example Source: https://github.com/romis2012/python-socks/blob/master/_autodocs/configuration.md Use this format for SOCKS5 proxies when no authentication is required. ```python 'socks5://127.0.0.1:1080' ``` -------------------------------- ### Proxy.connect() Source: https://github.com/romis2012/python-socks/blob/master/_autodocs/sync-api.md Establishes a connection to a destination host and port through the configured proxy. ```APIDOC ## Proxy.connect() ### Description Establishes a connection to a destination host and port through the configured proxy. Returns a standard Python socket in blocking mode. ### Method POST ### Endpoint `/connect` ### Parameters #### Path Parameters - **dest_host** (str) - Required - Destination hostname or IP address - **dest_port** (int) - Required - Destination port number #### Query Parameters - **timeout** (float | None) - Optional - Connection timeout in seconds. Defaults to 60. - **local_addr** (str | None) - Optional - Local address to bind to (keyword-only). ### Response #### Success Response (200) - **socket.socket** - A standard Python socket in blocking mode connected through the proxy to the destination. ### Request Example ```python from python_socks.sync import Proxy from python_socks import ProxyType import socket proxy = Proxy( proxy_type=ProxyType.SOCKS5, host='127.0.0.1', port=1080, username='user', password='pass' ) sock = proxy.connect( dest_host='example.com', dest_port=80, timeout=10 ) sock.sendall(b'GET / HTTP/1.1\r\nHost: example.com\r\n\r\n') data = sock.recv(4096) sock.close() ``` ### Error Handling - **ProxyConnectionError**: Cannot establish connection to proxy server. - **ProxyTimeoutError**: Connection times out (timeout parameter exceeded). - **ProxyError**: Proxy protocol error or authentication failure. ``` -------------------------------- ### Get Proxy Port Source: https://github.com/romis2012/python-socks/blob/master/_autodocs/sync-api.md Retrieve the port number of the proxy server from a Proxy instance. ```python from python_socks.sync import Proxy from python_socks import ProxyType proxy = Proxy(ProxyType.SOCKS5, '192.168.1.1', 1080) port = proxy.proxy_port print(f'Proxy port: {port}') ``` -------------------------------- ### Get Proxy Hostname Source: https://github.com/romis2012/python-socks/blob/master/_autodocs/sync-api.md Retrieve the hostname or IP address of the proxy server from a Proxy instance. ```python from python_socks.sync import Proxy from python_socks import ProxyType proxy = Proxy(ProxyType.SOCKS5, '192.168.1.1', 1080) host = proxy.proxy_host print(f'Proxy host: {host}') ``` -------------------------------- ### SyncProxy from_url() Class Method Source: https://github.com/romis2012/python-socks/blob/master/_autodocs/COMPLETION_REPORT.txt Create a SyncProxy instance from a proxy URL string. This simplifies configuration by parsing common URL formats, including authentication credentials. ```python proxy = SyncProxy.from_url('socks5://user:password@127.0.0.1:1080/') ``` -------------------------------- ### URL-encoded Special Characters Source: https://github.com/romis2012/python-socks/blob/master/_autodocs/configuration.md Special characters in passwords must be URL-encoded. This example shows a password with '@' and ':' encoded. ```python 'socks5://user:p%40ssw%3Ard@host:1080' ``` -------------------------------- ### Handle ProxyTimeoutError Source: https://github.com/romis2012/python-socks/blob/master/_autodocs/errors.md Example of catching ProxyTimeoutError during a proxy connection attempt. This error is also a subclass of TimeoutError. ```python from python_socks import ProxyTimeoutError from python_socks.sync import Proxy from python_socks import ProxyType proxy = Proxy(ProxyType.SOCKS5, '127.0.0.1', 1080) try: sock = proxy.connect('example.com', 80, timeout=1) except ProxyTimeoutError as e: print(f"Connection timed out: {e}") except Exception as e: print(f"Other error: {e}") ``` -------------------------------- ### Asyncio with asyncio.open_connection() Source: https://github.com/romis2012/python-socks/blob/master/_autodocs/COMPLETION_REPORT.txt Shows how to use AsyncioProxy in conjunction with asyncio.open_connection() for establishing asynchronous connections. This pattern is common in asyncio-based network programming. ```python import asyncio from python_socks.async_.asyncio.proxy import AsyncioProxy async def main(): proxy = AsyncioProxy(proxy_host='127.0.0.1', proxy_port=1080) # The proxy object itself can be used to establish the connection reader, writer = await proxy.connect('example.com', 80) # asyncio.open_connection is not directly used here, proxy.connect handles it. # This example clarifies that proxy.connect provides the streams. request = b"GET / HTTP/1.1\r\nHost: example.com\r\nConnection: close\r\n\r\n" writer.write(request) await writer.drain() response = await reader.read(4096) print(response.decode()) writer.close() asyncio.run(main()) ``` -------------------------------- ### Environment Variable Guidance Source: https://github.com/romis2012/python-socks/blob/master/_autodocs/COMPLETION_REPORT.txt Provides guidance on using environment variables for proxy configuration. This allows for flexible deployment and runtime configuration without code changes. ```python # Example: Set environment variables before running the script # export PROXY_HOST='127.0.0.1' # export PROXY_PORT='1080' # export PROXY_TYPE='socks5' # export PROXY_USERNAME='user' # export PROXY_PASSWORD='password' # The library can be configured to read these variables. ``` -------------------------------- ### SyncProxy connect() Method Source: https://github.com/romis2012/python-socks/blob/master/_autodocs/COMPLETION_REPORT.txt Establish a connection to a remote host through the configured proxy using the connect() method. This method takes the destination host and port as arguments. ```python proxy = SyncProxy(proxy_host='127.0.0.1', proxy_port=1080) reader, writer = proxy.connect('example.com', 80) ``` -------------------------------- ### Handle ProxyConnectionError Source: https://github.com/romis2012/python-socks/blob/master/_autodocs/errors.md Example of catching ProxyConnectionError when a connection to the proxy server fails. Accesses error details via args. ```python from python_socks import ProxyConnectionError from python_socks.sync import Proxy from python_socks import ProxyType proxy = Proxy(ProxyType.SOCKS5, '127.0.0.1', 1080) try: sock = proxy.connect('example.com', 80, timeout=5) except ProxyConnectionError as e: print(f"Cannot connect to proxy: {e.args[1]}") print(f"Error code: {e.args[0]}") except Exception as e: print(f"Other error: {e}") ``` -------------------------------- ### Multiple Proxies with Fallback Source: https://github.com/romis2012/python-socks/blob/master/_autodocs/COMPLETION_REPORT.txt Demonstrates a pattern for using multiple proxies in sequence, with fallback to the next proxy if a previous one fails. This enhances connection reliability. ```python from python_socks.sync_proxy import SyncProxy from python_socks.errors import ProxyError proxies_config = [ {'proxy_host': '1.1.1.1', 'proxy_port': 1080}, {'proxy_host': '2.2.2.2', 'proxy_port': 1080}, {'proxy_host': '3.3.3.3', 'proxy_port': 1080} ] target_host = 'example.com' target_port = 80 reader, writer = None, None for config in proxies_config: try: proxy = SyncProxy(**config) reader, writer = proxy.connect(target_host, target_port) print(f"Successfully connected via {config['proxy_host']}:{config['proxy_port']}") break except ProxyError as e: print(f"Failed to connect via {config['proxy_host']}:{config['proxy_port']}: {e}") if reader and writer: # ... use reader/writer ... writer.close() ``` -------------------------------- ### Parse URL-encoded Credentials Example Source: https://github.com/romis2012/python-socks/blob/master/_autodocs/types.md Illustrates parsing a proxy URL where username and password contain special characters that are URL-encoded. ```python from python_socks import parse_proxy_url, ProxyType # URL-encoded credentials ptype, host, port, user, pwd = parse_proxy_url('socks5://user%40name:pass%40word@host:1080') # Result: (ProxyType.SOCKS5, 'host', 1080, 'user@name', 'pass@word') ``` -------------------------------- ### ProxyTimeoutError Example Source: https://github.com/romis2012/python-socks/blob/master/_autodocs/COMPLETION_REPORT.txt Demonstrates catching a ProxyTimeoutError, which is raised when a proxy operation exceeds the configured timeout. This is crucial for robust error handling. ```python from python_socks.errors import ProxyTimeoutError try: # ... proxy connection attempt ... pass except ProxyTimeoutError: print('Proxy connection timed out.') ``` -------------------------------- ### Parse Proxy URL Basic Example Source: https://github.com/romis2012/python-socks/blob/master/_autodocs/types.md Parses a basic proxy URL string to extract the proxy type, host, and port. ```python from python_socks import parse_proxy_url, ProxyType # Basic URL ptype, host, port, user, pwd = parse_proxy_url('socks5://127.0.0.1:1080') # Result: (ProxyType.SOCKS5, '127.0.0.1', 1080, '', '') ``` -------------------------------- ### AnyIO HTTP GET via SOCKS5 Source: https://github.com/romis2012/python-socks/blob/master/README.md Designed for AnyIO applications. The `connect` method directly returns an `AnyioSocketStream` with SSL context. ```python import ssl from python_socks.async_.anyio import Proxy proxy = Proxy.from_url('socks5://user:password@127.0.0.1:1080') # `connect` returns AnyioSocketStream stream = await proxy.connect( dest_host='check-host.net', dest_port=443, dest_ssl=ssl.create_default_context(), ) request = ( b'GET /ip HTTP/1.1\r\n' b'Host: check-host.net\r\n' b'Connection: close\r\n\r\n' ) await stream.write_all(request) response = await stream.read() print(response) ``` -------------------------------- ### Asynchronous Connection Establishment Flow Source: https://github.com/romis2012/python-socks/blob/master/_autodocs/architecture.md Outlines the asynchronous connection process, emphasizing the use of `await` for I/O operations and framework-specific timeout handling. ```plaintext User calls: await proxy.connect(dest_host, dest_port, timeout) ↓ Wrap timeout with framework-specific mechanism ↓ [Async TCP connection to proxy_host:proxy_port] ↓ Wrap socket in AsyncioSocketStream (or variant) ↓ Get Connector via factory ↓ [Await connector.connect() coroutine] ├─ [Await stream.write_all()] ├─ [Await stream.read_exact()] └─ [Await stream.close() on error] ↓ Return socket/stream to caller ``` -------------------------------- ### Asyncio HTTP GET via SOCKS5 Source: https://github.com/romis2012/python-socks/blob/master/README.md Suitable for asyncio applications. The `connect` method returns a non-blocking socket compatible with `asyncio.open_connection`. ```python import ssl import asyncio from python_socks.async_.asyncio import Proxy proxy = Proxy.from_url('socks5://user:password@127.0.0.1:1080') # `connect` returns standard Python socket in non-blocking mode # so we can pass it to asyncio.open_connection(...) sock = await proxy.connect(dest_host='check-host.net', dest_port=443) reader, writer = await asyncio.open_connection( host=None, port=None, sock=sock, ssl=ssl.create_default_context(), server_hostname='check-host.net', ) request = ( b'GET /ip HTTP/1.1\r\n' b'Host: check-host.net\r\n' b'Connection: close\r\n\r\n' ) writer.write(request) response = await reader.read(-1) print(response) ``` -------------------------------- ### Proxy Constructor Source: https://github.com/romis2012/python-socks/blob/master/_autodocs/other-async-frameworks.md Initializes a Proxy instance for Anyio. This object is used to establish connections through a specified proxy server. ```APIDOC ## Proxy.__init__() ### Description Initializes a Proxy instance for Anyio. This object is used to establish connections through a specified proxy server. ### Method Proxy.__init__ ### Parameters #### Path Parameters - **proxy_type** (ProxyType) - Required - Proxy protocol (SOCKS4, SOCKS5, or HTTP) - **host** (str) - Required - Proxy server hostname or IP address - **port** (int) - Required - Proxy server port number - **username** (str | None) - Optional - Username for authentication (SOCKS5/HTTP) - **password** (str | None) - Optional - Password for authentication (SOCKS5/HTTP) - **rdns** (bool | None) - Optional - Remote DNS resolution flag; None uses protocol default - **proxy_ssl** (ssl.SSLContext | None) - Optional - SSL context for TLS connection to proxy server ### Notes The `proxy_ssl` parameter is unique to anyio; use it to encrypt the connection to the proxy server itself. ``` -------------------------------- ### Synchronous HTTP GET via SOCKS5 Source: https://github.com/romis2012/python-socks/blob/master/README.md Use this for standard blocking socket operations. Ensure the proxy URL is correctly formatted. ```python import ssl from python_socks.sync import Proxy proxy = Proxy.from_url('socks5://user:password@127.0.0.1:1080') # `connect` returns standard Python socket in blocking mode sock = proxy.connect(dest_host='check-host.net', dest_port=443) sock = ssl.create_default_context().wrap_socket( sock=sock, server_hostname='check-host.net' ) request = ( b'GET /ip HTTP/1.1\r\n' b'Host: check-host.net\r\n' b'Connection: close\r\n\r\n' ) sock.sendall(request) response = sock.recv(4096) print(response) ``` -------------------------------- ### Create Proxy from URL Source: https://github.com/romis2012/python-socks/blob/master/_autodocs/INDEX.md Instantiate a Proxy object by providing a proxy URL. This is the primary way to configure your proxy settings. ```APIDOC ## Create Proxy from URL ### Description Instantiate a Proxy object by providing a proxy URL. This is the primary way to configure your proxy settings. ### Method Signature ```python Proxy.from_url(url: str) ``` ### Parameters - **url** (str) - The URL of the proxy server (e.g., 'socks5://user:pass@host:port'). ### Returns - A Proxy object configured with the provided URL. ### Example ```python proxy = Proxy.from_url('socks5://user:pass@host:port') ``` ``` -------------------------------- ### Trio HTTP GET via SOCKS5 Source: https://github.com/romis2012/python-socks/blob/master/README.md For use with the Trio framework. The `connect` method returns a Trio-compatible socket which is then wrapped in `trio.SocketStream` and `trio.SSLStream`. ```python import ssl import trio from python_socks.async_.trio import Proxy proxy = Proxy.from_url('socks5://user:password@127.0.0.1:1080') # `connect` returns trio socket # so we can pass it to trio.SocketStream sock = await proxy.connect(dest_host='check-host.net', dest_port=443) stream = trio.SocketStream(sock) stream = trio.SSLStream( stream, ssl.create_default_context(), server_hostname='check-host.net' ) await stream.do_handshake() request = ( b'GET /ip HTTP/1.1\r\n' b'Host: check-host.net\r\n' b'Connection: close\r\n\r\n' ) await stream.send_all(request) response = await stream.receive_some(4096) print(response) ``` -------------------------------- ### Anyio with Automatic SSL Wrapping Source: https://github.com/romis2012/python-socks/blob/master/_autodocs/COMPLETION_REPORT.txt Illustrates using AnyioProxy with automatic SSL wrapping for HTTPS connections. Anyio's stream abstraction simplifies secure connection handling. ```python import anyio from python_socks.sync_proxy import SyncProxy async def main(): proxy = SyncProxy(proxy_host='127.0.0.1', proxy_port=1080, proxy_ssl=True) # AnyioProxy integrates with anyio streams stream = await proxy.connect('example.com', 443) request = b"GET / HTTP/1.1\r\nHost: example.com\r\nConnection: close\r\n\r\n" await stream.send(request) response = await stream.receive(4096) print(response.decode()) await stream.close() anyio.run(main) ``` -------------------------------- ### Asyncio Proxy Connection with SSL Source: https://github.com/romis2012/python-socks/blob/master/_autodocs/asyncio-api.md Demonstrates establishing an asynchronous connection through a SOCKS5 proxy to a destination, then using the resulting socket with `asyncio.open_connection` for an HTTPS connection. Includes sending an HTTP request and reading the response. ```python import asyncio import ssl from python_socks.async_.asyncio import Proxy from python_socks import ProxyType async def main(): # Create proxy instance proxy = Proxy( proxy_type=ProxyType.SOCKS5, host='127.0.0.1', port=1080, username='user', password='pass' ) # Connect through proxy sock = await proxy.connect( dest_host='example.com', dest_port=443, timeout=10 ) # Use with asyncio.open_connection() context = ssl.create_default_context() reader, writer = await asyncio.open_connection( host=None, port=None, sock=sock, ssl=context, server_hostname='example.com' ) # Send HTTP request writer.write(b'GET / HTTP/1.1\r\nHost: example.com\r\n\r\n') await writer.drain() # Read response data = await reader.read(4096) writer.close() await writer.wait_closed() asyncio.run(main()) ```