将代码块移至函数后出现OSError: [Errno 22]无效参数错误

问题 #

将BPF过滤器的创建逻辑移到独立函数create_filters_struct()后，调用sock.setsockopt()时出现OSError: [Errno 22] Invalid argument错误，但所有代码写在主逻辑里时运行正常。

原代码如下：

from ctypes import create_string_buffer, addressof
from struct import pack, unpack

import socket


SHOW_ERROR: bool = True


def bpf_jump(code, k, jt, jf):
    return pack('HBBI', code, jt, jf, k)

def bpf_stmt(code, k):
    return bpf_jump(code, k, 0, 0)

# 指令类别
BPF_LD = 0x00
BPF_JMP = 0x05
BPF_RET = 0x06

# ld/ldx 字段
BPF_H = 0x08
BPF_B = 0x10
BPF_ABS = 0x20

# alu/jmp 字段
BPF_JEQ = 0x10
BPF_K = 0x00

# 定义于asm/socket.h
SO_ATTACH_FILTER = 26

# 出于性能考虑，过滤器的顺序与直觉相反：最可能失败的检查放在最前面。
filters_list = [
    # 必须是UDP（检查字节偏移23处的协议字段）
    bpf_stmt(BPF_LD | BPF_B | BPF_ABS, 23),
    bpf_jump(BPF_JMP | BPF_JEQ | BPF_K, 0x11, 0, 3),
    # 必须是IPv4（检查字节偏移12处的以太网类型字段）
    bpf_stmt(BPF_LD | BPF_H | BPF_ABS, 12),
    bpf_jump(BPF_JMP | BPF_JEQ | BPF_K, 0x0800, 0, 1),
    bpf_stmt(BPF_RET | BPF_K, 0x0FFFFFFF), # 通过
    bpf_stmt(BPF_RET | BPF_K, 0), # 拒绝
]

str_buf = None
def create_filters_struct() -> bytes:
    filters = b''.join(filters_list)
    str_buf = create_string_buffer(filters)
    mem_addr_of_filters = addressof(str_buf)
    fprog = pack('HL', len(filters_list), mem_addr_of_filters)
    return fprog

if SHOW_ERROR is False:
    filters = b''.join(filters_list)
    b = create_string_buffer(filters)
    mem_addr_of_filters = addressof(b)
    fprog = pack('HL', len(filters_list), mem_addr_of_filters)
else:
    fprog = create_filters_struct()

net_if = "eth0"

sock = socket.socket(socket.AF_PACKET, socket.SOCK_RAW, socket.ntohs(0x0003))
sock.setsockopt(socket.SOL_SOCKET,
                SO_ATTACH_FILTER,
                fprog)

错误回溯信息：

Traceback (most recent call last):
  File "error_demo.py", line 68, in <module>
    fprog)
OSError: [Errno 22] Invalid argument

解决方案 #

问题根源 #

函数create_filters_struct()内的str_buf是局部变量，函数执行完毕后，这个c_char_Array对象会被Python垃圾回收机制销毁，对应的内存地址变为无效地址。而你打包进fprog的正是这个内存地址，当setsockopt尝试访问该地址时，内存已被释放，从而触发参数无效错误。

主逻辑中运行正常的原因是变量b处于全局作用域，直到setsockopt调用完成后才会被回收，内存地址始终有效。

修复方法 #

需要保证str_buf的生命周期覆盖到setsockopt调用完成，以下两种方式均可：

方式1：将str_buf提升为全局变量

修改create_filters_struct()函数，声明使用全局str_buf，避免局部变量被提前回收：

str_buf = None
def create_filters_struct() -> bytes:
    global str_buf  # 声明使用全局变量
    filters = b''.join(filters_list)
    str_buf = create_string_buffer(filters)
    mem_addr_of_filters = addressof(str_buf)
    fprog = pack('HL', len(filters_list), mem_addr_of_filters)
    return fprog

方式2：返回缓冲区对象和打包结构

让函数同时返回fprog和str_buf，在主逻辑中持有str_buf的引用，直到setsockopt执行完成：

def create_filters_struct() -> tuple[bytes, object]:
    filters = b''.join(filters_list)
    str_buf = create_string_buffer(filters)
    mem_addr_of_filters = addressof(str_buf)
    fprog = pack('HL', len(filters_list), mem_addr_of_filters)
    return fprog, str_buf

# 主逻辑部分修改
if SHOW_ERROR is False:
    filters = b''.join(filters_list)
    b = create_string_buffer(filters)
    mem_addr_of_filters = addressof(b)
    fprog = pack('HL', len(filters_list), mem_addr_of_filters)
else:
    fprog, str_buf = create_filters_struct()  # 持有缓冲区引用，避免被回收

两种方式都能确保str_buf对应的内存地址在setsockopt调用时仍有效，解决参数无效的问题。

内容的提问来源于stack exchange，提问作者Jim Fell