• 连接到指定的主代理服务器,使用给定的认证信息进行身份验证。
• 检测连接状态,如果无法连接到主代理，自动切换到备用代理服务器。
• 支持两种认证协议：NTLM和Digest。
• 提供配置文件存储代理和认证信息,支持动态更改配置。
• 日志记录,帮助诊断连接问题。

创建配置文件

创建一个JSON配置文件config.json如下：

{
    "proxy": {
        "main": {
            "url": "http://main-proxy.example.com:808",
            "username": "proxyuser",
            "password": "proxy123",
            "ntlm": true
        },
        "fallback": {
            "url": "http://fallback-proxy.example.com:808",
            "username": "fallbackuser",
            "password": "fallback123"
        }
    },
    "retries": 3,
    "timeout": 500
}

编写客户端代码

使用Python编写客户端代码,利用http.client库进行HTTP请求，并处理认证。

import http.client
import json
import socket
import time
class ProxyClient:
    def __init__(self, config_path):
        self.config = self._load_config(config_path)
        self.current_proxy = self._get_current_proxy()
        self._setup_logging()
    def _load_config(self, config_path):
        with open(config_path, 'r') as f:
            return json.load(f)
    def _setup_logging(self):
        import logging
        logging.basicConfig(level=logging.INFO,
                          format='%(asctime)s - %(levelname)s - %(message)s')
        self.logger = logging.getLogger(__name__)
    def _get_current_proxy(self):
        if self.config['current_proxy'] == 'main':
            return self.config['proxy']['main']
        elif self.config['current_proxy'] == 'fallback':
            return self.config['proxy']['fallback']
        else:
            raise ValueError("Invalid current proxy")
    def _make_request(self, method, url, body=None):
        try:
            # 使用配置中的当前代理
            proxy_info = self.current_proxy
            proxy_url = f"{proxy_info['url']}{url}"
            proxy_user = proxy_info['username']
            proxy_pass = proxy_info['password']
            auth_type = proxy_info.get('ntlm', False)
            # 初始化连接
            client = http.client.HTTPClient(proxy_info['url'])
            request = http.client.Request(method, proxy_url, headers={
                'Proxy-Auth': 'Basic' if not auth_type else 'NTLM',
                'Authorization': f"{auth_type} {proxy_user}:{proxy_pass}"
            })
            if body:
                request.body = body
            # 发送请求
            response = client.get(request)
            self.logger.info(f"Request to {url} successfully made via {proxy_url}")
            return response.status, response.body
        except (http.client.HTTPError, socket.error, Exception) as e:
            self.logger.error(f"Request to {url} failed: {str(e)}")
            raise
    def _check_connection(self):
        try:
            # 发送一个测试请求
            status, body = self._make_request('GET', '/health')
            if status == 200:
                return True
            else:
                return False
        except:
            return False
    def switch_proxy(self):
        # 检查当前是否是备用代理
        if self.current_proxy == 'fallback':
            self.logger.warning("Already on fallback proxy")
            return
        # 尝试切换到备用代理
        try:
            self.current_proxy = 'fallback'
            self.logger.info("Switching to fallback proxy")
            # 尝试连接备用代理
            status, _ = self._make_request('GET', '/', body=None)
            if status == 200:
                self.current_proxy = 'fallback'
                self.logger.info("Successfully switched to fallback proxy")
            else:
                self.logger.error("Failed to switch to fallback proxy")
        except:
            self.logger.error("Error switching to fallback proxy")
    def auto_proxy_switch(self):
        # 首先尝试连接主代理
        connected = self._check_connection()
        if not connected:
            # 切换到备用代理
            self.switch_proxy()

处理可能的错误和异常

在代码中添加异常处理机制,确保客户端能够稳定运行：

def except_block(func):
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except Exception as e:
            self.logger.error(f"Exception in {func.__name__}: {str(e)}")
            raise
    return wrapper
class ProxyClient:
    # 在方法中装饰
    @except_block
    def _make_request(self, method, url, body=None):
        # 方法体

测试客户端

编写测试用例,确保客户端能够正确连接代理并切换：

def test_connect_proxy():
    config = {
        "proxy": {
            "main": {"url": "http://main-proxy:808", "username": "user", "password": "pass", "ntlm": True},
            "fallback": {"url": "http://fallback-proxy:808", "username": "user", "password": "pass"}
        },
        "retries": 1,
        "timeout": 200
    }
    client = ProxyClient("config.json")
    # 测试连接主代理
    client._check_connection()
    # 切换到备用代理并测试
    client.switch_proxy()

部署和使用

将客户端部署在目标机器上,配置好config.json，然后运行：

python proxy_client.py

配置示例

提供一个config.json示例，用户可根据实际情况修改：

{
    "proxy": {
        "main": {
            "url": "http://proxy1.example.com:808",
            "username": "proxyuser",
            "password": "proxy123",
            "ntlm": true
        },
        "fallback": {
            "url": "http://proxy2.example.com:808",
            "username": "fallbackuser",
            "password": "fallback123"
        }
    },
    "retries": 3,
    "timeout": 500
}

日志管理

确保客户端在不同情况下都能正确记录日志,帮助用户诊断问题：

self.logger = logging.getLogger(__name__)
self.logger.setLevel(logging.INFO)
handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(logging.Formatter('%(asctime)s - %(levelname)s - %(message)s'))
self.logger.addHandler(handler)

高级功能扩展

如果需要,可以考虑以下扩展：

• 支持多个备用代理,优先级顺序。
• 动态更新配置文件,自动重载。
• 模糊化代理密码,防止被解析。
• 支持认证缓存,减少认证时间。
• 添加健康检查,自动切换到最佳可用代理。

注意事项

• 确保代理服务器和客户端在同一网络或有正确的网络配置。
• 部署时注意权限,确保代理用户有足够的访问权限。
• 测试在不同网络环境下,确保切换逻辑正确。
• 定期检查日志,监控代理切换情况。

通过以上步骤,您可以创建一个自动代理客户端，满足连接到主代理并切换到备用代理的需求。