Improper Validation of Array Index in Packetbeat Leading to Denial of Service

@@ -0,0 +1,45 @@
+# REQUIRED
+# Kind can be one of:
+# - breaking-change: a change to previously-documented behavior
+# - deprecation: functionality that is being removed in a later release
+# - bug-fix: fixes a problem in a previous version
+# - enhancement: extends functionality but does not break or fix existing behavior
+# - feature: new functionality
+# - known-issue: problems that we are aware of in a given version
+# - security: impacts on the security of a product or a user’s deployment.
+# - upgrade: important information for someone upgrading from a prior version
+# - other: does not fit into any of the other categories
+kind: bug-fix
+
+# REQUIRED for all kinds
+# Change summary; a 80ish characters long description of the change.
+summary: fix potential array access panics & infinite loops in postgres parser
+
+# REQUIRED for breaking-change, deprecation, known-issue
+# Long description; in case the summary is not enough to describe the change
+# this field accommodate a description without length limits.
+# description:
+
+# REQUIRED for breaking-change, deprecation, known-issue
+# impact:
+
+# REQUIRED for breaking-change, deprecation, known-issue
+# action:
+
+# REQUIRED for all kinds
+# Affected component; usually one of "elastic-agent", "fleet-server", "filebeat", "metricbeat", "auditbeat", "all", etc.
+component: packetbeat
+
+# AUTOMATED
+# OPTIONAL to manually add other PR URLs
+# PR URL: A link the PR that added the changeset.
+# If not present is automatically filled by the tooling finding the PR where this changelog fragment has been added.
+# NOTE: the tooling supports backports, so it's able to fill the original PR number instead of the backport PR number.
+# Please provide it if you are adding a fragment for a different PR.
+# pr: https://github.com/owner/repo/1234
+
+# AUTOMATED
+# OPTIONAL to manually add other issue URLs
+# Issue URL; optional; the GitHub issue related to this changeset (either closes or is part of).
+# If not present is automatically filled by the tooling with the issue linked to the PR number.
+# issue: https://github.com/owner/repo/1234

