analyzer: make http3/quic handling more reliable

engine/udp.go

@@ -131,13 +131,25 @@ func (m *udpStreamManager) MatchWithContext(streamID uint32, ipFlow gopacket.Flo
 	rev := false
 	value, ok := m.streams.Get(streamID)
 	if !ok {
-		// New stream
+		// Fallback: conntrack IDs can change during early flow lifetime on some systems.
-		value = &udpStreamValue{
+		// Try to find an existing stream by 5-tuple before creating a new stream.
-			Stream:  m.factory.New(ipFlow, udp.TransportFlow(), udp, uc),
+		matchedKey, matchedValue, matchedRev, found := m.findByFlow(ipFlow, udp.TransportFlow())
-			IPFlow:  ipFlow,
+		if found {
-			UDPFlow: udp.TransportFlow(),
+			value = matchedValue
+			rev = matchedRev
+			if matchedKey != streamID {
+				m.streams.Remove(matchedKey)
+				m.streams.Add(streamID, matchedValue)
+			}
+		} else {
+			// New stream
+			value = &udpStreamValue{
+				Stream:  m.factory.New(ipFlow, udp.TransportFlow(), udp, uc),
+				IPFlow:  ipFlow,
+				UDPFlow: udp.TransportFlow(),
+			}
+			m.streams.Add(streamID, value)
 		}
-		m.streams.Add(streamID, value)
 	} else {
 		// Stream ID exists, but is it really the same stream?
 		ok, rev = value.Match(ipFlow, udp.TransportFlow())
@@ -157,6 +169,19 @@ func (m *udpStreamManager) MatchWithContext(streamID uint32, ipFlow gopacket.Flo
 	}
 }
 
+func (m *udpStreamManager) findByFlow(ipFlow, udpFlow gopacket.Flow) (key uint32, value *udpStreamValue, rev bool, found bool) {
+	for _, k := range m.streams.Keys() {
+		v, ok := m.streams.Peek(k)
+		if !ok || v == nil {
+			continue
+		}
+		if ok2, rev2 := v.Match(ipFlow, udpFlow); ok2 {
+			return k, v, rev2, true
+		}
+	}
+	return 0, nil, false, false
+}
+
 type udpStream struct {
 	info          ruleset.StreamInfo
 	virgin        bool // true if no packets have been processed