[libbeat] Decode wineventlog processor

libbeat/cmd/instance/imports_common.go

@@ -31,6 +31,7 @@ import (
 	_ "github.com/elastic/beats/v7/libbeat/processors/communityid"
 	_ "github.com/elastic/beats/v7/libbeat/processors/convert"
 	_ "github.com/elastic/beats/v7/libbeat/processors/decode_xml"
+	_ "github.com/elastic/beats/v7/libbeat/processors/decode_xml_wineventlog"
 	_ "github.com/elastic/beats/v7/libbeat/processors/dissect"
 	_ "github.com/elastic/beats/v7/libbeat/processors/dns"
 	_ "github.com/elastic/beats/v7/libbeat/processors/extract_array"

libbeat/processors/decode_xml/config.go

@@ -25,6 +25,7 @@ type decodeXMLConfig struct {
 	ToLower       bool    `config:"to_lower"`
 	IgnoreMissing bool    `config:"ignore_missing"`
 	IgnoreFailure bool    `config:"ignore_failure"`
+	Schema        string  `config:"schema"`
 }
 
 func defaultConfig() decodeXMLConfig {

libbeat/processors/decode_xml/decode_xml.go

@@ -21,12 +21,10 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"strings"
 
 	"github.com/elastic/beats/v7/libbeat/beat"
 	"github.com/elastic/beats/v7/libbeat/common"
 	"github.com/elastic/beats/v7/libbeat/common/cfgwarn"
-	"github.com/elastic/beats/v7/libbeat/common/encoding/xml"
 	"github.com/elastic/beats/v7/libbeat/common/jsontransform"
 	"github.com/elastic/beats/v7/libbeat/logp"
 	"github.com/elastic/beats/v7/libbeat/processors"
@@ -36,7 +34,9 @@ import (
 
 type decodeXML struct {
 	decodeXMLConfig
-	log *logp.Logger
+
+	decode decoder
+	log    *logp.Logger
 }
 
 var (
@@ -51,9 +51,15 @@ const (
 func init() {
 	processors.RegisterPlugin(procName,
 		checks.ConfigChecked(New,
-			checks.RequireFields("fields"),
-			checks.AllowedFields("fields", "overwrite_keys", "add_error_key", "target", "document_id")))
+			checks.RequireFields("field"),
+			checks.AllowedFields(
+				"field", "target_field",
+				"overwrite_keys", "document_id",
+				"to_lower", "ignore_missing",
+				"ignore_failure", "schema",
+			)))
 	jsprocessor.RegisterPlugin(procName, New)
+	registerDecoders()
 }
 
 // New constructs a new decode_xml processor.
@@ -77,6 +83,7 @@ func newDecodeXML(config decodeXMLConfig) (processors.Processor, error) {
 
 	return &decodeXML{
 		decodeXMLConfig: config,
+		decode:          newDecoder(config),
 		log:             logp.NewLogger(logName),
 	}, nil
 }
@@ -104,7 +111,7 @@ func (x *decodeXML) run(event *beat.Event) error {
 		return errFieldIsNotString
 	}
 
-	xmlOutput, err := x.decodeField(text)
+	xmlOutput, err := x.decode([]byte(text))
 	if err != nil {
 		return err
 	}
@@ -131,19 +138,6 @@ func (x *decodeXML) run(event *beat.Event) error {
 	return nil
 }
 
-func (x *decodeXML) decodeField(data string) (decodedData map[string]interface{}, err error) {
-	dec := xml.NewDecoder(strings.NewReader(data))
-	if x.ToLower {
-		dec.LowercaseKeys()
-	}
-
-	out, err := dec.Decode()
-	if err != nil {
-		return nil, fmt.Errorf("error decoding XML field: %w", err)
-	}
-	return out, nil
-}
-
 func (x *decodeXML) String() string {
 	json, _ := json.Marshal(x.decodeXMLConfig)
 	return procName + "=" + string(json)

libbeat/processors/decode_xml/decode_xml_test.go

@@ -57,7 +57,7 @@ func TestDecodeXML(t *testing.T) {
 				</catalog>`,
 			},
 			Output: common.MapStr{
-				"xml": map[string]interface{}{
+				"xml": common.MapStr{
 					"catalog": map[string]interface{}{
 						"book": map[string]interface{}{
 							"author": "William H. Gaddis",
@@ -125,7 +125,7 @@ func TestDecodeXML(t *testing.T) {
 				</catalog>`,
 			},
 			Output: common.MapStr{
-				"message": map[string]interface{}{
+				"message": common.MapStr{
 					"catalog": map[string]interface{}{
 						"book": map[string]interface{}{
 							"author": "William H. Gaddis",
@@ -158,7 +158,7 @@ func TestDecodeXML(t *testing.T) {
 				</catalog>`,
 			},
 			Output: common.MapStr{
-				"message": map[string]interface{}{
+				"message": common.MapStr{
 					"catalog": map[string]interface{}{
 						"book": []interface{}{
 							map[string]interface{}{
@@ -203,7 +203,7 @@ func TestDecodeXML(t *testing.T) {
 				</catalog>`,
 			},
 			Output: common.MapStr{
-				"message": map[string]interface{}{
+				"message": common.MapStr{
 					"catalog": map[string]interface{}{
 						"book": []interface{}{
 							map[string]interface{}{

libbeat/processors/decode_xml/docs/decode_xml.asciidoc

@@ -55,15 +55,13 @@ Example XML input:
 
 [source,xml]
 -------------------------------------------------------------------------------
-{
-  <catalog>
-    <book seq="1">
-      <author>William H. Gaddis</author>
-      <title>The Recognitions</title>
-      <review>One of the great seminal American novels of the 20th century.</review>
-    </book>
-  </catalog>
-}
+<catalog>
+  <book seq="1">
+    <author>William H. Gaddis</author>
+    <title>The Recognitions</title>
+    <review>One of the great seminal American novels of the 20th century.</review>
+  </book>
+</catalog>
 -------------------------------------------------------------------------------
 
 Will produce the following output:
@@ -97,10 +95,13 @@ value (`target_field:`) is treated as if the field was not set at all.
 
 `overwrite_keys`:: (Optional) A boolean that specifies whether keys that already
 exist in the event are overwritten by keys from the decoded XML object. The
-default value is false.
+default value is `true`.
 
 `to_lower`:: (Optional) Converts all keys to lowercase. Accepts either true or
-false. The default value is true.
+false. The default value is `true`.
+
+`schema`:: (Optional) Specifies the schema of the message. Accepted schemas: `wineventlog`.
+The default value is ``.
 
 `document_id`:: (Optional) XML key to use as the document ID. If configured, the
 field will be removed from the original XML document and stored in

libbeat/processors/decode_xml/schema.go

@@ -0,0 +1,94 @@
+// 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 decode_xml
+
+import (
+	"bytes"
+	"errors"
+	"fmt"
+
+	"github.com/elastic/beats/v7/libbeat/common"
+	"github.com/elastic/beats/v7/libbeat/common/encoding/xml"
+	"github.com/elastic/beats/v7/libbeat/logp"
+	"github.com/elastic/beats/v7/winlogbeat/sys/winevent"
+)
+
+type newDecoderFunc func(cfg decodeXMLConfig) decoder
+type decoder func(p []byte) (common.MapStr, error)
+
+var (
+	registeredDecoders                = map[string]newDecoderFunc{}
+	newDefaultDecoder  newDecoderFunc = newSchemaLessDecoder
+)
+
+func registerDecoder(schema string, dec newDecoderFunc) error {
+	if schema == "" {
+		return errors.New("schema can't be empty")
+	}
+
+	if dec == nil {
+		return errors.New("decoder can't be nil")
+	}
+
+	if _, found := registeredDecoders[schema]; found {
+		return errors.New("already registered")
+	}
+
+	registeredDecoders[schema] = dec
+
+	return nil
+}
+
+func newDecoder(cfg decodeXMLConfig) decoder {
+	newDec, found := registeredDecoders[cfg.Schema]
+	if !found {
+		return newDefaultDecoder(cfg)
+	}
+	return newDec(cfg)
+}
+
+func registerDecoders() {
+	log := logp.L().Named(logName)
+	log.Debug(registerDecoder("wineventlog", newWineventlogDecoder))
+}
+
+func newSchemaLessDecoder(cfg decodeXMLConfig) decoder {
+	return func(p []byte) (common.MapStr, error) {
+		dec := xml.NewDecoder(bytes.NewReader(p))
+		if cfg.ToLower {
+			dec.LowercaseKeys()
+		}
+
+		out, err := dec.Decode()
+		if err != nil {
+			return nil, fmt.Errorf("error decoding XML field: %w", err)
+		}
+
+		return common.MapStr(out), nil
+	}
+}
+
+func newWineventlogDecoder(decodeXMLConfig) decoder {
+	return func(p []byte) (common.MapStr, error) {
+		evt, err := winevent.UnmarshalXML(p)
+		if err != nil {
+			return nil, err
+		}
+		return evt.Fields(), nil
+	}
+}

libbeat/processors/decode_xml_wineventlog/config.go

@@ -0,0 +1,33 @@
+// 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 decode_xml_wineventlog
+
+type config struct {
+	Field         string  `config:"field" validate:"required"`
+	Target        *string `config:"target_field"`
+	OverwriteKeys bool    `config:"overwrite_keys"`
+	IgnoreMissing bool    `config:"ignore_missing"`
+	IgnoreFailure bool    `config:"ignore_failure"`
+}
+
+func defaultConfig() config {
+	return config{
+		Field:         "message",
+		OverwriteKeys: true,
+	}
+}