[cef] Generate processor tags and normalize error handler (#15526)

- Generate tags for processors missing tags
- Normalize the pipeline error handler

Author: taylor-swanson

packages/cef/changelog.yml

@@ -1,4 +1,9 @@
 # newer versions go on top
+- version: "2.21.3"
+  changes:
+    - description: Generate processor tags and normalize error handler.
+      type: enhancement
+      link: https://github.com/elastic/integrations/pull/15526
 - version: "2.21.2"
   changes:
     - description: Changed owners.

packages/cef/data_stream/log/elasticsearch/ingest_pipeline/cp-pipeline.yml

@@ -276,26 +276,32 @@ processors:
         ctx["_tmp_copy"] = actions;
 
   - foreach:
+      tag: foreach__tmp_copy_ac4d5b1c
       field: _tmp_copy
       processor:
         set:
           field: '{{{_ingest._value.to}}}'
           value: '{{{_ingest._value.value}}}'
   - remove:
+      tag: remove__tmp_copy_02a3c909
       field: _tmp_copy
   - set:
+      tag: set_email_to_address_a2fb0fab
       if: ctx?.destination?.user?.email != null
       field: email.to.address
       value: ['{{{destination.user.email}}}']
   - set:
+      tag: set_email_from_address_865dddfa
       if: ctx?.source?.user?.email != null
       field: email.from.address
       value: ['{{{source.user.email}}}']
   - set:
+      tag: set_email_subject_d55a1a90
       if: ctx?.checkpoint?.email_subject != null
       field: email.subject
       copy_from: checkpoint.email_subject
   - set:
+      tag: set_email_message_id_70b21e23
       if: ctx?.checkpoint?.email_session_id != null
       field: email.message_id
       copy_from: checkpoint.email_session_id
@@ -306,6 +312,7 @@ processors:
       type: float
       on_failure:
         - remove:
+            tag: remove_event_risk_score_6ee4e8c9
             field: event.risk_score
   - convert:
       field: event.severity
@@ -314,6 +321,7 @@ processors:
       type: long
       on_failure:
         - remove:
+            tag: remove_event_severity_8251ac98
             field: event.severity
   # event.duration is a string and contains seconds. Convert to long nanos.
   - script:
@@ -327,41 +335,50 @@ processors:
       tag: calculate duration
       on_failure:
         - remove:
+            tag: remove_event_duration_8107f601
             field: event.duration
             ignore_missing: true
   # checkpoint.file_hash can be either MD5, SHA1 or SHA256.
   - rename:
+      tag: rename_checkpoint_file_hash_to_file_hash_md5_00c600f1
       if: ctx.checkpoint?.file_hash != null && ctx.checkpoint.file_hash.length()==32
       field: checkpoint.file_hash
       target_field: file.hash.md5
   - rename:
+      tag: rename_checkpoint_file_hash_to_file_hash_sha1_5b63a3eb
       if: ctx.checkpoint?.file_hash != null && ctx.checkpoint.file_hash.length()==40
       field: checkpoint.file_hash
       target_field: file.hash.sha1
   - rename:
+      tag: rename_checkpoint_file_hash_to_file_hash_sha256_9082ab2f
       if: ctx.checkpoint?.file_hash != null && ctx.checkpoint.file_hash.length()==64
       field: checkpoint.file_hash
       target_field: file.hash.sha256
   # Event kind is 'event' by default. 'alert' when a risk score and rule info
   # is present.
   - set:
+      tag: set_event_kind_de80643c
       field: event.kind
       value: event
   - set:
+      tag: set_event_kind_97cac9f4
       if: ctx.cef?.extensions?.cp_app_risk != null && ctx.rule != null
       field: event.kind
       value: alert
   # Set event.category to network/malware/intrusion_detection depending on which
   # fields have been populated.
   - append:
+      tag: append_event_category_01d5fb4a
       if: ctx.source?.ip != null && ctx.destination?.ip != null
       field: event.category
       value: network
   - append:
+      tag: append_event_category_3f3c21f2
       if: ctx.checkpoint?.protection_id != null || ctx.checkpoint?.spyware_name != null || ctx.checkpoint?.malware_family != null || ctx.checkpoint?.spyware_status != null
       field: event.category
       value: malware
   - append:
+      tag: append_event_category_762d9429
       if: ctx.event?.category != null && !(ctx.event.action.contains("malware")) && (ctx.checkpoint?.protection_type != null || ctx.cef.extensions?.flexString2Label == "Attack Information")
       field: event.category
       value: intrusion_detection
@@ -378,8 +395,11 @@ processors:
 on_failure:
   - append:
       field: error.message
-      value: |-
-        Processor "{{{ _ingest.on_failure_processor_type }}}" with tag "{{{ _ingest.on_failure_processor_tag }}}" in pipeline "{{{ _ingest.on_failure_pipeline }}}" failed with message "{{{ _ingest.on_failure_message }}}"
+      value: >-
+        Processor '{{{ _ingest.on_failure_processor_type }}}'
+        {{#_ingest.on_failure_processor_tag}}with tag '{{{ _ingest.on_failure_processor_tag }}}'
+        {{/_ingest.on_failure_processor_tag}}in pipeline '{{{ _ingest.pipeline }}}'
+        failed with message '{{{ _ingest.on_failure_message }}}'
   - set:
       field: event.kind
       value: pipeline_error

packages/cef/data_stream/log/elasticsearch/ingest_pipeline/default.yml

@@ -2,6 +2,7 @@
 description: Pipeline for CEF logs. CEF decoding happens in the Agent. This performs additional enrichment and vendor specific transformations.
 processors:
   - set:
+      tag: set_ecs_version_f5923549
       field: ecs.version
       value: 8.17.0
   - convert:
@@ -22,6 +23,7 @@ processors:
       target_field: destination.geo
   # IP Autonomous System (AS) Lookup
   - geoip:
+      tag: geoip_source_ip_to_source_as_28d69883
       field: source.ip
       database_file: GeoLite2-ASN.mmdb
       ignore_missing: true
@@ -30,6 +32,7 @@ processors:
       - organization_name
       target_field: source.as
   - geoip:
+      tag: geoip_destination_ip_to_destination_as_8a007787
       field: destination.ip
       database_file: GeoLite2-ASN.mmdb
       ignore_missing: true
@@ -38,61 +41,74 @@ processors:
       - organization_name
       target_field: destination.as
   - rename:
+      tag: rename_source_as_asn_to_source_as_number_a917047d
       field: source.as.asn
       ignore_missing: true
       target_field: source.as.number
   - rename:
+      tag: rename_source_as_organization_name_to_source_as_organization_name_f1362d0b
       field: source.as.organization_name
       ignore_missing: true
       target_field: source.as.organization.name
   - rename:
+      tag: rename_destination_as_asn_to_destination_as_number_3b459fcd
       field: destination.as.asn
       ignore_missing: true
       target_field: destination.as.number
   - rename:
+      tag: rename_destination_as_organization_name_to_destination_as_organization_name_814bd459
       field: destination.as.organization_name
       ignore_missing: true
       target_field: destination.as.organization.name
   - append:
+      tag: append_related_hash_592251e0
       if: ctx?.cef?.extensions?.fileHash != null && ctx?.cef?.extensions?.fileHash != ''
       field: related.hash
       allow_duplicates: false
       value: '{{{cef.extensions.fileHash}}}'
   - append:
+      tag: append_related_hash_be4900bb
       if: ctx?.cef?.extensions?.oldFileHash != null && ctx?.cef?.extensions?.oldFileHash != ''
       field: related.hash
       allow_duplicates: false
       value: '{{{cef.extensions.oldFileHash}}}'
   - append:
+      tag: append_related_ip_73d5506a
       if: ctx?.destination?.ip != null && ctx?.destination?.ip != ''
       field: related.ip
       allow_duplicates: false
       value: '{{{destination.ip}}}'
   - append:
+      tag: append_related_ip_3da81053
       if: ctx?.destination?.nat?.ip != null && ctx?.destination?.nat?.ip != ''
       field: related.ip
       allow_duplicates: false
       value: '{{{destination.nat.ip}}}'
   - append:
+      tag: append_related_ip_74a1d0ad
       if: ctx?.source?.ip != null && ctx?.source?.ip != ''
       field: related.ip
       allow_duplicates: false
       value: '{{{source.ip}}}'
   - append:
+      tag: append_related_ip_78428a7e
       if: ctx?.source?.nat?.ip != null && ctx?.source?.nat?.ip != ''
       field: related.ip
       allow_duplicates: false
       value: '{{{source.nat.ip}}}'
   - append:
+      tag: append_related_user_75c0abfc
       if: ctx?.destination?.user?.name != null
       field: related.user
       value: '{{{destination.user.name}}}'
   - append:
+      tag: append_related_user_afcddc50
       if: ctx?.source?.user?.name != null && ctx?.source?.user?.name != ''
       field: related.user
       allow_duplicates: false
       value: '{{{source.user.name}}}'
   - append:
+      tag: append_related_hosts_c0e4bbd6
       if: ctx?.observer?.hostname != null && ctx?.observer?.hostname != ''
       field: related.hosts
       allow_duplicates: false
@@ -121,11 +137,13 @@ processors:
       ignore_missing: true
       pattern: '[:.]'
       replacement: '-'
-      tag: gsub mac
+      tag: gsub_source_mac_1b14f2a4
   - uppercase:
+      tag: uppercase_destination_mac_04de3657
       field: destination.mac
       ignore_missing: true
   - uppercase:
+      tag: uppercase_source_mac_5b4e7be2
       field: source.mac
       ignore_missing: true
   #
@@ -146,17 +164,20 @@ processors:
         - '^%{ECS_SYSLOG_PRI}%{NONNEGINT} %{SYSLOG_TIMESTAMP} ' # RFC5224
       tag: timestamp extract grok
   - date:
+      tag: date__tmp_timestamp8601_c38c0806
       if: ctx?._tmp?.timestamp8601 != null
       field: _tmp.timestamp8601
       formats:
         - ISO8601
   - date:
+      tag: date__tmp_timestamp_771f472e
       if: ctx?._tmp?.timestamp != null
       field: _tmp.timestamp
       formats:
         - MMM  d HH:mm:ss
         - MMM dd HH:mm:ss
   - remove:
+      tag: remove_event_original_9f895a30
       if: ctx?.tags == null || !(ctx.tags.contains('preserve_original_event'))
       field: event.original
       ignore_failure: true
@@ -176,15 +197,18 @@ processors:
       value: '{{{_tmp.observer}}}'
   # Set ECS event outcome from ArcSight outcomes
   - set:
+      tag: set_event_outcome_da09bb71
       if: ctx.cef?.extensions?.categoryOutcome == "/Success"
       field: event.outcome
       value: success
   - set:
+      tag: set_event_outcome_cb9ede45
       if: ctx.cef?.extensions?.categoryOutcome == "/Failure"
       field: event.outcome
       value: failure
   # Cleanup
   - remove:
+      tag: remove_8c701636
       field:
         - cef.extensions._cefVer
         - _tmp
@@ -196,8 +220,11 @@ on_failure:
       ignore_missing: true
   - append:
       field: error.message
-      value: |-
-        Processor "{{{ _ingest.on_failure_processor_type }}}" with tag "{{{ _ingest.on_failure_processor_tag }}}" in pipeline "{{{ _ingest.on_failure_pipeline }}}" failed with message "{{{ _ingest.on_failure_message }}}"
+      value: >-
+        Processor '{{{ _ingest.on_failure_processor_type }}}'
+        {{#_ingest.on_failure_processor_tag}}with tag '{{{ _ingest.on_failure_processor_tag }}}'
+        {{/_ingest.on_failure_processor_tag}}in pipeline '{{{ _ingest.pipeline }}}'
+        failed with message '{{{ _ingest.on_failure_message }}}'
   - set:
       field: event.kind
       value: pipeline_error

packages/cef/data_stream/log/elasticsearch/ingest_pipeline/fp-pipeline.yml

@@ -3,29 +3,36 @@ description: Pipeline for Forcepoint CEF
 processors:
   # cs1 is ruleID
   - set:
+      tag: set_rule_id_7a577460
       field: rule.id
       ignore_empty_value: true
       value: '{{{cef.extensions.deviceCustomString1}}}'
   # cs2 is natRuleID
   - set:
+      tag: set_rule_id_c76c7491
       field: rule.id
       ignore_empty_value: true
       value: '{{{cef.extensions.deviceCustomString2}}}'
   # cs3 is VulnerabilityReference
   - set:
+      tag: set_vulnerability_reference_0b703e9a
       field: vulnerability.reference
       ignore_empty_value: true
       value: '{{{cef.extensions.deviceCustomString3}}}'
   # cs4 is virusID
   - set:
+      tag: set_cef_forcepoint_virus_id_ce0473c6
       field: cef.forcepoint.virus_id
       ignore_empty_value: true
       value: '{{{cef.extensions.deviceCustomString4}}}'
 on_failure:
   - append:
       field: error.message
-      value: |-
-        Processor "{{{ _ingest.on_failure_processor_type }}}" with tag "{{{ _ingest.on_failure_processor_tag }}}" in pipeline "{{{ _ingest.on_failure_pipeline }}}" failed with message "{{{ _ingest.on_failure_message }}}"
+      value: >-
+        Processor '{{{ _ingest.on_failure_processor_type }}}'
+        {{#_ingest.on_failure_processor_tag}}with tag '{{{ _ingest.on_failure_processor_tag }}}'
+        {{/_ingest.on_failure_processor_tag}}in pipeline '{{{ _ingest.pipeline }}}'
+        failed with message '{{{ _ingest.on_failure_message }}}'
   - set:
       field: event.kind
       value: pipeline_error

packages/cef/manifest.yml

@@ -1,6 +1,6 @@
 name: cef
 title: Common Event Format (CEF)
-version: "2.21.2"
+version: "2.21.3"
 description: Collect logs from CEF Logs with Elastic Agent.
 categories:
   - security