[Packetbeat] rpc fragment bounds checking (#47803)

* nfs xdr sanitization

* Add integration test

This test caused a crash prior to this PR
pcap captured when running the following:
```bash
nc -l 12049 >/dev/null
```
and in a different shell
```python
import socket, struct, time

dest = ("127.0.0.1", 12049)
frag_header = struct.pack("!I", 0x80000001)
payload = b"\x00"

with socket.create_connection(dest, timeout=5) as sock:
    sock.sendall(frag_header + payload)
    time.sleep(0.2)
```

* Add changelog fragment

* Update changelog/fragments/1764181634-rpc_fragment_sanitization.yaml

Co-authored-by: Mykola Kmet <[email protected]>

* review suggestions

 - add return for consistency
 - add failure case unit tests

* Appease the linter

---------

Co-authored-by: Mykola Kmet <[email protected]>

Author: nicholasberlin

changelog/fragments/1764181634-rpc_fragment_sanitization.yaml

@@ -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: rpc_fragment_sanitization
+
+# 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/elastic/beats/pull/47803
+
+# 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

packetbeat/protos/nfs/malformed.go

@@ -0,0 +1,28 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you 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.
+
+package nfs
+
+import "github.com/elastic/elastic-agent-libs/logp"
+
+func dropMalformed(context string, err error) bool {
+	if err != nil {
+		logp.Warn("nfs: dropping malformed %s: %v", context, err)
+		return true
+	}
+	return false
+}

packetbeat/protos/nfs/nfs.go

@@ -30,7 +30,7 @@ type nfs struct {
 	event beat.Event
 }
 
-func (nfs *nfs) getRequestInfo(xdr *xdr) mapstr.M {
+func (nfs *nfs) getRequestInfo(xdr *xdr) (mapstr.M, error) {
 	nfsInfo := mapstr.M{
 		"version": nfs.vers,
 	}
@@ -43,21 +43,35 @@ func (nfs *nfs) getRequestInfo(xdr *xdr) mapstr.M {
 		case 0:
 			nfsInfo["opcode"] = "NULL"
 		case 1:
-			tag := xdr.getDynamicOpaque()
+			tag, err := xdr.getDynamicOpaque()
+			if err != nil {
+				return nil, err
+			}
 			nfsInfo["tag"] = string(tag)
-			nfsInfo["minor_version"] = xdr.getUInt()
-			nfsInfo["opcode"] = nfs.findV4MainOpcode(xdr)
+			minor, err := xdr.getUInt()
+			if err != nil {
+				return nil, err
+			}
+			nfsInfo["minor_version"] = minor
+			opcode, err := nfs.findV4MainOpcode(xdr)
+			if err != nil {
+				return nil, err
+			}
+			nfsInfo["opcode"] = opcode
 		}
 	}
-	return nfsInfo
+	return nfsInfo, nil
 }
 
-func (nfs *nfs) getNFSReplyStatus(xdr *xdr) string {
+func (nfs *nfs) getNFSReplyStatus(xdr *xdr) (string, error) {
 	switch nfs.proc {
 	case 0:
-		return nfsStatus[0]
+		return nfsStatus[0], nil
 	default:
-		stat := int(xdr.getUInt())
-		return nfsStatus[stat]
+		stat, err := xdr.getUInt()
+		if err != nil {
+			return "", err
+		}
+		return nfsStatus[int(stat)], nil
 	}
 }

packetbeat/protos/nfs/nfs4.go

@@ -163,46 +163,61 @@ var nfsOpnum4 = map[int]string{
 	10044: "ILLEGAL",
 }
 
-func (nfs *nfs) eatData(op int, xdr *xdr) {
+func (nfs *nfs) eatData(op int, xdr *xdr) error {
 	switch op {
 	case opGetattr:
-		xdr.getUIntVector()
+		_, err := xdr.getUIntVector()
+		return err
 	case opGetfh:
-		// nothing to eat
+		return nil
 	case opLookup:
-		xdr.getDynamicOpaque()
+		_, err := xdr.getDynamicOpaque()
+		return err
 	case opLookupp:
-		// nothing to eat
+		return nil
 	case opNverify:
-		xdr.getUIntVector()
-		xdr.getDynamicOpaque()
+		if _, err := xdr.getUIntVector(); err != nil {
+			return err
+		}
+		_, err := xdr.getDynamicOpaque()
+		return err
 	case opPutfh:
-		xdr.getDynamicOpaque()
+		_, err := xdr.getDynamicOpaque()
+		return err
 	case opPutpubfh:
-		// nothing to eat
+		return nil
 	case opPutrootfh:
-		// nothing to eat
+		return nil
 	case opReadlink:
-		// nothing to eat
+		return nil
 	case opRenew:
-		xdr.getUHyper()
+		_, err := xdr.getUHyper()
+		return err
 	case opRestorefh:
-		// nothing to eat
+		return nil
 	case opSavefh:
-		// nothing to eat
+		return nil
 	case opSecinfo:
-		xdr.getDynamicOpaque()
+		_, err := xdr.getDynamicOpaque()
+		return err
 	case opVerify:
-		xdr.getUIntVector()
-		xdr.getDynamicOpaque()
+		if _, err := xdr.getUIntVector(); err != nil {
+			return err
+		}
+		_, err := xdr.getDynamicOpaque()
+		return err
 	case opSequence:
-		xdr.getOpaque(16)
-		xdr.getUInt()
-		xdr.getUInt()
-		xdr.getUInt()
-		xdr.getUInt()
-
+		if _, err := xdr.getOpaque(16); err != nil {
+			return err
+		}
+		for i := 0; i < 4; i++ {
+			if _, err := xdr.getUInt(); err != nil {
+				return err
+			}
+		}
+		return nil
 	}
+	return nil
 }
 
 // findV4MainOpcode finds the main operation in a compound call. If no main operation can be found, the last operation
@@ -219,20 +234,28 @@ func (nfs *nfs) eatData(op int, xdr *xdr) {
 // PUTFH + GETATTR
 //
 // GETATTR is the main operation.
-func (nfs *nfs) findV4MainOpcode(xdr *xdr) string {
+func (nfs *nfs) findV4MainOpcode(xdr *xdr) (string, error) {
 	// did we find a main operation opcode?
 	found := false
 
 	// default op code
 	currentOpname := "ILLEGAL"
 
-	opcount := int(xdr.getUInt())
+	opcountVal, err := xdr.getUInt()
+	if err != nil {
+		return "", err
+	}
+	opcount := int(opcountVal)
 	for i := 0; !found && i < opcount; i++ {
-		op := int(xdr.getUInt())
+		opVal, err := xdr.getUInt()
+		if err != nil {
+			return "", err
+		}
+		op := int(opVal)
 		opname, ok := nfsOpnum4[op]
 
 		if !ok {
-			return "ILLEGAL"
+			return "ILLEGAL", nil
 		}
 		currentOpname = opname
 
@@ -302,8 +325,10 @@ func (nfs *nfs) findV4MainOpcode(xdr *xdr) string {
 
 			found = true
 		default:
-			nfs.eatData(op, xdr)
+			if err := nfs.eatData(op, xdr); err != nil {
+				return "", err
+			}
 		}
 	}
-	return currentOpname
+	return currentOpname, nil
 }