Python“请求”库-定义特定的DNS？
声明：本页面是StackOverFlow热门问题的中英对照翻译，遵循CC BY-SA 4.0协议，如果您需要使用它，必须同样遵循CC BY-SA许可，注明原文地址和作者信息，同时你必须将它归于原作者（不是我）：StackOverFlow
原文地址: http://stackoverflow.com/questions/22609385/
Warning: these are provided under cc-by-sa 4.0 license. You are free to use/share it, But you must attribute it to the original authors (not me):
StackOverFlow
Python 'requests' library - define specific DNS?
提问by Taku
In my project I'm handling all HTTP requests with python requestslibrary.
在我的项目中，我使用 python requestslibrary处理所有 HTTP 请求。
Now, I need to query the http server using specific DNS - there are two environments, each using its own DNS, and changes are made independently.
现在，我需要使用特定的 DNS 查询 http 服务器 - 有两个环境，每个环境都使用自己的 DNS，并且更改是独立进行的。
So, when the code is running, it should use DNS specific to the environment, and not the DNS specified in my internet connection.
因此，当代码运行时，它应该使用特定于环境的 DNS，而不是我的 Internet 连接中指定的 DNS。
Has anyone tried this using python-requests? I've only found solution for urllib2:
https://stackoverflow.com/questions/4623090/python-set-custom-dns-server-for-urllib-requests
有没有人使用 python-requests 尝试过这个？我只找到了 urllib2 的解决方案：https:
//stackoverflow.com/questions/4623090/python-set-custom-dns-server-for-urllib-requests
采纳答案by Martijn Pieters
requestsuses urllib3, which ultimately uses httplib.HTTPConnectionas well, so the techniques from https://stackoverflow.com/questions/4623090/python-set-custom-dns-server-for-urllib-requests(now deleted, it merely linked to Tell urllib2 to use custom DNS) still apply, to a certain extent.
requests使用urllib3，最终httplib.HTTPConnection也会使用，因此来自https://stackoverflow.com/questions/4623090/python-set-custom-dns-server-for-urllib-requests的技术（现已删除，它仅链接到Tell urllib2 to使用自定义 DNS) 在一定程度上仍然适用。
The urllib3.connectionmodule subclasses httplib.HTTPConnectionunder the same name, having replaced the .connect()method with one that calls self._new_conn. In turn, this delegates to urllib3.util.connection.create_connection(). It is perhaps easiest to patch thatfunction:
该urllib3.connection模块httplib.HTTPConnection以相同的名称子类化，将.connect()方法替换为调用self._new_conn. 反过来，这委托给urllib3.util.connection.create_connection(). 修补该功能可能最简单：
from urllib3.util import connection
_orig_create_connection = connection.create_connection
def patched_create_connection(address, *args, **kwargs):
"""Wrap urllib3's create_connection to resolve the name elsewhere"""
# resolve hostname to an ip address; use your own
# resolver here, as otherwise the system resolver will be used.
host, port = address
hostname = your_dns_resolver(host)
return _orig_create_connection((hostname, port), *args, **kwargs)
connection.create_connection = patched_create_connection
and you'd provide your own code to resolve the hostportion of the address into an ip address instead of relying on the connection.create_connection()call (which wraps socket.create_connection()) to resolve the hostname for you.
并且您需要提供自己的代码来host将地址的一部分解析为 ip 地址，而不是依靠connection.create_connection()调用（包装socket.create_connection()）来为您解析主机名。
Like all monkeypatching, be careful that the code hasn't significantly changed in later releases; the patch here was created against urllib3version 1.21.1. but should work for versions as far back as 1.9.
与所有monkeypatching 一样，请注意代码在以后的版本中没有显着更改；这里的补丁是针对urllib31.21.1 版创建的。但应该适用于早至 1.9 的版本。
Note that this answer was re-written to work with newer urllib3releases, which have added a much more convenient patching location. See the edit history for the old method, applicable to version < 1.9, as a patch to the vendored urllib3version rather than a stand-alone installation.
请注意，此答案已重新编写以适用于较新urllib3版本，这些版本添加了更方便的修补位置。请参阅旧方法的编辑历史记录，适用于版本 < 1.9，作为供应商urllib3版本的补丁而不是独立安装。
回答by Sarah Messer
You should look into the TransportAdapters, including the source code. The documentation on them isn't great, but they give low-level access to a lot of the functionality described in RFC 2818and RFC 6125. In particular, those documents encourage (require?) client-side code to support application-specific DNS for the purpose of checking certificates' CommonName and SubjectAltName. The keyword argument you need in those calls is "assert_hostname". Here's how to set it with the requests library:
您应该查看TransportAdapters，包括源代码。关于它们的文档不是很好，但它们提供了对RFC 2818和RFC 6125 中描述的许多功能的低级访问。特别是，这些文档鼓励（要求？）客户端代码支持特定于应用程序的 DNS，以便检查证书的 CommonName 和 SubjectAltName。您在这些调用中需要的关键字参数是“assert_hostname”。以下是使用 requests 库设置它的方法：
from requests import Session, HTTPError
from requests.adapters import HTTPAdapter, DEFAULT_POOLSIZE, DEFAULT_RETRIES, DEFAULT_POOLBLOCK
class DNSResolverHTTPSAdapter(HTTPAdapter):
def __init__(self, common_name, host, pool_connections=DEFAULT_POOLSIZE, pool_maxsize=DEFAULT_POOLSIZE,
max_retries=DEFAULT_RETRIES, pool_block=DEFAULT_POOLBLOCK):
self.__common_name = common_name
self.__host = host
super(DNSResolverHTTPSAdapter, self).__init__(pool_connections=pool_connections, pool_maxsize=pool_maxsize,
max_retries=max_retries, pool_block=pool_block)
def get_connection(self, url, proxies=None):
redirected_url = url.replace(self.__common_name, self.__host)
return super(DNSResolverHTTPSAdapter, self).get_connection(redirected_url, proxies=proxies)
def init_poolmanager(self, connections, maxsize, block=DEFAULT_POOLBLOCK, **pool_kwargs):
pool_kwargs['assert_hostname'] = self.__common_name
super(DNSResolverHTTPSAdapter, self).init_poolmanager(connections, maxsize, block=block, **pool_kwargs)
common_name = 'SuperSecretSarahServer'
host = '192.168.33.51'
port = 666
base_url = 'https://{}:{}/api/'.format(common_name, port)
my_session = Session()
my_session.mount(self.base_url.lower(), DNSResolverHTTPSAdapter(common_name, host))
user_name = 'sarah'
url = '{}users/{}'.format(self.base_url, user_name)
default_response_kwargs = {
'auth': (NAME, PASSWORD),
'headers': {'Content-Type': 'application/json'},
'verify': SSL_OPTIONS['ca_certs'],
'cert': (SSL_OPTIONS['certfile'], SSL_OPTIONS['keyfile'])
}
response = my_session.get(url, **default_response_kwargs)
I use common_namefor the name expected to be on the certificate and how your code will reference the desired machine. I use hostfor a name recognized by the external world - FQDN, IP, DNS entry, ... Of course, the SSL_OPTIONS dictionary (in my example) must list appropriate certificate / key filenames on your machine. (Plus, NAME and PASSWORD should resolve to correct strings.)
我使用common_name证书上的名称以及您的代码将如何引用所需的机器。我host用于外部世界识别的名称 - FQDN、IP、DNS 条目……当然，SSL_OPTIONS 字典（在我的示例中）必须在您的机器上列出适当的证书/密钥文件名。（另外，NAME 和 PASSWORD 应该解析为正确的字符串。）
回答by Thibault Reuille
I know this is an old thread but here is my python3 compatible solution using tldextract and dnspython. I've left some commented out code to illustrate how to debug and set additional session parameters.
我知道这是一个旧线程，但这是我使用 tldextract 和 dnspython 的与 python3 兼容的解决方案。我留下了一些注释掉的代码来说明如何调试和设置额外的会话参数。
#!/usr/bin/env python3
import sys
from pprint import pprint as pp
import requests
import dns.resolver # NOTE: dnspython package
import tldextract
class CustomAdapter(requests.adapters.HTTPAdapter):
def __init__(self, nameservers):
self.nameservers = nameservers
super().__init__()
def resolve(self, host, nameservers, record_type):
dns_resolver = dns.resolver.Resolver()
dns_resolver.nameservers = nameservers
answers = dns_resolver.query(host, record_type)
for rdata in answers:
return str(rdata)
def get_connection(self, url, proxies=None):
ext = tldextract.extract(url)
fqdn = ".".join([ ext.subdomain, ext.domain, ext.suffix ])
print("FQDN: {}".format(fqdn))
a_record = self.resolve(fqdn, nameservers, 'A')
print("A record: {}".format(a_record))
resolved_url = url.replace(fqdn, a_record) # NOTE: Replace first occurrence only
print("Resolved URL: {}".format(resolved_url))
return super().get_connection(resolved_url, proxies=proxies)
if __name__ == "__main__":
if len(sys.argv) != 2:
print("Usage: {} <url>".format(sys.argv[0]))
sys.exit(0)
url = sys.argv[1]
nameservers = [
'208.67.222.222', # NOTE: OpenDNS
'8.8.8.8' # NOTE: Google
]
session = requests.Session()
session.mount(url, CustomAdapter(nameservers))
parameters = {
# "headers": {'Content-Type': 'application/json'},
# "timeout" : 45,
# "stream" : True
# "proxies" : {
# "http": "http://your_http_proxy:8080/",
# "https": "http://your_https_proxy:8081/"
# },
# "auth": (name, password),
# ...
}
response = session.get(url, **parameters)
pp(response.__dict__)
And here it the console output :
这里是控制台输出：
$ ./run.py http://www.test.com
FQDN: www.test.com
A record: 69.172.200.235
Resolved URL: http://69.172.200.235/
{'_content': b'<!DOCTYPE HTML PUBLIC "-//W3C//DTD HTML 4.01//EN" "http://www.w3'
b'.org/TR/html4/strict.dtd">\n<html>\n<head>\n<meta http-equiv="C'
b'ontent-Type" content="text/html; charset=iso-8859-1">\n<meta '
b'http-equiv="Content-Script-Type" content="text/javascript">\n'
b'<script type="text/javascript">\nfunction getCookie(c_name) {'
b' // Local function for getting a cookie value\n if (docume'
b'nt.cookie.length > 0) {\n c_start = document.cookie.in'
b'dexOf(c_name + "=");\n if (c_start!=-1) {\n c_st'
b'art=c_start + c_name.length + 1;\n c_end=document.cook'
b'ie.indexOf(";", c_start);\n\n if (c_end==-1) \n '
b' c_end = document.cookie.length;\n\n return unescape('
b'document.cookie.substring(c_start,c_end));\n }\n }\n '
b' return "";\n}\nfunction setCookie(c_name, value, expiredays'
b') { // Local function for setting a value of a cookie\n va'
b'r exdate = new Date();\n exdate.setDate(exdate.getDate()+e'
b'xpiredays);\n document.cookie = c_name + "=" + escape(valu'
b'e) + ((expiredays==null) ? "" : ";expires=" + exdate.toGMTString'
b'()) + ";path=/";\n}\nfunction getHostUri() {\n var loc = doc'
b"ument.location;\n return loc.toString();\n}\nsetCookie('YPF8"
b"827340282Jdskjhfiw_928937459182JAX666', '171.68.244.56', 10)"
b';\ntry { \n location.reload(true); \n} catch (err1) { \n '
b' try { \n location.reload(); \n } catch (err2) { '
b' \n \tlocation.href = getHostUri(); \n } \n}\n</scrip'
b't>\n</head>\n<body>\n<noscript>This site requires JavaScript an'
b'd Cookies to be enabled. Please change your browser settings or '
b'upgrade your browser.</noscript>\n</body>\n</html>\n',
'_content_consumed': True,
'_next': None,
'connection': <requests.adapters.HTTPAdapter object at 0x109130e48>,
'cookies': <RequestsCookieJar[]>,
'elapsed': datetime.timedelta(microseconds=992676),
'encoding': 'ISO-8859-1',
'headers': {'Server': 'nginx/1.14.2', 'Date': 'Wed, 01 May 2019 18:01:58 GMT', 'Content-Type': 'text/html', 'Transfer-Encoding': 'chunked', 'Connection': 'keep-alive', 'Keep-Alive': 'timeout=20', 'X-DIS-Request-ID': '2a5057a7c7b8a93dd700856c48fda74a', 'P3P': 'CP="NON DSP COR ADMa OUR IND UNI COM NAV INT"', 'Cache-Control': 'no-cache', 'Content-Encoding': 'gzip'},
'history': [<Response [302]>],
'raw': <urllib3.response.HTTPResponse object at 0x1095b90b8>,
'reason': 'OK',
'request': <PreparedRequest [GET]>,
'status_code': 200,
'url': 'https://www.test.com/'}
Hope this helps.
希望这可以帮助。
回答by ft9
A customized HTTPAdapter will do the trick.
定制的 HTTPAdapter 可以解决问题。
Don't forget to set server_hostnameto enable SNI.
不要忘记设置server_hostname启用SNI。
import requests
class HostHeaderSSLAdapter(requests.adapters.HTTPAdapter):
def resolve(self, hostname):
# a dummy DNS resolver
import random
ips = [
'104.16.89.20', # CloudFlare
'151.101.2.109', # Fastly
]
resolutions = {
'cdn.jsdelivr.net': random.choice(ips),
}
return resolutions.get(hostname)
def send(self, request, **kwargs):
from urllib.parse import urlparse
connection_pool_kwargs = self.poolmanager.connection_pool_kw
result = urlparse(request.url)
resolved_ip = self.resolve(result.hostname)
if result.scheme == 'https' and resolved_ip:
request.url = request.url.replace(
'https://' + result.hostname,
'https://' + resolved_ip,
)
connection_pool_kwargs['server_hostname'] = result.hostname # SNI
connection_pool_kwargs['assert_hostname'] = result.hostname
# overwrite the host header
request.headers['Host'] = result.hostname
else:
# theses headers from a previous request may have been left
connection_pool_kwargs.pop('server_hostname', None)
connection_pool_kwargs.pop('assert_hostname', None)
return super(HostHeaderSSLAdapter, self).send(request, **kwargs)
url = 'https://cdn.jsdelivr.net/npm/bootstrap/LICENSE'
session = requests.Session()
session.mount('https://', HostHeaderSSLAdapter())
r = session.get(url)
print(r.headers)
r = session.get(url)
print(r.headers)
