相当于Python Requests库中的`curl --connect-to`

ef1yzkbh  于 2023-06-30  发布在  Python
关注(0)|答案(1)|浏览(112)

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库中的等价物是什么?

aiazj4mn

aiazj4mn1#

没有这样的等价物,但您可以在创建连接之前修补较低级别以重写远程地址。
这在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或使用自定义适配器这样的解决方案是不够的,因为URL也会被重写(因此Host:头也会被重写)。当你使用curl --connect-to时,在HTTP级别没有任何改变。
我还需要选择性地强制http连接,尽管有URL方案。这是它的工作增强版本:

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?

相关问题