## python-socks
[](https://travis-ci.com/github/romis2012/python-socks)
[](https://coveralls.io/github/romis2012/python-socks?branch=master)
[](https://badge.fury.io/py/python-socks)
<!--
[](https://pepy.tech/project/python-socks)
-->
The `python-socks` package provides a core proxy client functionality for Python.
Supports SOCKS4(a), SOCKS5(h), HTTP (tunneling) 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](https://github.com/romis2012/aiohttp-socks) and [httpx-socks](https://github.com/romis2012/httpx-socks) packages.
## Requirements
- Python >= 3.6
- async-timeout >= 3.0.1 (optional)
- trio >= 0.16.0 (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
```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)
```
#### Async (asyncio)
```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)
```
#### Async (trio)
```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)
```
#### Async (curio)
```python
import curio.ssl as curiossl
from python_socks.async_.curio import Proxy
proxy = Proxy.from_url('socks5://user:password@127.0.0.1:1080')
# `connect` returns curio.io.Socket
sock = await proxy.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 = await ssl_context.wrap_socket(
sock, do_handshake_on_connect=False, server_hostname='check-host.net'
)
await sock.do_handshake()
stream = sock.as_stream()
await stream.write(request)
response = await stream.read(1024)
print(response)
```
#### Async (anyio)
```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)
```
## More complex example
#### A urllib3 PoolManager that routes connections via the proxy
```python
from urllib3 import PoolManager, HTTPConnectionPool, HTTPSConnectionPool
from urllib3.connection import HTTPConnection, HTTPSConnection
from python_socks.sync import Proxy
class ProxyHTTPConnection(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)
return proxy.connect(
dest_host=self.host,
dest_port=self.port,
timeout=self.timeout
)
class ProxyHTTPSConnection(ProxyHTTPConnection, HTTPSConnection):
pass
class ProxyHTTPConnectionPool(HTTPConnectionPool):
ConnectionCls = ProxyHTTPConnection
class ProxyHTTPSConnectionPool(HTTPSConnectionPool):
ConnectionCls = ProxyHTTPSConnection
class ProxyPoolManager(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'] = timeout
super().__init__(num_pools, headers, **connection_pool_kw)
self.pool_classes_by_scheme = {
'http': ProxyHTTPConnectionPool,
'https': ProxyHTTPSConnectionPool,
}
### and how to use it
manager = ProxyPoolManager('socks5://user:password@127.0.0.1:1080')
response = manager.request('GET', 'https://check-host.net/ip')
print(response.data)
```
Raw data
{
"_id": null,
"home_page": "https://github.com/romis2012/python-socks",
"name": "python-socks",
"maintainer": "",
"docs_url": null,
"requires_python": "",
"maintainer_email": "",
"keywords": "socks socks5 socks4 http proxy asyncio trio curio anyio",
"author": "Roman Snegirev",
"author_email": "snegiryev@gmail.com",
"download_url": "https://files.pythonhosted.org/packages/3f/dd/f8c98f6ae1c57d8f1889913ad3762468968762bf9eef2590a86d4cf4f393/python-socks-2.4.4.tar.gz",
"platform": null,
"description": "## python-socks\n\n[](https://travis-ci.com/github/romis2012/python-socks)\n[](https://coveralls.io/github/romis2012/python-socks?branch=master)\n[](https://badge.fury.io/py/python-socks)\n<!--\n[](https://pepy.tech/project/python-socks)\n-->\n\nThe `python-socks` package provides a core proxy client functionality for Python.\nSupports SOCKS4(a), SOCKS5(h), HTTP (tunneling) proxy and provides sync and async (asyncio, trio, curio, anyio) APIs.\nYou probably don't need to use `python-socks` directly. \nIt is used internally by \n[aiohttp-socks](https://github.com/romis2012/aiohttp-socks) and [httpx-socks](https://github.com/romis2012/httpx-socks) packages. \n\n## Requirements\n- Python >= 3.6\n- async-timeout >= 3.0.1 (optional)\n- trio >= 0.16.0 (optional)\n- curio >= 1.4 (optional)\n- anyio >= 3.3.4 (optional)\n\n## Installation\n\nonly sync proxy support:\n```\npip install python-socks\n```\n\nto include optional asyncio support:\n```\npip install python-socks[asyncio]\n```\n\nto include optional trio support:\n```\npip install python-socks[trio]\n```\n\nto include optional curio support:\n```\npip install python-socks[curio]\n```\n\nto include optional anyio support:\n```\npip install python-socks[anyio]\n```\n\n## Simple usage\nWe are making secure HTTP GET request via SOCKS5 proxy\n \n#### Sync\n```python\nimport ssl\nfrom python_socks.sync import Proxy\n\nproxy = Proxy.from_url('socks5://user:password@127.0.0.1:1080')\n\n# `connect` returns standard Python socket in blocking mode\nsock = proxy.connect(dest_host='check-host.net', dest_port=443)\n\nsock = ssl.create_default_context().wrap_socket(\n sock=sock,\n server_hostname='check-host.net'\n)\n\nrequest = (\n b'GET /ip HTTP/1.1\\r\\n'\n b'Host: check-host.net\\r\\n'\n b'Connection: close\\r\\n\\r\\n'\n)\nsock.sendall(request)\nresponse = sock.recv(4096)\nprint(response)\n```\n\n#### Async (asyncio)\n```python\nimport ssl\nimport asyncio\nfrom python_socks.async_.asyncio import Proxy\n\nproxy = Proxy.from_url('socks5://user:password@127.0.0.1:1080')\n\n# `connect` returns standard Python socket in non-blocking mode \n# so we can pass it to asyncio.open_connection(...)\nsock = await proxy.connect(dest_host='check-host.net', dest_port=443)\n\nreader, writer = await asyncio.open_connection(\n host=None,\n port=None,\n sock=sock,\n ssl=ssl.create_default_context(),\n server_hostname='check-host.net',\n)\n\nrequest = (\n b'GET /ip HTTP/1.1\\r\\n'\n b'Host: check-host.net\\r\\n'\n b'Connection: close\\r\\n\\r\\n'\n)\n\nwriter.write(request)\nresponse = await reader.read(-1)\nprint(response)\n```\n\n#### Async (trio)\n```python\nimport ssl\nimport trio\nfrom python_socks.async_.trio import Proxy\n\nproxy = Proxy.from_url('socks5://user:password@127.0.0.1:1080')\n\n# `connect` returns trio socket \n# so we can pass it to trio.SocketStream\nsock = await proxy.connect(dest_host='check-host.net', dest_port=443)\n\nstream = trio.SocketStream(sock)\n\nstream = trio.SSLStream(\n stream, ssl.create_default_context(),\n server_hostname='check-host.net'\n)\nawait stream.do_handshake()\n\nrequest = (\n b'GET /ip HTTP/1.1\\r\\n'\n b'Host: check-host.net\\r\\n'\n b'Connection: close\\r\\n\\r\\n'\n)\n\nawait stream.send_all(request)\nresponse = await stream.receive_some(4096)\nprint(response)\n```\n\n#### Async (curio)\n```python\nimport curio.ssl as curiossl\nfrom python_socks.async_.curio import Proxy\n\nproxy = Proxy.from_url('socks5://user:password@127.0.0.1:1080')\n# `connect` returns curio.io.Socket\nsock = await proxy.connect(\n dest_host='check-host.net',\n dest_port=443\n)\n\nrequest = (\n b'GET /ip HTTP/1.1\\r\\n'\n b'Host: check-host.net\\r\\n'\n b'Connection: close\\r\\n\\r\\n'\n)\n\nssl_context = curiossl.create_default_context()\nsock = await ssl_context.wrap_socket(\n sock, do_handshake_on_connect=False, server_hostname='check-host.net'\n)\n\nawait sock.do_handshake()\n\nstream = sock.as_stream()\n\nawait stream.write(request)\nresponse = await stream.read(1024)\nprint(response)\n```\n\n#### Async (anyio)\n```python\nimport ssl\nfrom python_socks.async_.anyio import Proxy\n\nproxy = Proxy.from_url('socks5://user:password@127.0.0.1:1080')\n\n# `connect` returns AnyioSocketStream\nstream = await proxy.connect(\n dest_host='check-host.net',\n dest_port=443,\n dest_ssl=ssl.create_default_context(),\n)\n\nrequest = (\n b'GET /ip HTTP/1.1\\r\\n'\n b'Host: check-host.net\\r\\n'\n b'Connection: close\\r\\n\\r\\n'\n)\n\nawait stream.write_all(request)\nresponse = await stream.read()\nprint(response)\n```\n\n## More complex example\n\n#### A urllib3 PoolManager that routes connections via the proxy\n\n```python\nfrom urllib3 import PoolManager, HTTPConnectionPool, HTTPSConnectionPool\nfrom urllib3.connection import HTTPConnection, HTTPSConnection\nfrom python_socks.sync import Proxy\n\n\nclass ProxyHTTPConnection(HTTPConnection):\n def __init__(self, *args, **kwargs):\n socks_options = kwargs.pop('_socks_options')\n self._proxy_url = socks_options['proxy_url']\n super().__init__(*args, **kwargs)\n\n def _new_conn(self):\n proxy = Proxy.from_url(self._proxy_url)\n return proxy.connect(\n dest_host=self.host,\n dest_port=self.port,\n timeout=self.timeout\n )\n\n\nclass ProxyHTTPSConnection(ProxyHTTPConnection, HTTPSConnection):\n pass\n\n\nclass ProxyHTTPConnectionPool(HTTPConnectionPool):\n ConnectionCls = ProxyHTTPConnection\n\n\nclass ProxyHTTPSConnectionPool(HTTPSConnectionPool):\n ConnectionCls = ProxyHTTPSConnection\n\n\nclass ProxyPoolManager(PoolManager):\n def __init__(self, proxy_url, timeout=5, num_pools=10, headers=None,\n **connection_pool_kw):\n\n connection_pool_kw['_socks_options'] = {'proxy_url': proxy_url}\n connection_pool_kw['timeout'] = timeout\n\n super().__init__(num_pools, headers, **connection_pool_kw)\n\n self.pool_classes_by_scheme = {\n 'http': ProxyHTTPConnectionPool,\n 'https': ProxyHTTPSConnectionPool,\n }\n\n\n### and how to use it\nmanager = ProxyPoolManager('socks5://user:password@127.0.0.1:1080')\nresponse = manager.request('GET', 'https://check-host.net/ip')\nprint(response.data)\n```\n\n\n",
"bugtrack_url": null,
"license": "Apache 2",
"summary": "Core proxy (SOCKS4, SOCKS5, HTTP tunneling) functionality for Python",
"version": "2.4.4",
"project_urls": {
"Homepage": "https://github.com/romis2012/python-socks"
},
"split_keywords": [
"socks",
"socks5",
"socks4",
"http",
"proxy",
"asyncio",
"trio",
"curio",
"anyio"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "0c82cbb43250a471a98511c982a2d795fb99f7e2e91f0c5077f034cd50496d21",
"md5": "a55576d957e2b713b20c27de1ccff29e",
"sha256": "fda465d3ef229119ee614eb85f2b7c0ad28be6dd40e0ef8dd317c49e8725e514"
},
"downloads": -1,
"filename": "python_socks-2.4.4-py3-none-any.whl",
"has_sig": false,
"md5_digest": "a55576d957e2b713b20c27de1ccff29e",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": null,
"size": 52433,
"upload_time": "2023-12-08T14:26:37",
"upload_time_iso_8601": "2023-12-08T14:26:37.163445Z",
"url": "https://files.pythonhosted.org/packages/0c/82/cbb43250a471a98511c982a2d795fb99f7e2e91f0c5077f034cd50496d21/python_socks-2.4.4-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "3fddf8c98f6ae1c57d8f1889913ad3762468968762bf9eef2590a86d4cf4f393",
"md5": "fa04b54df356eb40cb8fe022ef4d264c",
"sha256": "e5a8e4f78203612c813946feacd87b98943965a04389fe221fa1e9ab263ad22e"
},
"downloads": -1,
"filename": "python-socks-2.4.4.tar.gz",
"has_sig": false,
"md5_digest": "fa04b54df356eb40cb8fe022ef4d264c",
"packagetype": "sdist",
"python_version": "source",
"requires_python": null,
"size": 25792,
"upload_time": "2023-12-08T14:26:39",
"upload_time_iso_8601": "2023-12-08T14:26:39.620268Z",
"url": "https://files.pythonhosted.org/packages/3f/dd/f8c98f6ae1c57d8f1889913ad3762468968762bf9eef2590a86d4cf4f393/python-socks-2.4.4.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2023-12-08 14:26:39",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "romis2012",
"github_project": "python-socks",
"travis_ci": true,
"coveralls": true,
"github_actions": false,
"lcname": "python-socks"
}
JSON
