pwru:一款基于eBPF的细粒度网络数据包排查工具

pwru是 Cilium 推出的基于 eBPF 开发的网络数据包排查工具，它提供了更细粒度的网络数据包排查方案。本文将介绍pwru的使用方法和经典场景，并介绍其实现原理。

部署要求

pwru要求内核代码在 5.5 版本之上，--output-skb要求内核版本在 5.9 之上，并且要求内核开启以下配置：

OptionNoteCONFIG_DEBUG_INFO_BTF=yAvailable since >= 5.3CONFIG_KPROBES=y
CONFIG_PERF_EVENTS=y
CONFIG_BPF=y
CONFIG_BPF_SYSCALL=y

使用方法

Usage of ./pwru:
--filter-dst-ip string        filter destination IP addr
--filter-dst-port uint16      filter destination port
--filter-func string          filter kernelfunctionsto be probed by name (exact match, supports RE2 regular expression)
--filter-mark uint32          filter skb mark
--filter-netns uint32         filter netns inode
--filter-proto string         filter L4 protocol (tcp, udp, icmp)
--filter-src-ip string        filtersourceIP addr
--filter-src-port uint16      filtersourceport
--output-limit-lines uintexitthe program after the number of events has been received/printed
--output-metaprintskb metadata
--output-relative-timestampprintrelative timestamp per skb
--output-skbprintskb
--output-stackprintstack
--output-tupleprintL4 tuple

下图案例演示了pwru展现出快速定位出数据包被 iptables 规则 drop 掉的原因：

在不设置 iptables 规则之前：

添加了 iptables 规则之后

iptables -t filter -I OUTPUT 1 -m tcp --proto tcp --dst 1.1.1.1/32 -j DROP

可以看到在nf_hook_slow函数后发生了变化：

我们可以看到数据包在nf_hook_slow判决为NF_DROP，调用了kfree_skb

int nf_hook_slow(struct sk_buff *skb, struct nf_hook_state *state,
const struct nf_hook_entries *e, unsigned int s)
{
unsigned int verdict;
int ret;

for(; s < e->num_hook_entries; s++) {
verdict = nf_hook_entry_hookfn(&e->hooks[s], skb, state);
switch (verdict & NF_VERDICT_MASK) {
caseNF_ACCEPT:
break;
caseNF_DROP:
kfree_skb(skb);
ret = NF_DROP_GETERR(verdict);
if(ret == 0)
ret = -EPERM;
returnret;
caseNF_QUEUE:
ret = nf_queue(skb, state, s, verdict);
if(ret == 1)
continue;
returnret;
default:
/* Implicit handlingforNF_STOLEN, as well as any other
* non conventional verdicts.
*/
return0;
}
}

return1;
}

pwru本质上是向 kprobe 注册了一些 eBPF code，根据pwru传入的参数可以更新eBPF Map，改变限制条件，从而更新输出。

比如在FilterCfg里面制定了过滤的 IP 地址和协议等条件

typeFilterCfg struct {
FilterMark uint32

//Filter l3
FilterIPv6  uint8
FilterSrcIP [16]byte
FilterDstIP [16]byte

//Filter l4
FilterProto   uint8
FilterSrcPort uint16
FilterDstPort uint16

//TODO:ifthere are more options later,thenyou can consider using a bit map
OutputRelativeTS uint8
OutputMeta       uint8
OutputTuple      uint8
OutputSkb        uint8
OutputStack      uint8

Pad byte
}

会根据pwru传入的参数更新这个 eBPF Map

func ConfigBPFMap(flags *Flags, cfgMap *ebpf.Map) {
cfg := FilterCfg{
FilterMark: flags.FilterMark,
}

ifflags.FilterSrcPort > 0 {
cfg.FilterSrcPort = byteorder.HostToNetwork16(flags.FilterSrcPort)
}
ifflags.FilterDstPort > 0 {
cfg.FilterDstPort = byteorder.HostToNetwork16(flags.FilterDstPort)
}


switch strings.ToLower(flags.FilterProto) {
case"tcp":
cfg.FilterProto = syscall.IPPROTO_TCP
case"udp":
cfg.FilterProto = syscall.IPPROTO_UDP
case"icmp":
cfg.FilterProto = syscall.IPPROTO_ICMP
case"icmp6":
cfg.FilterProto = syscall.IPPROTO_ICMPV6
}

// ...

iferr := cfgMap.Update(uint32(0), cfg, 0); err != nil {
log.Fatalf("Failed to set filter map: %v", err)
}
}

在 eBPF code 中，可以看到会读取配置bpf_map_lookup_elem，然后进而执行真正的 filter：

struct config {
u32 mark;
u8 ipv6;
union addr saddr;
union addr daddr;
u8 l4_proto;
u16 sport;
u16 dport;
u8 output_timestamp;
u8 output_meta;
u8 output_tuple;
u8 output_skb;
u8 output_stack;
u8 pad;
} __attribute__((packed));

static __always_inline int
handle_everything(struct sk_buff *skb, struct pt_regs *ctx) {
struct event_t event = {};

u32 index = 0;
struct config *cfg = bpf_map_lookup_elem(&cfg_map, &index);

if(cfg) {
if(!filter(skb, cfg))
return0;

set_output(ctx, skb, &event, cfg);
}

event.pid = bpf_get_current_pid_tgid();
event.addr = PT_REGS_IP(ctx);
event.skb_addr = (u64) skb;
event.ts = bpf_ktime_get_ns();
bpf_perf_event_output(ctx, &events, BPF_F_CURRENT_CPU, &event, sizeof(event));

return0;
}

可以看到，这里通过bpf_perf_event_output将过滤结果以 Perf event 传递上来。

rd, err := perf.NewReader(events, os.Getpagesize())
iferr != nil {
log.Fatalf("Creating perf event reader: %s", err)
}
defer rd.Close()

// ...
var event pwru.Event
for{
record, err := rd.Read()
iferr != nil {
ifperf.IsClosed(err) {
return
}
log.Printf("Reading from perf event reader: %s", err)
}

ifrecord.LostSamples != 0 {
log.Printf("Perf event ring buffer full, dropped %d samples", record.LostSamples)
continue
}

iferr := binary.Read(bytes.NewBuffer(record.RawSample), binary.LittleEndian, &event); err != nil {
log.Printf("Parsing perf event: %s", err)
continue
}

output.Print(&event)

select {
case<-ctx.Done():
break
default:
continue
}
}