pwru的使用方法、经典场景及实现原理

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

安装部署

部署要求

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

使用方法

Usageof./pwru:
--filter-dst-ipstringfilterdestinationIPaddr
--filter-dst-portuint16filterdestinationport
--filter-funcstringfilterkernelfunctionstobeprobedbyname(exactmatch,supportsRE2regularexpression)
--filter-markuint32filterskbmark
--filter-netnsuint32filternetnsinode
--filter-protostringfilterL4protocol(tcp,udp,icmp)
--filter-src-ipstringfiltersourceIPaddr
--filter-src-portuint16filtersourceport
--output-limit-linesuintexittheprogramafterthenumberofeventshasbeenreceived/printed
--output-metaprintskbmetadata
--output-relative-timestampprintrelativetimestampperskb
--output-skbprintskb
--output-stackprintstack
--output-tupleprintL4tuple

案例演示

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

在不设置 iptables 规则之前：

添加了 iptables 规则之后

iptables-tfilter-IOUTPUT1-mtcp--prototcp--dst1.1.1.1/32-jDROP

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

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

intnf_hook_slow(structsk_buff*skb,structnf_hook_state*state,
conststructnf_hook_entries*e,unsignedints)
{
unsignedintverdict;
intret;

for(;snum_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:
/*ImplicithandlingforNF_STOLEN,aswellasanyother
*nonconventionalverdicts.
*/
return0;
}
}

return1;
}

原理实现

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

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

typeFilterCfgstruct{
FilterMarkuint32

//Filterl3
FilterIPv6uint8
FilterSrcIP[16]byte
FilterDstIP[16]byte

//Filterl4
FilterProtouint8
FilterSrcPortuint16
FilterDstPortuint16

//TODO:iftherearemoreoptionslater,thenyoucanconsiderusingabitmap
OutputRelativeTSuint8
OutputMetauint8
OutputTupleuint8
OutputSkbuint8
OutputStackuint8

Padbyte
}

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

funcConfigBPFMap(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)
}


switchstrings.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("Failedtosetfiltermap:%v",err)
}
}

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

structconfig{
u32mark;
u8ipv6;
unionaddrsaddr;
unionaddrdaddr;
u8l4_proto;
u16sport;
u16dport;
u8output_timestamp;
u8output_meta;
u8output_tuple;
u8output_skb;
u8output_stack;
u8pad;
}__attribute__((packed));

static__always_inlineint
handle_everything(structsk_buff*skb,structpt_regs*ctx){
structevent_tevent={};

u32index=0;
structconfig*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("Creatingperfeventreader:%s",err)
}
deferrd.Close()

//...
vareventpwru.Event
for{
record,err:=rd.Read()
iferr!=nil{
ifperf.IsClosed(err){
return
}
log.Printf("Readingfromperfeventreader:%s",err)
}

ifrecord.LostSamples!=0{
log.Printf("Perfeventringbufferfull,dropped%dsamples",record.LostSamples)
continue
}

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

output.Print(&event)

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