chore: Enable G110 rule for gosec

.golangci.yml

@@ -94,6 +94,7 @@ linters-settings:
       - G107
       - G108
       - G109
+      - G110
       - G111
       - G112
       - G114

internal/content_coding.go

@@ -6,6 +6,7 @@ import (
 	"compress/gzip"
 	"compress/zlib"
 	"errors"
+	"fmt"
 	"io"
 )
 
@@ -92,11 +93,11 @@ func (a *AutoDecoder) SetEncoding(encoding string) {
 	a.encoding = encoding
 }
 
-func (a *AutoDecoder) Decode(data []byte) ([]byte, error) {
+func (a *AutoDecoder) Decode(data []byte, maxDecodedSize int64) ([]byte, error) {
 	if a.encoding == "gzip" {
-		return a.gzip.Decode(data)
+		return a.gzip.Decode(data, maxDecodedSize)
 	}
-	return a.identity.Decode(data)
+	return a.identity.Decode(data, maxDecodedSize)
 }
 
 func NewAutoContentDecoder() *AutoDecoder {
@@ -199,7 +200,7 @@ func (*IdentityEncoder) Encode(data []byte) ([]byte, error) {
 // ContentDecoder removes a wrapper encoding from byte buffers.
 type ContentDecoder interface {
 	SetEncoding(string)
-	Decode([]byte) ([]byte, error)
+	Decode([]byte, int64) ([]byte, error)
 }
 
 // GzipDecoder decompresses buffers with gzip compression.
@@ -217,17 +218,20 @@ func NewGzipDecoder() *GzipDecoder {
 
 func (*GzipDecoder) SetEncoding(string) {}
 
-func (d *GzipDecoder) Decode(data []byte) ([]byte, error) {
+func (d *GzipDecoder) Decode(data []byte, maxDecodedSize int64) ([]byte, error) {
 	err := d.reader.Reset(bytes.NewBuffer(data))
 	if err != nil {
 		return nil, err
 	}
 	d.buf.Reset()
 
-	_, err = d.buf.ReadFrom(d.reader)
+	n, err := io.CopyN(d.buf, d.reader, maxDecodedSize)
 	if err != nil && !errors.Is(err, io.EOF) {
 		return nil, err
+	} else if n == maxDecodedSize {
+		return nil, fmt.Errorf("size of decoded data must be smaller than allowed size: '%d'", maxDecodedSize)
 	}
+
 	err = d.reader.Close()
 	if err != nil {
 		return nil, err
@@ -247,18 +251,22 @@ func NewZlibDecoder() *ZlibDecoder {
 
 func (*ZlibDecoder) SetEncoding(string) {}
 
-func (d *ZlibDecoder) Decode(data []byte) ([]byte, error) {
+func (d *ZlibDecoder) Decode(data []byte, maxDecodedSize int64) ([]byte, error) {
 	d.buf.Reset()
 
 	b := bytes.NewBuffer(data)
 	r, err := zlib.NewReader(b)
 	if err != nil {
 		return nil, err
 	}
-	_, err = io.Copy(d.buf, r)
+
+	n, err := io.CopyN(d.buf, r, maxDecodedSize)
 	if err != nil && !errors.Is(err, io.EOF) {
 		return nil, err
+	} else if n == maxDecodedSize {
+		return nil, fmt.Errorf("size of decoded data must be smaller than allowed size: '%d'", maxDecodedSize)
 	}
+
 	err = r.Close()
 	if err != nil {
 		return nil, err
@@ -275,6 +283,10 @@ func NewIdentityDecoder() *IdentityDecoder {
 
 func (*IdentityDecoder) SetEncoding(string) {}
 
-func (*IdentityDecoder) Decode(data []byte) ([]byte, error) {
+func (*IdentityDecoder) Decode(data []byte, maxDecodedSize int64) ([]byte, error) {
+	size := int64(len(data))
+	if size > maxDecodedSize {
+		return nil, fmt.Errorf("size of decoded data '%d' will be bigger than allowed size '%q'", size, maxDecodedSize)
+	}
 	return data, nil
 }

internal/content_coding_test.go

@@ -8,14 +8,16 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+const maxDecodedSize = 1024
+
 func TestGzipEncodeDecode(t *testing.T) {
 	enc := NewGzipEncoder()
 	dec := NewGzipDecoder()
 
 	payload, err := enc.Encode([]byte("howdy"))
 	require.NoError(t, err)
 
-	actual, err := dec.Decode(payload)
+	actual, err := dec.Decode(payload, maxDecodedSize)
 	require.NoError(t, err)
 
 	require.Equal(t, "howdy", string(actual))
@@ -28,15 +30,15 @@ func TestGzipReuse(t *testing.T) {
 	payload, err := enc.Encode([]byte("howdy"))
 	require.NoError(t, err)
 
-	actual, err := dec.Decode(payload)
+	actual, err := dec.Decode(payload, maxDecodedSize)
 	require.NoError(t, err)
 
 	require.Equal(t, "howdy", string(actual))
 
 	payload, err = enc.Encode([]byte("doody"))
 	require.NoError(t, err)
 
-	actual, err = dec.Decode(payload)
+	actual, err = dec.Decode(payload, maxDecodedSize)
 	require.NoError(t, err)
 
 	require.Equal(t, "doody", string(actual))
@@ -49,20 +51,31 @@ func TestZlibEncodeDecode(t *testing.T) {
 	payload, err := enc.Encode([]byte("howdy"))
 	require.NoError(t, err)
 
-	actual, err := dec.Decode(payload)
+	actual, err := dec.Decode(payload, maxDecodedSize)
 	require.NoError(t, err)
 
 	require.Equal(t, "howdy", string(actual))
 }
 
+func TestZlibEncodeDecodeWithTooLargeMessage(t *testing.T) {
+	enc := NewZlibEncoder()
+	dec := NewZlibDecoder()
+
+	payload, err := enc.Encode([]byte("howdy"))
+	require.NoError(t, err)
+
+	_, err = dec.Decode(payload, 3)
+	require.ErrorContains(t, err, "size of decoded data must be smaller than allowed size: '3'")
+}
+
 func TestIdentityEncodeDecode(t *testing.T) {
 	enc := NewIdentityEncoder()
 	dec := NewIdentityDecoder()
 
 	payload, err := enc.Encode([]byte("howdy"))
 	require.NoError(t, err)
 
-	actual, err := dec.Decode(payload)
+	actual, err := dec.Decode(payload, maxDecodedSize)
 	require.NoError(t, err)
 
 	require.Equal(t, "howdy", string(actual))

plugins/inputs/amqp_consumer/README.md

@@ -111,6 +111,11 @@ See the [CONFIGURATION.md][CONFIGURATION.md] for more details.
   ## - Use "auto" determine the encoding using the ContentEncoding header
   # content_encoding = "identity"
 
+  ## Maximum size of decoded message.
+  ## Acceptable units are B, KiB, KB, MiB, MB...
+  ## Without quotes and units, interpreted as size in bytes.
+  # max_decoded_size = "500MB"
+
   ## Data format to consume.
   ## Each data format has its own unique set of configuration options, read
   ## more about them here:

plugins/inputs/amqp_consumer/amqp_consumer.go

@@ -14,6 +14,7 @@ import (
 	amqp "github.com/rabbitmq/amqp091-go"
 
 	"github.com/influxdata/telegraf"
+	"github.com/influxdata/telegraf/config"
 	"github.com/influxdata/telegraf/internal"
 	"github.com/influxdata/telegraf/plugins/common/tls"
 	"github.com/influxdata/telegraf/plugins/inputs"
@@ -23,6 +24,8 @@ import (
 //go:embed sample.conf
 var sampleConfig string
 
+const defaultMaxDecodedSize = 500 * 1024 * 1024 //500MB
+
 type empty struct{}
 type semaphore chan empty
 
@@ -55,7 +58,8 @@ type AMQPConsumer struct {
 	AuthMethod string
 	tls.ClientConfig
 
-	ContentEncoding string `toml:"content_encoding"`
+	ContentEncoding string      `toml:"content_encoding"`
+	MaxDecodedSize  config.Size `toml:"max_decoded_size"`
 	Log             telegraf.Logger
 
 	deliveries map[telegraf.TrackingID]amqp.Delivery
@@ -113,6 +117,10 @@ func (a *AMQPConsumer) Init() error {
 		a.MaxUndeliveredMessages = 1000
 	}
 
+	if a.MaxDecodedSize <= 0 {
+		a.MaxDecodedSize = defaultMaxDecodedSize
+	}
+
 	return nil
 }
 
@@ -144,11 +152,11 @@ func (a *AMQPConsumer) createConfig() (*amqp.Config, error) {
 		}
 	}
 
-	config := amqp.Config{
+	amqpConfig := amqp.Config{
 		TLSClientConfig: tlsCfg,
 		SASL:            auth, // if nil, it will be PLAIN
 	}
-	return &config, nil
+	return &amqpConfig, nil
 }
 
 // Start satisfies the telegraf.ServiceInput interface
@@ -412,7 +420,7 @@ func (a *AMQPConsumer) onMessage(acc telegraf.TrackingAccumulator, d amqp.Delive
 	}
 
 	a.decoder.SetEncoding(d.ContentEncoding)
-	body, err := a.decoder.Decode(d.Body)
+	body, err := a.decoder.Decode(d.Body, int64(a.MaxDecodedSize))
 	if err != nil {
 		onError()
 		return err

plugins/inputs/amqp_consumer/amqp_consumer_test.go

@@ -23,6 +23,7 @@ func TestAutoEncoding(t *testing.T) {
 	a.deliveries = make(map[telegraf.TrackingID]amqp091.Delivery)
 	a.parser = parser
 	a.decoder, err = internal.NewContentDecoder("auto")
+	a.MaxDecodedSize = defaultMaxDecodedSize
 	require.NoError(t, err)
 
 	acc := &testutil.Accumulator{}

plugins/inputs/amqp_consumer/sample.conf

@@ -72,6 +72,11 @@
   ## - Use "auto" determine the encoding using the ContentEncoding header
   # content_encoding = "identity"
 
+  ## Maximum size of decoded message.
+  ## Acceptable units are B, KiB, KB, MiB, MB...
+  ## Without quotes and units, interpreted as size in bytes.
+  # max_decoded_size = "500MB"
+
   ## Data format to consume.
   ## Each data format has its own unique set of configuration options, read
   ## more about them here:

plugins/inputs/socket_listener/README.md

@@ -89,6 +89,11 @@ See the [CONFIGURATION.md][CONFIGURATION.md] for more details.
   ## "identity" to apply no encoding.
   # content_encoding = "identity"
 
+  ## Maximum size of decoded packet.
+  ## Acceptable units are B, KiB, KB, MiB, MB...
+  ## Without quotes and units, interpreted as size in bytes.
+  # max_decoded_size = "500MB"
+
   ## Message splitting strategy and corresponding settings for stream sockets
   ## (tcp, tcp4, tcp6, unix or unixpacket). The setting is ignored for packet
   ## listeners such as udp.

plugins/inputs/socket_listener/packet_listener.go

@@ -15,6 +15,7 @@ import (
 
 type packetListener struct {
 	Encoding       string
+	MaxDecodedSize int64
 	SocketMode     string
 	ReadBufferSize int
 	Parser         telegraf.Parser
@@ -36,7 +37,7 @@ func (l *packetListener) listen(acc telegraf.Accumulator) {
 			break
 		}
 
-		body, err := l.decoder.Decode(buf[:n])
+		body, err := l.decoder.Decode(buf[:n], l.MaxDecodedSize)
 		if err != nil {
 			acc.AddError(fmt.Errorf("unable to decode incoming packet: %w", err))
 		}

plugins/inputs/socket_listener/sample.conf

@@ -58,6 +58,11 @@
   ## "identity" to apply no encoding.
   # content_encoding = "identity"
 
+  ## Maximum size of decoded packet.
+  ## Acceptable units are B, KiB, KB, MiB, MB...
+  ## Without quotes and units, interpreted as size in bytes.
+  # max_decoded_size = "500MB"
+
   ## Message splitting strategy and corresponding settings for stream sockets
   ## (tcp, tcp4, tcp6, unix or unixpacket). The setting is ignored for packet
   ## listeners such as udp.

plugins/inputs/socket_listener/socket_listener.go

@@ -23,6 +23,8 @@ import (
 //go:embed sample.conf
 var sampleConfig string
 
+const defaultMaxDecodedSize = 500 * 1024 * 1024 //500MB
+
 type listener interface {
 	listen(acc telegraf.Accumulator)
 	addr() net.Addr
@@ -45,6 +47,7 @@ type SocketListener struct {
 	KeepAlivePeriod      *config.Duration `toml:"keep_alive_period"`
 	SocketMode           string           `toml:"socket_mode"`
 	ContentEncoding      string           `toml:"content_encoding"`
+	MaxDecodedSize       config.Size      `toml:"max_decoded_size"`
 	SplittingStrategy    string           `toml:"splitting_strategy"`
 	SplittingDelimiter   string           `toml:"splitting_delimiter"`
 	SplittingLength      int              `toml:"splitting_length"`
@@ -159,6 +162,10 @@ func (sl *SocketListener) Start(acc telegraf.Accumulator) error {
 		return fmt.Errorf("parsing address failed: %w", err)
 	}
 
+	if sl.MaxDecodedSize <= 0 {
+		sl.MaxDecodedSize = defaultMaxDecodedSize
+	}
+
 	switch u.Scheme {
 	case "tcp", "tcp4", "tcp6":
 		ssl := &streamListener{
@@ -195,26 +202,29 @@ func (sl *SocketListener) Start(acc telegraf.Accumulator) error {
 
 	case "udp", "udp4", "udp6":
 		psl := &packetListener{
-			Encoding: sl.ContentEncoding,
-			Parser:   sl.parser,
+			Encoding:       sl.ContentEncoding,
+			MaxDecodedSize: int64(sl.MaxDecodedSize),
+			Parser:         sl.parser,
 		}
 		if err := psl.setupUDP(u, ifname, int(sl.ReadBufferSize)); err != nil {
 			return err
 		}
 		sl.listener = psl
 	case "ip", "ip4", "ip6":
 		psl := &packetListener{
-			Encoding: sl.ContentEncoding,
-			Parser:   sl.parser,
+			Encoding:       sl.ContentEncoding,
+			MaxDecodedSize: int64(sl.MaxDecodedSize),
+			Parser:         sl.parser,
 		}
 		if err := psl.setupIP(u); err != nil {
 			return err
 		}
 		sl.listener = psl
 	case "unixgram":
 		psl := &packetListener{
-			Encoding: sl.ContentEncoding,
-			Parser:   sl.parser,
+			Encoding:       sl.ContentEncoding,
+			MaxDecodedSize: int64(sl.MaxDecodedSize),
+			Parser:         sl.parser,
 		}
 		if err := psl.setupUnixgram(u, sl.SocketMode); err != nil {
 			return err

plugins/outputs/graylog/graylog_test_linux.go

@@ -7,6 +7,7 @@ import (
 	"compress/zlib"
 	"crypto/tls"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"net"
@@ -150,11 +151,15 @@ func UDPServer(t *testing.T, wg *sync.WaitGroup, namefieldnoprefix bool) string
 			return err
 		}
 
+		var maxDecodedSize int64 = 500 * 1024 * 1024
 		bufW := bytes.NewBuffer(nil)
-		_, err = io.Copy(bufW, r)
-		if err != nil {
+		written, err := io.CopyN(bufW, r, maxDecodedSize)
+		if err != nil && !errors.Is(err, io.EOF) {
 			return err
+		} else if written == maxDecodedSize {
+			return fmt.Errorf("size of decoded data must be smaller than allowed size: '%d'", maxDecodedSize)
 		}
+
 		err = r.Close()
 		if err != nil {
 			return err