Equivalent of `curl --connect-to` in Python Requests library

Question

curl has an option 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.

What is the equivalent in Python Requests library?

Answer 1

There is no such equivalent but you can patch lower levels to rewrite the remote address before creating the connection.

This works in 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/')

Solutions like patching PoolManager or using a custom adapter are not enough because the URL is rewritten as well (and thus the Host: header). When you use curl --connect-to, nothing is altered at HTTP level.

I also needed to optionally force http connections despite the URL scheme. This is the working augmented version for that:

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/')

see also Python 'requests' library - define specific DNS?.