Fix correctness issues in eBPF programs and userland

- bpf/counter.bpf.c: use bpf_xdp_get_buff_len() for accurate XDP
  packet length instead of data_end - data arithmetic
- bpf/counter.bpf.c: mask PID with 0xFFFFFFFF to extract tgid only
  from bpf_get_current_pid_tgid() result
- bpf/counter.bpf.c: convert process_tcp/process_udp_recv return type
  to bool so callers can short-circuit on unsupported address families
- bpf/counter.bpf.c: fix IPv4-in-IPv6 address construction to use the
  ip4in6 prefix constant via memcpy instead of partial s6_addr16 writes
- bpf/counter.bpf.c: fix sport byte order in process_tcp using
  bpf_ntohs(inet_sport) and read iphdr->protocol for UDP kprobe filter
- bpf/counter.bpf.c: add src/dst swap for UDP receive direction and
  fix UDP length with bpf_ntohs(); guard skb_consume_udp on len <= 0
- bpf/counter.bpf.c: fix ICMP payload length with underflow guard and
  set type/code directly from __icmp_send kprobe arguments
- bpf/cgroup.bpf.c: set perf_cgroup_event max_entries to 0 for
  dynamic per-CPU sizing
- cgroup.go: add 10ms sleep on perf read errors to avoid busy-looping;
  refresh cgroup cache on lost perf samples
- probe.go: move dead error check for RawTracepoint support to correct
  location before AttachRawTracepoint call
- output.go: fix misleading comments in srcIPSort/dstIPSort (ascending,
  not descending)

Author: dkorunic

bpf/cgroup.bpf.c

