PySocks Monkey Patch with NO_PROXY Support #175
Description
PySocks doesn't natively support the NO_PROXY environment variable, which causes
local/private network requests to be incorrectly routed through the proxy, leading
to connection errors like:
error: reject loopback connection to: 192.168.1.90:8000
This is the correct way to use monkey patching with PySocks.
"""
This patch adds NO_PROXY support by intercepting the connect() method.
"""
import os
import re
import ipaddress
def is_local_address(host: str) -> bool:
"""
Determine if an address should bypass the proxy based on NO_PROXY environment variable.
If NO_PROXY is not set, uses default local address patterns:
- localhost
- 127.* (loopback)
- 10.* (private class A)
- 172.16.* to 172.31.* (private class B)
- 192.168.* (private class C)
- ::1 (IPv6 loopback)
Args:
host: Hostname or IP address
Returns:
bool: True if the address should bypass proxy
"""
if not host:
return False
# Read NO_PROXY environment variable (case-insensitive)
no_proxy = os.environ.get('NO_PROXY') or os.environ.get('no_proxy') or ''
# If NO_PROXY is not set, use default local address patterns
if not no_proxy:
host_lower = host.lower()
# localhost domain
if host_lower in ('localhost', 'localhost.localdomain'):
return True
# IPv6 loopback
if host == '::1':
return True
# 127.* (loopback)
if host.startswith('127.'):
return True
# 10.* (private class A)
if host.startswith('10.'):
return True
# 172.16.* to 172.31.* (private class B)
if host.startswith('172.'):
try:
parts = host.split('.')
if len(parts) >= 2:
second_octet = int(parts[1])
if 16 <= second_octet <= 31:
return True
except (ValueError, IndexError):
pass
# 192.168.* (private class C)
if host.startswith('192.168.'):
return True
return False
# Parse NO_PROXY list (comma-separated)
no_proxy_list = [item.strip() for item in no_proxy.split(',') if item.strip()]
host_lower = host.lower()
for pattern in no_proxy_list:
pattern = pattern.strip()
if not pattern:
continue
# Handle patterns with port (e.g., localhost:8080)
if ':' in pattern:
pattern_host, _ = pattern.rsplit(':', 1)
else:
pattern_host = pattern
pattern_host_lower = pattern_host.lower()
# Exact match
if host_lower == pattern_host_lower:
return True
# Wildcard match (e.g., *.local)
if '*' in pattern_host:
regex_pattern = pattern_host.replace('.', r'\.').replace('*', '.*')
if re.match(regex_pattern + '$', host, re.IGNORECASE):
return True
# CIDR match (e.g., 192.168.0.0/16)
if '/' in pattern_host:
try:
network = ipaddress.ip_network(pattern_host, strict=False)
try:
host_ip = ipaddress.ip_address(host)
if host_ip in network:
return True
except ValueError:
# host is not a valid IP, skip CIDR matching
pass
except ValueError:
# pattern_host is not a valid CIDR, skip
pass
# Prefix match (e.g., 192.168. or 10.)
if host.startswith(pattern_host):
return True
return False
def setup_proxy_with_no_proxy_support(proxy_host, proxy_port, proxy_type='SOCKS5'):
"""
Set up a global SOCKS proxy with NO_PROXY support via monkey patching.
This patches PySocks to respect the NO_PROXY environment variable, which
the library doesn't support natively.
Supported NO_PROXY formats:
- Exact match: localhost, 127.0.0.1
- Wildcards: *.local
- CIDR notation: 192.168.0.0/16, 10.0.0.0/8
- Prefix match: 192.168., 10.
- With port: localhost:8080
Args:
proxy_host: Proxy server hostname/IP
proxy_port: Proxy server port
proxy_type: Proxy type ('SOCKS5', 'SOCKS4', or 'HTTP')
Example:
>>> import os
>>> os.environ['NO_PROXY'] = 'localhost,127.0.0.1,192.168.0.0/16'
>>> setup_proxy_with_no_proxy_support('127.0.0.1', 1080)
# Now all socket connections will use the proxy except NO_PROXY addresses
>>> import requests
>>> requests.get('http://example.com') # Uses proxy
>>> requests.get('http://192.168.1.100:8000') # Bypasses proxy
"""
import socks
import socket
# Map proxy type string to constant
proxy_type_map = {
'SOCKS5': socks.SOCKS5,
'SOCKS4': socks.SOCKS4,
'HTTP': socks.HTTP,
}
proxy_type_const = proxy_type_map.get(proxy_type.upper(), socks.SOCKS5)
# Set default proxy
socks.set_default_proxy(proxy_type_const, proxy_host, proxy_port, True)
# Save original connect method
_original_socks_connect = socks.socksocket.connect
def patched_connect(self, dest_pair, catch_errors=None):
"""
Patched connect method that bypasses proxy for local addresses.
Args:
self: socksocket instance
dest_pair: (host, port) tuple
catch_errors: Error handling parameter
"""
host = dest_pair[0]
if is_local_address(host):
# Temporarily disable proxy for local addresses
old_proxy = self.proxy
self.proxy = (None, None, None, None, None, None)
try:
return _original_socks_connect(self, dest_pair, catch_errors)
finally:
self.proxy = old_proxy
else:
return _original_socks_connect(self, dest_pair, catch_errors)
# Apply the patch
socks.socksocket.connect = patched_connect
# Replace standard socket with socksocket
socket.socket = socks.socksocket
# Log configuration
no_proxy = os.environ.get('NO_PROXY') or os.environ.get('no_proxy')
if no_proxy:
print(f"Proxy configured: {proxy_host}:{proxy_port} (type: {proxy_type})")
print(f"NO_PROXY: {no_proxy}")
else:
print(f"Proxy configured: {proxy_host}:{proxy_port} (type: {proxy_type})")
print("NO_PROXY not set - using default local address patterns")
# Usage Example
if __name__ == '__main__':
import os
# Set NO_PROXY environment variable
os.environ['NO_PROXY'] = 'localhost,127.0.0.1,192.168.0.0/16,10.0.0.0/8,*.local'
# Setup proxy with NO_PROXY support
setup_proxy_with_no_proxy_support('127.0.0.1', 1080, 'SOCKS5')
# Now use any library that uses sockets
import requests
# This will use the proxy
response = requests.get('http://example.com')
# This will bypass the proxy (local address)
response = requests.get('http://192.168.1.100:8000')
# For aiohttp, it will also respect the patched socket
import aiohttp
import asyncio
async def fetch():
async with aiohttp.ClientSession() as session:
# Uses proxy
async with session.get('http://example.com') as resp:
print(await resp.text())
# Bypasses proxy
async with session.get('http://192.168.1.100:8000') as resp:
print(await resp.text())
asyncio.run(fetch())