[Python] BPF in Python sockets to filter packets by a specific port

I created this class to capture specific packets by their destination port, but it does not work. When I use this filter without the port verification, the filter works. I'm trying to create a sniffer to capture TCP packets with specific ports. I want to use BPF to capture these packets to avoid processing them with Python, as sometimes it can capture hundreds of packets and slow down the code.

class Sniffer:

def __init__(self, interface:str, protocol:str, ports:list=None) -> None:
self._interface:str = interface
self._protocol:str = protocol
self._ports:list[int] = ports



def _sniffer(self) -> BPF_Configured_Socket:
sniffer = socket.socket(socket.AF_PACKET, socket.SOCK_RAW, socket.ntohs(0x0003))
sniffer.bind((self._interface, 0))

bpf_filter = self._define_filter()
for i in bpf_filter:
print(i)

filter_array = (sock_filter * len(bpf_filter))()
for i, (code, jt, jf, k) in enumerate(bpf_filter):
filter_array = sock_filter(code, jt, jf, k)

prog = sock_fprog(len(bpf_filter), filter_array)

SO_ATTACH_FILTER = 26
libc = ctypes.cdll.LoadLibrary("libc.so.6")
libc.setsockopt(sniffer.fileno(), socket.SOL_SOCKET, SO_ATTACH_FILTER, ctypes.byref(prog), ctypes.sizeof(prog))

return sniffer



def _define_filter(self) -> BPF_Instruction:
FILTERS = {
'IP': self._get_ip_filter_parameters(),
'ARP': [(0x15, 0, 3, 0x0806)] #.......: Jump if EtherType == ARP (0x0806)
}
filter = [(0x28, 0, 0, 0x0000000c)] #....: Load EtherType (offset 12)
filter += FILTERS.get(self._protocol) #...: Specific parameters
filter += [(0x06, 0, 0, 0xFFFF), #........: Accept packet
(0x06, 0, 0, 0x0000)] #........: Discard packet
return filter



def _get_ip_filter_parameters(self) -> BPF_Instruction:
port_parameters = self._create_port_filter()
num = len(port_parameters)
parameters = [
(0x15, 0, num + 4, 0x0800), #...: Jump if EtherType == IPv4 (0x0800)
(0x28, 0, 0, 0x09), #...........: Load IP Protocol (offset 9)
(0x15, 0, num + 2, 0x06), #.....: Jump if Protocol == TCP (0x06)
(0x28, 0, 0, 35) #............: Load Destination Port (offset 36)
]
return parameters + port_parameters



def _create_port_filter(self) -> BPF_Instruction:
len_ports = len(self._ports)
return [(0x15, 0 , len_ports - i, port) for i, port in enumerate(self._ports)]


I found many sites describing how to use it in C. I need to use it in Python.

Continue reading...