如何通过Socks5代理让Python访问Kubernetes API Client？

无代理时正常运行的代码 #

不使用代理时，访问Kubernetes API Client完全正常，代码示例：

configuration = client.Configuration()
configuration.verify_ssl = False
configuration.host = "xxx"
configuration.api_key = {"authorization": "Bearer " + self.token}
c = api_client.ApiClient(configuration=configuration)
api = core_v1_api.CoreV1Api(c)
# 查询命名空间，执行成功
result = api.list_namespace()

遇到的问题 #

Kubernetes Python客户端由代码生成工具自动生成，无法直接配置SOCKS代理。尝试替换rest_client.pool_manager为SOCKSProxyManager的多种方式均未生效，调用接口时始终超时：

第一次尝试的代码 #

configuration = client.Configuration()
configuration.verify_ssl = False
configuration.api_key = {"authorization": "Bearer " + self.token}
configuration.host = "xxx"
c = api_client.ApiClient(configuration=configuration)
proxy = "socks5://xxx:1080"
c.rest_client.pool_manager = self.build_socks_proxy_manager(configuration)
api = core_v1_api.CoreV1Api(c)
# 查询命名空间，超时无法连接
result = api.list_namespace()

def build_socks_proxy_manager(self, configuration, pools_size=4, maxsize=None):
    # 省略部分初始化步骤
    return SOCKSProxyManager(num_pools=pools_size,
                      maxsize=maxsize,
                      cert_reqs = cert_reqs,
                      ca_certs=ca_certs,
                      cert_file=configuration.cert_file,
                      key_file=configuration.key_file,
                      proxy_url=configuration.proxy,
                      **addition_pool_args)

参考优化后的代码（仍超时） #

configuration = client.Configuration()
configuration.verify_ssl = False
configuration.api_key = {"authorization": "Bearer " + self.token}
configuration.host = "xxx"
c = api_client.ApiClient(configuration=configuration)
proxy = "socks5://xxx:1080"
c.rest_client.pool_manager = SOCKSProxyManager(proxy_url=proxy)
api = core_v1_api.CoreV1Api(c)
# 查询命名空间，仍然超时无法连接
result = api.list_namespace()

可行解决方案 #

方案1：完整配置SOCKSProxyManager参数 #

之前的尝试忽略了原pool_manager的SSL相关配置，导致请求无法正确建立连接。需要完整继承原配置的SSL参数，同时指定代理：

首先安装依赖：

pip install urllib3[socks]

修改后的代码：

import urllib3
from urllib3.contrib.socks import SOCKSProxyManager

configuration = client.Configuration()
configuration.verify_ssl = False
configuration.host = "xxx"
configuration.api_key = {"authorization": "Bearer " + self.token}

# 初始化ApiClient
c = api_client.ApiClient(configuration=configuration)

# 构建带SOCKS代理的PoolManager，复用原配置的SSL参数
proxy_url = "socks5://xxx:1080"
c.rest_client.pool_manager = SOCKSProxyManager(
    proxy_url=proxy_url,
    num_pools=c.rest_client.pool_manager.num_pools,
    maxsize=c.rest_client.pool_manager.maxsize,
    cert_reqs="CERT_NONE" if not configuration.verify_ssl else "CERT_REQUIRED",
    ca_certs=configuration.ssl_ca_cert,
    cert_file=configuration.cert_file,
    key_file=configuration.key_file
)

api = core_v1_api.CoreV1Api(c)
# 执行查询
result = api.list_namespace()

方案2：替换requests适配器（适用于基于requests的客户端） #

如果客户端底层依赖requests库，可以通过替换会话适配器实现代理：

import requests
from requests.adapters import HTTPAdapter
from urllib3.contrib.socks import SOCKSProxyManager

# 创建带SOCKS代理的适配器
adapter = HTTPAdapter(
    poolmanager=SOCKSProxyManager("socks5://xxx:1080")
)

configuration = client.Configuration()
configuration.verify_ssl = False
configuration.host = "xxx"
configuration.api_key = {"authorization": "Bearer " + self.token}

c = api_client.ApiClient(configuration=configuration)
# 为HTTP/HTTPS请求挂载代理适配器
c.session.mount("https://", adapter)
c.session.mount("http://", adapter)

api = core_v1_api.CoreV1Api(c)
result = api.list_namespace()

方案3：环境变量全局代理（无需修改代码） #

设置系统环境变量让urllib3自动使用SOCKS代理，适用于全局需要代理的场景：

export ALL_PROXY=socks5://xxx:1080

设置完成后直接运行原无代理代码即可。

内容的提问来源于stack exchange，提问作者robin_zhang