[cef] Generate processor tags and normalize error handler

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

Author: taylor-swanson

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

@@ -281,24 +281,30 @@ processors:
  set:
  field: '{{{_ingest._value.to}}}'
  value: '{{{_ingest._value.value}}}'
+ tag: foreach_ac4d5b1c
  - remove:
  field: _tmp_copy
+ tag: remove_02a3c909
  - set:
  if: ctx?.destination?.user?.email != null
  field: email.to.address
  value: ['{{{destination.user.email}}}']
+ tag: set_a2fb0fab
  - set:
  if: ctx?.source?.user?.email != null
  field: email.from.address
  value: ['{{{source.user.email}}}']
+ tag: set_865dddfa
  - set:
  if: ctx?.checkpoint?.email_subject != null
  field: email.subject
  copy_from: checkpoint.email_subject
+ tag: set_d55a1a90
  - set:
  if: ctx?.checkpoint?.email_session_id != null
  field: email.message_id
  copy_from: checkpoint.email_session_id
+ tag: set_70b21e23
  - convert:
  field: event.risk_score
  ignore_missing: true
@@ -334,37 +340,45 @@ processors:
  if: ctx.checkpoint?.file_hash != null && ctx.checkpoint.file_hash.length()==32
  field: checkpoint.file_hash
  target_field: file.hash.md5
+ tag: rename_00c600f1
  - rename:
  if: ctx.checkpoint?.file_hash != null && ctx.checkpoint.file_hash.length()==40
  field: checkpoint.file_hash
  target_field: file.hash.sha1
+ tag: rename_5b63a3eb
  - rename:
  if: ctx.checkpoint?.file_hash != null && ctx.checkpoint.file_hash.length()==64
  field: checkpoint.file_hash
  target_field: file.hash.sha256
+ tag: rename_9082ab2f
  # Event kind is 'event' by default. 'alert' when a risk score and rule info
  # is present.
  - set:
  field: event.kind
  value: event
+ tag: set_de80643c
  - set:
  if: ctx.cef?.extensions?.cp_app_risk != null && ctx.rule != null
  field: event.kind
  value: alert
+ tag: set_97cac9f4
  # Set event.category to network/malware/intrusion_detection depending on which
  # fields have been populated.
  - append:
  if: ctx.source?.ip != null && ctx.destination?.ip != null
  field: event.category
  value: network
+ tag: append_01d5fb4a
  - append:
  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
+ tag: append_3f3c21f2
  - append:
  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
+ tag: append_762d9429
  - convert:
  field: checkpoint.event_count
  ignore_missing: true
@@ -378,8 +392,10 @@ 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}}failed with message '{{{ _ingest.on_failure_message }}}'
  - set:
  field: event.kind
  value: pipeline_error

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

@@ -4,6 +4,7 @@ processors:
  - set:
  field: ecs.version
  value: 8.17.0
+ tag: set_f5923549
  - convert:
  field: event.id
  ignore_missing: true
@@ -29,6 +30,7 @@ processors:
  - asn
  - organization_name
  target_field: source.as
+ tag: geoip_28d69883
  - geoip:
  field: destination.ip
  database_file: GeoLite2-ASN.mmdb
@@ -37,66 +39,80 @@ processors:
  - asn
  - organization_name
  target_field: destination.as
+ tag: geoip_8a007787
  - rename:
  field: source.as.asn
  ignore_missing: true
  target_field: source.as.number
+ tag: rename_a917047d
  - rename:
  field: source.as.organization_name
  ignore_missing: true
  target_field: source.as.organization.name
+ tag: rename_f1362d0b
  - rename:
  field: destination.as.asn
  ignore_missing: true
  target_field: destination.as.number
+ tag: rename_3b459fcd
  - rename:
  field: destination.as.organization_name
  ignore_missing: true
  target_field: destination.as.organization.name
+ tag: rename_814bd459
  - append:
  if: ctx?.cef?.extensions?.fileHash != null && ctx?.cef?.extensions?.fileHash != ''
  field: related.hash
  allow_duplicates: false
  value: '{{{cef.extensions.fileHash}}}'
+ tag: append_592251e0
  - append:
  if: ctx?.cef?.extensions?.oldFileHash != null && ctx?.cef?.extensions?.oldFileHash != ''
  field: related.hash
  allow_duplicates: false
  value: '{{{cef.extensions.oldFileHash}}}'
