Protect Endpoint.dispatcher with an RWMutex.

Many Endpoint implementations had unsafe accesses of dispatcher field.

PiperOrigin-RevId: 502675235

pkg/tcpip/link/channel/channel.go

@@ -135,12 +135,15 @@ var _ stack.GSOEndpoint = (*Endpoint)(nil)
 // Endpoint is link layer endpoint that stores outbound packets in a channel
 // and allows injection of inbound packets.
 type Endpoint struct {
-	dispatcher         stack.NetworkDispatcher
 	mtu                uint32
 	linkAddr           tcpip.LinkAddress
 	LinkEPCapabilities stack.LinkEndpointCapabilities
 	SupportedGSOKind   stack.SupportedGSO
 
+	mu sync.RWMutex
+	// +checklocks:mu
+	dispatcher stack.NetworkDispatcher
+
 	// Outbound packet queue.
 	q *queue
 }
@@ -191,17 +194,24 @@ func (e *Endpoint) NumQueued() int {
 
 // InjectInbound injects an inbound packet.
 func (e *Endpoint) InjectInbound(protocol tcpip.NetworkProtocolNumber, pkt stack.PacketBufferPtr) {
-	e.dispatcher.DeliverNetworkPacket(protocol, pkt)
+	e.mu.RLock()
+	d := e.dispatcher
+	e.mu.RUnlock()
+	d.DeliverNetworkPacket(protocol, pkt)
 }
 
 // Attach saves the stack network-layer dispatcher for use later when packets
 // are injected.
 func (e *Endpoint) Attach(dispatcher stack.NetworkDispatcher) {
+	e.mu.Lock()
+	defer e.mu.Unlock()
 	e.dispatcher = dispatcher
 }
 
 // IsAttached implements stack.LinkEndpoint.IsAttached.
 func (e *Endpoint) IsAttached() bool {
+	e.mu.RLock()
+	defer e.mu.RUnlock()
 	return e.dispatcher != nil
 }
 

pkg/tcpip/link/fdbased/endpoint.go

@@ -135,7 +135,10 @@ type endpoint struct {
 	closed func(tcpip.Error)
 
 	inboundDispatchers []linkDispatcher
-	dispatcher         stack.NetworkDispatcher
+
+	mu sync.RWMutex
+	// +checklocks:mu
+	dispatcher stack.NetworkDispatcher
 
 	// packetDispatchMode controls the packet dispatcher used by this
 	// endpoint.
@@ -405,6 +408,8 @@ func isSocketFD(fd int) (bool, error) {
 //
 // Attach implements stack.LinkEndpoint.Attach.
 func (e *endpoint) Attach(dispatcher stack.NetworkDispatcher) {
+	e.mu.Lock()
+	defer e.mu.Unlock()
 	// nil means the NIC is being removed.
 	if dispatcher == nil && e.dispatcher != nil {
 		for _, dispatcher := range e.inboundDispatchers {
@@ -431,6 +436,8 @@ func (e *endpoint) Attach(dispatcher stack.NetworkDispatcher) {
 
 // IsAttached implements stack.LinkEndpoint.IsAttached.
 func (e *endpoint) IsAttached() bool {
+	e.mu.RLock()
+	defer e.mu.RUnlock()
 	return e.dispatcher != nil
 }
 
@@ -765,18 +772,25 @@ func (e *endpoint) ARPHardwareType() header.ARPHardwareType {
 type InjectableEndpoint struct {
 	endpoint
 
+	mu sync.RWMutex
+	// +checklocks:mu
 	dispatcher stack.NetworkDispatcher
 }
 
 // Attach saves the stack network-layer dispatcher for use later when packets
 // are injected.
 func (e *InjectableEndpoint) Attach(dispatcher stack.NetworkDispatcher) {
+	e.mu.Lock()
+	defer e.mu.Unlock()
 	e.dispatcher = dispatcher
 }
 
 // InjectInbound injects an inbound packet.
 func (e *InjectableEndpoint) InjectInbound(protocol tcpip.NetworkProtocolNumber, pkt stack.PacketBufferPtr) {
-	e.dispatcher.DeliverNetworkPacket(protocol, pkt)
+	e.mu.RLock()
+	d := e.dispatcher
+	e.mu.RUnlock()
+	d.DeliverNetworkPacket(protocol, pkt)
 }
 
 // NewInjectable creates a new fd-based InjectableEndpoint.

pkg/tcpip/link/fdbased/mmap.go

@@ -198,6 +198,9 @@ func (d *packetMMapDispatcher) dispatch() (bool, tcpip.Error) {
 			panic(fmt.Sprintf("LinkHeader().Consume(%d) must succeed", d.e.hdrSize))
 		}
 	}
-	d.e.dispatcher.DeliverNetworkPacket(p, pbuf)
+	d.e.mu.RLock()
+	dsp := d.e.dispatcher
+	d.e.mu.RUnlock()
+	dsp.DeliverNetworkPacket(p, pbuf)
 	return true, nil
 }

pkg/tcpip/link/fdbased/packet_dispatchers.go

@@ -212,7 +212,10 @@ func (d *readVDispatcher) dispatch() (bool, tcpip.Error) {
 		}
 	}
 
-	d.e.dispatcher.DeliverNetworkPacket(p, pkt)
+	d.e.mu.RLock()
+	dsp := d.e.dispatcher
+	d.e.mu.RUnlock()
+	dsp.DeliverNetworkPacket(p, pkt)
 
 	return true, nil
 }
@@ -291,6 +294,10 @@ func (d *recvMMsgDispatcher) dispatch() (bool, tcpip.Error) {
 	// Keep a list of packets so we can DecRef outside of the loop.
 	var pkts stack.PacketBufferList
 
+	d.e.mu.RLock()
+	dsp := d.e.dispatcher
+	d.e.mu.RUnlock()
+
 	defer func() { pkts.DecRef() }()
 	for k := 0; k < nMsgs; k++ {
 		n := int(d.msgHdrs[k].Len)
@@ -329,7 +336,7 @@ func (d *recvMMsgDispatcher) dispatch() (bool, tcpip.Error) {
 			}
 		}
 
-		d.e.dispatcher.DeliverNetworkPacket(p, pkt)
+		dsp.DeliverNetworkPacket(p, pkt)
 	}
 
 	return true, nil

pkg/tcpip/link/loopback/loopback.go

@@ -21,12 +21,16 @@
 package loopback
 
 import (
+	"sync"
+
 	"gvisor.dev/gvisor/pkg/tcpip"
 	"gvisor.dev/gvisor/pkg/tcpip/header"
 	"gvisor.dev/gvisor/pkg/tcpip/stack"
 )
 
 type endpoint struct {
+	mu sync.RWMutex
+	// +checklocks:mu
 	dispatcher stack.NetworkDispatcher
 }
 
@@ -39,11 +43,15 @@ func New() stack.LinkEndpoint {
 // Attach implements stack.LinkEndpoint.Attach. It just saves the stack network-
 // layer dispatcher for later use when packets need to be dispatched.
 func (e *endpoint) Attach(dispatcher stack.NetworkDispatcher) {
+	e.mu.Lock()
+	defer e.mu.Unlock()
 	e.dispatcher = dispatcher
 }
 
 // IsAttached implements stack.LinkEndpoint.IsAttached.
 func (e *endpoint) IsAttached() bool {
+	e.mu.RLock()
+	defer e.mu.RUnlock()
 	return e.dispatcher != nil
 }
 
@@ -75,6 +83,9 @@ func (*endpoint) Wait() {}
 
 // WritePackets implements stack.LinkEndpoint.WritePackets.
 func (e *endpoint) WritePackets(pkts stack.PacketBufferList) (int, tcpip.Error) {
+	e.mu.RLock()
+	d := e.dispatcher
+	e.mu.RUnlock()
 	for _, pkt := range pkts.AsSlice() {
 		// In order to properly loop back to the inbound side we must create a
 		// fresh packet that only contains the underlying payload with no headers
@@ -82,7 +93,7 @@ func (e *endpoint) WritePackets(pkts stack.PacketBufferList) (int, tcpip.Error)
 		newPkt := stack.NewPacketBuffer(stack.PacketBufferOptions{
 			Payload: pkt.ToBuffer(),
 		})
-		e.dispatcher.DeliverNetworkPacket(pkt.NetworkProtocolNumber, newPkt)
+		d.DeliverNetworkPacket(pkt.NetworkProtocolNumber, newPkt)
 		newPkt.DecRef()
 	}
 	return pkts.Len(), nil

pkg/tcpip/link/muxed/injectable.go

@@ -16,6 +16,8 @@
 package muxed
 
 import (
+	"sync"
+
 	"gvisor.dev/gvisor/pkg/bufferv2"
 	"gvisor.dev/gvisor/pkg/tcpip"
 	"gvisor.dev/gvisor/pkg/tcpip/header"
@@ -27,7 +29,10 @@ import (
 // will be written to. Note that HandleLocal works differently for this
 // endpoint (see WritePacket).
 type InjectableEndpoint struct {
-	routes     map[tcpip.Address]stack.InjectableLinkEndpoint
+	routes map[tcpip.Address]stack.InjectableLinkEndpoint
+
+	mu sync.RWMutex
+	// +checklocks:mu
 	dispatcher stack.NetworkDispatcher
 }
 
@@ -72,17 +77,24 @@ func (m *InjectableEndpoint) Attach(dispatcher stack.NetworkDispatcher) {
 	for _, endpoint := range m.routes {
 		endpoint.Attach(dispatcher)
 	}
+	m.mu.Lock()
 	m.dispatcher = dispatcher
+	m.mu.Unlock()
 }
 
 // IsAttached implements stack.LinkEndpoint.
 func (m *InjectableEndpoint) IsAttached() bool {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
 	return m.dispatcher != nil
 }
 
 // InjectInbound implements stack.InjectableLinkEndpoint.
 func (m *InjectableEndpoint) InjectInbound(protocol tcpip.NetworkProtocolNumber, pkt stack.PacketBufferPtr) {
-	m.dispatcher.DeliverNetworkPacket(protocol, pkt)
+	m.mu.RLock()
+	d := m.dispatcher
+	m.mu.RUnlock()
+	d.DeliverNetworkPacket(protocol, pkt)
 }
 
 // WritePackets writes outbound packets to the appropriate

pkg/tcpip/link/pipe/pipe.go

@@ -17,6 +17,8 @@
 package pipe
 
 import (
+	"sync"
+
 	"gvisor.dev/gvisor/pkg/tcpip"
 	"gvisor.dev/gvisor/pkg/tcpip/header"
 	"gvisor.dev/gvisor/pkg/tcpip/stack"
@@ -41,10 +43,13 @@ func New(linkAddr1, linkAddr2 tcpip.LinkAddress, mtu uint32) (*Endpoint, *Endpoi
 
 // Endpoint is one end of a pipe.
 type Endpoint struct {
+	linked   *Endpoint
+	linkAddr tcpip.LinkAddress
+	mtu      uint32
+
+	mu sync.RWMutex
+	// +checklocks:mu
 	dispatcher stack.NetworkDispatcher
-	linked     *Endpoint
-	linkAddr   tcpip.LinkAddress
-	mtu        uint32
 }
 
 func (e *Endpoint) deliverPackets(pkts stack.PacketBufferList) {
@@ -59,7 +64,10 @@ func (e *Endpoint) deliverPackets(pkts stack.PacketBufferList) {
 		newPkt := stack.NewPacketBuffer(stack.PacketBufferOptions{
 			Payload: pkt.ToBuffer(),
 		})
-		e.linked.dispatcher.DeliverNetworkPacket(pkt.NetworkProtocolNumber, newPkt)
+		e.linked.mu.RLock()
+		d := e.linked.dispatcher
+		e.linked.mu.RUnlock()
+		d.DeliverNetworkPacket(pkt.NetworkProtocolNumber, newPkt)
 		newPkt.DecRef()
 	}
 }
@@ -73,11 +81,15 @@ func (e *Endpoint) WritePackets(pkts stack.PacketBufferList) (int, tcpip.Error)
 
 // Attach implements stack.LinkEndpoint.
 func (e *Endpoint) Attach(dispatcher stack.NetworkDispatcher) {
+	e.mu.Lock()
+	defer e.mu.Unlock()
 	e.dispatcher = dispatcher
 }
 
 // IsAttached implements stack.LinkEndpoint.
 func (e *Endpoint) IsAttached() bool {
+	e.mu.RLock()
+	defer e.mu.RUnlock()
 	return e.dispatcher != nil
 }
 

pkg/tcpip/link/waitable/waitable.go

@@ -34,7 +34,10 @@ var _ stack.LinkEndpoint = (*Endpoint)(nil)
 // Endpoint is a waitable link-layer endpoint.
 type Endpoint struct {
 	dispatchGate sync.Gate
-	dispatcher   stack.NetworkDispatcher
+
+	mu sync.RWMutex
+	// +checklocks:mu
+	dispatcher stack.NetworkDispatcher
 
 	writeGate sync.Gate
 	lower     stack.LinkEndpoint
@@ -57,8 +60,10 @@ func (e *Endpoint) DeliverNetworkPacket(protocol tcpip.NetworkProtocolNumber, pk
 	if !e.dispatchGate.Enter() {
 		return
 	}
-
-	e.dispatcher.DeliverNetworkPacket(protocol, pkt)
+	e.mu.RLock()
+	d := e.dispatcher
+	e.mu.RUnlock()
+	d.DeliverNetworkPacket(protocol, pkt)
 	e.dispatchGate.Leave()
 }
 
@@ -67,8 +72,10 @@ func (e *Endpoint) DeliverLinkPacket(protocol tcpip.NetworkProtocolNumber, pkt s
 	if !e.dispatchGate.Enter() {
 		return
 	}
-
-	e.dispatcher.DeliverLinkPacket(protocol, pkt)
+	e.mu.RLock()
+	d := e.dispatcher
+	e.mu.RUnlock()
+	d.DeliverLinkPacket(protocol, pkt)
 	e.dispatchGate.Leave()
 }
 
@@ -76,12 +83,16 @@ func (e *Endpoint) DeliverLinkPacket(protocol tcpip.NetworkProtocolNumber, pkt s
 // registers with the lower endpoint as its dispatcher so that "e" is called
 // for inbound packets.
 func (e *Endpoint) Attach(dispatcher stack.NetworkDispatcher) {
+	e.mu.Lock()
 	e.dispatcher = dispatcher
+	e.mu.Unlock()
 	e.lower.Attach(e)
 }
 
 // IsAttached implements stack.LinkEndpoint.IsAttached.
 func (e *Endpoint) IsAttached() bool {
+	e.mu.RLock()
+	defer e.mu.RUnlock()
 	return e.dispatcher != nil
 }
 

pkg/tcpip/link/xdp/endpoint.go

@@ -54,6 +54,8 @@ type endpoint struct {
 	// its end of the communication pipe.
 	closed func(tcpip.Error)
 
+	mu sync.RWMutex
+	// +checkloks:mu
 	networkDispatcher stack.NetworkDispatcher
 
 	// wg keeps track of running goroutines.
@@ -169,6 +171,8 @@ func New(opts *Options) (stack.LinkEndpoint, error) {
 //
 // Attach implements stack.LinkEndpoint.Attach.
 func (ep *endpoint) Attach(networkDispatcher stack.NetworkDispatcher) {
+	ep.mu.Lock()
+	defer ep.mu.Unlock()
 	// nil means the NIC is being removed.
 	if networkDispatcher == nil && ep.IsAttached() {
 		ep.stopFD.Stop()
@@ -199,6 +203,8 @@ func (ep *endpoint) Attach(networkDispatcher stack.NetworkDispatcher) {
 
 // IsAttached implements stack.LinkEndpoint.IsAttached.
 func (ep *endpoint) IsAttached() bool {
+	ep.mu.RLock()
+	defer ep.mu.RUnlock()
 	return ep.networkDispatcher != nil
 }
 
@@ -341,6 +347,9 @@ func (ep *endpoint) dispatch() (bool, tcpip.Error) {
 			ep.control.UMEM.Unlock()
 
 			// Process each packet.
+			ep.mu.RLock()
+			d := ep.networkDispatcher
+			ep.mu.RUnlock()
 			for i := uint32(0); i < nReceived; i++ {
 				view := views[i]
 				data := view.AsSlice()
@@ -355,7 +364,7 @@ func (ep *endpoint) dispatch() (bool, tcpip.Error) {
 				if _, ok := pkt.LinkHeader().Consume(header.EthernetMinimumSize); !ok {
 					panic(fmt.Sprintf("LinkHeader().Consume(%d) must succeed", header.EthernetMinimumSize))
 				}
-				ep.networkDispatcher.DeliverNetworkPacket(netProto, pkt)
+				d.DeliverNetworkPacket(netProto, pkt)
 				pkt.DecRef()
 			}
 			// Tell the kernel that we're done with these