python-proxy 项目

# 什么是 socks5代理，它有什么优势？ socks5代理是一种网络协议，它可以在客户端和服务器之间建立一个中间层，使得客户端可以通过中间层来访问任何目标服务器，而不需要直接连接。socks5代理的全称是 Socket Secure version 5，它是 socks 协议的第五个版本，也是目前最新和最完善的版本。 socks5代理有以下几个优势： - 支持多种认证方式。socks5代理可以根据客户端和服务器的协商，选择合适的认证方式，如无需认证、用户名密码认证、GSS-API 认证等。这样可以提高安全性和灵活性。 - 支持多种地址类型。socks5代理可以处理 IPv4、IPv6、域名和其他任意格式的地址，而不需要进行 DNS 解析。这样可以提高效率和兼容性。 - 支持多种数据传输模式。socks5代理可以支持 TCP 和 UDP 两种数据传输模式，以及 BIND 和 ASSOCIATE 两种连接模式。这样可以满足不同的应用场景和需求。 - 支持透明代理。socks5代理可以在不修改客户端或服务器的情况下，实现透明地转发数据包。这样可以保持原始的网络结构和功能。 ## 如何使用 socks5代理？ 要使用 socks5代理，你需要以下几个条件： - 一个可用的 socks5代理服务器。你可以自己搭建一个 socks5代理服务器，或者使用第三方提供的服务。 - 一个支持 socks5协议的客户端软件。你可以使用浏览器、邮件客户端、即时通讯软件等任何支持 socks5协议的软件。 - 一个正确的配置方法。你需要根据你的具体情况，配置好你的客户端软件和 socks5代理服务器之间的连接参数。 具体的配置方法可能因为不同的软件而有所差异，但一般来说，你需要设置以下几个参数： - 代理类型：选择 SOCKS 或 SOCKS v.5 - 代理地址：输入你的 socks5代理服务器的 IP 地址或域名 - 代理端口：输入你的 socks5代理服务器监听的端口号 - 认证方式：选择你和你的 socks5代理服务器协商好的认证方式，并输入相应的用户名和密码（如果有） 完成以上设置后，你就可以通过 socks5代理来访问任何目标网站了。 ## 总结 socks5代理是一种强大而灵活的网络协议，它可以帮助我们实现匿名上网、突破网络封锁、加速网络访问等功能。

pycryptodome是一个可选库，可以启用更快（C版本）的密码学lib。pproxy有许多内置的纯Python密码学功能，它们是稳定的轻量级库，但速度比C版本慢。使用PyPy进行加速后，纯Python密码学模块的性能可以达到与C版本类似的水平。如果性能很重要且不使用PyPy，请安装pycryptodome替代纯Python模块。 asyncssh是一个可选库，可以启用SSH隧道客户端支持。

``` python class WindowsSetting(object): KEY = r'Software\Microsoft\Windows\CurrentVersion\Internet Settings\Connections' SUBKEY = 'DefaultConnectionSettings' def __init__(self, args): self.listen = None for option in args.listen: protos = [x.name for x in option.protos] if option.unix or 'ssl' in protos or 'secure' in protos: continue if 'http' in protos: self.listen = option break if self.listen is None: print('No server listen on localhost by http') import winreg key = winreg.OpenKey(winreg.HKEY_CURRENT_USER, self.KEY, 0, winreg.KEY_ALL_ACCESS) value, regtype = winreg.QueryValueEx(key, self.SUBKEY) assert regtype == winreg.REG_BINARY server = f'localhost:{self.listen.port}'.encode() bypass = '<local>'.encode() counter = int.from_bytes(value[4:8], 'little') + 1 value = value[:4] + struct.pack('<III', counter, 3, len(server)) + server + struct.pack('<I', len(bypass)) + bypass + b'\x00'*36 winreg.SetValueEx(key, self.SUBKEY, None, regtype, value) winreg.CloseKey(key) def clear(self): if self.listen is None: return import winreg key = winreg.OpenKey(winreg.HKEY_CURRENT_USER, self.KEY, 0, winreg.KEY_ALL_ACCESS) value, regtype = winreg.QueryValueEx(key, self.SUBKEY) assert regtype == winreg.REG_BINARY counter = int.from_bytes(value[4:8], 'little') + 1 value = value[:4] + struct.pack('<II', counter, 1) + b'\x00'*44 winreg.SetValueEx(key, self.SUBKEY, None, regtype, value) winreg.CloseKey(key) ``` 用于设置或清除 Windows 系统代理设置，代理一个监听 HTTP 协议的本地主机。 该类有一个常量 KEY，用于保存 Windows 网络设置的注册表键路径，以及一个常量 SUBKEY，用于存储系统代理设置的注册表子键名称。这些常量用于构造函数 __init__ 和方法 clear 中来访问和修改 Windows 注册表。 构造函数需要一个 args 参数，该参数预计是一个具有 listen 属性的对象。listen 属性是一个包含 port、protos 和 unix 属性的对象，用于确定一个监听 HTTP 的 localhost 服务器的配置。如果没有找到这样的服务器，则构造函数会打印一条消息。 在构造函数中，打开注册表键，并读取当前值的 SUBKEY 子键。该值预计是一个二进制值，其中包含若干特定格式的字段。修改该值以将 localhost 服务器添加到代理设置中，并将修改后的值写回注册表。 clear 方法重新打开注册表键，读取 SUBKEY 子键的当前值，修改值以从代理设置中删除 localhost 服务器，并将修改后的值写回注册表。

