I have two network interfaces (wifi and ethernet) both with internet access. Let's say my interfaces are eth
(ethernet) and wlp2
(wifi). I need specific requests to go through eth
interface and others through wpl2
.
Something like:
// Through "eth"
request.post(url="http://myapi.com/store_ip", iface="eth")
// Through "wlp2"
request.post(url="http://myapi.com/log", iface="wlp2")
I'm using requests
, but I can use pycurl
or urllib
if there isn't any way to do it with requests
.
How to specify source interface in python requests module? refers to Requests, bind to an ip and it doesn't work.
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)
Try changing the internal IP (192.168.0.200) to the corresponding iface in the code below.
import requests
from requests_toolbelt.adapters import source
def check_ip(inet_addr):
s = requests.Session()
iface = source.SourceAddressAdapter(inet_addr)
s.mount('http://', iface)
s.mount('https://', iface)
url = 'https://emapp.cc/get_my_ip'
resp = s.get(url)
print(resp.text)
if __name__ == '__main__':
check_ip('192.168.0.200')
If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!
Donate Us With