Send http request through specific network interface

Here is the solution for Requests library without monkey-patching anything.

This function will create a Session bound to the given IP address. It is up to you to determine IP address of the desired network interface.

Tested to work with requests==2.23.0.

import requests


def session_for_src_addr(addr: str) -> requests.Session:
    """
    Create `Session` which will bind to the specified local address
    rather than auto-selecting it.
    """
    session = requests.Session()
    for prefix in ('http://', 'https://'):
        session.get_adapter(prefix).init_poolmanager(
            # those are default values from HTTPAdapter's constructor
            connections=requests.adapters.DEFAULT_POOLSIZE,
            maxsize=requests.adapters.DEFAULT_POOLSIZE,
            # This should be a tuple of (address, port). Port 0 means auto-selection.
            source_address=(addr, 0),
        )

    return session


# usage example:
s = session_for_src_addr('192.168.1.12')
s.get('https://httpbin.org/ip')

Be warned though that this approach is identical to curl's --interface option, and won't help in some cases. Depending on your routing configuration, it might happen that even though you bind to the specific IP address, request will go through some other interface. So if this answer does not work for you then first check if curl http://httpbin.org/ip --interface myinterface will work as expected.

I found a way using pycurl. This works like a charm.

import pycurl
from io import BytesIO
import json


def curl_post(url, data, iface=None):
    c = pycurl.Curl()
    buffer = BytesIO()
    c.setopt(pycurl.URL, url)
    c.setopt(pycurl.POST, True)
    c.setopt(pycurl.HTTPHEADER, ['Content-Type: application/json'])
    c.setopt(pycurl.TIMEOUT, 10)
    c.setopt(pycurl.WRITEFUNCTION, buffer.write)
    c.setopt(pycurl.POSTFIELDS, data)
    if iface:
        c.setopt(pycurl.INTERFACE, iface)
    c.perform()

    # Json response
    resp = buffer.getvalue().decode('UTF-8')

    #  Check response is a JSON if not there was an error
    try:
        resp = json.loads(resp)
    except json.decoder.JSONDecodeError:
        pass

    buffer.close()
    c.close()
    return resp


if __name__ == '__main__':
    dat = {"id": 52, "configuration": [{"eno1": {"address": "192.168.1.1"}}]}
    res = curl_post("http://127.0.0.1:5000/network_configuration/", json.dumps(dat), "wlp2")
    print(res)

I'm leaving the question opened hopping that someone can give an answer using requests.

If you want to do this on Linux you could use SO_BINDTODEVICE flag for setsockopt (check man 7 socket, for more details). In fact, it's what used by curl, if you use --interface option on linux. But keep in mind that SO_BINDTODEVICE requires root permissions (CAP_NET_RAW, although there were some attempts to change this) and curl fallbacks to a regular bind trick if SO_BINDTODEVICE fails.

Here's sample curl strace when it fails:

strace -f -e setsockopt,bind curl --interface eth2 https://ifconfig.me/
strace: Process 18208 attached
[pid 18208] +++ exited with 0 +++
setsockopt(3, SOL_TCP, TCP_NODELAY, [1], 4) = 0
setsockopt(3, SOL_SOCKET, SO_KEEPALIVE, [1], 4) = 0
setsockopt(3, SOL_TCP, TCP_KEEPIDLE, [60], 4) = 0
setsockopt(3, SOL_TCP, TCP_KEEPINTVL, [60], 4) = 0
setsockopt(3, SOL_SOCKET, SO_BINDTODEVICE, "eth2\0", 5) = -1 EPERM (Operation not permitted)
bind(3, {sa_family=AF_INET, sin_port=htons(0), sin_addr=inet_addr("192.168.8.1")}, 16) = 0 # curl fallbacks to regular bind
127.0.0.1+++ exited with 0 +++

Also, wanted to say that using regular bind does not always guarantee that traffic would go through specified interface (@MarSoft answer uses plain bind). On linux, only SO_BINDTODEVICE guarantees that traffic would go through the specified device.

Here's an example how to use SO_BINDTODEVICE with requests and requests-toolbelt (as I said, it requires CAP_NET_RAW permissions).

import socket
import requests
from requests_toolbelt.adapters.socket_options import SocketOptionsAdapter


session = requests.Session()
# set interface here
options = [(socket.SOL_SOCKET, socket.SO_BINDTODEVICE, b"eth0")]
for prefix in ('http://', 'https://'):
    session.mount(prefix, SocketOptionsAdapter(socket_options=options))


print(session.get("https://ifconfig.me/").text)

Alternatively, if you don't want to use requests-toolbelt you can implement adapter class yourself:

import socket
import requests
from requests import adapters
from urllib3.poolmanager import PoolManager


class InterfaceAdapter(adapters.HTTPAdapter):

    def __init__(self, **kwargs):
        self.iface = kwargs.pop('iface', None)
        super(InterfaceAdapter, self).__init__(**kwargs)

    def _socket_options(self):
        if self.iface is None:
            return []
        else:
            return [(socket.SOL_SOCKET, socket.SO_BINDTODEVICE, self.iface)]

    def init_poolmanager(self, connections, maxsize, block=False):
        self.poolmanager = PoolManager(
            num_pools=connections,
            maxsize=maxsize,
            block=block,
            socket_options=self._socket_options()
        )


session = requests.Session()
for prefix in ('http://', 'https://'):
    session.mount(prefix, InterfaceAdapter(iface=b'eth0'))


print(session.get("https://ifconfig.me/").text)

Send http request through specific network interface

Related

Recent Posts