边学边做，法力无边。

tcpdump抓包

tcpdump是网络领域de.facto的工具和标准，网工们往往也比较熟悉其抓包语法。

tcpdump的实现是基于cbpf的vm来实现高性能地抓包。

众多文档都会提到cbpf的后继者ebpf，但当越来越多的ebpf网络程序和dpdk网络程序出现后，抓包往往只能是各个技术栈内提供，好在都有一些解决方案，比如

• dpdk的pdump，dumpcap [https://doc.dpdk.org/guides/tools/dumpcap.html] 。
• xdp里面的xdpcap [https://github.com/cloudflare/xdpcap]

实现原理

一、tcpdump 表达式 到 字节码

dpdk的dumpcap和xdp的xdpcap实现略有不同，但大体思路一致。

首先都是利用pcap库对tcpdump 表达式进行翻译，得到cbpf的字节码。

pcap = pcap_open_dead(DLT_EN10MB, intf->opts.snap_len);

if (!pcap)

rte_exit(EXIT_FAILURE, "can not open pcap ");

if (pcap_compile(pcap, &bf, intf->opts.filter, 1, PCAP_NETMASK_UNKNOWN) != 0) {

fprintf(stderr, "Invalid capture filter "%s": for interface '%s' ", intf->opts.filter, intf->name);

rte_exit(EXIT_FAILURE, " %s ",

pcap_geterr(pcap));

}

cbpf 的字节码很好阅读。cbpf的指令也能看出来比较简化易懂。

比如 `tcpdump -d ip` 可以看到表达式 `ip` 过滤条件对应的cbpf字节码

(000) ldh [12]

(001) jeq #0x800 jt 2 jf 3

(002) ret #262144

(003) ret #0

从000开始执行，首先到第12字节位置加载两个字节长度到寄存器（内存），然后比较这个双字节长度的数值是否等于0x0800.如果等于则跳转到002，不等则跳转到003. 这两个位置都是直接执行return，一个返回0，一个返回非0.

了解更多可以看内核cbpf文档「链接」

二、cbpf字节码翻译

dpdk是利用其内部集成的ubpf库，直接将cbpf翻译到了ebpf，并在其vm解释执行。

xdpcap则是利用https://github.com/cloudflare/cbpfc 实现了一个翻译为C代码的过程。而后利用ebpf的编译器实现c 代码翻译到 ebpf并执行的过程。

xdpcap由于其本身golang编译自带goebpf，所以这里就没有什么工作量了。

如果我们想要为bcc实现一个类似的功能，方便在在python代码来import使用。可以阅读cbpf文档，并实现反向逻辑。

由于cbpf指令有限，我们仅需要实现以下指令的翻译。

ld 1, 2, 3, 4, 12 Load word into A

ldi 4 Load word into A

ldh 1, 2 Load half-word into A

ldb 1, 2 Load byte into A

ldx 3, 4, 5, 12 Load word into X

ldxi 4 Load word into X

ldxb 5 Load byte into X

st 3 Store A into M[]

stx 3 Store X into M[]

jmp 6 Jump to label

ja 6 Jump to label

jeq 7, 8, 9, 10 Jump on A ==

jneq 9, 10 Jump on A !=

jne 9, 10 Jump on A !=

jlt 9, 10 Jump on A <

jle 9, 10 Jump on A <=

jgt 7, 8, 9, 10 Jump on A >

jge 7, 8, 9, 10 Jump on A >=

jset 7, 8, 9, 10 Jump on A &

add 0, 4 A +

sub 0, 4 A -

mul 0, 4 A *

p 0, 4 A /

mod 0, 4 A %

neg !A

and 0, 4 A &

or 0, 4 A |

xor 0, 4 A ^

lsh 0, 4 A <<

rsh 0, 4 A >>

tax Copy A into X

txa Copy X into A

ret 4, 11 Return

举例而言，

ret x ,直接翻译为 return x

ldh pos ,整个ld都是赋值，h代表位宽，翻译为 *((u16 *)({data} + {pos})))

jeq x, 跳转系列就是考虑不同的跳转条件。

add/sub/mul/p/mod/neg 四则运算

and/or/xor/lsh/rsh 逻辑运算

另外还有st/stx，tax/txa 涉及的赋值

ld系列和最后的st/stx，tax/txa这些操作要额外理解cbpf的vm内定义了A，X，M 主要寄存器，其中M有16个，操作就是拷贝来拷贝去。

A 32 bit wide accumulator

X 32 bit wide X register

M[] 16 x 32 bit wide misc registers aka "scratch memory

store", addressable from 0 to 15

到这里，cbpf已经没有其他不能解释清楚的了。我们既可以实现一个cbpf的vm，也可以实现反向翻译。

题外话，如果要实现一个ebpf的vm，情况就会稍微复杂。

翻译为C，首先定义一个函数给bcc编译器用

```

static inline u32

cbpf_filter_func (const u8 *const data, const u8 *const data_end) {

__attribute__((unused)) u32 A, X, M[16];

__attribute__((unused)) const u8 *indirect;

// 翻译的程序体

}

具体翻译代码就不详细介绍，参看源码 https://github.com/junka/pycbpf/blob/main/pycbpf/cbpf2c.py

到这里如果我们写一个 bcc 脚本, 用%s 预留一个过滤函数cbpf_filter_func的实现，这个是cbpf2c生成，

BPFTEXT = """

#include

#define MAX_PACKET_LEN (128)

struct filter_packet {

u8 packet[MAX_PACKET_LEN];

};

BPF_PERF_OUTPUT(filter_event);

%s

int filter_packets (struct pt_regs *ctx) {

struct filter_packet e = { };

struct sk_buff *skb;

u32 datalen = 0;

u32 ret = 0;

u8 *data;

skb = (struct sk_buff*)PT_REGS_PARM1(ctx);

data = skb->data;

datalen = skb->len;

/* use bpf_probe_read_user for uprobe OR bpf_probe_read_kernel for kprobe */

if (datalen > MAX_PACKET_LEN) {

datalen = MAX_PACKET_LEN;

}

bpf_probe_read_kernel(&e.packet, datalen, data);

/* cbpf filter packet that match */

ret = cbpf_filter_func(data, data + datalen);

if (!ret) {

return 0;

}

filter_event.perf_submit(ctx, &e, sizeof(e));

return 0;

}

之后，在python函数内补充完整cbpf_filter_func函数后，用bcc的attack_kprobe挂在到内核函数，或者attack_upobe,usdt等其他挂在点都可以，实现一个自定义程序内指定函数位置抓包能力。

def main(argv=None):

cfunc = 编译得到的函数主体

text = BPFTEXT % cfun

bctx = BPF(text=text.encode(), debug=0)

func_name = b"dev_queue_xmit"

bctx.attach_kprobe(event=func_name, fn_name=b"filter_packets")

为了保存为pcap格式，方便后面分析，还可以直接用pcap库才做来保存过滤报文。

pcap_dev = pcap.open_dead(pcap.DLT_EN10MB, 1000)

dumper = pcap.dump_open(pcap_dev, ctypes.c_char_p(args.file.encode("utf-8")))

print(f"Capturing packets from {func_name}... Hit Ctrl-C to end")

counter = 0

def filter_events_cb(_cpu, data, _size):

nonlocal counter

counter += 1

event = ctypes.cast(data, ctypes.POINTER(FilterPacket)).contents

now = time.time()

sec = int(now)

usec = int((now - sec) * 1e6)

tval = pcap.timeval(sec, usec)

hdr = pcap.pkthdr(tval, 100, 100)

pcap.dump(ctypes.cast(dumper, ctypes.POINTER(ctypes.c_ubyte)), hdr, event.packet)

bctx[b"filter_event"].open_perf_buffer(filter_events_cb)

while counter < args.count:

try:

bctx.perf_buffer_poll(timeout=1000)

except KeyboardInterrupt:

pcap.dump_close(dumper)

pcap.close(pcap_dev)

sys.exit()

print(f"{counter} packets cpatured")

pcap.dump_close(dumper)

pcap.close(pcap_dev)

至此完结。

知识回顾

- 学习了cbpf 指令和vm解析

- 了解了dpdkdumpcap，xdpcap实现

- 学习了bcc程序如何编写并在指定位置执行

- 学习了libpcap的用法，compile得到cbpf，dump_pcap实现抓包保存

- 输出了python下cbpf2c最终可以做ebpf的工具