[Filebeat] Add mime type detection for Elasticsearch module (#22975)

* Add mime type resolution for elasticsearch filebeat module

* Update changelog

CHANGELOG.next.asciidoc

@@ -100,6 +100,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - API address and shard ID are required settings in the Cloud Foundry input. {pull}21759[21759]
 - Remove `suricata.eve.timestamp` alias field. {issue}10535[10535] {pull}22095[22095]
 - Rename bad ECS field name tracing.trace.id to trace.id in aws elb fileset. {pull}22571[22571]
+- Fix parsing issues with nested JSON payloads in Elasticsearch audit log fileset. {pull}22975[22975]
 
 *Heartbeat*
 
@@ -738,6 +739,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Add support for UNIX datagram sockets in `unix` input. {issues}18632[18632] {pull}22699[22699]
 - Add new httpjson input features and mark old config ones for deprecation {pull}22320[22320]
 - Add logic for external network.direction in sophos xg fileset {pull}22973[22973]
+- Add `http.request.mime_type` for Elasticsearch audit log fileset. {pull}22975[22975]
 
 *Heartbeat*
 

filebeat/module/elasticsearch/audit/config/audit.yml

@@ -11,3 +11,39 @@ processors:
       target: ''
       fields:
         ecs.version: 1.7.0
+  - if:
+      regexp:
+        message: "^{"
+    then:
+      - decode_json_fields:
+          fields: [ "message" ]
+          target: _json
+      - rename:
+          fields:
+            - from: _json.request.body
+              to: _request
+          ignore_missing: true
+      - drop_fields:
+          fields: [ "_json" ]
+    else:
+      - script:
+          lang: javascript
+          id: elasticsearch_audit
+          source: >
+            var requestRegex = new RegExp("request_body=\\\[(.*)\\\]$");
+            function process(event) {
+              var message = event.Get("message");
+              if (message !== null) {
+                var matches = message.match(requestRegex);
+                if (matches && matches.length > 1) {
+                  event.Put("_request", matches[1]);
+                }
+              }
+            }
+  - detect_mime_type:
+      field: _request
+      target: http.request.mime_type
+  - drop_fields:
+      fields: ['_request']
+      ignore_missing: true
+

filebeat/module/elasticsearch/audit/ingest/pipeline-plaintext.yml

@@ -19,7 +19,7 @@ processors:
       ES_AUDIT_REQUEST: (request\=\[%{WORD:elasticsearch.audit.request.name}\])?
       ES_AUDIT_REQUEST_BODY: (request_body\=\[%{DATA:http.request.body.content}\])?
     patterns:
-    - '%{ES_TIMESTAMP}\s*%{ES_NODE_NAME}\s*%{ES_AUDIT_LAYER}\s*%{ES_AUDIT_EVENT_TYPE}\s*%{ES_AUDIT_ORIGIN_TYPE},?\s*%{ES_AUDIT_ORIGIN_ADDRESS},?\s*%{ES_AUDIT_PRINCIPAL},?\s*%{ES_AUDIT_REALM},?\s*%{ES_AUDIT_ROLES},?\s*%{ES_AUDIT_ACTION},?\s*%{ES_AUDIT_INDICES},?\s*%{ES_AUDIT_URI},?\s*%{ES_AUDIT_URI_PARAMS},?\s*%{ES_AUDIT_REQUEST},?\s*%{ES_AUDIT_REQUEST_BODY},?'
+    - '%{ES_TIMESTAMP}\s*%{ES_NODE_NAME}\s*%{ES_AUDIT_LAYER}\s*%{ES_AUDIT_EVENT_TYPE}\s*%{ES_AUDIT_ORIGIN_TYPE},?\s*%{ES_AUDIT_ORIGIN_ADDRESS},?\s*%{ES_AUDIT_PRINCIPAL},?\s*%{ES_AUDIT_REALM},?\s*%{ES_AUDIT_ROLES},?\s*%{ES_AUDIT_ACTION},?\s*%{ES_AUDIT_INDICES},?\s*%{ES_AUDIT_URI},?\s*%{ES_AUDIT_URI_PARAMS},?\s*%{ES_AUDIT_REQUEST},?\s*%{ES_AUDIT_REQUEST_BODY}$'
 - split:
     field: elasticsearch.audit.user.roles
     separator: ','

filebeat/module/elasticsearch/audit/test/test-access.log-expected.json

@@ -153,6 +153,7 @@
         "event.type": "access",
         "fileset.name": "audit",
         "http.request.body.content": "body",
+        "http.request.mime_type": "text/plain; charset=utf-8",
         "input.type": "log",
         "log.offset": 986,
         "message": "[2018-06-19T05:24:15,190] [v_VJhjV] [rest] [authentication_failed]\torigin_address=[172.18.0.3], principal=[elastic], uri=[/_nodes?filter_path=nodes.*.version%2Cnodes.*.http.publish_address%2Cnodes.*.ip], request_body=[body]",
@@ -219,7 +220,8 @@
         "event.timezone": "-02:00",
         "event.type": "access",
         "fileset.name": "audit",
-        "http.request.body.content": "{\"metadata\":{\"intelligence\":7},\"full_name\":\"Jack Nicholson\",\"roles\":[\"admin\",\"other_role1\"",
+        "http.request.body.content": "{\"metadata\":{\"intelligence\":7},\"full_name\":\"Jack Nicholson\",\"roles\":[\"admin\",\"other_role1\"],\"email\":\"jacknich@example.com\"}",
+        "http.request.mime_type": "application/json",
         "input.type": "log",
         "log.offset": 1626,
         "message": "[2019-01-27T20:04:27,244] [node-0] [rest] [authentication_success]    origin_address=[::1], principal=[elastic-admin], realm=[default_file], uri=[/_xpack/security/user/jacknich2], params=[{username=jacknich2}], request_body=[{\"metadata\":{\"intelligence\":7},\"full_name\":\"Jack Nicholson\",\"roles\":[\"admin\",\"other_role1\"],\"email\":\"jacknich@example.com\"}]",

filebeat/module/elasticsearch/audit/test/test-audit.log-expected.json

@@ -203,6 +203,7 @@
         "host.id": "y8fa3M5zSSGo1M_KJRMUXw",
         "http.request.body.content": "\n{\n    \"query\" : {\n        \"term\" : { \"user\" : \"kimchy\" }\n    }\n}\n",
         "http.request.method": "GET",
+        "http.request.mime_type": "application/json",
         "input.type": "log",
         "log.offset": 2056,
         "message": "{\"@timestamp\":\"2019-01-27T20:15:10,380\", \"node.name\":\"node-0\", \"node.id\":\"y8fa3M5zSSGo1M_KJRMUXw\", \"event.type\":\"rest\", \"event.action\":\"authentication_success\", \"user.name\":\"elastic-admin\", \"origin.type\":\"rest\", \"origin.address\":\"[::1]:58955\", \"realm\":\"default_file\", \"url.path\":\"/_search\", \"request.method\":\"GET\", \"request.body\":\"\\n{\\n    \\\"query\\\" : {\\n        \\\"term\\\" : { \\\"user\\\" : \\\"kimchy\\\" }\\n    }\\n}\\n\", \"request.id\":\"WzL_kb6VSvOhAq0twPvHOQ\"}",