如何使用Python的requests模块通过代理进行身份验证(需要摘要身份验证)

35

我之前使用过Mechanize模块,现在想要使用Requests模块。
(Python mechanize doesn't work when HTTPS and Proxy Authentication required)

当我访问互联网时,我必须通过代理服务器。
代理服务器需要身份验证。我编写了以下代码。

import requests
from requests.auth import HTTPProxyAuth

proxies = {"http":"192.168.20.130:8080"}
auth = HTTPProxyAuth("username", "password")

r = requests.get("http://www.google.co.jp/", proxies=proxies, auth=auth)

上述代码在代理服务器需要基本身份验证时运行良好。
现在我想知道当代理服务器需要摘要身份验证时我该怎么做。
HTTPProxyAuth 在摘要身份验证中似乎无效(r.status_code 返回 407)。
9个回答

36

大多数情况下不需要自己实现! Requests已经内置了对代理和基本验证的支持。

proxies = { 'https' : 'https://user:password@proxyip:port' } 
r = requests.get('https://url', proxies=proxies) 

文档中查看更多信息。

如果需要摘要身份验证,HTTPDigestAuth 可能会有所帮助。
或者您可能需要像下面的 yutaka2487 一样扩展它。

注意:必须使用代理服务器的 IP 地址而不是名称!


3
它只适用于基本身份验证,而不是 OP 所要求的摘要身份验证。 - Tey'
2
它不会起作用,因为HTTPDigestAuth仅支持与最终网站服务器(WWW-Authenticate/Authorization标头,401状态)进行身份验证,而不是代理服务器(Proxy-Authenticate/Proxy-Authorization标头,407状态)。您需要类似于@yutaka2487提供的解决方案,但它仅适用于通过代理联系HTTP服务器,而不是HTTPS服务器,因为requests/urllib3后端在通过隧道传输HTTPS连接时不报告代理错误,因此摘要认证无法正常工作。 - Tey'
@Tey 只需考虑 或者你可能需要像 yutaka2487 下面所做的那样尝试扩展它 - imbr

22

我编写了一个类,可以用于代理身份验证(基于摘要认证)。
我几乎从 requests.auth.HTTPDigestAuth 中借用了所有的代码。

我编写了一个能够用于代理身份验证的类(基于摘要认证),几乎所有的代码都是从 requests.auth.HTTPDigestAuth 中借鉴而来。

import requests
import requests.auth

class HTTPProxyDigestAuth(requests.auth.HTTPDigestAuth):
    def handle_407(self, r):
        """Takes the given response and tries digest-auth, if needed."""

        num_407_calls = r.request.hooks['response'].count(self.handle_407)

        s_auth = r.headers.get('Proxy-authenticate', '')

        if 'digest' in s_auth.lower() and num_407_calls < 2:

            self.chal = requests.auth.parse_dict_header(s_auth.replace('Digest ', ''))

            # Consume content and release the original connection
            # to allow our new request to reuse the same one.
            r.content
            r.raw.release_conn()

            r.request.headers['Authorization'] = self.build_digest_header(r.request.method, r.request.url)
            r.request.send(anyway=True)
            _r = r.request.response
            _r.history.append(r)

            return _r

        return r

    def __call__(self, r):
        if self.last_nonce:
            r.headers['Proxy-Authorization'] = self.build_digest_header(r.method, r.url)
        r.register_hook('response', self.handle_407)
        return r

使用方法:

proxies = {
    "http" :"192.168.20.130:8080",
    "https":"192.168.20.130:8080",
}
auth = HTTPProxyDigestAuth("username", "password")

# HTTP
r = requests.get("http://www.google.co.jp/", proxies=proxies, auth=auth)
r.status_code # 200 OK

# HTTPS
r = requests.get("https://www.google.co.jp/", proxies=proxies, auth=auth)
r.status_code # 200 OK

10
我遇到了错误:“HTTPProxyDigestAuth”对象没有“last_nonce”属性,当我尝试使用你的类时。我会调查一下。 - MattCochrane
6
现在不需要自己编写代码了,requests 库现在已经内置了代理支持,例如 proxies = { 'https' : 'https://user:password@ip:port' } ; r = requests.get('https://url', proxies=proxies),请查看 http://docs.python-requests.org/en/latest/user/advanced/。 - BurnsBA
2
此代码片段现在无法与最近的 requests 版本一起使用,因为它现在将请求参数放在线程本地存储中。而且,即使修复了代码,也无法通过代理连接到 HTTPS 网站。详情请查看 https://github.com/psf/requests/issues/2526#issuecomment-89514502。 - Tey'
1
@omegastripes 如果它对你有用,那就意味着它使用的是基本身份验证,而不是 OP 所要求的摘要身份验证。 - robertspierre
1
@BurnsBA requests 文档 明确指出,内置的代理支持仅适用于基本认证,而不适用于 OP 所要求的摘要认证。 - robertspierre
显示剩余4条评论

9
我编写了一个Python模块(可在这里找到),可以使用摘要方案对HTTP代理进行身份验证。它能够在连接到HTTPS网站时工作 (通过猴子补丁),并允许对网站进行身份验证。这应该与最新的Python 2和3中的requests 库一起使用。
以下示例通过要求使用用户名user1和密码password1的HTTP摘要认证,通过HTTP代理1.2.3.4:8080获取页面https://httpbin.org/ip
import requests
from requests_digest_proxy import HTTPProxyDigestAuth

s = requests.Session()
s.proxies = {
        'http': 'http://1.2.3.4:8080/',
        'https': 'http://1.2.3.4:8080/'
}
s.auth = HTTPProxyDigestAuth('user1', 'password1')

print(s.get('https://httpbin.org/ip').text)

如果网站需要某种HTTP身份验证,可以通过以下方式将其指定给HTTPProxyDigestAuth构造函数:
# HTTP Basic authentication for website
s.auth = HTTPProxyDigestAuth(('user1', 'password1'),
        auth=requests.auth.HTTPBasicAuth('user1', 'password0'))
print(s.get('https://httpbin.org/basic-auth/user1/password0').text))

