endpoint.go

     1  // Copyright 2023 The Go Authors. All rights reserved.
     2  // Use of this source code is governed by a BSD-style
     3  // license that can be found in the LICENSE file.
     4  
     5  package quic
     6  
     7  import (
     8  	"context"
     9  	"crypto/rand"
    10  	"errors"
    11  	"fmt"
    12  	"net"
    13  	"net/netip"
    14  	"sync"
    15  	"sync/atomic"
    16  	"time"
    17  )
    18  
    19  // An Endpoint handles QUIC traffic on a network address.
    20  // It can accept inbound connections or create outbound ones.
    21  //
    22  // Multiple goroutines may invoke methods on an Endpoint simultaneously.
    23  type Endpoint struct {
    24  	listenConfig *Config
    25  	packetConn   packetConn
    26  	testHooks    endpointTestHooks
    27  	resetGen     statelessResetTokenGenerator
    28  	retry        retryState
    29  
    30  	acceptQueue queue[*Conn] // new inbound connections
    31  	connsMap    connsMap     // only accessed by the listen loop
    32  
    33  	connsMu sync.Mutex
    34  	conns   map[*Conn]struct{}
    35  	closing bool          // set when Close is called
    36  	closec  chan struct{} // closed when the listen loop exits
    37  }
    38  
    39  type endpointTestHooks interface {
    40  	newConn(c *Conn, cids newServerConnIDs)
    41  }
    42  
    43  // A packetConn is the interface to sending and receiving UDP packets.
    44  type packetConn interface {
    45  	Close() error
    46  	LocalAddr() netip.AddrPort
    47  	Read(f func(*datagram)) error
    48  	Write(datagram) error
    49  }
    50  
    51  // Listen listens on a local network address.
    52  //
    53  // The config is used to for connections accepted by the endpoint.
    54  // If the config is nil, the endpoint will not accept connections.
    55  func Listen(network, address string, listenConfig *Config) (*Endpoint, error) {
    56  	if listenConfig != nil && listenConfig.TLSConfig == nil {
    57  		return nil, errors.New("TLSConfig is not set")
    58  	}
    59  	a, err := net.ResolveUDPAddr(network, address)
    60  	if err != nil {
    61  		return nil, err
    62  	}
    63  	udpConn, err := net.ListenUDP(network, a)
    64  	if err != nil {
    65  		return nil, err
    66  	}
    67  	pc, err := newNetUDPConn(udpConn)
    68  	if err != nil {
    69  		return nil, err
    70  	}
    71  	return newEndpoint(pc, listenConfig, nil)
    72  }
    73  
    74  // NewEndpoint creates an endpoint using a net.PacketConn as the underlying transport.
    75  //
    76  // If the PacketConn is not a *net.UDPConn, the endpoint may be slower and lack
    77  // access to some features of the network.
    78  func NewEndpoint(conn net.PacketConn, config *Config) (*Endpoint, error) {
    79  	var pc packetConn
    80  	var err error
    81  	switch conn := conn.(type) {
    82  	case *net.UDPConn:
    83  		pc, err = newNetUDPConn(conn)
    84  	default:
    85  		pc, err = newNetPacketConn(conn)
    86  	}
    87  	if err != nil {
    88  		return nil, err
    89  	}
    90  	return newEndpoint(pc, config, nil)
    91  }
    92  
    93  func newEndpoint(pc packetConn, config *Config, hooks endpointTestHooks) (*Endpoint, error) {
    94  	e := &Endpoint{
    95  		listenConfig: config,
    96  		packetConn:   pc,
    97  		testHooks:    hooks,
    98  		conns:        make(map[*Conn]struct{}),
    99  		acceptQueue:  newQueue[*Conn](),
   100  		closec:       make(chan struct{}),
   101  	}
   102  	var statelessResetKey [32]byte
   103  	if config != nil {
   104  		statelessResetKey = config.StatelessResetKey
   105  	}
   106  	e.resetGen.init(statelessResetKey)
   107  	e.connsMap.init()
   108  	if config != nil && config.RequireAddressValidation {
   109  		if err := e.retry.init(); err != nil {
   110  			return nil, err
   111  		}
   112  	}
   113  	go e.listen()
   114  	return e, nil
   115  }
   116  
   117  // LocalAddr returns the local network address.
   118  func (e *Endpoint) LocalAddr() netip.AddrPort {
   119  	return e.packetConn.LocalAddr()
   120  }
   121  
   122  // Close closes the Endpoint.
   123  // Any blocked operations on the Endpoint or associated Conns and Stream will be unblocked
   124  // and return errors.
   125  //
   126  // Close aborts every open connection.
   127  // Data in stream read and write buffers is discarded.
   128  // It waits for the peers of any open connection to acknowledge the connection has been closed.
   129  func (e *Endpoint) Close(ctx context.Context) error {
   130  	e.acceptQueue.close(errors.New("endpoint closed"))
   131  
   132  	// It isn't safe to call Conn.Abort or conn.exit with connsMu held,
   133  	// so copy the list of conns.
   134  	var conns []*Conn
   135  	e.connsMu.Lock()
   136  	if !e.closing {
   137  		e.closing = true // setting e.closing prevents new conns from being created
   138  		for c := range e.conns {
   139  			conns = append(conns, c)
   140  		}
   141  		if len(e.conns) == 0 {
   142  			e.packetConn.Close()
   143  		}
   144  	}
   145  	e.connsMu.Unlock()
   146  
   147  	for _, c := range conns {
   148  		c.Abort(localTransportError{code: errNo})
   149  	}
   150  	select {
   151  	case <-e.closec:
   152  	case <-ctx.Done():
   153  	}
   154  	for _, c := range conns {
   155  		c.exit()
   156  	}
   157  	return ctx.Err() // nil if context hasn't expired
   158  }
   159  
   160  // Accept waits for and returns the next connection.
   161  func (e *Endpoint) Accept(ctx context.Context) (*Conn, error) {
   162  	return e.acceptQueue.get(ctx)
   163  }
   164  
   165  // Dial creates and returns a connection to a network address.
   166  // The config cannot be nil.
   167  func (e *Endpoint) Dial(ctx context.Context, network, address string, config *Config) (*Conn, error) {
   168  	u, err := net.ResolveUDPAddr(network, address)
   169  	if err != nil {
   170  		return nil, err
   171  	}
   172  	addr := u.AddrPort()
   173  	addr = netip.AddrPortFrom(addr.Addr().Unmap(), addr.Port())
   174  	c, err := e.newConn(time.Now(), config, clientSide, newServerConnIDs{}, address, addr)
   175  	if err != nil {
   176  		return nil, err
   177  	}
   178  	if err := c.waitReady(ctx); err != nil {
   179  		c.Abort(nil)
   180  		return nil, err
   181  	}
   182  	return c, nil
   183  }
   184  
   185  func (e *Endpoint) newConn(now time.Time, config *Config, side connSide, cids newServerConnIDs, peerHostname string, peerAddr netip.AddrPort) (*Conn, error) {
   186  	e.connsMu.Lock()
   187  	defer e.connsMu.Unlock()
   188  	if e.closing {
   189  		return nil, errors.New("endpoint closed")
   190  	}
   191  	c, err := newConn(now, side, cids, peerHostname, peerAddr, config, e)
   192  	if err != nil {
   193  		return nil, err
   194  	}
   195  	e.conns[c] = struct{}{}
   196  	return c, nil
   197  }
   198  
   199  // serverConnEstablished is called by a conn when the handshake completes
   200  // for an inbound (serverSide) connection.
   201  func (e *Endpoint) serverConnEstablished(c *Conn) {
   202  	e.acceptQueue.put(c)
   203  }
   204  
   205  // connDrained is called by a conn when it leaves the draining state,
   206  // either when the peer acknowledges connection closure or the drain timeout expires.
   207  func (e *Endpoint) connDrained(c *Conn) {
   208  	var cids [][]byte
   209  	for i := range c.connIDState.local {
   210  		cids = append(cids, c.connIDState.local[i].cid)
   211  	}
   212  	var tokens []statelessResetToken
   213  	for i := range c.connIDState.remote {
   214  		tokens = append(tokens, c.connIDState.remote[i].resetToken)
   215  	}
   216  	e.connsMap.updateConnIDs(func(conns *connsMap) {
   217  		for _, cid := range cids {
   218  			conns.retireConnID(c, cid)
   219  		}
   220  		for _, token := range tokens {
   221  			conns.retireResetToken(c, token)
   222  		}
   223  	})
   224  	e.connsMu.Lock()
   225  	defer e.connsMu.Unlock()
   226  	delete(e.conns, c)
   227  	if e.closing && len(e.conns) == 0 {
   228  		e.packetConn.Close()
   229  	}
   230  }
   231  
   232  func (e *Endpoint) listen() {
   233  	defer close(e.closec)
   234  	err := e.packetConn.Read(func(m *datagram) {
   235  		if e.connsMap.updateNeeded.Load() {
   236  			e.connsMap.applyUpdates()
   237  		}
   238  		e.handleDatagram(m)
   239  	})
   240  	e.acceptQueue.close(fmt.Errorf("packet connection closed: %v", err))
   241  }
   242  
   243  func (e *Endpoint) handleDatagram(m *datagram) {
   244  	dstConnID, ok := dstConnIDForDatagram(m.b)
   245  	if !ok {
   246  		m.recycle()
   247  		return
   248  	}
   249  	c := e.connsMap.byConnID[string(dstConnID)]
   250  	if c == nil {
   251  		// TODO: Move this branch into a separate goroutine to avoid blocking
   252  		// the endpoint while processing packets.
   253  		e.handleUnknownDestinationDatagram(m)
   254  		return
   255  	}
   256  
   257  	// TODO: This can block the endpoint while waiting for the conn to accept the dgram.
   258  	// Think about buffering between the receive loop and the conn.
   259  	c.sendMsg(m)
   260  }
   261  
   262  func (e *Endpoint) handleUnknownDestinationDatagram(m *datagram) {
   263  	defer func() {
   264  		if m != nil {
   265  			m.recycle()
   266  		}
   267  	}()
   268  	const minimumValidPacketSize = 21
   269  	if len(m.b) < minimumValidPacketSize {
   270  		return
   271  	}
   272  	now := time.Now()
   273  	// Check to see if this is a stateless reset.
   274  	var token statelessResetToken
   275  	copy(token[:], m.b[len(m.b)-len(token):])
   276  	if c := e.connsMap.byResetToken[token]; c != nil {
   277  		c.sendMsg(func(now time.Time, c *Conn) {
   278  			c.handleStatelessReset(now, token)
   279  		})
   280  		return
   281  	}
   282  	// If this is a 1-RTT packet, there's nothing productive we can do with it.
   283  	// Send a stateless reset if possible.
   284  	if !isLongHeader(m.b[0]) {
   285  		e.maybeSendStatelessReset(m.b, m.peerAddr)
   286  		return
   287  	}
   288  	p, ok := parseGenericLongHeaderPacket(m.b)
   289  	if !ok || len(m.b) < paddedInitialDatagramSize {
   290  		return
   291  	}
   292  	switch p.version {
   293  	case quicVersion1:
   294  	case 0:
   295  		// Version Negotiation for an unknown connection.
   296  		return
   297  	default:
   298  		// Unknown version.
   299  		e.sendVersionNegotiation(p, m.peerAddr)
   300  		return
   301  	}
   302  	if getPacketType(m.b) != packetTypeInitial {
   303  		// This packet isn't trying to create a new connection.
   304  		// It might be associated with some connection we've lost state for.
   305  		// We are technically permitted to send a stateless reset for
   306  		// a long-header packet, but this isn't generally useful. See:
   307  		// https://www.rfc-editor.org/rfc/rfc9000#section-10.3-16
   308  		return
   309  	}
   310  	if e.listenConfig == nil {
   311  		// We are not configured to accept connections.
   312  		return
   313  	}
   314  	if len(p.srcConnID) > maxConnIDLen || len(p.dstConnID) > maxConnIDLen {
   315  		// Enforce QUICv1 connection ID length limits.
   316  		// https://www.rfc-editor.org/rfc/rfc9000.html#section-17.2-3.12.1
   317  		// https://www.rfc-editor.org/rfc/rfc9000.html#section-17.2-3.16.1
   318  		return
   319  	}
   320  	cids := newServerConnIDs{
   321  		srcConnID: p.srcConnID,
   322  		dstConnID: p.dstConnID,
   323  	}
   324  	if e.listenConfig.RequireAddressValidation {
   325  		var ok bool
   326  		cids.retrySrcConnID = p.dstConnID
   327  		cids.originalDstConnID, ok = e.validateInitialAddress(now, p, m.peerAddr)
   328  		if !ok {
   329  			return
   330  		}
   331  	} else {
   332  		cids.originalDstConnID = p.dstConnID
   333  	}
   334  	var err error
   335  	c, err := e.newConn(now, e.listenConfig, serverSide, cids, "", m.peerAddr)
   336  	if err != nil {
   337  		// The accept queue is probably full.
   338  		// We could send a CONNECTION_CLOSE to the peer to reject the connection.
   339  		// Currently, we just drop the datagram.
   340  		// https://www.rfc-editor.org/rfc/rfc9000.html#section-5.2.2-5
   341  		return
   342  	}
   343  	c.sendMsg(m)
   344  	m = nil // don't recycle, sendMsg takes ownership
   345  }
   346  
   347  func (e *Endpoint) maybeSendStatelessReset(b []byte, peerAddr netip.AddrPort) {
   348  	if !e.resetGen.canReset {
   349  		// Config.StatelessResetKey isn't set, so we don't send stateless resets.
   350  		return
   351  	}
   352  	// The smallest possible valid packet a peer can send us is:
   353  	//   1 byte of header
   354  	//   connIDLen bytes of destination connection ID
   355  	//   1 byte of packet number
   356  	//   1 byte of payload
   357  	//   16 bytes AEAD expansion
   358  	if len(b) < 1+connIDLen+1+1+16 {
   359  		return
   360  	}
   361  	// TODO: Rate limit stateless resets.
   362  	cid := b[1:][:connIDLen]
   363  	token := e.resetGen.tokenForConnID(cid)
   364  	// We want to generate a stateless reset that is as short as possible,
   365  	// but long enough to be difficult to distinguish from a 1-RTT packet.
   366  	//
   367  	// The minimal 1-RTT packet is:
   368  	//   1 byte of header
   369  	//   0-20 bytes of destination connection ID
   370  	//   1-4 bytes of packet number
   371  	//   1 byte of payload
   372  	//   16 bytes AEAD expansion
   373  	//
   374  	// Assuming the maximum possible connection ID and packet number size,
   375  	// this gives 1 + 20 + 4 + 1 + 16 = 42 bytes.
   376  	//
   377  	// We also must generate a stateless reset that is shorter than the datagram
   378  	// we are responding to, in order to ensure that reset loops terminate.
   379  	//
   380  	// See: https://www.rfc-editor.org/rfc/rfc9000#section-10.3
   381  	size := min(len(b)-1, 42)
   382  	// Reuse the input buffer for generating the stateless reset.
   383  	b = b[:size]
   384  	rand.Read(b[:len(b)-statelessResetTokenLen])
   385  	b[0] &^= headerFormLong // clear long header bit
   386  	b[0] |= fixedBit        // set fixed bit
   387  	copy(b[len(b)-statelessResetTokenLen:], token[:])
   388  	e.sendDatagram(datagram{
   389  		b:        b,
   390  		peerAddr: peerAddr,
   391  	})
   392  }
   393  
   394  func (e *Endpoint) sendVersionNegotiation(p genericLongPacket, peerAddr netip.AddrPort) {
   395  	m := newDatagram()
   396  	m.b = appendVersionNegotiation(m.b[:0], p.srcConnID, p.dstConnID, quicVersion1)
   397  	m.peerAddr = peerAddr
   398  	e.sendDatagram(*m)
   399  	m.recycle()
   400  }
   401  
   402  func (e *Endpoint) sendConnectionClose(in genericLongPacket, peerAddr netip.AddrPort, code transportError) {
   403  	keys := initialKeys(in.dstConnID, serverSide)
   404  	var w packetWriter
   405  	p := longPacket{
   406  		ptype:     packetTypeInitial,
   407  		version:   quicVersion1,
   408  		num:       0,
   409  		dstConnID: in.srcConnID,
   410  		srcConnID: in.dstConnID,
   411  	}
   412  	const pnumMaxAcked = 0
   413  	w.reset(paddedInitialDatagramSize)
   414  	w.startProtectedLongHeaderPacket(pnumMaxAcked, p)
   415  	w.appendConnectionCloseTransportFrame(code, 0, "")
   416  	w.finishProtectedLongHeaderPacket(pnumMaxAcked, keys.w, p)
   417  	buf := w.datagram()
   418  	if len(buf) == 0 {
   419  		return
   420  	}
   421  	e.sendDatagram(datagram{
   422  		b:        buf,
   423  		peerAddr: peerAddr,
   424  	})
   425  }
   426  
   427  func (e *Endpoint) sendDatagram(dgram datagram) error {
   428  	return e.packetConn.Write(dgram)
   429  }
   430  
   431  // A connsMap is an endpoint's mapping of conn ids and reset tokens to conns.
   432  type connsMap struct {
   433  	byConnID     map[string]*Conn
   434  	byResetToken map[statelessResetToken]*Conn
   435  
   436  	updateMu     sync.Mutex
   437  	updateNeeded atomic.Bool
   438  	updates      []func(*connsMap)
   439  }
   440  
   441  func (m *connsMap) init() {
   442  	m.byConnID = map[string]*Conn{}
   443  	m.byResetToken = map[statelessResetToken]*Conn{}
   444  }
   445  
   446  func (m *connsMap) addConnID(c *Conn, cid []byte) {
   447  	m.byConnID[string(cid)] = c
   448  }
   449  
   450  func (m *connsMap) retireConnID(c *Conn, cid []byte) {
   451  	delete(m.byConnID, string(cid))
   452  }
   453  
   454  func (m *connsMap) addResetToken(c *Conn, token statelessResetToken) {
   455  	m.byResetToken[token] = c
   456  }
   457  
   458  func (m *connsMap) retireResetToken(c *Conn, token statelessResetToken) {
   459  	delete(m.byResetToken, token)
   460  }
   461  
   462  func (m *connsMap) updateConnIDs(f func(*connsMap)) {
   463  	m.updateMu.Lock()
   464  	defer m.updateMu.Unlock()
   465  	m.updates = append(m.updates, f)
   466  	m.updateNeeded.Store(true)
   467  }
   468  
   469  // applyUpdates is called by the datagram receive loop to update its connection ID map.
   470  func (m *connsMap) applyUpdates() {
   471  	m.updateMu.Lock()
   472  	defer m.updateMu.Unlock()
   473  	for _, f := range m.updates {
   474  		f(m)
   475  	}
   476  	clear(m.updates)
   477  	m.updates = m.updates[:0]
   478  	m.updateNeeded.Store(false)
   479  }
   480
View as plain text