python-socks

The python-socks package provides a core proxy client functionality for Python. Supports SOCKS4(a), SOCKS5(h), HTTP CONNECT proxy and provides sync and async (asyncio, trio, curio, anyio) APIs. You probably don't need to use python-socks directly. It is used internally by aiohttp-socks and httpx-socks packages.

Requirements

• Python >= 3.8
• async-timeout >= 4.0 (optional)
• trio >= 0.24 (optional)
• curio >= 1.4 (optional)
• anyio >= 3.3.4 (optional)

Installation

only sync proxy support:

pip install python-socks

to include optional asyncio support:

pip install python-socks[asyncio]

to include optional trio support:

pip install python-socks[trio]

to include optional curio support:

pip install python-socks[curio]

to include optional anyio support:

pip install python-socks[anyio]

Simple usage

We are making secure HTTP GET request via SOCKS5 proxy

Sync

importsslfrompython_socks.syncimportProxyproxy=Proxy.from_url('socks5://user:password@127.0.0.1:1080')
# `connect` returns standard Python socket in blocking modesock=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)

Async (asyncio)

importsslimportasynciofrompython_socks.async_.asyncioimportProxyproxy=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=awaitproxy.connect(dest_host='check-host.net', dest_port=443)
reader, writer=awaitasyncio.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=awaitreader.read(-1)
print(response)

Async (trio)

importsslimporttriofrompython_socks.async_.trioimportProxyproxy=Proxy.from_url('socks5://user:password@127.0.0.1:1080')
# `connect` returns trio socket # so we can pass it to trio.SocketStreamsock=awaitproxy.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'
)
awaitstream.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'
)
awaitstream.send_all(request)
response=awaitstream.receive_some(4096)
print(response)

Async (curio)

importcurio.sslascuriosslfrompython_socks.async_.curioimportProxyproxy=Proxy.from_url('socks5://user:password@127.0.0.1:1080')
# `connect` returns curio.io.Socketsock=awaitproxy.connect(
dest_host='check-host.net',
dest_port=443
)
request= (
b'GET /ip HTTP/1.1\r\n'b'Host: check-host.net\r\n'b'Connection: close\r\n\r\n'
)
ssl_context=curiossl.create_default_context()
sock=awaitssl_context.wrap_socket(
sock, do_handshake_on_connect=False, server_hostname='check-host.net'
)
awaitsock.do_handshake()
stream=sock.as_stream()
awaitstream.write(request)
response=awaitstream.read(1024)
print(response)

Async (anyio)

importsslfrompython_socks.async_.anyioimportProxyproxy=Proxy.from_url('socks5://user:password@127.0.0.1:1080')
# `connect` returns AnyioSocketStreamstream=awaitproxy.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'
)
awaitstream.write_all(request)
response=awaitstream.read()
print(response)

More complex example

A urllib3 PoolManager that routes connections via the proxy

fromurllib3importPoolManager, HTTPConnectionPool, HTTPSConnectionPoolfromurllib3.connectionimportHTTPConnection, HTTPSConnectionfrompython_socks.syncimportProxyclassProxyHTTPConnection(HTTPConnection):
def__init__(self, *args, **kwargs):
socks_options=kwargs.pop('_socks_options')
self._proxy_url=socks_options['proxy_url']
super().__init__(*args, **kwargs)
def_new_conn(self):
proxy=Proxy.from_url(self._proxy_url)
returnproxy.connect(
dest_host=self.host,
dest_port=self.port,
timeout=self.timeout
)
classProxyHTTPSConnection(ProxyHTTPConnection, HTTPSConnection):
passclassProxyHTTPConnectionPool(HTTPConnectionPool):
ConnectionCls=ProxyHTTPConnectionclassProxyHTTPSConnectionPool(HTTPSConnectionPool):
ConnectionCls=ProxyHTTPSConnectionclassProxyPoolManager(PoolManager):
def__init__(self, proxy_url, timeout=5, num_pools=10, headers=None,
**connection_pool_kw):
connection_pool_kw['_socks_options'] = {'proxy_url': proxy_url}
connection_pool_kw['timeout'] =timeoutsuper().__init__(num_pools, headers, **connection_pool_kw)
self.pool_classes_by_scheme= {
'http': ProxyHTTPConnectionPool,
'https': ProxyHTTPSConnectionPool,
}
### and how to use itmanager=ProxyPoolManager('socks5://user:password@127.0.0.1:1080')
response=manager.request('GET', 'https://check-host.net/ip')
print(response.data)