eBPF correctness improvements for IPv6, ICMP and CGroup

- Walk IPv6 extension headers (HopByHop, Routing, Fragment, Destination)
  to find the real transport protocol instead of using nexthdr directly;
  update TCP/UDP/ICMPv6 header pointers to use the resolved transport ptr
- Add IPv6 extension header protocol constants to counter.h
- Fix ip6_send_skb kprobe to use sk->sk_protocol instead of
  ip6hdr->nexthdr so packets with extension headers are counted correctly
- Fix process_udp_send to validate skb before reading udphdr fields
- Populate cgroupid in process_cgroup_skb via bpf_get_current_cgroup_id()
- Compute accurate ICMP/ICMPv6 error response sizes per RFC 792/4443
  (8-byte header + IP header + 8 bytes of original data) instead of
  reusing the full triggering payload length
- Skip update_val in icmp_rcv when msglen is zero
- Log perf read errors in cGroupWatcher instead of silently sleeping
- Return partial batch results alongside error in listMapBatch

Author: dkorunic

bpf/counter.bpf.c

@@ -182,14 +182,42 @@ process_ip6(struct ipv6hdr *ip6, void *data_end, statkey *key) {
     return NOK;
   }
 
-  // IPv6 copy of source IP, destination IP and transport protocol
+  // IPv6 copy of source IP and destination IP
   key->srcip = ip6->saddr;
   key->dstip = ip6->daddr;
-  key->proto = ip6->nexthdr;
 