@@ -45,7 +45,7 @@ struct {
 struct {
   __uint(type,
          BPF_MAP_TYPE_PERF_EVENT_ARRAY); // perf event array requires 4.3 kernel
-  __uint(max_entries, MAX_ENTRIES);
+  __uint(max_entries, 0);
   __type(key, int);
   __type(value, __u32);
 } perf_cgroup_event SEC(".maps");

bpf/counter.bpf.c

@@ -382,7 +382,7 @@ xdp_process_packet(struct xdp_md *xdp) {
   void *data = (void *)(long)xdp->data;
   void *data_end = (void *)(long)xdp->data_end;
 
-  process_eth(data, data_end, data_end - data);
+  process_eth(data, data_end, bpf_xdp_get_buff_len(xdp));
 }
 
 /**
@@ -441,7 +441,7 @@ SEC("cgroup/sock_create")
 int cgroup_sock_create(struct bpf_sock *sk) {
   __u64 cookie = bpf_get_socket_cookie(sk);
   sockinfo ski = {
-      .pid = bpf_get_current_pid_tgid(),
+      .pid = bpf_get_current_pid_tgid() & 0xFFFFFFFF,
       .comm = {0},
   };
 
@@ -511,21 +511,21 @@ int cgroup_skb_egress(struct __sk_buff *skb) {
  *
  * @throws none
  */
-static inline __attribute__((always_inline)) void
+static inline __attribute__((always_inline)) bool
 process_tcp(bool receive, struct sock *sk, statkey *key, pid_t pid) {
   __u16 family = BPF_CORE_READ(sk, __sk_common.skc_family);
 
   switch (family) {
   case AF_INET: {
     // convert to V4MAPPED address
     __be32 ip4_src = BPF_CORE_READ(sk, __sk_common.skc_rcv_saddr);
-    key->srcip.s6_addr16[5] = bpf_htons(0xffff);
-    __builtin_memcpy(&key->srcip.s6_addr32[3], &ip4_src, sizeof(ip4_src));
+    __builtin_memcpy(key->srcip.s6_addr, ip4in6, sizeof(ip4in6));
+    __builtin_memcpy(key->srcip.s6_addr + sizeof(ip4in6), &ip4_src, sizeof(ip4_src));
 
     // convert to V4MAPPED address
     __be32 ip4_dst = BPF_CORE_READ(sk, __sk_common.skc_daddr);
-    key->dstip.s6_addr16[5] = bpf_htons(0xffff);
-    __builtin_memcpy(&key->dstip.s6_addr32[3], &ip4_dst, sizeof(ip4_dst));
+    __builtin_memcpy(key->dstip.s6_addr, ip4in6, sizeof(ip4in6));
+    __builtin_memcpy(key->dstip.s6_addr + sizeof(ip4in6), &ip4_dst, sizeof(ip4_dst));
 
     break;
   }
@@ -536,14 +536,14 @@ process_tcp(bool receive, struct sock *sk, statkey *key, pid_t pid) {
     break;
   }
   default: {
-    return;
+    return false;
   }
   }
 
   __u16 sport = BPF_CORE_READ(sk, __sk_common.skc_num);
   if (sport == 0) {
     struct inet_sock *isk = (struct inet_sock *)sk;
-    BPF_CORE_READ_INTO(&sport, isk, inet_sport);
+    sport = bpf_ntohs(BPF_CORE_READ(isk, inet_sport));
   }
   key->src_port = sport;
   key->dst_port = bpf_ntohs(BPF_CORE_READ(sk, __sk_common.skc_dport));
@@ -561,6 +561,8 @@ process_tcp(bool receive, struct sock *sk, statkey *key, pid_t pid) {
     key->src_port = key->dst_port;
     key->dst_port = tmp_port;
   }
+
+  return true;
 }
 
 /**
@@ -579,7 +581,7 @@ process_tcp(bool receive, struct sock *sk, statkey *key, pid_t pid) {
  *
  * @throws none
  */
-static inline __attribute__((always_inline)) void
+static inline __attribute__((always_inline)) bool
 process_udp_recv(bool receive, struct sk_buff *skb, statkey *key, pid_t pid) {
   struct udphdr *udphdr =
       (struct udphdr *)(BPF_CORE_READ(skb, head) +
@@ -594,13 +596,13 @@ process_udp_recv(bool receive, struct sk_buff *skb, statkey *key, pid_t pid) {
 
     // convert to V4MAPPED address
     __be32 ip4_src = BPF_CORE_READ(iphdr, saddr);
-    key->srcip.s6_addr16[5] = bpf_htons(0xffff);
-    __builtin_memcpy(&key->srcip.s6_addr32[3], &ip4_src, sizeof(ip4_src));
+    __builtin_memcpy(key->srcip.s6_addr, ip4in6, sizeof(ip4in6));
+    __builtin_memcpy(key->srcip.s6_addr + sizeof(ip4in6), &ip4_src, sizeof(ip4_src));
 
     // convert to V4MAPPED address
     __be32 ip4_dst = BPF_CORE_READ(iphdr, daddr);
-    key->dstip.s6_addr16[5] = bpf_htons(0xffff);
-    __builtin_memcpy(&key->dstip.s6_addr32[3], &ip4_dst, sizeof(ip4_dst));
+    __builtin_memcpy(key->dstip.s6_addr, ip4in6, sizeof(ip4in6));
+    __builtin_memcpy(key->dstip.s6_addr + sizeof(ip4in6), &ip4_dst, sizeof(ip4_dst));
     break;
   }
   case ETH_P_IPV6: {
@@ -614,14 +616,27 @@ process_udp_recv(bool receive, struct sk_buff *skb, statkey *key, pid_t pid) {
     break;
   }
   default:
-    return;
+    return false;
   }
 
   key->src_port = bpf_ntohs(BPF_CORE_READ(udphdr, source));
   key->dst_port = bpf_ntohs(BPF_CORE_READ(udphdr, dest));
 
   key->proto = IPPROTO_UDP;
   key->pid = pid;
+
+  /* we need to swap the source and destination IP addresses and ports */
+  if (receive) {
+    struct in6_addr tmp_ip = key->srcip;
+    key->srcip = key->dstip;
+    key->dstip = tmp_ip;
+
+    __u16 tmp_port = key->src_port;
+    key->src_port = key->dst_port;
+    key->dst_port = tmp_port;
+  }
+
+  return true;
 }
 
 /**
@@ -648,13 +663,13 @@ process_icmp4(struct sk_buff *skb, statkey *key, pid_t pid) {
 
   // convert to V4MAPPED address
   __be32 ip4_src = BPF_CORE_READ(iphdr, saddr);
-  key->srcip.s6_addr16[5] = bpf_htons(0xffff);
-  __builtin_memcpy(&key->srcip.s6_addr32[3], &ip4_src, sizeof(ip4_src));
+  __builtin_memcpy(key->srcip.s6_addr, ip4in6, sizeof(ip4in6));
+  __builtin_memcpy(key->srcip.s6_addr + sizeof(ip4in6), &ip4_src, sizeof(ip4_src));
 
   // convert to V4MAPPED address
   __be32 ip4_dst = BPF_CORE_READ(iphdr, daddr);
-  key->dstip.s6_addr16[5] = bpf_htons(0xffff);
-  __builtin_memcpy(&key->dstip.s6_addr32[3], &ip4_dst, sizeof(ip4_dst));
+  __builtin_memcpy(key->dstip.s6_addr, ip4in6, sizeof(ip4in6));
+  __builtin_memcpy(key->dstip.s6_addr + sizeof(ip4in6), &ip4_dst, sizeof(ip4_dst));
 
   // store ICMP type in src port
   key->src_port = BPF_CORE_READ(icmphdr, type);
@@ -664,8 +679,9 @@ process_icmp4(struct sk_buff *skb, statkey *key, pid_t pid) {
   key->proto = IPPROTO_ICMP;
   key->pid = pid;
 
-  size_t msglen = bpf_ntohs(BPF_CORE_READ(iphdr, tot_len)) -
-                  BPF_CORE_READ_BITFIELD_PROBED(iphdr, ihl) * 4;
+  __u16 tot_len = bpf_ntohs(BPF_CORE_READ(iphdr, tot_len));
+  __u16 ihl_bytes = (__u16)(BPF_CORE_READ_BITFIELD_PROBED(iphdr, ihl) * 4);
+  size_t msglen = (ihl_bytes <= tot_len) ? (tot_len - ihl_bytes) : 0;
 
   return msglen;
 }
@@ -735,8 +751,9 @@ process_udp_send(struct sk_buff *skb, statkey *key, pid_t pid) {
       (struct udphdr *)(BPF_CORE_READ(skb, head) +
                         BPF_CORE_READ(skb, transport_header));
 
-  process_udp_recv(false, skb, key, pid);
-  size_t msglen = BPF_CORE_READ(udphdr, len);
+  if (!process_udp_recv(false, skb, key, pid))
+    return 0;
+  size_t msglen = bpf_ntohs(BPF_CORE_READ(udphdr, len));
 
   return msglen;
 }
@@ -891,7 +908,8 @@ int BPF_KPROBE(tcp_sendmsg, struct sock *sk, struct msghdr *msg, size_t size) {
 
   key.cgroupid = bpf_get_current_cgroup_id();
 
-  process_tcp(false, sk, &key, pid);
+  if (!process_tcp(false, sk, &key, pid))
+    return 0;
   update_val(&key, size);
 
   return 0;
@@ -927,7 +945,8 @@ int BPF_KPROBE(tcp_cleanup_rbuf, struct sock *sk, int copied) {
 
   key.cgroupid = bpf_get_current_cgroup_id();
 
-  process_tcp(true, sk, &key, pid);
+  if (!process_tcp(true, sk, &key, pid))
+    return 0;
   update_val(&key, copied);
 
   return 0;
@@ -949,8 +968,9 @@ int BPF_KPROBE(tcp_cleanup_rbuf, struct sock *sk, int copied) {
  */
 SEC("kprobe/ip_send_skb")
 int BPF_KPROBE(ip_send_skb, struct net *net, struct sk_buff *skb) {
-  __u16 protocol = BPF_CORE_READ(skb, protocol);
-  if (protocol != IPPROTO_UDP) {
+  struct iphdr *iphdr = (struct iphdr *)(BPF_CORE_READ(skb, head) +
+                                         BPF_CORE_READ(skb, network_header));
+  if (BPF_CORE_READ(iphdr, protocol) != IPPROTO_UDP) {
     return 0;
   }
 
@@ -987,6 +1007,10 @@ int BPF_KPROBE(ip_send_skb, struct net *net, struct sk_buff *skb) {
  */
 SEC("kprobe/skb_consume_udp")
 int BPF_KPROBE(skb_consume_udp, struct sock *sk, struct sk_buff *skb, int len) {
+  if (len <= 0) {
+    return 0;
+  }
+
   statkey key;
   __builtin_memset(&key, 0, sizeof(key));
 
@@ -997,7 +1021,8 @@ int BPF_KPROBE(skb_consume_udp, struct sock *sk, struct sk_buff *skb, int len) {
 
   key.cgroupid = bpf_get_current_cgroup_id();
 
-  process_udp_recv(true, skb, &key, pid);
+  if (!process_udp_recv(true, skb, &key, pid))
+    return 0;
   update_val(&key, len);
 
   return 0;
@@ -1034,6 +1059,11 @@ 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);
+  /* __icmp_send is called with the triggering packet's skb; its
+   * transport_header points to that packet's transport layer (e.g. TCP/UDP),
+   * not to an ICMP header. Use the kprobe arguments directly for type/code. */
+  key.src_port = type;
+  key.dst_port = code;
   update_val(&key, msglen);
 
   return 0;

cgroup.go

@@ -31,6 +31,7 @@ import (
 	"strings"
 	"sync"
 	"syscall"
+	"time"
 
 	"github.com/cilium/ebpf/perf"
 )
@@ -197,6 +198,17 @@ func cGroupWatcher(objs cgroupObjects) (*perf.Reader, error) {
 					return
 				}
 
+				// avoid tight-spinning on persistent errors
+				time.Sleep(10 * time.Millisecond)
+
+				continue
+			}
+
+			// Perf buffer overflowed: lost samples mean we may have missed
+			// cgroup mkdir events, so rebuild the cache from the filesystem.
+			if r.LostSamples > 0 {
+				cgroupCacheRefresh(CGroupRootPath)
+
 				continue
 			}
 

cgroup_arm64_bpfel.o

@@ -96,7 +96,7 @@ func bytesSort(stats []statEntry) {
 	})
 }
 
-// srcIPSort sorts a slice of statEntry objects by their SrcIP field in descending order.
+// srcIPSort sorts a slice of statEntry objects by their SrcIP field in ascending order.
 //
 // Parameters:
 //
@@ -107,7 +107,7 @@ func srcIPSort(stats []statEntry) {
 	})
 }
 
-// dstIPSort sorts a slice of statEntry objects by their DstIP field in descending order.
+// dstIPSort sorts a slice of statEntry objects by their DstIP field in ascending order.
 //
 // Parameters:
 //

cgroup_x86_bpfel.o

@@ -292,6 +292,10 @@ func startCGroupTrace(objs cgroupObjects, links []link.Link) []link.Link {
 		return links
 	}
 
+	if err != nil {
+		log.Fatalf("Error checking RawTracepoint support: %v", err)
+	}
+
 	l, err = link.AttachRawTracepoint(link.RawTracepointOptions{
 		Program: objs.TraceCgroupMkdir,
 		Name:    "cgroup_mkdir",