diff --git a/minecraft/conn.go b/minecraft/conn.go index b88340795..fb925a742 100644 --- a/minecraft/conn.go +++ b/minecraft/conn.go @@ -653,6 +653,7 @@ func (conn *Conn) takeDeferredPacket() (*packetData, bool) { // deferPacket defers a packet so that it is obtained in the next ReadPacket call func (conn *Conn) deferPacket(pk *packetData) { + pk = pk.ensureOwned() conn.deferredPacketMu.Lock() conn.deferredPackets = append(conn.deferredPackets, pk) conn.deferredPacketMu.Unlock() @@ -685,7 +686,7 @@ func (conn *Conn) receive(data []byte) error { } select { case <-conn.ctx.Done(): - case conn.packets <- pkData: + case conn.packets <- pkData.ensureOwned(): } return nil } diff --git a/minecraft/dial.go b/minecraft/dial.go index da701d551..94a81c5bc 100644 --- a/minecraft/dial.go +++ b/minecraft/dial.go @@ -417,26 +417,12 @@ func listenConn(conn *Conn, readyForLogin, connected chan struct{}, cancel conte for { // We finally arrived at the packet decoding loop. We constantly decode packets that arrive // and push them to the Conn so that they may be processed. - packets, err := conn.dec.Decode() - if err != nil { - if !errors.Is(err, net.ErrClosed) { - if cancelContext { - cancel(err) - } else { - conn.log.Error(err.Error()) - } - } - return - } - for _, data := range packets { + receiveErr := false + if err := conn.dec.DecodeFunc(func(data []byte) error { loggedInBefore, readyToLoginBefore := conn.loggedIn, conn.readyToLogin if err := conn.receive(data); err != nil { - if cancelContext { - cancel(err) - } else { - conn.log.Error(err.Error()) - } - return + receiveErr = true + return err } if !readyToLoginBefore && conn.readyToLogin { // This is the signal that the connection is ready to login, so we put a value in the channel so that @@ -449,6 +435,16 @@ func listenConn(conn *Conn, readyForLogin, connected chan struct{}, cancel conte cancelContext = false connected <- struct{}{} } + return nil + }); err != nil { + if receiveErr || !errors.Is(err, net.ErrClosed) { + if cancelContext { + cancel(err) + } else { + conn.log.Error(err.Error()) + } + } + return } } } diff --git a/minecraft/listener.go b/minecraft/listener.go index 62b5b8ef9..743d0077a 100644 --- a/minecraft/listener.go +++ b/minecraft/listener.go @@ -26,6 +26,10 @@ import ( "golang.org/x/oauth2" ) +// errListenerClosed stops a connection's read loop when the listener closes before the connection could be +// delivered to Accept. +var errListenerClosed = errors.New("listener closed before connection delivery") + // ListenConfig holds settings that may be edited to change behaviour of a Listener. type ListenConfig struct { // ErrorLog is a log.Logger that errors that occur during packet handling of @@ -531,32 +535,32 @@ func (listener *Listener) handleConn(conn *Conn) { for { // We finally arrived at the packet decoding loop. We constantly decode packets that arrive // and push them to the Conn so that they may be processed. - packets, err := conn.dec.Decode() - if err != nil { - // A login timeout was logged when it fired. - if !errors.Is(err, net.ErrClosed) && !errors.Is(context.Cause(conn.ctx), errLoginTimeout) { - conn.log.Error(err.Error()) - } - return - } - for _, data := range packets { + receiveErr := false + if err := conn.dec.DecodeFunc(func(data []byte) error { loggedInBefore := conn.loggedIn if err := conn.receive(data); err != nil { - conn.log.Error(err.Error()) - return + receiveErr = true + return err } if !loggedInBefore && conn.loggedIn { select { case <-listener.close: // The listener was closed while this one was logged in, so the incoming channel will be // closed. Just return so the connection is closed and cleaned up. - return + return errListenerClosed case listener.incoming <- conn: // The connection was previously not logged in, but was after receiving this packet, // meaning the connection is fully completely now. We add it to the channel so that // a call to Accept() can receive it. } } + return nil + }); err != nil { + // A login timeout was logged when it fired. + if receiveErr || (!errors.Is(err, net.ErrClosed) && !errors.Is(err, errListenerClosed) && !errors.Is(context.Cause(conn.ctx), errLoginTimeout)) { + conn.log.Error(err.Error()) + } + return } } } diff --git a/minecraft/packet.go b/minecraft/packet.go index 89c68f186..629258819 100644 --- a/minecraft/packet.go +++ b/minecraft/packet.go @@ -12,6 +12,9 @@ type packetData struct { h *packet.Header full []byte payload *bytes.Buffer + // owned specifies if full and payload no longer alias the Decoder's pooled batch buffer, meaning the + // packetData may be retained after the read callback it was created in returns. + owned bool } // parseData parses the packet data slice passed into a packetData struct. @@ -24,12 +27,34 @@ func parseData(data []byte, conn *Conn) (*packetData, error) { return nil, fmt.Errorf("read packet header: %w", err) } if conn.packetFunc != nil { - // The packet func was set, so we call it. - conn.packetFunc(*header, buf.Bytes(), conn.RemoteAddr(), conn.LocalAddr()) + // The packet func was set, so we call it. The payload is copied so that the function may retain it. + conn.packetFunc(*header, bytes.Clone(buf.Bytes()), conn.RemoteAddr(), conn.LocalAddr()) } return &packetData{h: header, full: data, payload: buf}, nil } +// ensureOwned returns a packetData that owns its underlying data, copying it if it still aliases the +// Decoder's pooled batch buffer. +func (p *packetData) ensureOwned() *packetData { + if p.owned { + return p + } + full := bytes.Clone(p.full) + payloadOffset := len(p.full) - p.payload.Len() + var payload []byte + if payloadOffset < 0 { + payload = bytes.Clone(p.payload.Bytes()) + } else { + payload = full[payloadOffset:] + } + return &packetData{ + h: p.h, + full: full, + payload: bytes.NewBuffer(payload), + owned: true, + } +} + type unknownPacketError struct { id uint32 } diff --git a/minecraft/protocol/packet/compression.go b/minecraft/protocol/packet/compression.go index a4a4cd7df..7e8ac0ff3 100644 --- a/minecraft/protocol/packet/compression.go +++ b/minecraft/protocol/packet/compression.go @@ -28,6 +28,14 @@ type appendCompression interface { MaxCompressedLen(decompressedLen int) int } +// appendDecompression is implemented by Compression algorithms that can decompress into an existing +// destination buffer instead of allocating a new one on every call. +type appendDecompression interface { + // DecompressAppend appends the decompressed form of compressed to dst and returns the result. It + // returns an error if the decompressed size exceeds limit. + DecompressAppend(dst, compressed []byte, limit int) ([]byte, error) +} + var ( // NopCompression is an empty implementation that does not compress data. NopCompression nopCompression @@ -110,7 +118,34 @@ func (flateCompression) Compress(decompressed []byte) ([]byte, error) { } // Decompress ... -func (flateCompression) Decompress(compressed []byte, limit int) ([]byte, error) { +func (c flateCompression) Decompress(compressed []byte, limit int) ([]byte, error) { + pooled := getDecompressBuffer() + defer putDecompressBuffer(pooled) + + data, err := c.DecompressAppend(*pooled, compressed, limit) + if err != nil { + return nil, err + } + *pooled = data + return append([]byte(nil), data...), nil +} + +// DecompressAppend decompresses flate data into dst. Flate has no length prefix, so dst is grown to double +// the compressed size up front and appendReader grows it further if needed. +func (flateCompression) DecompressAppend(dst, compressed []byte, limit int) ([]byte, error) { + limit = max(limit, 0) + hint := len(compressed) + if hint > math.MaxInt/2 { + hint = math.MaxInt + } else { + hint *= 2 + } + hint = max(hint, 32*1024) + if limit != math.MaxInt { + hint = min(hint, limit+1) + } + dst = slices.Grow(dst, hint) + r := flateDecompressPool.Get().(io.ReadCloser) defer func() { _ = r.Close() @@ -120,40 +155,11 @@ func (flateCompression) Decompress(compressed []byte, limit int) ([]byte, error) if err := r.(flate.Resetter).Reset(bytes.NewReader(compressed), nil); err != nil { return nil, fmt.Errorf("reset flate: %w", err) } - - decompressed := internal.BufferPool.Get().(*bytes.Buffer) - defer func() { - // Only return reasonably sized buffers to the pool to avoid retaining very large arrays. - if decompressed.Cap() <= 1<<20 { // 1 MiB cap - decompressed.Reset() - internal.BufferPool.Put(decompressed) - } - }() - - // Handle no limit - if limit == math.MaxInt { - if _, err := io.Copy(decompressed, r); err != nil { - return nil, fmt.Errorf("decompress flate: %w", err) - } - return append([]byte(nil), decompressed.Bytes()...), nil - } - - // If the compressed data is less than half the limit, we can safely assume l*2, otherwise cap at limit. - capHint := limit - if l := len(compressed); l <= limit/2 { - capHint = l * 2 - } - decompressed.Grow(capHint) - - // Read limit+1 bytes to detect overflow - lr := &io.LimitedReader{R: r, N: int64(limit) + 1} - if _, err := io.Copy(decompressed, lr); err != nil { + decompressed, err := appendReader(dst, r, limit) + if err != nil { return nil, fmt.Errorf("decompress flate: %w", err) } - if lr.N <= 0 { - return nil, fmt.Errorf("decompress flate: size exceeds limit %d", limit) - } - return append([]byte(nil), decompressed.Bytes()...), nil + return decompressed, nil } // EncodeCompression ... @@ -187,10 +193,13 @@ func (snappyCompression) MaxCompressedLen(decompressedLen int) int { } // Decompress ... -func (snappyCompression) Decompress(compressed []byte, limit int) ([]byte, error) { - // Snappy writes a decoded data length prefix, so it can allocate the - // perfect size right away and only needs to allocate once. No need to pool - // byte slices here either. +func (c snappyCompression) Decompress(compressed []byte, limit int) ([]byte, error) { + return c.DecompressAppend(nil, compressed, limit) +} + +// DecompressAppend decompresses snappy data into dst. Snappy writes a decoded data length prefix, so the +// exact size can be checked against limit and reserved in dst before decompressing. +func (snappyCompression) DecompressAppend(dst, compressed []byte, limit int) ([]byte, error) { decodedLen, err := s2.DecodedLen(compressed) if err != nil { return nil, fmt.Errorf("snappy decoded length: %w", err) @@ -198,11 +207,47 @@ func (snappyCompression) Decompress(compressed []byte, limit int) ([]byte, error if decodedLen > limit { return nil, fmt.Errorf("snappy decoded size %d exceeds limit %d", decodedLen, limit) } - decompressed, err := s2.Decode(nil, compressed) + offset := len(dst) + dst = slices.Grow(dst, decodedLen) + decompressed, err := s2.Decode(dst[offset:offset:cap(dst)], compressed) if err != nil { return nil, fmt.Errorf("decompress snappy: %w", err) } - return decompressed, nil + return append(dst[:offset], decompressed...), nil +} + +// appendReader appends everything read from r to dst and returns the result. It returns an error if more +// than limit bytes are read. +func appendReader(dst []byte, r io.Reader, limit int) ([]byte, error) { + const chunkSize = 32 * 1024 + + // Reading limit+1 bytes detects overflow; cap the limit so the +1 cannot itself overflow. + limit = max(min(limit, math.MaxInt-1), 0) + start := len(dst) + for { + remaining := limit + 1 - (len(dst) - start) + if cap(dst) == len(dst) { + dst = slices.Grow(dst, min(chunkSize, remaining)) + } + readLen := min(cap(dst)-len(dst), chunkSize, remaining) + + n := len(dst) + dst = dst[:n+readLen] + read, err := r.Read(dst[n:]) + dst = dst[:n+read] + if len(dst)-start > limit { + return nil, fmt.Errorf("size exceeds limit %d", limit) + } + if err == io.EOF { + return dst, nil + } + if err != nil { + return nil, err + } + if read == 0 { + return nil, io.ErrNoProgress + } + } } // init registers all valid compressions with the protocol. diff --git a/minecraft/protocol/packet/decoder.go b/minecraft/protocol/packet/decoder.go index 4eb01a3dd..ccd7b6bb6 100644 --- a/minecraft/protocol/packet/decoder.go +++ b/minecraft/protocol/packet/decoder.go @@ -7,8 +7,7 @@ import ( "fmt" "io" "math" - - "github.com/sandertv/gophertunnel/minecraft/protocol" + "sync" ) // Decoder handles the decoding of Minecraft packets sent through an io.Reader. These packets in turn contain @@ -106,14 +105,65 @@ const ( // maximumInBatch is the maximum amount of packets that may be found in a batch. If a compressed batch has // more than this amount, decoding will fail. maximumInBatch = 812 + // maxPooledDecompressBufferCap is the maximum capacity of a decompression buffer that is returned to + // decompressBufferPool. Larger buffers are dropped so a single big batch does not pin memory forever. + maxPooledDecompressBufferCap = 1 << 20 // DefaultMaxDecompressedLen is the default maximum decompressed batch size. DefaultMaxDecompressedLen = 16 * 1024 * 1024 ) -// Decode decodes one 'packet' from the io.Reader passed in NewDecoder(), producing a slice of packets that it -// held and an error if not successful. +// decompressBufferPool holds byte slices that decompressed batches are read into, so that a new buffer does +// not have to be allocated for every batch. +var decompressBufferPool = sync.Pool{ + New: func() any { + b := make([]byte, 0, 32*1024) + return &b + }, +} + +// Decode decodes one packet batch from the io.Reader passed in NewDecoder(), producing a slice of packets +// that it held and an error if not successful. The returned packet slices are owned by the caller. func (decoder *Decoder) Decode() (packets [][]byte, err error) { - var data []byte + data, pooled, err := decoder.readBatch() + if pooled != nil { + defer putDecompressBuffer(pooled) + } + if err != nil { + return nil, err + } + data = bytes.Clone(data) + err = decoder.walkBatch(data, func(packet []byte) error { + packets = append(packets, packet) + return nil + }) + if err != nil { + return nil, err + } + return packets, nil +} + +// DecodeFunc decodes one packet batch and calls f for each packet in the batch. The packet slice passed to f +// is only valid until f returns: it may point into a pooled buffer that is reused for the next batch, so f +// must copy the slice if it needs to keep it. The batch is validated in full before the first call to f, so a +// malformed batch dispatches no packets at all. +func (decoder *Decoder) DecodeFunc(f func(packet []byte) error) error { + data, pooled, err := decoder.readBatch() + if pooled != nil { + defer putDecompressBuffer(pooled) + } + if err != nil { + return err + } + if err := decoder.walkBatch(data, nil); err != nil { + return err + } + return decoder.walkBatch(data, f) +} + +// readBatch reads a single batch and returns its plain payload, decrypted and decompressed as necessary. If +// the payload was decompressed into a pooled buffer, that buffer is returned too and must be released by the +// caller with putDecompressBuffer, whether or not an error is returned. +func (decoder *Decoder) readBatch() (data []byte, pooled *[]byte, err error) { if decoder.pr == nil { var n int n, err = decoder.r.Read(decoder.buf) @@ -122,15 +172,15 @@ func (decoder *Decoder) Decode() (packets [][]byte, err error) { data, err = decoder.pr.ReadPacket() } if err != nil { - return nil, fmt.Errorf("read batch: %w", err) + return nil, nil, fmt.Errorf("read batch: %w", err) } if len(data) == 0 { - return nil, nil + return nil, nil, nil } h := data[:min(len(decoder.header), len(data))] if !bytes.Equal(h, decoder.header) { - return nil, fmt.Errorf("decode batch: invalid header %x: expected %x", h, decoder.header) + return nil, nil, fmt.Errorf("decode batch: invalid header %x: expected %x", h, decoder.header) } data = data[len(decoder.header):] @@ -138,53 +188,116 @@ func (decoder *Decoder) Decode() (packets [][]byte, err error) { decoder.encrypt.decrypt(data) if err := decoder.encrypt.verify(data); err != nil { // The packet did not have a correct checksum. - return nil, fmt.Errorf("verify batch: %w", err) + return nil, nil, fmt.Errorf("verify batch: %w", err) } data = data[:len(data)-8] } if decoder.decompress { if len(data) == 0 { - return nil, fmt.Errorf("decompress batch: missing compression algorithm") + return nil, nil, fmt.Errorf("decompress batch: missing compression algorithm") } if data[0] == 0xff { data = data[1:] } else { compression, ok := CompressionByID(uint16(data[0])) if !ok { - return nil, fmt.Errorf("decompress batch: unknown compression algorithm %v", data[0]) + return nil, nil, fmt.Errorf("decompress batch: unknown compression algorithm %v", data[0]) } if compression != decoder.compression { - return nil, fmt.Errorf("decompress batch: unexpected compression algorithm: got %v, expected %v", compression, decoder.compression) + return nil, nil, fmt.Errorf("decompress batch: unexpected compression algorithm: got %v, expected %v", compression, decoder.compression) } - data, err = compression.Decompress(data[1:], decoder.maxDecompressedLen) + data, pooled, err = decoder.decompressBatch(compression, data[1:]) if err != nil { - return nil, fmt.Errorf("decompress batch: %w", err) + return nil, pooled, fmt.Errorf("decompress batch: %w", err) } } } - // Uncompressed batches skip Decompress, so the limit is enforced on the final payload too. if len(data) > decoder.maxDecompressedLen { - return nil, fmt.Errorf("decode batch: size %v exceeds limit %v", len(data), decoder.maxDecompressedLen) + return nil, pooled, fmt.Errorf("decode batch: size %v exceeds limit %v", len(data), decoder.maxDecompressedLen) } + return data, pooled, nil +} - b := bytes.NewBuffer(data) - for b.Len() != 0 { - var length uint32 - if err := protocol.Varuint32(b, &length); err != nil { - return nil, fmt.Errorf("decode batch: read packet length: %w", err) +// decompressBatch decompresses a batch, borrowing a pooled buffer when the compression supports it. A non-nil +// pooled buffer must be released by the caller, whether or not an error is returned. +func (decoder *Decoder) decompressBatch(compression Compression, compressed []byte) ([]byte, *[]byte, error) { + if decompressor, ok := compression.(appendDecompression); ok { + pooled := getDecompressBuffer() + data, err := decompressor.DecompressAppend(*pooled, compressed, decoder.maxDecompressedLen) + if err != nil { + return nil, pooled, err } + *pooled = data + return data, pooled, nil + } + + data, err := compression.Decompress(compressed, decoder.maxDecompressedLen) + if err != nil { + return nil, nil, err + } + return data, nil, nil +} + +// walkBatch validates the batch payload passed and calls f, if non-nil, for each packet in it. Passing a nil +// f validates the batch without dispatching its packets. +func (decoder *Decoder) walkBatch(data []byte, f func(packet []byte) error) error { + var packetCount int + for len(data) != 0 { + length, n, err := readPacketLength(data) + if err != nil { + return fmt.Errorf("decode batch: read packet length: %w", err) + } + data = data[n:] if length == 0 { - return nil, fmt.Errorf("decode batch: empty packet") + return fmt.Errorf("decode batch: empty packet") + } + if length > uint32(len(data)) { + return fmt.Errorf("decode batch: packet length %v exceeds remaining %v", length, len(data)) } - if length > uint32(b.Len()) { - return nil, fmt.Errorf("decode batch: packet length %v exceeds remaining %v", length, b.Len()) + if packetCount >= maximumInBatch && decoder.checkPacketLimit { + return fmt.Errorf("decode batch: number of packets exceeds max=%v", maximumInBatch) } - if len(packets) >= maximumInBatch && decoder.checkPacketLimit { - return nil, fmt.Errorf("decode batch: number of packets exceeds max=%v", maximumInBatch) + if f != nil { + if err := f(data[:length:length]); err != nil { + return err + } } - packets = append(packets, b.Next(int(length))) + packetCount++ + data = data[length:] } - return packets, nil + return nil +} + +// getDecompressBuffer returns an empty buffer from decompressBufferPool. +func getDecompressBuffer() *[]byte { + b := decompressBufferPool.Get().(*[]byte) + *b = (*b)[:0] + return b +} + +// putDecompressBuffer returns a buffer to decompressBufferPool, unless it has grown too large to retain. +func putDecompressBuffer(b *[]byte) { + if cap(*b) <= maxPooledDecompressBufferCap { + *b = (*b)[:0] + decompressBufferPool.Put(b) + } +} + +// readPacketLength reads the varuint32 length prefix of a packet in a batch and returns it along with the +// number of bytes it occupied. +func readPacketLength(data []byte) (uint32, int, error) { + var length uint32 + for i := range 5 { + if i >= len(data) { + return 0, 0, io.ErrUnexpectedEOF + } + b := data[i] + length |= uint32(b&0x7f) << (7 * i) + if b&0x80 == 0 { + return length, i + 1, nil + } + } + return 0, 0, fmt.Errorf("varuint32 did not terminate after 5 bytes") }