-  switch (ip6->nexthdr) {
+  // Walk optional extension headers to find the actual transport protocol.
+  // Variable-length extension headers (HopByHop, Routing, Destination):
+  //   total size = (hdr_ext_len + 1) * 8 bytes; next_header at offset 0.
+  // Fragment header: fixed 8 bytes; next_header at offset 0.
+  __u8 nexthdr = ip6->nexthdr;
+  void *transport = (void *)ip6 + sizeof(*ip6);
+
+#pragma unroll
+  for (int i = 0; i < 6; i++) {
+    if (nexthdr != IPPROTO_HOPOPTS && nexthdr != IPPROTO_ROUTING &&
+        nexthdr != IPPROTO_FRAGMENT && nexthdr != IPPROTO_DSTOPTS) {
+      break;
+    }
+    if (nexthdr == IPPROTO_FRAGMENT) {
+      if (transport + 8 > data_end)
+        return NOK;
+      nexthdr = ((__u8 *)transport)[0];
+      transport += 8;
+    } else {
+      if (transport + 2 > data_end)
+        return NOK;
+      __u8 hdrlen = ((__u8 *)transport)[1];
+      nexthdr = ((__u8 *)transport)[0];
+      transport += ((__u32)(hdrlen + 1) * 8);
+    }
+  }
+
+  key->proto = nexthdr;
+
+  switch (nexthdr) {
   case IPPROTO_TCP: {
-    struct tcphdr *tcp = (void *)ip6 + sizeof(*ip6);
+    struct tcphdr *tcp = transport;
 
     // validate TCP size
     if ((void *)tcp + sizeof(*tcp) > data_end) {
@@ -202,7 +230,7 @@ process_ip6(struct ipv6hdr *ip6, void *data_end, statkey *key) {
     break;
   }
   case IPPROTO_UDP: {
-    struct udphdr *udp = (void *)ip6 + sizeof(*ip6);
+    struct udphdr *udp = transport;
 
     // validate UDP size
     if ((void *)udp + sizeof(*udp) > data_end) {
@@ -215,7 +243,7 @@ process_ip6(struct ipv6hdr *ip6, void *data_end, statkey *key) {
     break;
   }
   case IPPROTO_ICMPV6: {
-    struct icmp6hdr *icmp = (void *)ip6 + sizeof(*ip6);
+    struct icmp6hdr *icmp = transport;
 
     // validate ICMPv6 size
     if ((void *)icmp + sizeof(*icmp) > data_end) {
@@ -355,6 +383,8 @@ process_cgroup_skb(struct __sk_buff *skb) {
     __builtin_memcpy(key.comm, ski->comm, sizeof(key.comm));
   }
 
+  key.cgroupid = bpf_get_current_cgroup_id();
+
   statvalue *val = (statvalue *)bpf_map_lookup_elem(&pkt_count, &key);
   if (val) {
     // atomic XADD, doesn't need bpf_spin_lock()
@@ -769,12 +799,12 @@ process_icmp6(struct sk_buff *skb, statkey *key, pid_t pid) {
  */
 static inline __attribute__((always_inline)) size_t
 process_udp_send(struct sk_buff *skb, statkey *key, pid_t pid) {
+  if (!process_udp_recv(false, skb, key, pid))
+    return 0;
+
   struct udphdr *udphdr =
       (struct udphdr *)(BPF_CORE_READ(skb, head) +
                         BPF_CORE_READ(skb, transport_header));
-
-  if (!process_udp_recv(false, skb, key, pid))
-    return 0;
   size_t msglen = bpf_ntohs(BPF_CORE_READ(udphdr, len));
 
   return msglen;
@@ -1028,10 +1058,11 @@ int BPF_KPROBE(ip_send_skb, struct net *net, struct sk_buff *skb) {
  */
 SEC("kprobe/ip6_send_skb")
 int BPF_KPROBE(ip6_send_skb, struct sk_buff *skb) {
-  struct ipv6hdr *ip6hdr =
-      (struct ipv6hdr *)(BPF_CORE_READ(skb, head) +
-                         BPF_CORE_READ(skb, network_header));
-  if (BPF_CORE_READ(ip6hdr, nexthdr) != IPPROTO_UDP) {
+  // Use sk_protocol from the socket rather than ip6hdr->nexthdr so that
+  // IPv6 packets with extension headers (where nexthdr != IPPROTO_UDP) are
+  // still accounted for correctly.
+  struct sock *sk = BPF_CORE_READ(skb, sk);
+  if (!sk || BPF_CORE_READ(sk, sk_protocol) != IPPROTO_UDP) {
     return 0;
   }
 
@@ -1120,7 +1151,10 @@ int BPF_KPROBE(__icmp_send, struct sk_buff *skb, __u8 type, __u8 code,
 
   key.cgroupid = bpf_get_current_cgroup_id();
 
-  size_t msglen = process_icmp4(skb, &key, pid);
+  /* process_icmp4 populates src/dst IPs from the triggering packet's IP
+   * header; its returned msglen (triggering payload) is not used here. */
+  process_icmp4(skb, &key, pid);
+
   /* __icmp_send is called with the triggering packet's skb, so src/dst IPs
    * come from the original packet (src=sender, dst=us). Swap them to reflect
    * the actual ICMP error direction (src=us, dst=original sender).
@@ -1131,6 +1165,20 @@ int BPF_KPROBE(__icmp_send, struct sk_buff *skb, __u8 type, __u8 code,
   key.dstip = tmp_ip;
   key.src_port = type;
   key.dst_port = code;
+
+  /* Compute the actual ICMP error response size per RFC 792 / RFC 1812:
+   * 8-byte ICMP header + original IP header + first 8 bytes of original data.
+   * Using the triggering packet's tot_len (as process_icmp4 did) over-counts
+   * by including the full original payload instead of just the 8-byte excerpt.
+   */
+  struct iphdr *iphdr = (struct iphdr *)(BPF_CORE_READ(skb, head) +
+                                         BPF_CORE_READ(skb, network_header));
+  __u16 tot_len = bpf_ntohs(BPF_CORE_READ(iphdr, tot_len));
+  __u16 ihl_bytes = (__u16)(BPF_CORE_READ_BITFIELD_PROBED(iphdr, ihl) * 4);
+  __u16 payload = (ihl_bytes <= tot_len) ? (tot_len - ihl_bytes) : 0;
+  size_t msglen =
+      sizeof(struct icmphdr) + ihl_bytes + (payload > 8 ? 8 : payload);
+
   update_val(&key, msglen);
 
   return 0;
@@ -1165,7 +1213,10 @@ int BPF_KPROBE(icmp6_send, struct sk_buff *skb, __u8 type, __u8 code,
 
   key.cgroupid = bpf_get_current_cgroup_id();
 
-  size_t msglen = process_icmp6(skb, &key, pid);
+  /* process_icmp6 populates src/dst IPs from the triggering packet's IPv6
+   * header; its returned msglen (triggering payload_len) is not used here. */
+  process_icmp6(skb, &key, pid);
+
   /* icmp6_send is called with the triggering packet's skb, so src/dst IPs
    * come from the original packet (src=sender, dst=us). Swap them to reflect
    * the actual ICMPv6 error direction (src=us, dst=original sender).
@@ -1176,6 +1227,13 @@ int BPF_KPROBE(icmp6_send, struct sk_buff *skb, __u8 type, __u8 code,
   key.dstip = tmp_ip;
   key.src_port = type;
   key.dst_port = code;
+
+  /* Compute the actual ICMPv6 error response size per RFC 4443:
+   * 8-byte ICMPv6 header + fixed 40-byte IPv6 header + first 8 bytes of
+   * original data. The triggering payload_len (as process_icmp6 returned)
+   * over-counts by including the full original payload. */
+  size_t msglen = sizeof(struct icmp6hdr) + sizeof(struct ipv6hdr) + 8;
+
   update_val(&key, msglen);
 
   return 0;
@@ -1207,7 +1265,8 @@ int BPF_KPROBE(icmp_rcv, struct sk_buff *skb) {
   key.cgroupid = bpf_get_current_cgroup_id();
 
   size_t msglen = process_icmp4(skb, &key, pid);
-  update_val(&key, msglen);
+  if (msglen > 0)
+    update_val(&key, msglen);
 
   return 0;
 }

bpf/counter.h

@@ -40,3 +40,9 @@
 
 #define TASK_COMM_LEN 16
 #define MAX_ENTRIES 131072
+
+// IPv6 extension header protocol numbers
+#define IPPROTO_HOPOPTS  0  // Hop-by-Hop Options
+#define IPPROTO_ROUTING  43 // Routing Header
+#define IPPROTO_FRAGMENT 44 // Fragment Header
+#define IPPROTO_DSTOPTS  60 // Destination Options

cgroup.go

@@ -25,6 +25,7 @@ import (
 	"encoding/binary"
 	"errors"
 	"io/fs"
+	"log"
 	"os"
 	"path/filepath"
 	"strconv"
@@ -198,7 +199,8 @@ func cGroupWatcher(objs cgroupObjects) (*perf.Reader, error) {
 					return
 				}
 
-				// avoid tight-spinning on persistent errors
+				// log and avoid tight-spinning on persistent errors
+				log.Printf("cGroupWatcher: perf read error: %v", err)
 				time.Sleep(10 * time.Millisecond)
 
 				continue

counter_arm64_bpfel.o

@@ -132,7 +132,7 @@ func listMapBatch(m *ebpf.Map, start time.Time) ([]statEntry, error) {
 				break
 			}
 
-			return nil, err
+			return stats, err
 		}
 	}