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, 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.9
• async-timeout >= 5.0 (optional)
• trio >= 0.30 (optional)
• anyio >= 4.12 (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 anyio support:

pip install python-socks[anyio]

Simple usage

We are making secure HTTP GET request via SOCKS5 proxy

Sync

importsslfrompython_socks.syncimportProxydeffetch():
proxy=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",
)
# fmt: offrequest= (
b"GET /ip HTTP/1.1\r\n"b"Host: check-host.net\r\n"b"Connection: close\r\n\r\n"
)
# fmt: onsock.sendall(request)
response=sock.recv(4096)
print(response)
fetch()

Async (asyncio)

importasyncioimportsslfrompython_socks.async_.asyncioimportProxyasyncdeffetch():
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=awaitproxy.connect(
dest_host="check-host.net",
dest_port=443,
)
reader, writer=awaitasyncio.open_connection(
sock=sock,
ssl=ssl.create_default_context(),
server_hostname="check-host.net",
)
# fmt: offrequest= (
b"GET /ip HTTP/1.1\r\n"b"Host: check-host.net\r\n"b"Connection: close\r\n\r\n"
)
# fmt: onwriter.write(request)
response=awaitreader.read(-1)
print(response)
writer.close()
awaitwriter.wait_closed()
asyncio.run(fetch())

Async (trio)

importsslimporttriofrompython_socks.async_.trioimportProxyasyncdeffetch():
proxy=Proxy.from_url("socks5://user:password@127.0.0.1:1080")
# `connect` returns trio.socket.SocketType# 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_context=ssl.create_default_context(),
server_hostname="check-host.net",
)
awaitstream.do_handshake()
# fmt: offrequest= (
b"GET /ip HTTP/1.1\r\n"b"Host: check-host.net\r\n"b"Connection: close\r\n\r\n"
)
# fmt: onawaitstream.send_all(request)
response=awaitstream.receive_some(4096)
print(response)
awaitstream.aclose()
trio.run(fetch)

Async (anyio)

importsslimportanyiofromanyio.streams.tlsimportTLSStreamfrompython_socks.async_.anyioimportProxyasyncdeffetch():
proxy=Proxy.from_url("socks5://user:password@127.0.0.1:1080")
# `connect` returns anyio.abc.SocketStream# we can use it directlystream=awaitproxy.connect(
dest_host="check-host.net",
dest_port=443,
)
stream=awaitTLSStream.wrap(
stream,
ssl_context=ssl.create_default_context(),
hostname="check-host.net",
)
# fmt: offrequest= (
b"GET /ip HTTP/1.1\r\n"b"Host: check-host.net\r\n"b"Connection: close\r\n\r\n"
)
# fmt: onawaitstream.send(request)
response=awaitstream.receive(4096)
print(response)
awaitstream.aclose()
anyio.run(fetch)

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)

Proxy Chaining (sync example — same for asyncio, trio, anyio)

importsslfrompython_socks.syncimportProxydeffetch():
proxy1=Proxy.from_url("socks5://user:password@127.0.0.1:1080")
proxy2=Proxy.from_url("socks4://127.0.0.1:1081", forward=proxy1)
proxy3=Proxy.from_url("http://user:password@127.0.0.1:1082", forward=proxy2)
sock=proxy3.connect(
dest_host="check-host.net",
dest_port=443,
)
sock=ssl.create_default_context().wrap_socket(
sock=sock,
server_hostname="check-host.net",
)
# fmt: offrequest= (
b"GET /ip HTTP/1.1\r\n"b"Host: check-host.net\r\n"b"Connection: close\r\n\r\n"
)
# fmt: onsock.sendall(request)
response=sock.recv(4096)
print(response)
fetch()