@@ -18,6 +18,7 @@
 package pgsql
 
 import (
+	"encoding/binary"
 	"errors"
 	"fmt"
 	"strings"
@@ -59,9 +60,16 @@ func (pgsql *pgsqlPlugin) parseMessageStart(s *pgsqlStream) (bool, bool) {
 	pgsql.detailf("parseMessageStart")
 
 	m := s.message
+	//malformed streams can put us in an infinite loop,
+	// just stop if we get multiple iterations where
+	// isSpecialCommand can't find the length of the command
+	lengthFailed := 0
 
-	for len(s.data[s.parseOffset:]) >= 5 {
+	for len(s.data[s.parseOffset:]) >= 5 && (lengthFailed < 100) {
 		isSpecial, length, command := pgsql.isSpecialCommand(s.data[s.parseOffset:])
+		if length == 0 {
+			lengthFailed++
+		}
 		if !isSpecial {
 			return pgsql.parseCommand(s)
 		}
@@ -340,8 +348,15 @@ func (pgsql *pgsqlPlugin) parseExtReq(s *pgsqlStream, length int) (bool, bool) {
 	}
 	m.size = uint64(msgSize)
 	m.toExport = true
+	queryRequestPoint := m.start + 6
+
+	if len(s.data) < (queryRequestPoint) {
+		pgsql.detailf("invalid start point for extended request: len %v, start %v",
+			len(s.data), queryRequestPoint)
+		return false, false
+	}
 
-	query, err := common.ReadString(s.data[m.start+6:])
+	query, err := common.ReadString(s.data[queryRequestPoint:])
 	if err != nil {
 		pgsql.detailf("Invalid extended query request")
 		return false, false
@@ -573,7 +588,7 @@ func (pgsql *pgsqlPlugin) parseMessageData(s *pgsqlStream) (bool, bool) {
 }
 
 func (pgsql *pgsqlPlugin) parseDataRow(s *pgsqlStream, buf []byte) error {
-	m := s.message
+	msg := s.message
 
 	// read field count (int16)
 	off := 2
@@ -583,8 +598,8 @@ func (pgsql *pgsqlPlugin) parseDataRow(s *pgsqlStream, buf []byte) error {
 	rows := []string{}
 	rowLength := 0
 
-	if fieldCount > len(m.fieldsFormat) {
-		return fmt.Errorf("%w: DataRow field mismatch, got %d, expected %d", errFieldBufferBig, fieldCount, len(m.fieldsFormat))
+	if fieldCount > len(msg.fieldsFormat) {
+		return fmt.Errorf("%w: DataRow field mismatch, got %d, expected %d", errFieldBufferBig, fieldCount, len(msg.fieldsFormat))
 	}
 	for field := range fieldCount {
 		if len(buf) <= off {
@@ -603,7 +618,7 @@ func (pgsql *pgsqlPlugin) parseDataRow(s *pgsqlStream, buf []byte) error {
 
 		// read column value (byten)
 		var columnValue []byte
-		if m.fieldsFormat[field] == 0 {
+		if msg.fieldsFormat[field] == 0 {
 			// field value in text format
 			if columnLength > 0 {
 				columnValue = buf[off : off+columnLength]
@@ -626,9 +641,9 @@ func (pgsql *pgsqlPlugin) parseDataRow(s *pgsqlStream, buf []byte) error {
 		return errFieldBufferBig
 	}
 
-	m.numberOfRows++
-	if len(m.rows) < pgsql.maxStoreRows {
-		m.rows = append(m.rows, rows)
+	msg.numberOfRows++
+	if len(msg.rows) < pgsql.maxStoreRows {
+		msg.rows = append(msg.rows, rows)
 	}
 
 	return nil
@@ -743,11 +758,12 @@ func (pgsql *pgsqlPlugin) isSpecialCommand(data []byte) (bool, int, int) {
 // length field in pgsql counts total length of length field + payload, not
 // including the message identifier. => Always check buffer size >= length + 1
 func readLength(b []byte) int {
-	return int(common.BytesNtohl(b))
+	// big endian = network byte order
+	return int(binary.BigEndian.Uint32(b))
 }
 
 func readCount(b []byte) int {
-	return int(common.BytesNtohs(b))
+	return int(binary.BigEndian.Uint16(b))
 }
 
 func pgsqlString(b []byte, sz int) (string, error) {

@@ -0,0 +1,45 @@
+# REQUIRED
+# Kind can be one of:
+# - breaking-change: a change to previously-documented behavior
+# - deprecation: functionality that is being removed in a later release
+# - bug-fix: fixes a problem in a previous version
+# - enhancement: extends functionality but does not break or fix existing behavior
+# - feature: new functionality
+# - known-issue: problems that we are aware of in a given version
+# - security: impacts on the security of a product or a user’s deployment.
+# - upgrade: important information for someone upgrading from a prior version
+# - other: does not fit into any of the other categories
+kind: bug-fix
+
+# REQUIRED for all kinds
+# Change summary; a 80ish characters long description of the change.
+summary: fix procfs network parsers
+
+# REQUIRED for breaking-change, deprecation, known-issue
+# Long description; in case the summary is not enough to describe the change
+# this field accommodate a description without length limits.
+# description:
+
+# REQUIRED for breaking-change, deprecation, known-issue
+# impact:
+
+# REQUIRED for breaking-change, deprecation, known-issue
+# action:
+
+# REQUIRED for all kinds
+# Affected component; usually one of "elastic-agent", "fleet-server", "filebeat", "metricbeat", "auditbeat", "all", etc.
+component: packetbeat
+
+# AUTOMATED
+# OPTIONAL to manually add other PR URLs
+# PR URL: A link the PR that added the changeset.
+# If not present is automatically filled by the tooling finding the PR where this changelog fragment has been added.
+# NOTE: the tooling supports backports, so it's able to fill the original PR number instead of the backport PR number.
+# Please provide it if you are adding a fragment for a different PR.
+# pr: https://github.com/owner/repo/1234
+
+# AUTOMATED
+# OPTIONAL to manually add other issue URLs
+# Issue URL; optional; the GitHub issue related to this changeset (either closes or is part of).
+# If not present is automatically filled by the tooling with the issue linked to the PR number.
+# issue: https://github.com/owner/repo/1234

@@ -141,13 +141,13 @@ func findSocketsOfPid(prefix string, pid int) (inodes []uint64, err error) {
 		}
 
 		if strings.HasPrefix(link, "socket:[") {
-			inode, err := strconv.ParseInt(link[8:len(link)-1], 10, 64)
+			inode, err := strconv.ParseUint(link[8:len(link)-1], 10, 64)
 			if err != nil {
 				logp.Debug("procs", "%s", err.Error())
 				continue
 			}
 
-			inodes = append(inodes, uint64(inode))
+			inodes = append(inodes, inode)
 		}
 	}
 
@@ -204,10 +204,10 @@ func parseProcNetProto(input io.Reader, ipv6 bool) ([]*socketInfo, error) {
 			continue
 		}
 
-		uid, _ := strconv.Atoi(string(words[7]))
+		uid, _ := strconv.ParseUint(string(words[7]), 10, 32)
 		sock.uid = uint32(uid)
-		inode, _ := strconv.Atoi(string(words[9]))
-		sock.inode = uint64(inode)
+		inode, _ := strconv.ParseUint(string(words[9]), 10, 64)
+		sock.inode = inode
 
 		sockets = append(sockets, &sock)
 	}
@@ -230,7 +230,7 @@ func hexToIPPort(str []byte, ipv6 bool) (net.IP, uint16, error) {
 		return nil, 0, err
 	}
 
-	port, err := strconv.ParseInt(string(words[1]), 16, 32)
+	port, err := strconv.ParseUint(string(words[1]), 16, 16)
 	if err != nil {
 		return nil, 0, err
 	}
@@ -255,8 +255,13 @@ func hexToIpv4(word string) (net.IP, error) {
 
 func hexToIpv6(word string) (net.IP, error) {
 	p := make(net.IP, net.IPv6len)
+	if len(word) < 32 {
+		return nil, fmt.Errorf("got ip6 address of invalid length: %d", len(word))
+	}
 	for i := 0; i < 4; i++ {
-		part, err := strconv.ParseUint(word[i*8:(i+1)*8], 16, 32)
+		start := i * 8
+		end := (i + 1) * 8
+		part, err := strconv.ParseUint(word[start:end], 16, 32)
 		if err != nil {
 			return nil, err
 		}

@@ -20,16 +20,14 @@
 package procs
 
 import (
-	"io/ioutil"
 	"os"
 	"path/filepath"
 	"testing"
 
-	"github.com/elastic/elastic-agent-libs/logp"
+	"github.com/stretchr/testify/require"
 )
 
 func TestFindSocketsOfPid(t *testing.T) {
-	logp.TestingSetup()
 
 	proc := []testProcFile{
 		{path: "/proc/766/fd/0", isLink: true, contents: "/dev/null"},
@@ -44,7 +42,7 @@ func TestFindSocketsOfPid(t *testing.T) {
 	}
 
 	// Create fake proc file system
-	pathPrefix, err := ioutil.TempDir("", "find-sockets")
+	pathPrefix, err := os.MkdirTemp("", "find-sockets")
 	if err != nil {
 		t.Error("TempDir failed:", err)
 		return
@@ -65,6 +63,11 @@ func TestFindSocketsOfPid(t *testing.T) {
 	assertUint64ArraysAreEqual(t, []uint64{7619, 7620}, inodes)
 }
 
+func TestParseInvalidIp6Addr(t *testing.T) {
+	_, err := hexToIpv6("B80D012000000000FFFF2301EFCD")
+	require.Error(t, err)
+}
+
 func TestParse_Proc_Net_Tcp(t *testing.T) {
 	socketInfo, err := socketsFromProc("../tests/files/proc_net_tcp.txt", false)
 	if err != nil {
@@ -116,7 +119,7 @@ func createFakeDirectoryStructure(prefix string, files []testProcFile) error {
 		}
 
 		if !file.isLink {
-			err = ioutil.WriteFile(filepath.Join(prefix, file.path),
+			err = os.WriteFile(filepath.Join(prefix, file.path),
 				[]byte(file.contents), 0o644)
 			if err != nil {
 				return err