Skip the inbound firewall for the rest of a flow's run within one receive batch

interface.go

@@ -356,7 +356,19 @@ type rxContext struct {
 	h            *header.H
 	fwPacket     *firewall.ParsedPacket
 	hostmapCache map[uint32]*HostInfo
-	lhh          *LightHouseHandler
+
+	// lastAllowed is the last flow the inbound firewall let in during this batch. UDP GRO hands us segments
+	// in runs from the same flow, so the rest of a run can skip the firewall. Cleared by endBatch.
+	lastAllowed     firewall.Packet
+	lastAllowedHost *HostInfo
+	lhh             *LightHouseHandler
+}
+
+// endBatch forgets what was learned during a batch, so nothing outlives a hostmap or firewall change by more than one batch
+func (rxc *rxContext) endBatch() {
+	clear(rxc.hostmapCache)
+	rxc.lastAllowed = firewall.Packet{}
+	rxc.lastAllowedHost = nil
 }
 
 func newRxContext(f *Interface, q int) *rxContext {
@@ -389,7 +401,7 @@ func (f *Interface) listenOut(i int) {
 		if err := f.batchers[i].Flush(); err != nil {
 			f.l.Error("Failed to flush tun coalescer", "error", err)
 		}
-		clear(rxc.hostmapCache)
+		rxc.endBatch()
 	}
 
 	err := li.ListenOut(listener, flusher)

outside.go

@@ -477,7 +477,7 @@ func (f *Interface) handleOutsideMessagePacket(hostinfo *HostInfo, messageCounte
 		return
 	}
 
-	dropReason := f.firewall.Drop(rxc.fwPacket.Packet, true, hostinfo, f.pki.GetCAPool())
+	dropReason := f.allowInbound(hostinfo, rxc)
 	if dropReason != nil {
 		f.rejectOutside(out, hostinfo.ConnectionState, hostinfo, rxc.nb, rxc.scratch, rxc.q)
 		if f.l.Enabled(context.Background(), slog.LevelDebug) {
@@ -492,6 +492,23 @@ func (f *Interface) handleOutsideMessagePacket(hostinfo *HostInfo, messageCounte
 	}
 }
 
+// allowInbound runs the inbound firewall check on rxc.fwPacket. It skips the check when the packet belongs to
+// the same flow from the same host as the last one allowed in this batch.
+func (f *Interface) allowInbound(hostinfo *HostInfo, rxc *rxContext) error {
+	fp := rxc.fwPacket.Packet
+	if hostinfo == rxc.lastAllowedHost && fp == rxc.lastAllowed {
+		return nil
+	}
+
+	if err := f.firewall.Drop(fp, true, hostinfo, f.pki.GetCAPool()); err != nil {
+		return err
+	}
+
+	rxc.lastAllowed = fp
+	rxc.lastAllowedHost = hostinfo
+	return nil
+}
+
 func (f *Interface) maybeSendRecvError(endpoint netip.AddrPort, index uint32) {
 	if f.sendRecvErrorConfig.ShouldRecvError(endpoint) {
 		f.sendRecvError(endpoint, index)

outside_test.go

@@ -6,12 +6,16 @@ import (
 	"net"
 	"net/netip"
 	"testing"
+	"time"
 
+	"github.com/gaissmai/bart"
 	"github.com/google/gopacket"
 	"github.com/google/gopacket/layers"
+	"github.com/slackhq/nebula/cert"
 	"github.com/slackhq/nebula/iputil"
 
 	"github.com/slackhq/nebula/firewall"
+	"github.com/slackhq/nebula/test"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"golang.org/x/net/ipv4"
@@ -841,3 +845,49 @@ func Test_newPacket_parsedFields(t *testing.T) {
 	assert.True(t, p.Fragment)
 	assert.True(t, p.FragAny)
 }
+
+func TestInterface_allowInbound(t *testing.T) {
+	l := test.NewLoggerWithOutput(&bytes.Buffer{})
+	myNetworks := new(bart.Lite)
+	myNetworks.Insert(netip.MustParsePrefix("192.0.2.1/24"))
+	me := &dummyCert{name: "me", networks: []netip.Prefix{netip.MustParsePrefix("192.0.2.1/24")}}
+	peer := &cert.CachedCertificate{Certificate: &dummyCert{name: "peer", networks: []netip.Prefix{netip.MustParsePrefix("192.0.2.2/24")}}}
+	h := &HostInfo{ConnectionState: &ConnectionState{peerCert: peer}, vpnAddrs: []netip.Addr{netip.MustParseAddr("192.0.2.2")}}
+	h.buildNetworks(myNetworks, peer.Certificate)
+
+	allowAll := NewFirewall(l, time.Minute, time.Minute, time.Minute, me)
+	require.NoError(t, allowAll.AddRule(true, firewall.ProtoAny, 0, 0, []string{"any"}, "", "", "", "", ""))
+	allowNone := NewFirewall(l, time.Minute, time.Minute, time.Minute, me)
+
+	f := &Interface{firewall: allowAll, pki: &PKI{}, l: l}
+	rxc := &rxContext{fwPacket: &firewall.ParsedPacket{}, hostmapCache: map[uint32]*HostInfo{}}
+	flow := firewall.Packet{
+		LocalAddr:  netip.MustParseAddr("192.0.2.1"),
+		RemoteAddr: netip.MustParseAddr("192.0.2.2"),
+		LocalPort:  443,
+		RemotePort: 55000,
+		Protocol:   iputil.IPProtocolUDP,
+	}
+	rxc.fwPacket.Packet = flow
+	require.NoError(t, f.allowInbound(h, rxc))
+
+	// With a firewall that allows nothing, the rest of the flow's run in this batch still passes
+	f.firewall = allowNone
+	require.NoError(t, f.allowInbound(h, rxc))
+
+	// A different flow is checked
+	other := flow
+	other.RemotePort++
+	rxc.fwPacket.Packet = other
+	require.ErrorIs(t, f.allowInbound(h, rxc), ErrNoMatchingRule)
+
+	// So is the same flow from a different host
+	rxc.fwPacket.Packet = flow
+	h2 := &HostInfo{ConnectionState: h.ConnectionState, vpnAddrs: h.vpnAddrs}
+	h2.buildNetworks(myNetworks, peer.Certificate)
+	require.ErrorIs(t, f.allowInbound(h2, rxc), ErrNoMatchingRule)
+
+	// And the same flow in the next batch
+	rxc.endBatch()
+	require.ErrorIs(t, f.allowInbound(h, rxc), ErrNoMatchingRule)
+}