bpf:hubble: support policy verdict from L3 devices

When `bpf_host` is attached to a L3 device and there are policies
(e.g., when HostFw is enabled), verdict notification are emitted, but
monitor/Hubble incorrectly decodes the packet, starting from a non-existent
L2 header. Let's fix this by adding a flag, similarly to what we did for
drop/trace notifications.

From v1.18 onwards, `bpf_wireguard` has its own program rather than attaching
`bpf_host` to the `cilium_wg0@ingress`. Though, it still can happen that
`bpf_host` is attached to some L3 device (TUN).

Example of PolicyVerdict incorrectly decoded:

```
Policy verdict log: flow 0x0 local EP ID 285, remote ID remote-node, proto 1, ingress, action allow, auth: disabled, match L4-Only, 40:00:40:01:90:a3 -> 45:c0:00:34:51:3b UnknownEthernetType
```

Example of PolicyVerdict correctly decoded after PR:

```
Policy verdict log: flow 0x0 local EP ID 285, remote ID remote-node, proto 1, ingress, action allow, auth: disabled, match L4-Only, 172.18.0.3 -> 172.18.0.4 EchoRequest
```

Signed-off-by: Simone Magnani <simone.magnani@isovalent.com>

Author: smagnani96

bpf/lib/policy_log.h