# HTTP Digest authentication for website
s.auth = HTTPProxyDigestAuth(('user1', 'password1'),,
        auth=requests.auth.HTTPDigestAuth('user1', 'password0'))
print(s.get('https://httpbin.org/digest-auth/auth/user1/password0').text)

我尝试使用这个模块,但是它给了我一个407错误。请告诉我这个还能用吗? - Yash.S.Narang
@Yash.S.Narang 脚本在最新的 Python 和 requests 版本中仍然能够正常工作。如果代理凭据错误,您仍然可能会遇到 407 错误,请检查一下凭据是否正确。 - Tey'
非常感谢您分享Python模块! - Gotenks
这个很棒,连接到HTTPS网站时也很好用,而requests-toolbelt则失败了。非常感谢! - robertspierre
我在使用这个时,连接到HTTPS网站仍然存在问题。 - sourcream

3

这个片段适用于两种类型的请求(httphttps)。已在当前版本的 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):
        # Ensure state is initialized just once per-thread
        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 response is not 407, do not auth
        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))

            # Consume content and release the original connection
            # to allow our new request to reuse the same one.
            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):
        # Initialize per-thread state, if needed
        self.init_per_thread_state()
        # If we have a saved nonce, skip the 407
        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)。在我的解决方法版本中,我们通过向代理服务器发送请求并处理接收到的响应来预先准备必要的授权数据。

如果要组合一个应用程序,需要在需要Basic或Digest代理身份验证的两个环境中工作,是否有一种预防性的方法来确定您是否需要创建/使用基于Digest或Basic的身份验证? - Richard
如果这是网站上的身份验证,那么有一种方法。您发送一个请求,接收到一个带有401代码和所使用的身份验证类型指示的响应。收到此信息后,您可以根据需要形成包含身份验证数据的标头,并使用此标头发送新请求。但是,如果我们谈论请求中的代理身份验证,则这将不起作用。如果我们发送具有不正确数据的请求,我们将收到OSError。在错误文本中没有关于身份验证类型的任何信息。 - zanuda
谢谢!...那正是我的感觉/担忧。处理代理,特别是当它们使用身份验证(这是企业代理环境中最常见的情况)时,在请求中是否存在真正的问题?作为替代概念-至少对于Windows机器而言-是否有任何更简单的方法,只需依附于操作系统中配置的内容并使用它即可? - Richard
是的,我认为这确实是“requests”的问题。这个问题的根源在于Python中的“http”库。至于通过代理发送HTTP请求的简单方法,对于Windows - 我成功地通过Kerberos身份验证的代理使用curl完成了它,只需要指示该身份验证类型在代理服务器上被使用,然后curl本身就会找到存储在系统中的所有必要凭据。 - zanuda
没错。我也用Powershell编写了一个脚本,现在正在用Python替换它。在powershell中,这很简单 - 它还可以找出正在使用的身份验证方法并直接使用它。...简单易行。requests中这种不受支持的混乱情况并不好。 - Richard

2

您可以使用requests.auth.HTTPDigestAuth代替requests.auth.HTTPProxyAuth,以使用摘要认证。


我想通过代理身份验证(基于摘要认证)。这与通常的摘要认证不同。因此,我需要扩展HTTPDigestAuth(见下文)。 - yutaka2487

2

1
看起来这个在连接HTTPS网站时不起作用,只能用于HTTP。 - robertspierre

2

这对我来说有效。实际上,我不知道这种方法中用户:密码的安全性:

import requests
import os

http_proxyf = 'http://user:password@proxyip:port'
os.environ["http_proxy"] = http_proxyf
os.environ["https_proxy"] = http_proxyf

sess = requests.Session()
# maybe need sess.trust_env = True
print(sess.get('https://some.org').text)

这仅适用于基本身份验证,而不适用于摘要身份验证,这是OP所要求的。 - robertspierre
由于这里使用了基本认证,用户名和密码会以明文形式发送。 - Mark

1
import requests
import os


# in my case I had to add my local domain
proxies = {
  'http': 'proxy.myagency.com:8080',
  'https': 'user@localdomain:password@proxy.myagency.com:8080',
}


r=requests.get('https://api.github.com/events', proxies=proxies)
print(r.text)

这仅适用于基本身份验证,而不适用于摘要身份验证,因此不回答OP的问题。 - robertspierre

0
这是一个与HTTP基本身份验证无关的答案 - 例如组织内的透明代理。
import requests

url      = 'https://someaddress-behindproxy.com'
params   = {'apikey': '123456789'}                     #if you need params
proxies  = {'https': 'https://proxyaddress.com:3128'}  #or some other port
response = requests.get(url, proxies=proxies, params=params)

希望这能帮助到某个人。


网页内容由stack overflow 提供, 点击上面的
可以查看英文原文,
原文链接