[processor/transform/internal/logparsingfuncs] Add ParseCEF log parsing function (#49288)

Author: Caleb-Hurshman

.chloggen/feat_transform-processor-parse-cef-func.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. receiver/filelog)
+component: processor/transform
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add `ParseCEF` function for parsing Common Event Format (CEF) security event data.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [48351]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [user]

processor/transformprocessor/README.md

@@ -275,6 +275,7 @@ In addition to the common OTTL functions, the processor defines its own function
 
 **Logs only functions**
 
+- [ParseCEF](#parsecef)
 - [ParseCLF](#parseclf)
 - [ParseLEEF](#parseleef)
 
@@ -711,6 +712,32 @@ Examples:
 # counts: [84, 126, 5, 50, 1]
 ```
 
+### ParseCEF
+
+`ParseCEF(target)`
+
+The `ParseCEF` function returns a `pcommon.Map` that is the result of parsing the `target` string as a [Common Event Format (CEF)](https://www.microfocus.com/documentation/arcsight/arcsight-smartconnectors-8.4/cef-implementation-standard/Content/CEF/Chapter%201%20What%20is%20CEF.htm) message.
+
+`target` is a Getter that returns a string. If the returned string is empty, or cannot be parsed as CEF, an error will be returned.
+
+`ParseCEF` is tolerant of an optional syslog header preceding the `CEF:` token; parsing begins at the first occurrence of `CEF:` in the input.
+
+The returned map has the following top-level fields:
+
+- `cef.version` — the CEF version (the integer following `CEF:`).
+- `cef.device_vendor`, `cef.device_product`, `cef.device_version`, `cef.device_event_class_id`, `cef.name`, `cef.severity` — the six CEF header fields.
+- `cef.extensions` — a map of the parsed key/value extension pairs.
+
+Within the header fields, the escape sequences `\|` (pipe) and `\\` (backslash) are unescaped. Within extension values, the escape sequences `\\` (backslash), `\=` (equals), `\n` (newline), and `\r` (carriage return) are unescaped.
+
+Extension parsing uses the position of the next `key=` token as the end of the current value, so values may contain spaces. All extension values are returned as strings.
+
+Examples:
+
+- `ParseCEF(body)`
+
+- `ParseCEF("CEF:0|Security|threatmanager|1.0|100|worm successfully stopped|10|src=10.0.0.1 dst=2.1.2.2 spt=1232")`
+
 ### ParseCLF
 
 `ParseCLF(target, Optional[format])`

processor/transformprocessor/internal/logparsingfuncs/func_parse_cef.go

@@ -0,0 +1,283 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package logparsingfuncs // import "github.com/open-telemetry/opentelemetry-collector-contrib/processor/transformprocessor/internal/logparsingfuncs"
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strings"
+
+	"go.opentelemetry.io/collector/pdata/pcommon"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottllog"
+)
+
+type parseCEFArguments struct {
+	Target ottl.StringGetter[*ottllog.TransformContext]
+}
+
+func NewParseCEFFactory() ottl.Factory[*ottllog.TransformContext] {
+	return ottl.NewFactory("ParseCEF", &parseCEFArguments{}, createParseCEFFunction)
+}
+
+func createParseCEFFunction(_ ottl.FunctionContext, oArgs ottl.Arguments) (ottl.ExprFunc[*ottllog.TransformContext], error) {
+	args, ok := oArgs.(*parseCEFArguments)
+	if !ok {
+		return nil, errors.New("parseCEFFactory args must be of type *parseCEFArguments")
+	}
+
+	return parseCEF(args.Target), nil
+}
+
+func parseCEF(target ottl.StringGetter[*ottllog.TransformContext]) ottl.ExprFunc[*ottllog.TransformContext] {
+	return func(ctx context.Context, tCtx *ottllog.TransformContext) (any, error) {
+		source, err := target.Get(ctx, tCtx)
+		if err != nil {
+			return nil, err
+		}
+
+		if source == "" {
+			return nil, errors.New("cannot parse empty CEF message")
+		}
+
+		return parseCEFMessage(source)
+	}
+}
+
+type cefHeader struct {
+	version            string
+	deviceVendor       string
+	deviceProduct      string
+	deviceVersion      string
+	deviceEventClassID string
+	name               string
+	severity           string
+}
+
+// isCEFKeyChar reports whether c may appear in a CEF extension key. Keys
+// consist of ASCII alphanumerics and underscores.
+func isCEFKeyChar(c byte) bool {
+	return c >= 'A' && c <= 'Z' || c >= 'a' && c <= 'z' || c >= '0' && c <= '9' || c == '_'
+}
+
+// cefKeyLen returns the length of the CEF extension key starting at extension[i]
+// followed by an `=`, or 0 if no valid key begins at that position. A key is a
+// run of key characters terminated by an `=`.
+func cefKeyLen(extension string, i int) int {
+	j := i
+	for j < len(extension) && isCEFKeyChar(extension[j]) {
+		j++
+	}
+	if j > i && j < len(extension) && extension[j] == '=' {
+		return j - i
+	}
+	return 0
+}
+
+// countCEFExtensionKeys counts the number of `key=` tokens in the extension so
+// the destination map can be sized once up front.
+func countCEFExtensionKeys(extension string) int {
+	count := 0
+	for i := 0; i < len(extension); {
+		if i == 0 || extension[i-1] == ' ' {
+			if keyLen := cefKeyLen(extension, i); keyLen > 0 {
+				count++
+				i += keyLen + 1
+				continue
+			}
+		}
+		i++
+	}
+	return count
+}
+
+func parseCEFMessage(message string) (pcommon.Map, error) {
+	cefStart := strings.Index(message, "CEF:")
+	if cefStart == -1 {
+		return pcommon.Map{}, errors.New("invalid CEF message: 'CEF:' not found")
+	}
+
+	cefMessage := message[cefStart:]
+
+	fields, err := splitCEFHeader(cefMessage)
+	if err != nil {
+		return pcommon.Map{}, err
+	}
+
+	versionField := fields[0]
+	version := strings.TrimPrefix(versionField, "CEF:")
+	if version == "" {
+		return pcommon.Map{}, errors.New("invalid CEF message: missing version")
+	}
+
+	header := cefHeader{
+		version:            version,
+		deviceVendor:       unescapeCEFHeader(fields[1]),
+		deviceProduct:      unescapeCEFHeader(fields[2]),
+		deviceVersion:      unescapeCEFHeader(fields[3]),
+		deviceEventClassID: unescapeCEFHeader(fields[4]),
+		name:               unescapeCEFHeader(fields[5]),
+		severity:           unescapeCEFHeader(fields[6]),
+	}
+
+	var extension string
+	if len(fields) == 8 {
+		extension = fields[7]
+	}
+
+	return buildCEFResult(header, extension), nil
+}
+
+// splitCEFHeader splits a CEF message on unescaped pipes. The first seven
+// fields are the prefix (CEF:Version) and the six header fields. Everything
+// after the seventh pipe is the extension, returned as the eighth field if
+// present.
+func splitCEFHeader(message string) ([]string, error) {
+	const headerFieldCount = 7
+
+	fields := make([]string, 0, headerFieldCount+1)
+	var current strings.Builder
+
+	for i := 0; i < len(message); i++ {
+		c := message[i]
+		if c == '\\' && i+1 < len(message) {
+			next := message[i+1]
+			if next == '|' || next == '\\' {
+				current.WriteByte(c)
+				current.WriteByte(next)
+				i++
+				continue
+			}
+		}
+		if c == '|' {
+			fields = append(fields, current.String())
+			current.Reset()
+			if len(fields) == headerFieldCount {
+				fields = append(fields, message[i+1:])
+				return fields, nil
+			}
+			continue
+		}
+		current.WriteByte(c)
+	}
+	fields = append(fields, current.String())
+
+	if len(fields) < headerFieldCount {
+		return nil, fmt.Errorf("invalid CEF header: expected at least %d pipe-delimited fields (CEF:Version, Device Vendor, Device Product, Device Version, Device Event Class ID, Name, Severity), got %d", headerFieldCount, len(fields))
+	}
+	return fields, nil
+}
+
+// unescapeCEFHeader unescapes the two characters that may be escaped inside a
+// CEF header field: pipe and backslash.
+func unescapeCEFHeader(s string) string {
+	if !strings.Contains(s, `\`) {
+		return s
+	}
+	var b strings.Builder
+	b.Grow(len(s))
+	for i := 0; i < len(s); i++ {
+		c := s[i]
+		if c == '\\' && i+1 < len(s) {
+			next := s[i+1]
+			if next == '|' || next == '\\' {
+				b.WriteByte(next)
+				i++
+				continue
+			}
+		}
+		b.WriteByte(c)
+	}
+	return b.String()
+}
+
+// unescapeCEFValue unescapes the four sequences that may appear inside a CEF
+// extension value: backslash, equals, newline, and carriage return.
+func unescapeCEFValue(s string) string {
+	if !strings.Contains(s, `\`) {
+		return s
+	}
+	var b strings.Builder
+	b.Grow(len(s))
+	for i := 0; i < len(s); i++ {
+		c := s[i]
+		if c == '\\' && i+1 < len(s) {
+			switch s[i+1] {
+			case '\\':
+				b.WriteByte('\\')
+				i++
+				continue
+			case '=':
+				b.WriteByte('=')
+				i++
+				continue
+			case 'n':
+				b.WriteByte('\n')
+				i++
+				continue
+			case 'r':
+				b.WriteByte('\r')
+				i++
+				continue
+			}
+		}
+		b.WriteByte(c)
+	}
+	return b.String()
+}
+
+// parseCEFExtensions parses the extension portion of a CEF message directly
+// into dest as key/value pairs. Keys are runs of alphanumerics and underscores
+// preceded by the start of the string or a space. Values may contain spaces and
+// extend until the next `key=` token or the end of the string.
+func parseCEFExtensions(extension string, dest pcommon.Map) {
+	dest.EnsureCapacity(countCEFExtensionKeys(extension))
+
+	haveKey := false
+	var key string
+	var valueStart int
+
+	for i := 0; i < len(extension); {
+		if i == 0 || extension[i-1] == ' ' {
+			if keyLen := cefKeyLen(extension, i); keyLen > 0 {
+				if haveKey {
+					value := strings.TrimRight(extension[valueStart:i], " ")
+					dest.PutStr(key, unescapeCEFValue(value))
+				}
+				key = extension[i : i+keyLen]
+				valueStart = i + keyLen + 1
+				haveKey = true
+				i = valueStart
+				continue
+			}
+		}
+		i++
+	}
+
+	if haveKey {
+		value := strings.TrimRight(extension[valueStart:], " ")
+		dest.PutStr(key, unescapeCEFValue(value))
+	}
+}
+
+func buildCEFResult(header cefHeader, extension string) pcommon.Map {
+	result := pcommon.NewMap()
+
+	result.PutStr("cef.version", header.version)
+	result.PutStr("cef.device_vendor", header.deviceVendor)
+	result.PutStr("cef.device_product", header.deviceProduct)
+	result.PutStr("cef.device_version", header.deviceVersion)
+	result.PutStr("cef.device_event_class_id", header.deviceEventClassID)
+	result.PutStr("cef.name", header.name)
+	result.PutStr("cef.severity", header.severity)
+
+	extensionsMap := result.PutEmptyMap("cef.extensions")
+	if extension != "" {
+		parseCEFExtensions(extension, extensionsMap)
+	}
+
+	return result
+}