@@ -30,7 +30,7 @@ struct policy_verdict_notify {
 		ipv6:1,
 		match_type:3,
 		audited:1,
-		pad0:1;
+		l3:1;
 	__u8	auth_type;
 	__u8	pad1[3]; /* align with 64 bits */
 	__u32	cookie;
@@ -107,6 +107,7 @@ send_policy_verdict_notify(struct __ctx_buff *ctx, __u32 remote_label, __u16 dst
 		.audited	= is_audited,
 		.auth_type      = auth_type,
 		.cookie		= cookie,
+		.l3		= THIS_IS_L3_DEV,
 	};
 
 	ctx_event_output(ctx, &cilium_events,

pkg/hubble/parser/threefour/parser.go

@@ -214,8 +214,8 @@ func (p *Parser) Decode(data []byte, decoded *pb.Flow) error {
 		return fmt.Errorf("not enough bytes to decode %d", data)
 	}
 
-	isL3Device := tn != nil && tn.IsL3Device() || dn != nil && dn.IsL3Device()
-	isIPv6 := tn != nil && tn.IsIPv6() || dn != nil && dn.IsIPv6()
+	isL3Device := tn != nil && tn.IsL3Device() || dn != nil && dn.IsL3Device() || pvn != nil && pvn.IsTrafficL3Device()
+	isIPv6 := tn != nil && tn.IsIPv6() || dn != nil && dn.IsIPv6() || pvn != nil && pvn.IsTrafficIPv6()
 	isVXLAN := tn != nil && tn.IsVXLAN() || dn != nil && dn.IsVXLAN()
 	isGeneve := tn != nil && tn.IsGeneve() || dn != nil && dn.IsGeneve()
 	ether, ip, l4, tunnel, srcIP, dstIP, srcPort, dstPort, summary, err := decodeLayers(data[packetOffset:], p.packet, isL3Device, isIPv6, isVXLAN, isGeneve)

pkg/hubble/parser/threefour/parser_test.go

@@ -2345,23 +2345,29 @@ func TestDecode_PolicyVerdictNotify(t *testing.T) {
 	parser, err := New(hivetest.Logger(t), defaultEndpointGetter, nil, nil, nil, nil, nil)
 	require.NoError(t, err)
 
-	template := &flowpb.Flow{
-		EventType:   &flowpb.CiliumEventType{Type: 5},
-		Summary:     flowpb.IPVersion_IPv4.String(),
-		Type:        flowpb.FlowType_L3_L4,
-		Verdict:     flowpb.Verdict_FORWARDED,
-		Source:      &flowpb.Endpoint{},
-		Destination: &flowpb.Endpoint{},
-		Ethernet: &flowpb.Ethernet{
-			Source:      srcMAC.String(),
-			Destination: dstMAC.String(),
-		},
-		IP: &flowpb.IP{
-			IpVersion:   flowpb.IPVersion_IPv4,
-			Source:      localIP.String(),
-			Destination: remoteIP.String(),
-		},
-		IsReply: wrapperspb.Bool(false),
+	getTemplate := func(isL3Device bool) *flowpb.Flow {
+		template := &flowpb.Flow{
+			EventType:   &flowpb.CiliumEventType{Type: 5},
+			Summary:     flowpb.IPVersion_IPv4.String(),
+			Type:        flowpb.FlowType_L3_L4,
+			Verdict:     flowpb.Verdict_FORWARDED,
+			Source:      &flowpb.Endpoint{},
+			Destination: &flowpb.Endpoint{},
+			Ethernet: &flowpb.Ethernet{
+				Source:      srcMAC.String(),
+				Destination: dstMAC.String(),
+			},
+			IP: &flowpb.IP{
+				IpVersion:   flowpb.IPVersion_IPv4,
+				Source:      localIP.String(),
+				Destination: remoteIP.String(),
+			},
+			IsReply: wrapperspb.Bool(false),
+		}
+		if isL3Device {
+			template.Ethernet = nil
+		}
+		return template
 	}
 
 	testCases := []struct {
@@ -2386,7 +2392,7 @@ func TestDecode_PolicyVerdictNotify(t *testing.T) {
 			},
 		},
 		{
-			name: "ingresss",
+			name: "ingress",
 			event: monitor.PolicyVerdictNotify{
 				Type:   byte(monitorAPI.MessageTypePolicyVerdict),
 				Source: localEP,
@@ -2398,20 +2404,41 @@ func TestDecode_PolicyVerdictNotify(t *testing.T) {
 				TrafficDirection: flowpb.TrafficDirection_INGRESS,
 			},
 		},
+		{
+			name: "ingress_l3_device",
+			event: monitor.PolicyVerdictNotify{
+				Type:   byte(monitorAPI.MessageTypePolicyVerdict),
+				Source: localEP,
+				Flags:  monitorAPI.PolicyIngress | monitor.PolicyVerdictNotifyFlagIsL3,
+			},
+			ipTuple: egressTuple,
+			want: &flowpb.Flow{
+				Source:           &flowpb.Endpoint{ID: 1234},
+				TrafficDirection: flowpb.TrafficDirection_INGRESS,
+			},
+		},
 	}
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
-			want := proto.Clone(template)
-			proto.Merge(want, tc.want)
+			isL3Device := false
+			if ev, ok := tc.event.(monitor.PolicyVerdictNotify); ok {
+				isL3Device = ev.IsTrafficL3Device()
+			}
 
-			data, err := testutils.CreateL3L4Payload(tc.event,
-				&layers.Ethernet{
+			var l []gopacket.SerializableLayer
+			if !isL3Device {
+				l = append(l, &layers.Ethernet{
 					SrcMAC:       srcMAC,
 					DstMAC:       dstMAC,
 					EthernetType: layers.EthernetTypeIPv4,
-				},
-				&layers.IPv4{SrcIP: tc.ipTuple.src.AsSlice(), DstIP: tc.ipTuple.dst.AsSlice()},
-			)
+				})
+			}
+			l = append(l, &layers.IPv4{SrcIP: tc.ipTuple.src.AsSlice(), DstIP: tc.ipTuple.dst.AsSlice()})
+
+			want := proto.Clone(getTemplate(isL3Device))
+			proto.Merge(want, tc.want)
+
+			data, err := testutils.CreateL3L4Payload(tc.event, l...)
 			if err != nil {
 				t.Fatalf("Unexpected error from CreateL3L4Payload(%T, ...): %v", tc.event, err)
 			}

pkg/monitor/datapath_policy.go

@@ -37,6 +37,10 @@ const (
 	// corresponds to whether the traffic was allowed due to the audit mode
 	PolicyVerdictNotifyFlagIsAudited = 0x40
 
+	// PolicyVerdictNotifyFlagIsL3 is the bit mask in Flags that
+	// corresponds to whether the traffic is from a L3 device or not
+	PolicyVerdictNotifyFlagIsL3 = 0x80
+
 	// PolicyVerdictNotifyFlagMatchTypeBitOffset is the bit offset in Flags that
 	// corresponds to the policy match type
 	PolicyVerdictNotifyFlagMatchTypeBitOffset = 3
@@ -114,6 +118,11 @@ func (n *PolicyVerdictNotify) IsTrafficIPv6() bool {
 	return (n.Flags&PolicyVerdictNotifyFlagIsIPv6 > 0)
 }
 
+// IsTrafficL3Device returns true if this notify is from a L3 device
+func (n *PolicyVerdictNotify) IsTrafficL3Device() bool {
+	return (n.Flags&PolicyVerdictNotifyFlagIsL3 > 0)
+}
+
 // GetPolicyMatchType returns how the traffic matched the policy
 func (n *PolicyVerdictNotify) GetPolicyMatchType() api.PolicyMatchType {
 	return api.PolicyMatchType((n.Flags & PolicyVerdictNotifyFlagMatchType) >>
@@ -161,5 +170,5 @@ func (n *PolicyVerdictNotify) DumpInfo(buf *bufio.Writer, data []byte, numeric a
 	fmt.Fprintf(buf, ", proto %d, %s, action %s, auth: %s, match %s, %s\n", n.Proto, dir,
 		GetPolicyActionString(n.Verdict, n.IsTrafficAudited()),
 		n.GetAuthType(), n.GetPolicyMatchType(),
-		GetConnectionSummary(data[PolicyVerdictNotifyLen:], nil))
+		GetConnectionSummary(data[PolicyVerdictNotifyLen:], &decodeOpts{IsL3Device: n.IsTrafficL3Device(), IsIPv6: n.IsTrafficIPv6()}))
 }