Starting with 3.3 python supports sendmsg as well as recvmsg.
Over the next lines, I'll outline how to use send/recvmsg on datagram sockets to
In case the socket is “connected” - using connect(), receiving the destination address is not required, as the socket is fixed to a single address, specifying a source address when sending a packet is not required too, the socket is already defined to serve a single host.
In case the socket is bound to 0.0.0.0 or ::, receiving the destination address allows using the same address as source address when responding to a packet. In case of multi homed setups, or IPv6 with temporary addresses, providing knowing the address a peer used to talk to us, and using the same address to sent data to the peer is required, as the peer may get the data from a different address else, and discard the data on arrival, as the src-host does not match the expectations.
Using recvmsg it is possible to retrieve the destination address of a incoming packet.
First, we have to tell the socket we want to receive this additional data. As the socket module lacks some constants, we need to define them too.
IP_PKTINFO = 8 IPV6_PKTINFO = 50 IPV6_RECVPKTINFO = 49 SOL_IPV6 = 41 def preparefromto(s): if s.family in (socket.AF_INET,socket.AF_INET6): s.setsockopt(socket.SOL_IP, IP_PKTINFO, 1) if s.family == socket.AF_INET6: s.setsockopt(SOL_IPV6, IPV6_RECVPKTINFO, 1)
Now, recvmsg will carry the address of the remote peer, all we have to do is extracting this information.
Unfortunately the information is exported as a list of tuples, and the data in the tuples is bytes data. Therefore need to define some ctype Structures, so we can make use of the data returned by recvmsg.
uint32_t = ctypes.c_uint32 in_addr_t = uint32_t class in_addr(ctypes.Structure): _fields_ = [('s_addr', in_addr_t)] class in6_addr_U(ctypes.Union): _fields_ = [ ('__u6_addr8', ctypes.c_uint8 * 16), ('__u6_addr16', ctypes.c_uint16 * 8), ('__u6_addr32', ctypes.c_uint32 * 4), ] class in6_addr(ctypes.Structure): _fields_ = [ ('__in6_u', in6_addr_U), ] class in_pktinfo(ctypes.Structure): _fields_ = [ ('ipi_ifindex', ctypes.c_int), ('ipi_spec_dst', in_addr), ('ipi_addr', in_addr), ] class in6_pktinfo(ctypes.Structure): _fields_ = [ ('ipi6_addr', in6_addr), ('ipi6_ifindex', ctypes.c_uint), ]
Now, we can write a wrapper function for recvmsg, which will return the data, the source and destination host tuples.
For mapped IPv4, the destination address is not mapped IPv4 but plain IPv4, which is why we do not use the sockets family to limit the scope of SOL_IP to AF_INET.
def recvfromto(s): _to = None data, ancdata, msg_flags, _from = s.recvmsg(5120, socket.CMSG_LEN(5120 * 5)) for anc in ancdata: if anc[0] == socket.SOL_IP and anc[1] == IP_PKTINFO: addr = in_pktinfo.from_buffer_copy(anc[2]) addr = ipaddress.IPv4Address(memoryview(addr.ipi_addr).tobytes()) _to = (str(addr),s.getsockname()[1]) elif anc[0] == SOL_IPV6 and anc[1] == IPV6_PKTINFO: addr = in6_pktinfo.from_buffer_copy(anc[2]) addr = ipaddress.ip_address(memoryview(addr.ipi6_addr).tobytes()) _to = (str(addr),s.getsockname()[1]) return data,_from,_to
The new python module ipaddress capability to parse ip addresses from bytes come in very handy here.
As recvfromto works, we can get over to wrap sendmsg into a function which allows providing the source address - sendtofrom. sendtofrom takes a socket, the data, the destination and source information as arguments, sets up the required data for sendmsg and returns the result of sendmsg.
def sendtofrom(s, _data, _to, _from): ancdata = [] if type(_from) == tuple: _from = _from[0] addr = ipaddress.ip_address(_from) if type(addr) == ipaddress.IPv4Address: _f = in_pktinfo() _f.ipi_spec_dst = in_addr.from_buffer_copy(addr.packed) ancdata = [(socket.SOL_IP, IP_PKTINFO, memoryview(_f).tobytes())] elif s.family == socket.AF_INET6 and type(addr) == ipaddress.IPv6Address: _f = in6_pktinfo() _f.ipi6_addr = in6_addr.from_buffer_copy(addr.packed) ancdata = [(SOL_IPV6, IPV6_PKTINFO, memoryview(_f).tobytes())] return s.sendmsg([_data], ancdata, 0, _to)
Once again ipaddress is used, here to retrieve the bytes of the address.
Full testing is rather complicated, it requires two hosts, A and B. A and B need to be able to connect each via IPv4 and IPv6. A will be used to connect our “test” service on B. B requires multiple IPv4 and IPv6 addresses, each of those addresses needs to reachable from A.
To test IPv6:
if __name__ == '__main__': r = socket.socket(socket.AF_INET6, socket.SOCK_DGRAM) r.bind(("::", 5005)) preparefromto(r) while True: a = recvfromto(r) print(a) data,_from,_to = a print(sendtofrom(r, data, _from, _to[0]))
Using nc6 to connect to B from A and typing something will return the data, for every address B has and which is reachable from A - even IPv4 (use nc for IPv4). IPv4 is mapped in such cases.
For IPv4:
if __name__ == '__main__': r = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) r.bind(("0.0.0.0", 5005)) preparefromto(r) while True: a = recvfromto(r) print(a) data,_from,_to = a print(sendtofrom(r, data, _from, _to[0]))
This way it is possible to create a multi-homed udp service for many clients using a single socket.
For the ease of use … the full code:
Click to display ⇲
Click to hide ⇱
import ctypes import socket import ipaddress IP_PKTINFO = 8 IPV6_PKTINFO = 50 IPV6_RECVPKTINFO = 49 SOL_IPV6 = 41 uint32_t = ctypes.c_uint32 in_addr_t = uint32_t class in_addr(ctypes.Structure): _fields_ = [('s_addr', in_addr_t)] class in6_addr_U(ctypes.Union): _fields_ = [ ('__u6_addr8', ctypes.c_uint8 * 16), ('__u6_addr16', ctypes.c_uint16 * 8), ('__u6_addr32', ctypes.c_uint32 * 4), ] class in6_addr(ctypes.Structure): _fields_ = [ ('__in6_u', in6_addr_U), ] class in_pktinfo(ctypes.Structure): _fields_ = [ ('ipi_ifindex', ctypes.c_int), ('ipi_spec_dst', in_addr), ('ipi_addr', in_addr), ] class in6_pktinfo(ctypes.Structure): _fields_ = [ ('ipi6_addr', in6_addr), ('ipi6_ifindex', ctypes.c_uint), ] def preparefromto(s): if s.family in (socket.AF_INET,socket.AF_INET6): s.setsockopt(socket.SOL_IP, IP_PKTINFO, 1) if s.family == socket.AF_INET6: s.setsockopt(SOL_IPV6, IPV6_RECVPKTINFO, 1) def recvfromto(s): _to = None data, ancdata, msg_flags, _from = s.recvmsg(5120, socket.CMSG_LEN(5120 * 5)) for anc in ancdata: if anc[0] == socket.SOL_IP and anc[1] == IP_PKTINFO: addr = in_pktinfo.from_buffer_copy(anc[2]) addr = ipaddress.IPv4Address(memoryview(addr.ipi_addr).tobytes()) _to = (str(addr),s.getsockname()[1]) elif anc[0] == SOL_IPV6 and anc[1] == IPV6_PKTINFO: addr = in6_pktinfo.from_buffer_copy(anc[2]) addr = ipaddress.ip_address(memoryview(addr.ipi6_addr).tobytes()) _to = (str(addr),s.getsockname()[1]) return data,_from,_to def sendtofrom(s, _data, _to, _from): ancdata = [] if type(_from) == tuple: _from = _from[0] addr = ipaddress.ip_address(_from) if type(addr) == ipaddress.IPv4Address: _f = in_pktinfo() _f.ipi_spec_dst = in_addr.from_buffer_copy(addr.packed) ancdata = [(socket.SOL_IP, IP_PKTINFO, memoryview(_f).tobytes())] elif s.family == socket.AF_INET6 and type(addr) == ipaddress.IPv6Address: _f = in6_pktinfo() _f.ipi6_addr = in6_addr.from_buffer_copy(addr.packed) ancdata = [(SOL_IPV6, IPV6_PKTINFO, memoryview(_f).tobytes())] return s.sendmsg([_data], ancdata, 0, _to) class xsocket(socket.socket): def __init__(self, family=socket.AF_INET, type=socket.SOCK_STREAM, proto=0): socket.socket.__init__(self, family, type, proto) preparefromto(self) def recvfromto(self): return recvfromto(self) def sendtofrom(self, data, _to, _from): return sendtofrom(self, data, _to, _from) if __name__ == '__main__': r = xsocket(socket.AF_INET6, socket.SOCK_DGRAM) r.bind(("::", 5005)) while True: a = r.recvfromto() print(a) data,_from,_to = a print(r.sendtofrom(data, _from, _to)) print(_to)
Just in case you ever wanted to debug your python service interactively, and pdb did not do the job you've been looking for, how about the standard python interpreter on stdin?
The one below basically wraps PyRun_InteractiveOne via ctypes, sets some tty modes, things not that interesting, they just have to work.
But, it even autocompletes on tab, and restores the tty settings via atexit.
Combined with objgraph1), this is was rather valueable to me. For objgraph, use python 3.3 and replace __name__ in objgraph with __qualname__2).
Given I was unable to find something, here is the code.
pyev3) can be replaced with your favourite event loop easily - it is just a single socket to poll for readability.
#!/usr/bin/python3.2 import pyev import sys class cli: def __init__(self, loop): sys.ps1 = ">>>" sys.ps2 = "..." import ctypes libc = ctypes.cdll.LoadLibrary("libc.so.6") libc.fdopen.restype = ctypes.c_void_p libc.fdopen.argtype = [ctypes.c_int, ctypes.c_char_p] self.stdin = libc.fdopen(sys.stdin.fileno(),"r") self._io = pyev.Io(sys.stdin.fileno(), pyev.EV_READ, loop, self._io_cb) self._io.start() import termios # [iflag, oflag, cflag, lflag, ispeed, ospeed, cc] self.read_termios = termios.tcgetattr(sys.stdin.fileno()); self.poll_termios = termios.tcgetattr(sys.stdin.fileno()); self.read_termios[3] |= (termios.ICANON|termios.ECHOCTL|termios.ECHO); self.poll_termios[3] &= ~(termios.ICANON|termios.ECHOCTL|termios.ECHO); try: import readline except ImportError: print("Module readline not available.") else: import rlcompleter readline.parse_and_bind("tab: complete") import atexit atexit.register(self._atexit) def _atexit(self): self.mode(self.read_termios) def setblocking(self,blocking): import fcntl import os if blocking == True: fcntl.fcntl(sys.stdin.fileno(), fcntl.F_SETFL, fcntl.fcntl(sys.stdin.fileno(), fcntl.F_GETFL) | os.O_NONBLOCK) else: fcntl.fcntl(sys.stdin.fileno(), fcntl.F_SETFL, fcntl.fcntl(sys.stdin.fileno(), fcntl.F_GETFL) & ~os.O_NONBLOCK) def mode(self, m): import termios termios.tcsetattr(sys.stdin.fileno(), termios.TCSANOW, m) def _io_cb(self, watcher, revents): from ctypes import pythonapi self.mode(self.read_termios) # self.setblocking(False) # nonblocking stdin causes very interesting problems here! pythonapi.PyRun_InteractiveOne(self.stdin, b"<stdin>") # self.setblocking(True) # even if disabled during runtime operation self.mode(self.poll_termios) if __name__ == '__main__': loop = pyev.default_loop() c = cli(loop) def signals(watcher, revents): watcher.loop.stop() import signal sigint = pyev.Signal(signal.SIGINT, loop, signals) sigint.start() loop.start() sys.exit(0)
The BSD or Berkely Packet Filter is a register-based filter evaluator and network tap invented 1990 by Steven McCanne and Van Jacobson to replace the CMU/Stanford Packet Filter (CSPF) and Sun NIT filter technology with a faster alternative1). While bpf consists of two components, the filter evaluator and the network tap, we'll ignore the network tap and focus on the filter evaluator instead.
For whatever reason python lacks a binding for getifaddrs. For dionaea I created the binding myself, but the code can not be used without dionaea, and I don't like having to install additional bindings to get some basic functionality.
Therefore, I decided to create the functionality provided by the dionaea getifaddrs binding using python ctypes, easy to install, no compiling, copy&paste does the trick.
The code works with python2/3, Linux works, Darwin/OSX may work, Windows? good question.
If you want to use a native library in python, but there is no binding, you can 'try' to interface the library with ctypes.
As I wanted to play with bpf, which is part of libpcap, which lacks a python3 binding, I decided to try ctypes.
What I wanted to do:
