Python Requests 库中的 curl --connect-to 等价物答案

【问题标题】：Equivalent of `curl --connect-to` in Python Requests libraryPython Requests 库中的 curl --connect-to 等价物
【发布时间】：2020-06-25 05:07:30
【问题描述】：

curl 有一个选项connect-to

   --connect-to <HOST1:PORT1:HOST2:PORT2>

          For  a  request  to  the  given  HOST:PORT pair, connect to CONNECT-TO-HOST:CONNECT-TO-PORT instead.  This option is suitable to direct requests at a specific
          server, e.g. at a specific cluster node in a cluster of servers.  This option is only used to establish the network connection. It does NOT affect  the  host-
          name/port  that is used for TLS/SSL (e.g. SNI, certificate verification) or for the application protocols.  "host" and "port" may be the empty string, meaning
          "any host/port".  "connect-to-host" and "connect-to-port" may also be the empty string, meaning "use the request's original host/port".

          This option can be used many times to add many connect rules.

Python Requests 库中的等价物是什么？

【问题讨论】：

标签： http curl python-requests

【解决方案1】：

没有这样的等价物，但您可以在创建连接时修补较低级别以重写远程地址。

这适用于 Python 3：

from unittest.mock import patch


# contextmanager for forcing a connection to a given host, port
def connect_to(host, port):
    from urllib3.util.connection import create_connection as orig_create_connection

    def _forced_address_connection(address, *args, **kwargs):
        forced_address = (host, port)
        return orig_create_connection(forced_address, *args, **kwargs)

    return patch('urllib3.util.connection.create_connection', _forced_address_connection)


# force connections to 127.0.0.1:8080
with connect_to('127.0.0.1', 8080):
    res = requests.get('http://service.example.com/')

其他解决方案（修补 PoolManager，使用 custom adapter）是不够的，因为 URL 也被重写（以及 Host: 标头）。当您使用curl --connect-to 时，不会在 HTTP 级别触及任何内容。

尽管有 URL 方案，我还需要选择性地强制 http 连接。这是适用的增强版本：

import contextlib
from unittest.mock import patch


@contextlib.contextmanager
def connect_to(host, port, force_http=False):
    from urllib3.connection import HTTPConnection
    from urllib3.util.connection import create_connection as orig_create_connection

    def _forced_address_connection(address, *args, **kwargs):
        forced_address = (host, port)
        return orig_create_connection(forced_address, *args, **kwargs)

    class ForcedHTTPConnection(HTTPConnection):
        def __init__(self, **kwargs):
            httpconn_kw = ('host', 'port', 'timeout', 'source_address', 'blocksize')
            httpconn_kwargs = dict([(k, kwargs[k]) for k in httpconn_kw if k in kwargs])
            super().__init__(**httpconn_kwargs)

    patchers = [patch('urllib3.util.connection.create_connection', _forced_address_connection)]
    if force_http:
        patchers.append(patch('urllib3.connectionpool.HTTPSConnectionPool.ConnectionCls', ForcedHTTPConnection))

    for p in patchers:
        p.start()

    yield

    for p in patchers:
        p.stop()


# force connections to 127.0.0.1:8080 using http
with connect_to('127.0.0.1', 8080, force_http=True):
    res = requests.get('https://service.example.com/')

另见Python 'requests' library - define specific DNS?。

【讨论】：