Библиотека запросов Python - определить конкретный DNS?

Ответ 1

requests использует urllib3, который в конечном итоге использует httplib.HTTPConnection, поэтому методы из https://stackoverflow.com/questions/4623090/python-set-custom-dns-server-for-urllib-requests (теперь удаленный, он просто связан с Tell urllib2 для использования пользовательского DNS) все еще применяется в определенной степени.

Подклассы urllib3.connection модуля urllib3.connection под тем же именем, заменив метод .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 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

и вы предоставили бы свой собственный код для разрешения части host адреса в IP-адрес вместо того, чтобы полагаться на вызов connection.create_connection() (который обертывает socket.create_connection()), чтобы разрешить имя хоста для вас.

Как и все monkeypatching, будьте осторожны, чтобы код не был значительно изменен в последующих выпусках; патч здесь был создан против urllib3 версии 1.21.1. но должен работать для версий еще в 1.9.

Обратите внимание, что этот ответ был переписан для работы с более новыми релизами urllib3, которые добавили гораздо более удобное место для исправления. См. Историю изменений для старого метода, применимого к версии < 1.9, в качестве патча для версии urllib3, а не для автономной установки.

Ответ 2

Вы должны изучить TransportAdapters, включая исходный код. Документация по ним невелика, но они дают доступ на низкий уровень к множеству функций, описанных в RFC 2818 и RFC 6125. В частности, эти документы поддерживают (требуют?) Клиентский код для поддержки DNS-приложений для проверки сертификатов "CommonName" и "SubjectAltName". Аргумент ключевого слова, который вам нужен в этих вызовах, - "assert_hostname". Здесь, как установить его с помощью библиотеки запросов:

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)

Я использую common_name для имени, которое должно быть указано в сертификате, и как ваш код будет ссылаться на нужный компьютер. Я использую host для имени, распознанного внешним миром - FQDN, IP, запись в DNS... Конечно, словарь SSL_OPTIONS (в моем примере) должен отображать соответствующие имена файлов сертификатов/ключей на вашем компьютере. (Кроме того, NAME и PASSWORD должны разрешить исправление строк.)

Ответ 3

Я знаю, что это старый поток, но вот мое Python3-совместимое решение с использованием tldextract и dnspython. Я оставил закомментированный код, чтобы проиллюстрировать, как отлаживать и устанавливать дополнительные параметры сеанса.

#!/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__)

И вот это вывод консоли:

$ ./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/'}

Надеюсь это поможет.

Ответ 4

Или просто используйте подпроцесс с curl и добавьте --dns-servers

Ответ 5

Настроенный HTTPAdapter сделает свое дело.

Не забудьте установить 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)