lib/connections: Update pfilter to pick up bugfix/oob stuff, support OOB connections (fixes #7636) (#7654)

lib/connections/quic_listen.go

@@ -83,18 +83,33 @@ func (t *quicListener) OnExternalAddressChanged(address *stun.Host, via string)
 func (t *quicListener) serve(ctx context.Context) error {
 	network := strings.ReplaceAll(t.uri.Scheme, "quic", "udp")
 
-	packetConn, err := net.ListenPacket(network, t.uri.Host)
+	udpAddr, err := net.ResolveUDPAddr(network, t.uri.Host)
 	if err != nil {
 		l.Infoln("Listen (BEP/quic):", err)
 		return err
 	}
-	defer packetConn.Close()
 
-	svc, conn := stun.New(t.cfg, t, packetConn)
+	udpConn, err := net.ListenUDP(network, udpAddr)
+	if err != nil {
+		l.Infoln("Listen (BEP/quic):", err)
+		return err
+	}
+	defer func() { _ = udpConn.Close() }()
+
+	svc, conn := stun.New(t.cfg, t, udpConn)
 	defer conn.Close()
-	wrapped := &stunConnQUICWrapper{
+
+	quicWrapper := quicWrapper{
 		PacketConn: conn,
-		underlying: packetConn.(*net.UDPConn),
+		underlying: udpConn,
+	}
+	var wrapped net.PacketConn = &quicWrapper
+
+	if oobConn, ok := conn.(oobConn); ok {
+		l.Debugf("wrapping in oob conn")
+		wrapped = &oobConnWrapper{
+			quicWrapper, oobConn,
+		}
 	}
 
 	go svc.Serve(ctx)
@@ -112,11 +127,11 @@ func (t *quicListener) serve(ctx context.Context) error {
 	t.notifyAddressesChanged(t)
 	defer t.clearAddresses(t)
 
-	l.Infof("QUIC listener (%v) starting", packetConn.LocalAddr())
-	defer l.Infof("QUIC listener (%v) shutting down", packetConn.LocalAddr())
+	l.Infof("QUIC listener (%v) starting", udpConn.LocalAddr())
+	defer l.Infof("QUIC listener (%v) shutting down", udpConn.LocalAddr())
 
 	t.mut.Lock()
-	t.laddr = packetConn.LocalAddr()
+	t.laddr = udpConn.LocalAddr()
 	t.mut.Unlock()
 	defer func() {
 		t.mut.Lock()
@@ -233,18 +248,32 @@ func (quicListenerFactory) Enabled(cfg config.Configuration) bool {
 	return true
 }
 
-// stunConnQUICWrapper provides methods used by quic.
-type stunConnQUICWrapper struct {
+// quicWrapper provides methods used by quic
+// https://github.com/lucas-clemente/quic-go/blob/master/packet_handler_map.go#L85
+type quicWrapper struct {
 	net.PacketConn
 	underlying *net.UDPConn
 }
 
-func (s *stunConnQUICWrapper) SetReadBuffer(size int) error {
-	// https://github.com/lucas-clemente/quic-go/blob/master/packet_handler_map.go#L85
+// SetReadBuffer is required by QUIC
+func (s *quicWrapper) SetReadBuffer(size int) error {
 	return s.underlying.SetReadBuffer(size)
 }
 
-func (s *stunConnQUICWrapper) SyscallConn() (syscall.RawConn, error) {
-	// https://github.com/lucas-clemente/quic-go/blob/84e03e59760ceee37359688871bb0688fcc4e98f/conn_windows.go#L18
+// SyscallConn is required by QUIC
+func (s *quicWrapper) SyscallConn() (syscall.RawConn, error) {
 	return s.underlying.SyscallConn()
 }
+
+// oobConn is used to assert that stun package returned a net.PacketConn that implements this interface.
+// If it does, we then wrap quicWrapper in oobConnWrapper, to expose those methods to QUIC package.
+type oobConn interface {
+	ReadMsgUDP(b, oob []byte) (n, oobn, flags int, addr *net.UDPAddr, err error)
+	WriteMsgUDP(b, oob []byte, addr *net.UDPAddr) (n, oobn int, err error)
+}
+
+// See: https://pkg.go.dev/github.com/lucas-clemente/quic-go#OOBCapablePacketConn
+type oobConnWrapper struct {
+	quicWrapper
+	oobConn
+}

lib/stun/stun.go

@@ -38,17 +38,35 @@ const (
 	NATSymmetricUDPFirewall = stun.NATSymmetricUDPFirewall
 )
 
-type writeTrackingPacketConn struct {
+type writeTrackingUdpConn struct {
 	lastWrite int64 // atomic, must remain 64-bit aligned
-	net.PacketConn
+	// Needs to be UDPConn not PacketConn, as pfilter checks for WriteMsgUDP/ReadMsgUDP
+	// and even if we embed UDPConn here, in place of a PacketConn, seems the interface
+	// check fails.
+	*net.UDPConn
 }
 
-func (c *writeTrackingPacketConn) WriteTo(p []byte, addr net.Addr) (n int, err error) {
+func (c *writeTrackingUdpConn) WriteTo(p []byte, addr net.Addr) (n int, err error) {
 	atomic.StoreInt64(&c.lastWrite, time.Now().Unix())
-	return c.PacketConn.WriteTo(p, addr)
+	return c.UDPConn.WriteTo(p, addr)
 }
 
-func (c *writeTrackingPacketConn) getLastWrite() time.Time {
+func (c *writeTrackingUdpConn) WriteMsgUDP(b, oob []byte, addr *net.UDPAddr) (n, oobn int, err error) {
+	atomic.StoreInt64(&c.lastWrite, time.Now().Unix())
+	return c.UDPConn.WriteMsgUDP(b, oob, addr)
+}
+
+func (c *writeTrackingUdpConn) WriteToUDP(b []byte, addr *net.UDPAddr) (int, error) {
+	atomic.StoreInt64(&c.lastWrite, time.Now().Unix())
+	return c.UDPConn.WriteToUDP(b, addr)
+}
+
+func (c *writeTrackingUdpConn) Write(b []byte) (int, error) {
+	atomic.StoreInt64(&c.lastWrite, time.Now().Unix())
+	return c.UDPConn.Write(b)
+}
+
+func (c *writeTrackingUdpConn) getLastWrite() time.Time {
 	unix := atomic.LoadInt64(&c.lastWrite)
 	return time.Unix(unix, 0)
 }
@@ -65,18 +83,18 @@ type Service struct {
 	stunConn   net.PacketConn
 	client     *stun.Client
 
-	writeTrackingPacketConn *writeTrackingPacketConn
+	writeTrackingUdpConn *writeTrackingUdpConn
 
 	natType NATType
 	addr    *Host
 }
 
-func New(cfg config.Wrapper, subscriber Subscriber, conn net.PacketConn) (*Service, net.PacketConn) {
+func New(cfg config.Wrapper, subscriber Subscriber, conn *net.UDPConn) (*Service, net.PacketConn) {
 	// Wrap the original connection to track writes on it
-	writeTrackingPacketConn := &writeTrackingPacketConn{lastWrite: 0, PacketConn: conn}
+	writeTrackingUdpConn := &writeTrackingUdpConn{lastWrite: 0, UDPConn: conn}
 
 	// Wrap it in a filter and split it up, so that stun packets arrive on stun conn, others arrive on the data conn
-	filterConn := pfilter.NewPacketFilter(writeTrackingPacketConn)
+	filterConn := pfilter.NewPacketFilter(writeTrackingUdpConn)
 	otherDataConn := filterConn.NewConn(otherDataPriority, nil)
 	stunConn := filterConn.NewConn(stunFilterPriority, &stunFilter{
 		ids: make(map[string]time.Time),
@@ -97,7 +115,7 @@ func New(cfg config.Wrapper, subscriber Subscriber, conn net.PacketConn) (*Servi
 		stunConn:   stunConn,
 		client:     client,
 
-		writeTrackingPacketConn: writeTrackingPacketConn,
+		writeTrackingUdpConn: writeTrackingUdpConn,
 
 		natType: NATUnknown,
 		addr:    nil,
@@ -241,7 +259,7 @@ func (s *Service) stunKeepAlive(ctx context.Context, addr string, extAddr *Host)
 		}
 
 		// Adjust the keepalives to fire only nextSleep after last write.
-		lastWrite := s.writeTrackingPacketConn.getLastWrite()
+		lastWrite := s.writeTrackingUdpConn.getLastWrite()
 		minSleep := time.Duration(s.cfg.Options().StunKeepaliveMinS) * time.Second
 		if nextSleep < minSleep {
 			nextSleep = minSleep
@@ -270,7 +288,7 @@ func (s *Service) stunKeepAlive(ctx context.Context, addr string, extAddr *Host)
 		}
 
 		// Check if any writes happened while we were sleeping, if they did, sleep again
-		lastWrite = s.writeTrackingPacketConn.getLastWrite()
+		lastWrite = s.writeTrackingUdpConn.getLastWrite()
 		if gap := time.Since(lastWrite); gap < nextSleep {
 			l.Debugf("%s stun last write gap less than next sleep: %s < %s. Will try later", s, gap, nextSleep)
 			goto tryLater