这个片段适用于两种类型的请求(http
和 https
)。已在当前版本的 requests(2.23.0)上进行了测试。
import re
import requests
from requests.utils import get_auth_from_url
from requests.auth import HTTPDigestAuth
from requests.utils import parse_dict_header
from urllib3.util import parse_url
def get_proxy_autorization_header(proxy, method):
username, password = get_auth_from_url(proxy)
auth = HTTPProxyDigestAuth(username, password)
proxy_url = parse_url(proxy)
proxy_response = requests.request(method, proxy_url, auth=auth)
return proxy_response.request.headers['Proxy-Authorization']
class HTTPSAdapterWithProxyDigestAuth(requests.adapters.HTTPAdapter):
def proxy_headers(self, proxy):
headers = {}
proxy_auth_header = get_proxy_autorization_header(proxy, 'CONNECT')
headers['Proxy-Authorization'] = proxy_auth_header
return headers
class HTTPAdapterWithProxyDigestAuth(requests.adapters.HTTPAdapter):
def proxy_headers(self, proxy):
return {}
def add_headers(self, request, **kwargs):
proxy = kwargs['proxies'].get('http', '')
if proxy:
proxy_auth_header = get_proxy_autorization_header(proxy, request.method)
request.headers['Proxy-Authorization'] = proxy_auth_header
class HTTPProxyDigestAuth(requests.auth.HTTPDigestAuth):
def init_per_thread_state(self):
if not hasattr(self._thread_local, 'init'):
self._thread_local.init = True
self._thread_local.last_nonce = ''
self._thread_local.nonce_count = 0
self._thread_local.chal = {}
self._thread_local.pos = None
self._thread_local.num_407_calls = None
def handle_407(self, r, **kwargs):
"""
Takes the given response and tries digest-auth, if needed.
:rtype: requests.Response
"""
if r.status_code != 407:
self._thread_local.num_407_calls = 1
return r
s_auth = r.headers.get('proxy-authenticate', '')
if 'digest' in s_auth.lower() and self._thread_local.num_407_calls < 2:
self._thread_local.num_407_calls += 1
pat = re.compile(r'digest ', flags=re.IGNORECASE)
self._thread_local.chal = requests.utils.parse_dict_header(
pat.sub('', s_auth, count=1))
r.content
r.close()
prep = r.request.copy()
requests.cookies.extract_cookies_to_jar(prep._cookies, r.request, r.raw)
prep.prepare_cookies(prep._cookies)
prep.headers['Proxy-Authorization'] = self.build_digest_header(prep.method, prep.url)
_r = r.connection.send(prep, **kwargs)
_r.history.append(r)
_r.request = prep
return _r
self._thread_local.num_407_calls = 1
return r
def __call__(self, r):
self.init_per_thread_state()
if self._thread_local.last_nonce:
r.headers['Proxy-Authorization'] = self.build_digest_header(r.method, r.url)
r.register_hook('response', self.handle_407)
self._thread_local.num_407_calls = 1
return r
session = requests.Session()
session.proxies = {
'http': 'http://username:password@proxyhost:proxyport',
'https': 'http://username:password@proxyhost:proxyport'
}
session.trust_env = False
session.mount('http://', HTTPAdapterWithProxyDigestAuth())
session.mount('https://', HTTPSAdapterWithProxyDigestAuth())
response_http = session.get("http://ww3.safestyle-windows.co.uk/the-secret-door/")
print(response_http.status_code)
response_https = session.get("https://dev59.com/dGYr5IYBdhLWcg3wpbyQ")
print(response_https.status_code)
一般来说,代理授权的问题也与其他类型的身份验证(ntlm、kerberos)在使用协议HTTPS进行连接时相关。尽管存在大量问题(自2013年以来,可能还有更早的问题我没有找到):
在requests中:摘要代理授权, NTLM代理授权, Kerberos代理授权
在urlib3中:NTLM代理授权, NTLM代理授权
以及许多其他问题,但这个问题仍未解决。
在模块httplib(python2)/http.client(python3)的函数_tunnel中存在问题的根源。如果连接尝试失败,则会引发一个OSError,但不返回响应代码(我们的情况下为407)和构建授权标头所需的其他数据。Lukasa在这里给出了解释
here。
只要urllib3(或requests)的维护者没有解决方案,我们就只能使用各种解决方法(例如,使用@Tey的方法
approach或像这样做
this)。在我的解决方法版本中,我们通过向代理服务器发送请求并处理接收到的响应来预先准备必要的授权数据。
HTTPDigestAuth
仅支持与最终网站服务器(WWW-Authenticate
/Authorization
标头,401状态)进行身份验证,而不是代理服务器(Proxy-Authenticate
/Proxy-Authorization
标头,407状态)。您需要类似于@yutaka2487提供的解决方案,但它仅适用于通过代理联系HTTP服务器,而不是HTTPS服务器,因为requests/urllib3后端在通过隧道传输HTTPS连接时不报告代理错误,因此摘要认证无法正常工作。 - Tey'或者你可能需要像 yutaka2487 下面所做的那样尝试扩展它
。 - imbr