``` python3 async def test_url(url, rserver): url = urllib.parse.urlparse(url) assert url.scheme in ('http', 'https'), f'Unknown scheme {url.scheme}' host_name, port = proto.netloc_split(url.netloc, default_port = 80 if url.scheme=='http' else 443) initbuf = f'GET {url.path or "/"} HTTP/1.1\r\nHost: {host_name}\r\nUser-Agent: pproxy-{__version__}\r\nAccept: */*\r\nConnection: close\r\n\r\n'.encode() for roption in rserver: print(f'============ {roption.bind} ============') try: reader, writer = await roption.open_connection(host_name, port, None, None) except asyncio.TimeoutError: raise Exception(f'Connection timeout {rserver}') try: reader, writer = await roption.prepare_connection(reader, writer, host_name, port) except Exception: writer.close() raise Exception('Unknown remote protocol') if url.scheme == 'https': import ssl sslclient = ssl.create_default_context(ssl.Purpose.SERVER_AUTH) sslclient.check_hostname = False sslclient.verify_mode = ssl.CERT_NONE reader, writer = proto.sslwrap(reader, writer, sslclient, False, host_name) writer.write(initbuf) headers = await reader.read_until(b'\r\n\r\n') print(headers.decode()[:-4]) print(f'--------------------------------') body = bytearray() while not reader.at_eof(): s = await reader.read(65536) if not s: break body.extend(s) print(body.decode('utf8', 'ignore')) print(f'============ success ============') ``` 这是一个异步函数 test_url，用于测试一个 URL 是否能够被代理到指定的远程服务端。 该函数接受两个参数，一个是要测试的 URL，一个是代理服务器列表 rserver。函数首先会解析 URL，并检查 URL 的方案是否为 http 或 https。然后构造一个 HTTP GET 请求，发送到 URL 的主机上。 函数会遍历 rserver 中的每个代理服务器，尝试连接代理服务器，并向代理服务器发送 HTTP 请求。代理服务器的连接方法和准备方法由 roption 对象提供，这个对象预计有 open_connection 和 prepare_connection 两个方法，用于连接和准备代理服务器连接。 如果 URL 的方案为 https，则会使用 SSL 包装，连接到代理服务器。最后，读取代理服务器的响应，并将响应头和响应体打印出来。 当函数成功测试 URL 时，将打印 success 消息。

