[Filebeat] Replace copy_from with templated value (#26631)

* Replace copy_from with templated value To ensure compatibility with Elasticsearch versions <7.13 this removes usage of `copy_from` in `set` processors. Relates #26629 * panw: replace copy_from usage with script Replaces the usage of a set processor with copy_from (ES 7.13+) with a painless script that performs the same operation and it's backwards compatible. * cyberarkpas: Replace usage of copy_from with script This updates the ID-mapping script to set fields instead of constructing and op-list that is latter processed with foreach/set. * Update CHANGELOG.next.asciidoc Co-authored-by: Adrian Serrano <adrisr83@gmail.com> Co-authored-by: Adrian Serrano <adrisr83@gmail.com> (cherry picked from commit a7b0110)
elastic · Jul 19, 2021 · b9d9638 · b9d9638
1 parent cda9e92
commit b9d9638
Show file tree

Hide file tree

Showing 12 changed files with 74 additions and 34 deletions.
diff --git a/CHANGELOG.next.asciidoc b/CHANGELOG.next.asciidoc
@@ -229,6 +229,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Fix default config template values for paths on oracle module: {pull}26276[26276]
 - Do not close filestream harvester if an unexpected error is returned when close.on_state_change.* is enabled. {pull}26411[26411]
 - Change type of max_bytes in all configs to be cfgtype.ByteSize {pull}26699[26699]
+- Fix Elasticsearch compatibility for modules that use `copy_from` in `set` processors. {issue}26629[26629]
 
 *Filebeat*
 

diff --git a/filebeat/module/elasticsearch/audit/ingest/pipeline-json.yml b/filebeat/module/elasticsearch/audit/ingest/pipeline-json.yml
@@ -156,7 +156,7 @@ processors:
       path: elasticsearch.audit
   - set:
       field: http.request.id
-      copy_from: elasticsearch.audit.request.id
+      value: '{{{elasticsearch.audit.request.id}}}'
       ignore_empty_value: true
   - dot_expander:
       field: cluster.name

diff --git a/filebeat/module/nginx/ingress_controller/ingest/pipeline.yml b/filebeat/module/nginx/ingress_controller/ingest/pipeline.yml
@@ -73,7 +73,7 @@ processors:
       value: ""
   - set:
       field: http.request.id
-      copy_from: nginx.ingress_controller.http.request.id
+      value: '{{{nginx.ingress_controller.http.request.id}}}'
       ignore_empty_value: true
       ignore_failure: true
   - script:

diff --git a/x-pack/filebeat/module/cyberarkpas/audit/ingest/pipeline.yml b/x-pack/filebeat/module/cyberarkpas/audit/ingest/pipeline.yml
@@ -914,40 +914,59 @@ processors:
             value: "success"
           - set: event.reason
             from: cyberarkpas.audit.reason
+      on_failure:
+        - append:
+            field: error.message
+            value: 'Failed to enrich based on ID #{{{ event.code }}}: {{{_ingest.on_failure_message}}}'
 
       source: >
-        def clone(def val) {
-            return val instanceof List? new ArrayList(val) : val;
+        def clone(def ref) {
+          if (ref == null) return ref;
+          if (ref instanceof Map) {
+            ref = ref.entrySet().stream().collect(
+                    Collectors.toMap(
+                      e -> e.getKey(),
+                      e -> clone(e.getValue())
+                    )
+                  );
+          } else if (ref instanceof List) {
+            ref = ref.stream().map(e -> clone(e)).collect(
+                    Collectors.toList()
+                  );
+          }
+          return ref;
         }
         def read_field(def map, String name) {
           if (map == null || !(map instanceof Map)) return null;
           int pos = name.indexOf(".");
           return pos == -1? map[name]
                           : read_field(map[name.substring(0, pos)], name.substring(pos+1));
         }
+        boolean set_field(Map map, String name, def value) {
+          int pos = name.indexOf(".");
+          if (pos == -1) {
+            map[name] = clone(value);
+            return true;
+          }
+          String key  = name.substring(0, pos),
+                 path = name.substring(pos+1);
+          if (!map.containsKey(key)) {
+            map[key] = new HashMap();
+          }
+          map = map[key];
+          return map instanceof Map? set_field(map, path, value)
+                                   : false;
+        }
         String msgID = ctx.event?.code;
         def actions = params.get(msgID);
         if (actions == null) return;
-        List values = new ArrayList();
         for (def item : actions) {
           def val = item.value;
           if (val == null && (val = read_field(ctx, item.from)) == null || val == "") continue;
-          values.add([
-            "to": item.set,
-            "value": clone(val)
-          ]);
+          if (!set_field(ctx, item.set, val)) {
+            throw new Exception("Failed to set field " + item.set);
+          }
         }
-        if (!values.isEmpty()) ctx._tmp["values"] = values;
-
-  - foreach:
-      field: _tmp.values
-      ignore_missing: true
-      processor:
-        set:
-          field: '{{{_ingest._value.to}}}'
-          copy_from: '_ingest._value.value'
-          ignore_empty_value: true
-          override: true
 
   #
   # Force event.outcome: unknown in case it gets a value other than one of the allowed.
@@ -994,7 +1013,7 @@ processors:
       on_failure:
         - set:
             field: source.domain
-            copy_from: source.address
+            value: '{{{source.address}}}'
   - convert:
       field: destination.address
       target_field: destination.ip
@@ -1003,7 +1022,7 @@ processors:
       on_failure:
         - set:
             field: destination.domain
-            copy_from: destination.address
+            value: '{{{destination.address}}}'
 
   #
   # Populate related.ip

diff --git a/x-pack/filebeat/module/panw/panos/ingest/pipeline.yml b/x-pack/filebeat/module/panw/panos/ingest/pipeline.yml
@@ -462,10 +462,30 @@ processors:
         ANY: '.*'
       if: 'ctx?.file?.name != null && ctx?.file?.name != ""'
 
-  - set:
-      field: user
-      copy_from: source.user
-      if: "ctx?.source?.user != null"
+  - script:
+      lang: painless
+      description: Copy source.user to user
+      source: >
+        def clone(def ref) {
+          if (ref == null) return ref;
+          if (ref instanceof Map) {
+            ref = ref.entrySet().stream().collect(
+                    Collectors.toMap(
+                      e -> e.getKey(),
+                      e -> clone(e.getValue())
+                    )
+                  );
+          } else if (ref instanceof List) {
+            ref = ref.stream().map(e -> clone(e)).collect(
+                    Collectors.toList()
+                  );
+          }
+          return ref;
+        }
+        def u = ctx?.source?.user;
+        if (u != null) {
+          ctx["user"] = clone(u);
+        }
 
   - append:
       field: related.user

diff --git a/x-pack/filebeat/module/threatintel/abuseurl/ingest/pipeline.yml b/x-pack/filebeat/module/threatintel/abuseurl/ingest/pipeline.yml
@@ -45,7 +45,7 @@ processors:
     remove_if_successful: true
 - set:
     field: threatintel.indicator.url.full
-    copy_from: threatintel.indicator.url.original
+    value: '{{{threatintel.indicator.url.original}}}'
     ignore_empty_value: true
 - rename:
     field: threatintel.abuseurl.host

diff --git a/x-pack/filebeat/module/threatintel/anomali/ingest/pipeline.yml b/x-pack/filebeat/module/threatintel/anomali/ingest/pipeline.yml
@@ -84,7 +84,7 @@ processors:
     if: ctx?.threatintel?.indicator?.type == 'url'
 - set:
     field: threatintel.indicator.url.full
-    copy_from: threatintel.indicator.url.original
+    value: '{{{threatintel.indicator.url.original}}}'
     ignore_empty_value: true
 - rename:
     field: _tmp.threatvalue

diff --git a/x-pack/filebeat/module/threatintel/anomalithreatstream/ingest/pipeline.yml b/x-pack/filebeat/module/threatintel/anomalithreatstream/ingest/pipeline.yml
@@ -265,7 +265,7 @@ processors:
 
 - set:
     field: threatintel.indicator.url.full
-    copy_from: threatintel.indicator.url.original
+    value: '{{{threatintel.indicator.url.original}}}'
     ignore_empty_value: true
 
 - rename:

diff --git a/x-pack/filebeat/module/threatintel/misp/ingest/pipeline.yml b/x-pack/filebeat/module/threatintel/misp/ingest/pipeline.yml
@@ -139,7 +139,7 @@ processors:
 
 - set:
     field: threatintel.indicator.url.full
-    copy_from: threatintel.indicator.url.original
+    value: '{{{threatintel.indicator.url.original}}}'
     ignore_empty_value: true
     if: "ctx?.threatintel?.indicator?.type == 'url' && ctx?.threatintel?.misp?.attribute?.type != 'uri'"
 

diff --git a/x-pack/filebeat/module/threatintel/otx/ingest/pipeline.yml b/x-pack/filebeat/module/threatintel/otx/ingest/pipeline.yml
@@ -87,7 +87,7 @@ processors:
     if: ctx?.threatintel?.indicator?.type == 'url'
 - set:
     field: threatintel.indicator.url.full
-    copy_from: threatintel.indicator.url.original
+    value: '{{{threatintel.indicator.url.original}}}'
     ignore_empty_value: true
     if: "ctx?.threatintel?.otx?.type == 'URL'"
 

diff --git a/x-pack/filebeat/module/threatintel/recordedfuture/ingest/pipeline.yml b/x-pack/filebeat/module/threatintel/recordedfuture/ingest/pipeline.yml
@@ -122,7 +122,7 @@ processors:
 
   - set:
       field: json.ip_range
-      copy_from: json.entity.name
+      value: '{{{json.entity.name}}}'
       if: 'ctx.json.entity?.type == "IpAddress" && ctx.json.entity.name != null && ctx.json.entity.name.contains("/")'
   - set:
       field: json.ip_range
@@ -134,7 +134,7 @@ processors:
       if: 'ctx.threatintel?.indicator?.type == "ipv6-addr" && ctx.json.entity.name != null && !ctx.json.entity.name.contains("/")'
   - set:
       field: json.ip_range
-      copy_from: json.entity.name
+      value: '{{{json.entity.name}}}'
       if: 'ctx.json.entity?.type == "IpAddress" && ctx.json.entity.name != null && ctx.json.entity.name.contains("/")'
 
   - rename:

diff --git a/x-pack/filebeat/module/zoom/webhook/ingest/meeting.yml b/x-pack/filebeat/module/zoom/webhook/ingest/meeting.yml
@@ -51,7 +51,7 @@ processors:
     if: ctx?.url?.original != null
 - set:
     field: url.full
-    copy_from: url.original
+    value: '{{{url.original}}}'
     ignore_failure: true
     if: ctx?.url?.original != null
 #