socket_SO_ATTACH_FILTER.py

import socket
import ctypes
import ctypes.util

SO_ATTACH_FILTER = 26


class BpfInsn(ctypes.Structure):
    # https://unix.superglobalmegacorp.com/BSD4.4/newsrc/net/bpf.h.html - bpf_insn
    # The instruction data structure.
    _fields_ = [
        ('code', ctypes.c_ushort),  # actual filter code
        ('jt', ctypes.c_ubyte),  # jump true
        ('jf', ctypes.c_ubyte),  # jump false
        ('k', ctypes.c_int)  # generic multiuse field
    ]


class BpfProgram(ctypes.Structure):
    # https://unix.superglobalmegacorp.com/BSD4.4/newsrc/net/bpf.h.html - bpf_program
    # Structure for BIOCSETF.
    _fields_ = [
        ('bf_len', ctypes.c_int),
        ('bpf_insn', ctypes.POINTER(BpfInsn))
    ]


class Pcap:
    # https://github.com/the-tcpdump-group/libpcap/blob/master/gencode.c
    # Wraps pcap library and exposes pcap_compile_nopcap.

    @staticmethod
    def _pcap_lib() -> ctypes.CDLL:
        lib_f = ctypes.util.find_library('pcap')
        return ctypes.CDLL(lib_f)

    def __init__(self) -> None:
        pcap = self._pcap_lib()
        self.__class__ = type(
            pcap.__class__.__name__,
            (self.__class__, pcap.__class__),
            {}
        )
        self.__dict__ = pcap.__dict__

    def compile_nopcap(
            self,
            snaplen: ctypes.c_int, linktype: ctypes.c_int,
            bpf_program: ctypes.Structure, buf: ctypes.c_char_p,
            optimize: ctypes.c_int, netmask: ctypes.c_int
    ) -> int:
        self.pcap_compile_nopcap.restype = ctypes.c_int
        self.pcap_compile_nopcap.argtypes = [
            ctypes.c_int,
            ctypes.c_int,
            ctypes.POINTER(BpfProgram),
            ctypes.c_char_p,
            ctypes.c_int,
            ctypes.c_int
        ]
        ret = self.pcap_compile_nopcap(snaplen, linktype, bpf_program, buf, optimize, netmask)
        if ret == -1:
            raise Exception('Failed to compile filter')
        return ret

def add_packet_filter(sock: socket.socket, filter_string: str) -> None:
    pcap = Pcap()
    filter_prog = BpfProgram()
    filter_buff = ctypes.create_string_buffer(filter_string.encode())
    pcap.compile_nopcap(65535, 1, ctypes.byref(filter_prog), filter_buff, 0, -1)
    sock.setsockopt(socket.SOL_SOCKET, SO_ATTACH_FILTER, filter_prog)