``` python3 def proxy_by_uri(uri, jump): scheme, _, uri = uri.partition('://') url = urllib.parse.urlparse('s://'+uri) rawprotos = [i.lower() for i in scheme.split('+')] err_str, protos = proto.get_protos(rawprotos) protonames = [i.name for i in protos] if err_str: raise argparse.ArgumentTypeError(err_str) if 'ssl' in rawprotos or 'secure' in rawprotos: import ssl sslserver = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH) sslclient = ssl.create_default_context(ssl.Purpose.SERVER_AUTH) if 'ssl' in rawprotos: sslclient.check_hostname = False sslclient.verify_mode = ssl.CERT_NONE sslcontexts.append(sslserver) sslcontexts.append(sslclient) else: sslserver = sslclient = None if 'quic' in rawprotos or 'h3' in protonames: try: import ssl, aioquic.quic.configuration except Exception: raise Exception('Missing library: "pip3 install aioquic"') quicserver = aioquic.quic.configuration.QuicConfiguration(is_client=False, max_stream_data=2**60, max_data=2**60, idle_timeout=SOCKET_TIMEOUT) quicclient = aioquic.quic.configuration.QuicConfiguration(max_stream_data=2**60, max_data=2**60, idle_timeout=SOCKET_TIMEOUT*5) quicclient.verify_mode = ssl.CERT_NONE sslcontexts.append(quicserver) sslcontexts.append(quicclient) if 'h2' in rawprotos: try: import h2 except Exception: raise Exception('Missing library: "pip3 install h2"') urlpath, _, plugins = url.path.partition(',') urlpath, _, lbind = urlpath.partition('@') plugins = plugins.split(',') if plugins else None cipher, _, loc = url.netloc.rpartition('@') if cipher: from .cipher import get_cipher if ':' not in cipher: try: cipher = base64.b64decode(cipher).decode() except Exception: pass if ':' not in cipher: raise argparse.ArgumentTypeError('userinfo must be "cipher:key"') err_str, cipher = get_cipher(cipher) if err_str: raise argparse.ArgumentTypeError(err_str) if plugins: from .plugin import get_plugin for name in plugins: if not name: continue err_str, plugin = get_plugin(name) if err_str: raise argparse.ArgumentTypeError(err_str) cipher.plugins.append(plugin) if loc: host_name, port = proto.netloc_split(loc, default_port=22 if 'ssh' in rawprotos else 8080) else: host_name = port = None if url.fragment.startswith('#'): with open(url.fragment[1:]) as f: auth = f.read().rstrip().encode() else: auth = url.fragment.encode() users = [i.rstrip() for i in auth.split(b'\n')] if auth else None if 'direct' in protonames: return ProxyDirect(lbind=lbind) else: params = dict(jump=jump, protos=protos, cipher=cipher, users=users, rule=url.query, bind=loc or urlpath, host_name=host_name, port=port, unix=not loc, lbind=lbind, sslclient=sslclient, sslserver=sslserver) if 'quic' in rawprotos: proxy = ProxyQUIC(quicserver, quicclient, **params) elif 'h3' in protonames: proxy = ProxyH3(quicserver, quicclient, **params) elif 'h2' in protonames: proxy = ProxyH2(**params) elif 'ssh' in protonames: proxy = ProxySSH(**params) else: proxy = ProxySimple(**params) if 'in' in rawprotos: proxy = ProxyBackward(proxy, rawprotos.count('in'), **params) return proxy ``` 这是一个函数proxy_by_uri，用于根据一个URI字符串创建不同类型的代理服务器对象，并返回给调用者。 该函数接受两个参数，一个是URI字符串uri，一个是代理跳板列表jump。函数首先按照:分割，将scheme（通常指代使用的传输协议）和uri获取。然后使用urllib.parse.urlparse解析uri，并得到scheme, path, netloc，fragment等信息。 接下来根据解析得到的scheme，使用proto.get_protos获取对应协议处理插件列表。如果是ssl或secure，代表使用SSL协议，函数将创建一个sslclient和sslserver对象。如果是quic或h3，代表使用QUIC或HTTP/3协议，函数将引入aioquic.quic.configuration模块，并创建对应的quicserver和quicclient对象。如果是h2，代表使用HTTP/2协议，函数将引入h2模块。如果插件列表不为空，函数将根据插件列表创建插件对象，并将其添加到cipher中。 接下来函数将解析uri中的各个字段，如path，lbind，netloc，fragment等，并根据这些字段构造代理对象的基本属性。 最后，根据scheme以及各个字段构造的代理对象属性，使用不同的代理对象类型来创建最终的代理对象。

``` python3 class AuthTable(object): _auth = {} _user = {} def __init__(self, remote_ip, authtime): self.remote_ip = remote_ip self.authtime = authtime def authed(self): if time.time() - self._auth.get(self.remote_ip, 0) <= self.authtime: return self._user[self.remote_ip] def set_authed(self, user): self._auth[self.remote_ip] = time.time() self._user[self.remote_ip] = user ``` 这是一个用于跟踪已验证用户的IP地址的表的类定义。它包括两个类变量“_auth”和“_user”，它们是字典，将远程IP地址映射到`身份验证时间`和`已验证的用户`。 init()方法使用远程IP地址和身份验证时间初始化类。authed()方法检查用户是否已经进行身份验证，它通过查找远程IP在_auth字典中，并检查身份验证时间是否已经过期。如果用户已经通过身份验证，则返回_auth字典中存储的用户。 set_authed() 方法在_auth和_user字典中为当前的远程IP地址设置身份验证时间和用户。

Socks是一种网络传输协议，允许客户端向代理服务器发送请求，让代理服务器代表客户端完成某些操作。Socks4和Socks5是其中两种版本。 Socks4和Socks5的主要区别包括： 1. Socks5支持IPv6地址，而Socks4只支持IPv4地址。 2. Socks5支持UDP数据包的发送和接收，Socks4只支持TCP数据包。 3. Socks5使用更加安全的GSS API机制进行认证，可以使用用户名和密码来进行身份验证，而Socks4只支持单向身份验证。 4. Socks5支持代理链，可以将多个代理服务器串连起来使用，而Socks4不支持代理链。 5. Socks5支持多种认证方式，包括用户名/密码、常见的各种认证方式（如NTLM、Kerberos等），而Socks4只支持基本的用户名/密码认证方式。 Socks5比Socks4更加功能强大和安全，但Socks4简单易用，比较适合一些简单的应用场景。

dnscrypt-proxy 是一种开源工具，用于加密 DNS 查询流量并保护用户的隐私。它可以将 DNS 查询流量从计算机或移动设备转发到加密的 DNS 服务器，以保护用户免受 DNS 劫持和窥视者的攻击。更妙的是由于DNS解析被缓存和加速，用户可以明显感觉到外网网页打开的速度变快了。 https://github.com/DNSCrypt/dnscrypt-proxy