+ tag: append_be4900bb
  - append:
  if: ctx?.destination?.ip != null && ctx?.destination?.ip != ''
  field: related.ip
  allow_duplicates: false
  value: '{{{destination.ip}}}'
+ tag: append_73d5506a
  - append:
  if: ctx?.destination?.nat?.ip != null && ctx?.destination?.nat?.ip != ''
  field: related.ip
  allow_duplicates: false
  value: '{{{destination.nat.ip}}}'
+ tag: append_3da81053
  - append:
  if: ctx?.source?.ip != null && ctx?.source?.ip != ''
  field: related.ip
  allow_duplicates: false
  value: '{{{source.ip}}}'
+ tag: append_74a1d0ad
  - append:
  if: ctx?.source?.nat?.ip != null && ctx?.source?.nat?.ip != ''
  field: related.ip
  allow_duplicates: false
  value: '{{{source.nat.ip}}}'
+ tag: append_78428a7e
  - append:
  if: ctx?.destination?.user?.name != null
  field: related.user
  value: '{{{destination.user.name}}}'
+ tag: append_75c0abfc
  - append:
  if: ctx?.source?.user?.name != null && ctx?.source?.user?.name != ''
  field: related.user
  allow_duplicates: false
  value: '{{{source.user.name}}}'
+ tag: append_afcddc50
  - append:
  if: ctx?.observer?.hostname != null && ctx?.observer?.hostname != ''
  field: related.hosts
  allow_duplicates: false
  value: '{{{observer.hostname}}}'
+ tag: append_c0e4bbd6
  - pipeline:
  if: ctx.cef?.device?.vendor == 'FORCEPOINT'
  name: '{{ IngestPipeline "fp-pipeline" }}'
@@ -125,9 +141,11 @@ processors:
  - uppercase:
  field: destination.mac
  ignore_missing: true
+ tag: uppercase_04de3657
  - uppercase:
  field: source.mac
  ignore_missing: true
+ tag: uppercase_5b4e7be2
  #
  # Timestamp parsing.
  #
@@ -150,17 +168,20 @@ processors:
  field: _tmp.timestamp8601
  formats:
  - ISO8601
+ tag: date_c38c0806
  - date:
  if: ctx?._tmp?.timestamp != null
  field: _tmp.timestamp
  formats:
  - MMM d HH:mm:ss
  - MMM dd HH:mm:ss
+ tag: date_771f472e
  - remove:
  if: ctx?.tags == null || !(ctx.tags.contains('preserve_original_event'))
  field: event.original
  ignore_failure: true
  ignore_missing: true
+ tag: remove_9f895a30
  # Converting the observer.ip to an array if its not, to provide correct ECS mapping with backwards compatibility.
  # Later this should be changed in the decode_cef processor as well.
  - rename:
@@ -179,25 +200,30 @@ processors:
  if: ctx.cef?.extensions?.categoryOutcome == "/Success"
  field: event.outcome
  value: success
+ tag: set_da09bb71
  - set:
  if: ctx.cef?.extensions?.categoryOutcome == "/Failure"
  field: event.outcome
  value: failure
+ tag: set_cb9ede45
  # Cleanup
  - remove:
  field:
  - cef.extensions._cefVer
  - _tmp
  ignore_missing: true
+ tag: remove_8c701636
 on_failure:
  - remove:
  field:
  - _tmp
  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}}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

@@ -6,26 +6,32 @@ processors:
  field: rule.id
  ignore_empty_value: true
  value: '{{{cef.extensions.deviceCustomString1}}}'
+ tag: set_7a577460
  # cs2 is natRuleID
  - set:
  field: rule.id
  ignore_empty_value: true
  value: '{{{cef.extensions.deviceCustomString2}}}'
+ tag: set_c76c7491
  # cs3 is VulnerabilityReference
  - set:
  field: vulnerability.reference
  ignore_empty_value: true
  value: '{{{cef.extensions.deviceCustomString3}}}'
+ tag: set_0b703e9a
  # cs4 is virusID
  - set:
  field: cef.forcepoint.virus_id
  ignore_empty_value: true
  value: '{{{cef.extensions.deviceCustomString4}}}'
+ tag: set_ce0473c6
 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}}failed with message '{{{ _ingest.on_failure_message }}}'
  - set:
  field: event.kind
  value: pipeline_error