MM-39315: Bump dependencies (#19039)

Note: We keep splitio/go-client untouched
because the dependency is broken.

See https://github.com/mattermost/mattermost-server/pull/18604

```release-note
NONE
```

vendor/github.com/hashicorp/memberlist/config.go

@@ -21,6 +21,17 @@ type Config struct {
 	// make a NetTransport using BindAddr and BindPort from this structure.
 	Transport Transport
 
+	// Label is an optional set of bytes to include on the outside of each
+	// packet and stream.
+	//
+	// If gossip encryption is enabled and this is set it is treated as GCM
+	// authenticated data.
+	Label string
+
+	// SkipInboundLabelCheck skips the check that inbound packets and gossip
+	// streams need to be label prefixed.
+	SkipInboundLabelCheck bool
+
 	// Configuration related to what address to bind to and ports to
 	// listen on. The port is used for both UDP and TCP gossip. It is
 	// assumed other nodes are running on this port, but they do not need

vendor/github.com/hashicorp/memberlist/label.go

@@ -0,0 +1,178 @@
+package memberlist
+
+import (
+	"bufio"
+	"fmt"
+	"io"
+	"net"
+)
+
+// General approach is to prefix all packets and streams with the same structure:
+//
+// magic type byte (244): uint8
+// length of label name:  uint8 (because labels can't be longer than 255 bytes)
+// label name:            []uint8
+
+// LabelMaxSize is the maximum length of a packet or stream label.
+const LabelMaxSize = 255
+
+// AddLabelHeaderToPacket prefixes outgoing packets with the correct header if
+// the label is not empty.
+func AddLabelHeaderToPacket(buf []byte, label string) ([]byte, error) {
+	if label == "" {
+		return buf, nil
+	}
+	if len(label) > LabelMaxSize {
+		return nil, fmt.Errorf("label %q is too long", label)
+	}
+
+	return makeLabelHeader(label, buf), nil
+}
+
+// RemoveLabelHeaderFromPacket removes any label header from the provided
+// packet and returns it along with the remaining packet contents.
+func RemoveLabelHeaderFromPacket(buf []byte) (newBuf []byte, label string, err error) {
+	if len(buf) == 0 {
+		return buf, "", nil // can't possibly be labeled
+	}
+
+	// [type:byte] [size:byte] [size bytes]
+
+	msgType := messageType(buf[0])
+	if msgType != hasLabelMsg {
+		return buf, "", nil
+	}
+
+	if len(buf) < 2 {
+		return nil, "", fmt.Errorf("cannot decode label; packet has been truncated")
+	}
+
+	size := int(buf[1])
+	if size < 1 {
+		return nil, "", fmt.Errorf("label header cannot be empty when present")
+	}
+
+	if len(buf) < 2+size {
+		return nil, "", fmt.Errorf("cannot decode label; packet has been truncated")
+	}
+
+	label = string(buf[2 : 2+size])
+	newBuf = buf[2+size:]
+
+	return newBuf, label, nil
+}
+
+// AddLabelHeaderToStream prefixes outgoing streams with the correct header if
+// the label is not empty.
+func AddLabelHeaderToStream(conn net.Conn, label string) error {
+	if label == "" {
+		return nil
+	}
+	if len(label) > LabelMaxSize {
+		return fmt.Errorf("label %q is too long", label)
+	}
+
+	header := makeLabelHeader(label, nil)
+
+	_, err := conn.Write(header)
+	return err
+}
+
+// RemoveLabelHeaderFromStream removes any label header from the beginning of
+// the stream if present and returns it along with an updated conn with that
+// header removed.
+//
+// Note that on error it is the caller's responsibility to close the
+// connection.
+func RemoveLabelHeaderFromStream(conn net.Conn) (net.Conn, string, error) {
+	br := bufio.NewReader(conn)
+
+	// First check for the type byte.
+	peeked, err := br.Peek(1)
+	if err != nil {
+		if err == io.EOF {
+			// It is safe to return the original net.Conn at this point because
+			// it never contained any data in the first place so we don't have
+			// to splice the buffer into the conn because both are empty.
+			return conn, "", nil
+		}
+		return nil, "", err
+	}
+
+	msgType := messageType(peeked[0])
+	if msgType != hasLabelMsg {
+		conn, err = newPeekedConnFromBufferedReader(conn, br, 0)
+		return conn, "", err
+	}
+
+	// We are guaranteed to get a size byte as well.
+	peeked, err = br.Peek(2)
+	if err != nil {
+		if err == io.EOF {
+			return nil, "", fmt.Errorf("cannot decode label; stream has been truncated")
+		}
+		return nil, "", err
+	}
+
+	size := int(peeked[1])
+	if size < 1 {
+		return nil, "", fmt.Errorf("label header cannot be empty when present")
+	}
+	// NOTE: we don't have to check this against LabelMaxSize because a byte
+	// already has a max value of 255.
+
+	// Once we know the size we can peek the label as well. Note that since we
+	// are using the default bufio.Reader size of 4096, the entire label header
+	// fits in the initial buffer fill so this should be free.
+	peeked, err = br.Peek(2 + size)
+	if err != nil {
+		if err == io.EOF {
+			return nil, "", fmt.Errorf("cannot decode label; stream has been truncated")
+		}
+		return nil, "", err
+	}
+
+	label := string(peeked[2 : 2+size])
+
+	conn, err = newPeekedConnFromBufferedReader(conn, br, 2+size)
+	if err != nil {
+		return nil, "", err
+	}
+
+	return conn, label, nil
+}
+
+// newPeekedConnFromBufferedReader will splice the buffer contents after the
+// offset into the provided net.Conn and return the result so that the rest of
+// the buffer contents are returned first when reading from the returned
+// peekedConn before moving on to the unbuffered conn contents.
+func newPeekedConnFromBufferedReader(conn net.Conn, br *bufio.Reader, offset int) (*peekedConn, error) {
+	// Extract any of the readahead buffer.
+	peeked, err := br.Peek(br.Buffered())
+	if err != nil {
+		return nil, err
+	}
+
+	return &peekedConn{
+		Peeked: peeked[offset:],
+		Conn:   conn,
+	}, nil
+}
+
+func makeLabelHeader(label string, rest []byte) []byte {
+	newBuf := make([]byte, 2, 2+len(label)+len(rest))
+	newBuf[0] = byte(hasLabelMsg)
+	newBuf[1] = byte(len(label))
+	newBuf = append(newBuf, []byte(label)...)
+	if len(rest) > 0 {
+		newBuf = append(newBuf, []byte(rest)...)
+	}
+	return newBuf
+}
+
+func labelOverhead(label string) int {
+	if label == "" {
+		return 0
+	}
+	return 2 + len(label)
+}

vendor/github.com/hashicorp/memberlist/memberlist.go

@@ -187,6 +187,17 @@ func newMemberlist(conf *Config) (*Memberlist, error) {
 		nodeAwareTransport = &shimNodeAwareTransport{transport}
 	}
 
+	if len(conf.Label) > LabelMaxSize {
+		return nil, fmt.Errorf("could not use %q as a label: too long", conf.Label)
+	}
+
+	if conf.Label != "" {
+		nodeAwareTransport = &labelWrappedTransport{
+			label:              conf.Label,
+			NodeAwareTransport: nodeAwareTransport,
+		}
+	}
+
 	m := &Memberlist{
 		config:               conf,
 		shutdownCh:           make(chan struct{}),
@@ -262,7 +273,7 @@ func (m *Memberlist) Join(existing []string) (int, error) {
 			hp := joinHostPort(addr.ip.String(), addr.port)
 			a := Address{Addr: hp, Name: addr.nodeName}
 			if err := m.pushPullNode(a, true); err != nil {
-				err = fmt.Errorf("Failed to join %s: %v", addr.ip, err)
+				err = fmt.Errorf("Failed to join %s: %v", a.Addr, err)
 				errs = multierror.Append(errs, err)
 				m.logger.Printf("[DEBUG] memberlist: %v", err)
 				continue

vendor/github.com/hashicorp/memberlist/net.go

@@ -42,6 +42,9 @@ const (
 type messageType uint8
 
 // The list of available message types.
+//
+// WARNING: ONLY APPEND TO THIS LIST! The numeric values are part of the
+// protocol itself.
 const (
 	pingMsg messageType = iota
 	indirectPingMsg
@@ -59,6 +62,13 @@ const (
 	errMsg
 )
 
+const (
+	// hasLabelMsg has a deliberately high value so that you can disambiguate
+	// it from the encryptionVersion header which is either 0/1 right now and
+	// also any of the existing messageTypes
+	hasLabelMsg messageType = 244
+)
+
 // compressionType is used to specify the compression algorithm
 type compressionType uint8
 
@@ -226,7 +236,32 @@ func (m *Memberlist) handleConn(conn net.Conn) {
 	metrics.IncrCounter([]string{"memberlist", "tcp", "accept"}, 1)
 
 	conn.SetDeadline(time.Now().Add(m.config.TCPTimeout))
-	msgType, bufConn, dec, err := m.readStream(conn)
+
+	var (
+		streamLabel string
+		err         error
+	)
+	conn, streamLabel, err = RemoveLabelHeaderFromStream(conn)
+	if err != nil {
+		m.logger.Printf("[ERR] memberlist: failed to receive and remove the stream label header: %s %s", err, LogConn(conn))
+		return
+	}
+
+	if m.config.SkipInboundLabelCheck {
+		if streamLabel != "" {
+			m.logger.Printf("[ERR] memberlist: unexpected double stream label header: %s", LogConn(conn))
+			return
+		}
+		// Set this from config so that the auth data assertions work below.
+		streamLabel = m.config.Label
+	}
+
+	if m.config.Label != streamLabel {
+		m.logger.Printf("[ERR] memberlist: discarding stream with unacceptable label %q: %s", streamLabel, LogConn(conn))
+		return
+	}
+
+	msgType, bufConn, dec, err := m.readStream(conn, streamLabel)
 	if err != nil {
 		if err != io.EOF {
 			m.logger.Printf("[ERR] memberlist: failed to receive: %s %s", err, LogConn(conn))
@@ -238,7 +273,7 @@ func (m *Memberlist) handleConn(conn net.Conn) {
 				return
 			}
 
-			err = m.rawSendMsgStream(conn, out.Bytes())
+			err = m.rawSendMsgStream(conn, out.Bytes(), streamLabel)
 			if err != nil {
 				m.logger.Printf("[ERR] memberlist: Failed to send error: %s %s", err, LogConn(conn))
 				return
@@ -269,7 +304,7 @@ func (m *Memberlist) handleConn(conn net.Conn) {
 			return
 		}
 
-		if err := m.sendLocalState(conn, join); err != nil {
+		if err := m.sendLocalState(conn, join, streamLabel); err != nil {
 			m.logger.Printf("[ERR] memberlist: Failed to push local state: %s %s", err, LogConn(conn))
 			return
 		}
@@ -297,7 +332,7 @@ func (m *Memberlist) handleConn(conn net.Conn) {
 			return
 		}
 
-		err = m.rawSendMsgStream(conn, out.Bytes())
+		err = m.rawSendMsgStream(conn, out.Bytes(), streamLabel)
 		if err != nil {
 			m.logger.Printf("[ERR] memberlist: Failed to send ack: %s %s", err, LogConn(conn))
 			return
@@ -322,10 +357,35 @@ func (m *Memberlist) packetListen() {
 }
 
 func (m *Memberlist) ingestPacket(buf []byte, from net.Addr, timestamp time.Time) {
+	var (
+		packetLabel string
+		err         error
+	)
+	buf, packetLabel, err = RemoveLabelHeaderFromPacket(buf)
+	if err != nil {
+		m.logger.Printf("[ERR] memberlist: %v %s", err, LogAddress(from))
+		return
+	}
+
+	if m.config.SkipInboundLabelCheck {
+		if packetLabel != "" {
+			m.logger.Printf("[ERR] memberlist: unexpected double packet label header: %s", LogAddress(from))
+			return
+		}
+		// Set this from config so that the auth data assertions work below.
+		packetLabel = m.config.Label
+	}
+
+	if m.config.Label != packetLabel {
+		m.logger.Printf("[ERR] memberlist: discarding packet with unacceptable label %q: %s", packetLabel, LogAddress(from))
+		return
+	}
+
 	// Check if encryption is enabled
 	if m.config.EncryptionEnabled() {
 		// Decrypt the payload
-		plain, err := decryptPayload(m.config.Keyring.GetKeys(), buf, nil)
+		authData := []byte(packetLabel)
+		plain, err := decryptPayload(m.config.Keyring.GetKeys(), buf, authData)
 		if err != nil {
 			if !m.config.GossipVerifyIncoming {
 				// Treat the message as plaintext
@@ -723,7 +783,7 @@ func (m *Memberlist) encodeAndSendMsg(a Address, msgType messageType, msg interf
 // opportunistically create a compoundMsg and piggy back other broadcasts.
 func (m *Memberlist) sendMsg(a Address, msg []byte) error {
 	// Check if we can piggy back any messages
-	bytesAvail := m.config.UDPBufferSize - len(msg) - compoundHeaderOverhead
+	bytesAvail := m.config.UDPBufferSize - len(msg) - compoundHeaderOverhead - labelOverhead(m.config.Label)
 	if m.config.EncryptionEnabled() && m.config.GossipVerifyOutgoing {
 		bytesAvail -= encryptOverhead(m.encryptionVersion())
 	}
@@ -795,9 +855,12 @@ func (m *Memberlist) rawSendMsgPacket(a Address, node *Node, msg []byte) error {
 	// Check if we have encryption enabled
 	if m.config.EncryptionEnabled() && m.config.GossipVerifyOutgoing {
 		// Encrypt the payload
-		var buf bytes.Buffer
-		primaryKey := m.config.Keyring.GetPrimaryKey()
-		err := encryptPayload(m.encryptionVersion(), primaryKey, msg, nil, &buf)
+		var (
+			primaryKey  = m.config.Keyring.GetPrimaryKey()
+			packetLabel = []byte(m.config.Label)
+			buf         bytes.Buffer
+		)
+		err := encryptPayload(m.encryptionVersion(), primaryKey, msg, packetLabel, &buf)
 		if err != nil {
 			m.logger.Printf("[ERR] memberlist: Encryption of message failed: %v", err)
 			return err
@@ -812,7 +875,7 @@ func (m *Memberlist) rawSendMsgPacket(a Address, node *Node, msg []byte) error {
 
 // rawSendMsgStream is used to stream a message to another host without
 // modification, other than applying compression and encryption if enabled.
-func (m *Memberlist) rawSendMsgStream(conn net.Conn, sendBuf []byte) error {
+func (m *Memberlist) rawSendMsgStream(conn net.Conn, sendBuf []byte, streamLabel string) error {
 	// Check if compression is enabled
 	if m.config.EnableCompression {
 		compBuf, err := compressPayload(sendBuf)
@@ -825,7 +888,7 @@ func (m *Memberlist) rawSendMsgStream(conn net.Conn, sendBuf []byte) error {
 
 	// Check if encryption is enabled
 	if m.config.EncryptionEnabled() && m.config.GossipVerifyOutgoing {
-		crypt, err := m.encryptLocalState(sendBuf)
+		crypt, err := m.encryptLocalState(sendBuf, streamLabel)
 		if err != nil {
 			m.logger.Printf("[ERROR] memberlist: Failed to encrypt local state: %v", err)
 			return err
@@ -871,7 +934,8 @@ func (m *Memberlist) sendUserMsg(a Address, sendBuf []byte) error {
 	if _, err := bufConn.Write(sendBuf); err != nil {
 		return err
 	}
-	return m.rawSendMsgStream(conn, bufConn.Bytes())
+
+	return m.rawSendMsgStream(conn, bufConn.Bytes(), m.config.Label)
 }
 
 // sendAndReceiveState is used to initiate a push/pull over a stream with a
@@ -891,12 +955,12 @@ func (m *Memberlist) sendAndReceiveState(a Address, join bool) ([]pushNodeState,
 	metrics.IncrCounter([]string{"memberlist", "tcp", "connect"}, 1)
 
 	// Send our state
-	if err := m.sendLocalState(conn, join); err != nil {
+	if err := m.sendLocalState(conn, join, m.config.Label); err != nil {
 		return nil, nil, err
 	}
 
 	conn.SetDeadline(time.Now().Add(m.config.TCPTimeout))
-	msgType, bufConn, dec, err := m.readStream(conn)
+	msgType, bufConn, dec, err := m.readStream(conn, m.config.Label)
 	if err != nil {
 		return nil, nil, err
 	}
@@ -921,7 +985,7 @@ func (m *Memberlist) sendAndReceiveState(a Address, join bool) ([]pushNodeState,
 }
 
 // sendLocalState is invoked to send our local state over a stream connection.
-func (m *Memberlist) sendLocalState(conn net.Conn, join bool) error {
+func (m *Memberlist) sendLocalState(conn net.Conn, join bool, streamLabel string) error {
 	// Setup a deadline
 	conn.SetDeadline(time.Now().Add(m.config.TCPTimeout))
 
@@ -978,11 +1042,11 @@ func (m *Memberlist) sendLocalState(conn net.Conn, join bool) error {
 	}
 
 	// Get the send buffer
-	return m.rawSendMsgStream(conn, bufConn.Bytes())
+	return m.rawSendMsgStream(conn, bufConn.Bytes(), streamLabel)
 }
 
 // encryptLocalState is used to help encrypt local state before sending
-func (m *Memberlist) encryptLocalState(sendBuf []byte) ([]byte, error) {
+func (m *Memberlist) encryptLocalState(sendBuf []byte, streamLabel string) ([]byte, error) {
 	var buf bytes.Buffer
 
 	// Write the encryptMsg byte
@@ -995,9 +1059,15 @@ func (m *Memberlist) encryptLocalState(sendBuf []byte) ([]byte, error) {
 	binary.BigEndian.PutUint32(sizeBuf, uint32(encLen))
 	buf.Write(sizeBuf)
 
+	// Authenticated Data is:
+	//
+	//   [messageType; byte] [messageLength; uint32] [stream_label; optional]
+	//
+	dataBytes := appendBytes(buf.Bytes()[:5], []byte(streamLabel))
+
 	// Write the encrypted cipher text to the buffer
 	key := m.config.Keyring.GetPrimaryKey()
-	err := encryptPayload(encVsn, key, sendBuf, buf.Bytes()[:5], &buf)
+	err := encryptPayload(encVsn, key, sendBuf, dataBytes, &buf)
 	if err != nil {
 		return nil, err
 	}
@@ -1005,7 +1075,7 @@ func (m *Memberlist) encryptLocalState(sendBuf []byte) ([]byte, error) {
 }
 
 // decryptRemoteState is used to help decrypt the remote state
-func (m *Memberlist) decryptRemoteState(bufConn io.Reader) ([]byte, error) {
+func (m *Memberlist) decryptRemoteState(bufConn io.Reader, streamLabel string) ([]byte, error) {
 	// Read in enough to determine message length
 	cipherText := bytes.NewBuffer(nil)
 	cipherText.WriteByte(byte(encryptMsg))
@@ -1027,8 +1097,13 @@ func (m *Memberlist) decryptRemoteState(bufConn io.Reader) ([]byte, error) {
 		return nil, err
 	}
 
-	// Decrypt the cipherText
-	dataBytes := cipherText.Bytes()[:5]
+	// Decrypt the cipherText with some authenticated data
+	//
+	// Authenticated Data is:
+	//
+	//   [messageType; byte] [messageLength; uint32] [label_data; optional]
+	//
+	dataBytes := appendBytes(cipherText.Bytes()[:5], []byte(streamLabel))
 	cipherBytes := cipherText.Bytes()[5:]
 
 	// Decrypt the payload
@@ -1036,15 +1111,18 @@ func (m *Memberlist) decryptRemoteState(bufConn io.Reader) ([]byte, error) {
 	return decryptPayload(keys, cipherBytes, dataBytes)
 }
 
-// readStream is used to read from a stream connection, decrypting and
+// readStream is used to read messages from a stream connection, decrypting and
 // decompressing the stream if necessary.
-func (m *Memberlist) readStream(conn net.Conn) (messageType, io.Reader, *codec.Decoder, error) {
+//
+// The provided streamLabel if present will be authenticated during decryption
+// of each message.
+func (m *Memberlist) readStream(conn net.Conn, streamLabel string) (messageType, io.Reader, *codec.Decoder, error) {
 	// Created a buffered reader
 	var bufConn io.Reader = bufio.NewReader(conn)
 
 	// Read the message type
 	buf := [1]byte{0}
-	if _, err := bufConn.Read(buf[:]); err != nil {
+	if _, err := io.ReadFull(bufConn, buf[:]); err != nil {
 		return 0, nil, nil, err
 	}
 	msgType := messageType(buf[0])
@@ -1056,7 +1134,7 @@ func (m *Memberlist) readStream(conn net.Conn) (messageType, io.Reader, *codec.D
 				fmt.Errorf("Remote state is encrypted and encryption is not configured")
 		}
 
-		plain, err := m.decryptRemoteState(bufConn)
+		plain, err := m.decryptRemoteState(bufConn, streamLabel)
 		if err != nil {
 			return 0, nil, nil, err
 		}
@@ -1236,11 +1314,11 @@ func (m *Memberlist) sendPingAndWaitForAck(a Address, ping ping, deadline time.T
 		return false, err
 	}
 
-	if err = m.rawSendMsgStream(conn, out.Bytes()); err != nil {
+	if err = m.rawSendMsgStream(conn, out.Bytes(), m.config.Label); err != nil {
 		return false, err
 	}
 
-	msgType, _, dec, err := m.readStream(conn)
+	msgType, _, dec, err := m.readStream(conn, m.config.Label)
 	if err != nil {
 		return false, err
 	}

vendor/github.com/hashicorp/memberlist/peeked_conn.go

@@ -0,0 +1,48 @@
+// Copyright 2017 Google Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// Originally from: https://github.com/google/tcpproxy/blob/master/tcpproxy.go
+// at f5c09fbedceb69e4b238dec52cdf9f2fe9a815e2
+
+package memberlist
+
+import "net"
+
+// peekedConn is an incoming connection that has had some bytes read from it
+// to determine how to route the connection. The Read method stitches
+// the peeked bytes and unread bytes back together.
+type peekedConn struct {
+	// Peeked are the bytes that have been read from Conn for the
+	// purposes of route matching, but have not yet been consumed
+	// by Read calls. It set to nil by Read when fully consumed.
+	Peeked []byte
+
+	// Conn is the underlying connection.
+	// It can be type asserted against *net.TCPConn or other types
+	// as needed. It should not be read from directly unless
+	// Peeked is nil.
+	net.Conn
+}
+
+func (c *peekedConn) Read(p []byte) (n int, err error) {
+	if len(c.Peeked) > 0 {
+		n = copy(p, c.Peeked)
+		c.Peeked = c.Peeked[n:]
+		if len(c.Peeked) == 0 {
+			c.Peeked = nil
+		}
+		return n, nil
+	}
+	return c.Conn.Read(p)
+}

vendor/github.com/hashicorp/memberlist/security.go

@@ -199,3 +199,22 @@ func decryptPayload(keys [][]byte, msg []byte, data []byte) ([]byte, error) {
 
 	return nil, fmt.Errorf("No installed keys could decrypt the message")
 }
+
+func appendBytes(first []byte, second []byte) []byte {
+	hasFirst := len(first) > 0
+	hasSecond := len(second) > 0
+
+	switch {
+	case hasFirst && hasSecond:
+		out := make([]byte, 0, len(first)+len(second))
+		out = append(out, first...)
+		out = append(out, second...)
+		return out
+	case hasFirst:
+		return first
+	case hasSecond:
+		return second
+	default:
+		return nil
+	}
+}

vendor/github.com/hashicorp/memberlist/state.go

@@ -274,6 +274,11 @@ func failedRemote(err error) bool {
 			case "dial", "read", "write":
 				return true
 			}
+		} else if strings.HasPrefix(t.Net, "udp") {
+			switch t.Op {
+			case "write":
+				return true
+			}
 		}
 	}
 	return false
@@ -587,7 +592,7 @@ func (m *Memberlist) gossip() {
 	m.nodeLock.RUnlock()
 
 	// Compute the bytes available
-	bytesAvail := m.config.UDPBufferSize - compoundHeaderOverhead
+	bytesAvail := m.config.UDPBufferSize - compoundHeaderOverhead - labelOverhead(m.config.Label)
 	if m.config.EncryptionEnabled() {
 		bytesAvail -= encryptOverhead(m.encryptionVersion())
 	}

vendor/github.com/hashicorp/memberlist/transport.go

@@ -111,3 +111,50 @@ func (t *shimNodeAwareTransport) WriteToAddress(b []byte, addr Address) (time.Ti
 func (t *shimNodeAwareTransport) DialAddressTimeout(addr Address, timeout time.Duration) (net.Conn, error) {
 	return t.DialTimeout(addr.Addr, timeout)
 }
+
+type labelWrappedTransport struct {
+	label string
+	NodeAwareTransport
+}
+
+var _ NodeAwareTransport = (*labelWrappedTransport)(nil)
+
+func (t *labelWrappedTransport) WriteToAddress(buf []byte, addr Address) (time.Time, error) {
+	var err error
+	buf, err = AddLabelHeaderToPacket(buf, t.label)
+	if err != nil {
+		return time.Time{}, fmt.Errorf("failed to add label header to packet: %w", err)
+	}
+	return t.NodeAwareTransport.WriteToAddress(buf, addr)
+}
+
+func (t *labelWrappedTransport) WriteTo(buf []byte, addr string) (time.Time, error) {
+	var err error
+	buf, err = AddLabelHeaderToPacket(buf, t.label)
+	if err != nil {
+		return time.Time{}, err
+	}
+	return t.NodeAwareTransport.WriteTo(buf, addr)
+}
+
+func (t *labelWrappedTransport) DialAddressTimeout(addr Address, timeout time.Duration) (net.Conn, error) {
+	conn, err := t.NodeAwareTransport.DialAddressTimeout(addr, timeout)
+	if err != nil {
+		return nil, err
+	}
+	if err := AddLabelHeaderToStream(conn, t.label); err != nil {
+		return nil, fmt.Errorf("failed to add label header to stream: %w", err)
+	}
+	return conn, nil
+}
+
+func (t *labelWrappedTransport) DialTimeout(addr string, timeout time.Duration) (net.Conn, error) {
+	conn, err := t.NodeAwareTransport.DialTimeout(addr, timeout)
+	if err != nil {
+		return nil, err
+	}
+	if err := AddLabelHeaderToStream(conn, t.label); err != nil {
+		return nil, fmt.Errorf("failed to add label header to stream: %w", err)
+	}
+	return conn, nil
+}

vendor/github.com/hashicorp/yamux/session.go

@@ -80,7 +80,7 @@ type Session struct {
 // or to directly send a header
 type sendReady struct {
 	Hdr  []byte
-	Body io.Reader
+	Body []byte
 	Err  chan error
 }
 
@@ -352,7 +352,7 @@ func (s *Session) keepalive() {
 }
 
 // waitForSendErr waits to send a header, checking for a potential shutdown
-func (s *Session) waitForSend(hdr header, body io.Reader) error {
+func (s *Session) waitForSend(hdr header, body []byte) error {
 	errCh := make(chan error, 1)
 	return s.waitForSendErr(hdr, body, errCh)
 }
@@ -360,7 +360,7 @@ func (s *Session) waitForSend(hdr header, body io.Reader) error {
 // waitForSendErr waits to send a header with optional data, checking for a
 // potential shutdown. Since there's the expectation that sends can happen
 // in a timely manner, we enforce the connection write timeout here.
-func (s *Session) waitForSendErr(hdr header, body io.Reader, errCh chan error) error {
+func (s *Session) waitForSendErr(hdr header, body []byte, errCh chan error) error {
 	t := timerPool.Get()
 	timer := t.(*time.Timer)
 	timer.Reset(s.config.ConnectionWriteTimeout)
@@ -440,7 +440,7 @@ func (s *Session) send() {
 
 			// Send data from a body if given
 			if ready.Body != nil {
-				_, err := io.Copy(s.conn, ready.Body)
+				_, err := s.conn.Write(ready.Body)
 				if err != nil {
 					s.logger.Printf("[ERR] yamux: Failed to write body: %v", err)
 					asyncSendErr(ready.Err, err)

vendor/github.com/hashicorp/yamux/stream.go

@@ -169,7 +169,7 @@ func (s *Stream) Write(b []byte) (n int, err error) {
 func (s *Stream) write(b []byte) (n int, err error) {
 	var flags uint16
 	var max uint32
-	var body io.Reader
+	var body []byte
 START:
 	s.stateLock.Lock()
 	switch s.state {
@@ -195,7 +195,7 @@ START:
 
 	// Send up to our send window
 	max = min(window, uint32(len(b)))
-	body = bytes.NewReader(b[:max])
+	body = b[:max]
 
 	// Send the header
 	s.sendHdr.encode(typeData, flags, s.id, max)