Fix cannot translate peer address in some UDP scenarios (#134)

CHANGES.md

@@ -15,6 +15,7 @@ Release Notes.
 * Fixed `ip_list_rcv` probe is not exist in older linux kernel.
 * Fix concurrent map operation in the access log module.
 * Fix the profiling cannot found process issue.
+* Fix cannot translate peer address in some UDP scenarios.
 
 #### Documentation
 

bpf/accesslog/common/connection.h

@@ -216,6 +216,8 @@ static __always_inline void submit_new_connection(void* ctx, bool success, __u32
     event->success = success;
 
     __u16 port;
+    event->local_port = 0;
+    event->remote_port = 0;
     if (socket != NULL) {
         // only get from accept function(server side)
         struct sock* s;
@@ -247,11 +249,14 @@ static __always_inline void submit_new_connection(void* ctx, bool success, __u32
             bpf_probe_read(&event->remote_addr_v4, sizeof(event->remote_addr_v4), &daddr->sin_addr.s_addr);
             bpf_probe_read(&port, sizeof(port), &daddr->sin_port);
             event->remote_port = bpf_ntohs(port);
+            // cleanup the local address
+            event->local_addr_v4 = 0;
         } else if (event->socket_family == AF_INET6) {
             struct sockaddr_in6 *daddr = (struct sockaddr_in6 *)addr;
             bpf_probe_read(&event->remote_addr_v6, sizeof(event->remote_addr_v6), &daddr->sin6_addr.s6_addr);
             bpf_probe_read(&port, sizeof(port), &daddr->sin6_port);
             event->remote_port = bpf_ntohs(port);
+            __builtin_memset(&event->local_addr_v6, 0, sizeof(event->local_addr_v6));
         }
     } else {
         event->socket_family = AF_UNKNOWN;

bpf/accesslog/syscalls/connect.c

@@ -146,4 +146,15 @@ int sock_alloc_ret(struct pt_regs *ctx) {
     return 0;
 }
 
+SEC("kprobe/ip4_datagram_connect")
+int ip4_udp_datagram_connect(struct pt_regs *ctx) {
+    __u64 id = bpf_get_current_pid_tgid();
+    struct connect_args_t *connect_args = bpf_map_lookup_elem(&conecting_args, &id);
+    if (connect_args) {
+        struct sock *sock = (struct sock*)PT_REGS_PARM1(ctx);
+        connect_args->sock = sock;
+    }
+    return 0;
+}
+
 #include "connect_conntrack.c"

bpf/accesslog/syscalls/connect_conntrack.c

@@ -140,3 +140,8 @@ SEC("kprobe/nf_confirm")
 int nf_confirm(struct pt_regs* ctx) {
     return nf_conn_aware(ctx, (struct nf_conn*)PT_REGS_PARM3(ctx));
 }
+
+SEC("kprobe/ctnetlink_fill_info")
+int nf_ctnetlink_fill_info(struct pt_regs* ctx) {
+    return nf_conn_aware(ctx, (struct nf_conn*)PT_REGS_PARM5(ctx));
+}

pkg/accesslog/collector/connect.go

@@ -68,10 +68,18 @@ func (c *ConnectCollector) Start(_ *module.Manager, context *common.AccessLogCon
 	context.BPF.AddLink(link.Kretprobe, map[string]*ebpf.Program{
 		"sock_alloc": context.BPF.SockAllocRet,
 	})
-
 	context.BPF.AddLink(link.Kprobe, map[string]*ebpf.Program{
+		"ip4_datagram_connect": context.BPF.Ip4UdpDatagramConnect,
+	})
+
+	_ = context.BPF.AddLinkOrError(link.Kprobe, map[string]*ebpf.Program{
 		"__nf_conntrack_hash_insert": context.BPF.NfConntrackHashInsert,
-		"nf_confirm":                 context.BPF.NfConfirm,
+	})
+	_ = context.BPF.AddLinkOrError(link.Kprobe, map[string]*ebpf.Program{
+		"nf_confirm": context.BPF.NfConfirm,
+	})
+	_ = context.BPF.AddLinkOrError(link.Kprobe, map[string]*ebpf.Program{
+		"ctnetlink_fill_info": context.BPF.NfCtnetlinkFillInfo,
 	})
 
 	context.BPF.ReadEventAsync(context.BPF.SocketConnectionEventQueue, func(data interface{}) {
@@ -127,11 +135,15 @@ func (c *ConnectCollector) buildSocketFromConnectEvent(event *events.SocketConne
 	}
 	socketPair := c.buildSocketPair(event)
 	if socketPair != nil && socketPair.IsValid() {
+		connectLogger.Debugf("found the connection from the connect event is valid, connection ID: %d, randomID: %d",
+			event.ConID, event.RandomID)
 		return socketPair
 	}
 	// if only the local port not success, maybe the upstream port is not open, so it could be continued
 	if c.isOnlyLocalPortEmpty(socketPair) {
 		event.ConnectSuccess = 0
+		connectLogger.Debugf("the connection from the connect event is only the local port is empty, connection ID: %d, randomID: %d",
+			event.ConID, event.RandomID)
 		return socketPair
 	}
 

pkg/tools/ip/bpf.go

@@ -23,6 +23,9 @@ import (
 )
 
 func ParseIPV4(bpfVal uint32) string {
+	if bpfVal == 0 {
+		return ""
+	}
 	return net.IP((*(*[net.IPv4len]byte)(unsafe.Pointer(&bpfVal)))[:]).String()
 }
 

pkg/tools/ip/conntrack.go

@@ -86,6 +86,8 @@ func (c *ConnTrack) UpdateRealPeerAddress(addr *SocketPair) bool {
 	}
 	if res := c.filterValidateReply(dump, tuple); res != nil {
 		addr.DestIP = res.Src.String()
+		log.Debugf("found the connection from the dump all conntrack, src: %s:%d, dst: %s:%d, proto number: %d",
+			tuple.Src, *tuple.Proto.SrcPort, tuple.Dst, *tuple.Proto.DstPort, *tuple.Proto.Number)
 		return true
 	}
 	return false