[8.x](backport #41229) De-duplicate pipeline steps in system.auth module

filebeat/module/system/auth/ingest/common.yml

@@ -0,0 +1,172 @@
+description: Common steps for Journald and log files from system/auth Filebeat module
+processors:
+  - grok:
+      description: Grok usernames from PAM messages.
+      tag: grok-pam-users
+      field: message
+      ignore_missing: true
+      ignore_failure: true
+      patterns:
+        - 'for user %{QUOTE}?%{DATA:_temp.foruser}%{QUOTE}? by %{QUOTE}?%{DATA:_temp.byuser}%{QUOTE}?(?:\(uid=%{NUMBER:_temp.byuid}\))?$'
+        - 'for user %{QUOTE}?%{DATA:_temp.foruser}%{QUOTE}?$'
+        - 'by user %{QUOTE}?%{DATA:_temp.byuser}%{QUOTE}?$'
+        - '%{BOUNDARY} user %{QUOTE}%{DATA:_temp.user}%{QUOTE}'
+      pattern_definitions:
+        QUOTE: "['\"]"
+        BOUNDARY: "(?<! )"
+      if: ctx.message != null && ctx.message != ""
+  - rename:
+      field: _temp.byuser
+      target_field: user.name
+      ignore_missing: true
+      ignore_failure: true
+  - rename:
+      field: _temp.byuid
+      target_field: user.id
+      ignore_missing: true
+      ignore_failure: true
+  - rename:
+      field: _temp.foruser
+      target_field: user.name
+      ignore_missing: true
+      ignore_failure: true
+      if: ctx.user?.name == null || ctx.user?.name == ""
+  - rename:
+      field: _temp.user
+      target_field: user.name
+      ignore_missing: true
+      ignore_failure: true
+      if: ctx.user?.name == null || ctx.user?.name == ""
+  - rename:
+      field: _temp.foruser
+      target_field: user.effective.name
+      ignore_missing: true
+      ignore_failure: true
+      if: ctx.user?.name != null
+  - remove:
+      field: _temp
+      ignore_missing: true
+  - convert:
+      field: source.address
+      target_field: source.ip
+      type: ip
+      ignore_missing: true
+      on_failure:
+        - set:
+            field: source.domain
+            copy_from: source.address
+            ignore_failure: true
+  - convert:
+      field: system.auth.sudo.user
+      target_field: user.effective.name
+      type: string
+      ignore_failure: true
+      if: ctx.system?.auth?.sudo?.user != null
+  - convert:
+      field: system.auth.ssh.dropped_ip
+      target_field: source.ip
+      type: ip
+      ignore_missing: true
+      on_failure:
+        - remove:
+            field: system.auth.ssh.dropped_ip
+  - geoip:
+      field: source.ip
+      target_field: source.geo
+      ignore_missing: true
+  - geoip:
+      database_file: GeoLite2-ASN.mmdb
+      field: source.ip
+      target_field: source.as
+      properties:
+        - asn
+        - organization_name
+      ignore_missing: true
+  - rename:
+      field: source.as.asn
+      target_field: source.as.number
+      ignore_missing: true
+  - rename:
+      field: source.as.organization_name
+      target_field: source.as.organization.name
+      ignore_missing: true
+  - set:
+      field: event.kind
+      value: event
+  - script:
+      description: Add event.category/action/output to SSH events.
+      tag: script-categorize-ssh-event
+      if: ctx.system?.auth?.ssh?.event != null
+      lang: painless
+      source: >-
+        if (ctx.system.auth.ssh.event == "Accepted") {
+          ctx.event.type = ["info"];
+          ctx.event.category = ["authentication", "session"];
+          ctx.event.action = "ssh_login";
+          ctx.event.outcome = "success";
+        } else if (ctx.system.auth.ssh.event == "Invalid" || ctx.system.auth.ssh.event == "Failed") {
+          ctx.event.type = ["info"];
+          ctx.event.category = ["authentication"];
+          ctx.event.action = "ssh_login";
+          ctx.event.outcome = "failure";
+        }
+  - append:
+      field: event.category
+      value: iam
+      if: ctx.process?.name != null && ['groupadd', 'groupdel', 'groupmod', 'useradd', 'userdel', 'usermod'].contains(ctx.process.name)
+  - set:
+      field: event.outcome
+      value: success
+      if: ctx.process?.name != null && (ctx.message == null || !ctx.message.contains("fail")) && ['groupadd', 'groupdel', 'groupmod', 'useradd', 'userdel', 'usermod'].contains(ctx.process.name)
+  - set:
+      field: event.outcome
+      value: failure
+      if: ctx.process?.name != null && (ctx.message != null && ctx.message.contains("fail")) && ['groupadd', 'groupdel', 'groupmod', 'useradd', 'userdel', 'usermod'].contains(ctx.process.name)
+  - append:
+      field: event.type
+      value: user
+      if: ctx.process?.name != null && ['useradd', 'userdel', 'usermod'].contains(ctx.process.name)
+  - append:
+      field: event.type
+      value: group
+      if: ctx.process?.name != null && ['groupadd', 'groupdel', 'groupmod'].contains(ctx.process.name)
+  - append:
+      field: event.type
+      value: creation
+      if: ctx.process?.name != null && ['useradd', 'groupadd'].contains(ctx.process.name)
+  - append:
+      field: event.type
+      value: deletion
+      if: ctx.process?.name != null && ['userdel', 'groupdel'].contains(ctx.process.name)
+  - append:
+      field: event.type
+      value: change
+      if: ctx.process?.name != null && ['usermod', 'groupmod'].contains(ctx.process.name)
+  - append:
+      field: related.user
+      value: "{{{ user.name }}}"
+      allow_duplicates: false
+      if: ctx.user?.name != null && ctx.user?.name != ''
+  - append:
+      field: related.user
+      value: "{{{ user.effective.name }}}"
+      allow_duplicates: false
+      if: ctx.user?.effective?.name != null && ctx.user?.effective?.name != ''
+  - append:
+      field: related.ip
+      value: "{{{ source.ip }}}"
+      allow_duplicates: false
+      if: ctx.source?.ip != null && ctx.source?.ip != ''
+  - append:
+      field: related.hosts
+      value: "{{{ host.hostname }}}"
+      allow_duplicates: false
+      if: ctx.host?.hostname != null && ctx.host?.hostname != ''
+  - set:
+      field: ecs.version
+      value: 8.0.0
+  - remove:
+      field: event.original
+      if: "ctx?.tags == null || !(ctx.tags.contains('preserve_original_event'))"
+      ignore_failure: true
+      ignore_missing: true

filebeat/module/system/auth/ingest/entrypoint.yml

@@ -1,5 +1,8 @@
 description: Entrypoint Pipeline for system/auth Filebeat module
 processors:
+  - set:
+      field: event.ingested
+      copy_from: _ingest.timestamp
   - script:
       source: |
               if(ctx?.journald != null){

filebeat/module/system/auth/ingest/files.yml

@@ -1,9 +1,6 @@
 ---
 description: Pipeline for parsing system authorization and secure logs.
 processors:
-  - set:
-      field: event.ingested
-      copy_from: _ingest.timestamp
   - rename:
       if: ctx.event?.original == null
       field: message
@@ -28,76 +25,8 @@ processors:
             target_field: message
   - remove:
       field: _temp
-  - grok:
-      description: Grok usernames from PAM messages.
-      tag: grok-pam-users
-      field: message
-      ignore_missing: true
-      ignore_failure: true
-      patterns:
-        - 'for user %{QUOTE}?%{DATA:_temp.foruser}%{QUOTE}? by %{QUOTE}?%{DATA:_temp.byuser}%{QUOTE}?(?:\(uid=%{NUMBER:_temp.byuid}\))?$'
-        - 'for user %{QUOTE}?%{DATA:_temp.foruser}%{QUOTE}?$'
-        - 'by user %{QUOTE}?%{DATA:_temp.byuser}%{QUOTE}?$'
-        - '%{BOUNDARY} user %{QUOTE}%{DATA:_temp.user}%{QUOTE}'
-      pattern_definitions:
-        QUOTE: "['\"]"
-        BOUNDARY: "(?<! )"
-      if: ctx.message != null && ctx.message != ""
-  - rename:
-      field: _temp.byuser
-      target_field: user.name
-      ignore_missing: true
-      ignore_failure: true
-  - rename:
-      field: _temp.byuid
-      target_field: user.id
-      ignore_missing: true
-      ignore_failure: true
-  - rename:
-      field: _temp.foruser
-      target_field: user.name
-      ignore_missing: true
-      ignore_failure: true
-      if: ctx.user?.name == null || ctx.user?.name == ""
-  - rename:
-      field: _temp.user
-      target_field: user.name
-      ignore_missing: true
-      ignore_failure: true
-      if: ctx.user?.name == null || ctx.user?.name == ""
-  - rename:
-      field: _temp.foruser
-      target_field: user.effective.name
-      ignore_missing: true
-      ignore_failure: true
-      if: ctx.user?.name != null
-  - remove:
-      field: _temp
-      ignore_missing: true
-  - convert:
-      field: source.address
-      target_field: source.ip
-      type: ip
-      ignore_missing: true
-      on_failure:
-        - set:
-            field: source.domain
-            copy_from: source.address
-            ignore_failure: true
-  - convert:
-      field: system.auth.sudo.user
-      target_field: user.effective.name
-      type: string
-      ignore_failure: true
-      if: ctx.system?.auth?.sudo?.user != null
-  - convert:
-      field: system.auth.ssh.dropped_ip
-      target_field: source.ip
-      type: ip
-      ignore_missing: true
-      on_failure:
-        - remove:
-            field: system.auth.ssh.dropped_ip
+  - pipeline:
+      name: "{< IngestPipeline "common" >}"
   - date:
       if: ctx.event?.timezone == null
       field: system.auth.timestamp
@@ -125,106 +54,6 @@ processors:
             value: '{{{ _ingest.on_failure_message }}}'
   - remove:
       field: system.auth.timestamp
-  - geoip:
-      field: source.ip
-      target_field: source.geo
-      ignore_missing: true
-  - geoip:
-      database_file: GeoLite2-ASN.mmdb
-      field: source.ip
-      target_field: source.as
-      properties:
-        - asn
-        - organization_name
-      ignore_missing: true
-  - rename:
-      field: source.as.asn
-      target_field: source.as.number
-      ignore_missing: true
-  - rename:
-      field: source.as.organization_name
-      target_field: source.as.organization.name
-      ignore_missing: true
-  - set:
-      field: event.kind
-      value: event
-  - script:
-      description: Add event.category/action/output to SSH events.
-      tag: script-categorize-ssh-event
-      if: ctx.system?.auth?.ssh?.event != null
-      lang: painless
-      source: >-
-        if (ctx.system.auth.ssh.event == "Accepted") {
-          ctx.event.type = ["info"];
-          ctx.event.category = ["authentication", "session"];
-          ctx.event.action = "ssh_login";
-          ctx.event.outcome = "success";
-        } else if (ctx.system.auth.ssh.event == "Invalid" || ctx.system.auth.ssh.event == "Failed") {
-          ctx.event.type = ["info"];
-          ctx.event.category = ["authentication"];
-          ctx.event.action = "ssh_login";
-          ctx.event.outcome = "failure";
-        }
-  - append:
-      field: event.category
-      value: iam
-      if: ctx.process?.name != null && ['groupadd', 'groupdel', 'groupmod', 'useradd', 'userdel', 'usermod'].contains(ctx.process.name)
-  - set:
-      field: event.outcome
-      value: success
-      if: ctx.process?.name != null && (ctx.message == null || !ctx.message.contains("fail")) && ['groupadd', 'groupdel', 'groupmod', 'useradd', 'userdel', 'usermod'].contains(ctx.process.name)
-  - set:
-      field: event.outcome
-      value: failure
-      if: ctx.process?.name != null && (ctx.message != null && ctx.message.contains("fail")) && ['groupadd', 'groupdel', 'groupmod', 'useradd', 'userdel', 'usermod'].contains(ctx.process.name)
-  - append:
-      field: event.type
-      value: user
-      if: ctx.process?.name != null && ['useradd', 'userdel', 'usermod'].contains(ctx.process.name)
-  - append:
-      field: event.type
-      value: group
-      if: ctx.process?.name != null && ['groupadd', 'groupdel', 'groupmod'].contains(ctx.process.name)
-  - append:
-      field: event.type
-      value: creation
-      if: ctx.process?.name != null && ['useradd', 'groupadd'].contains(ctx.process.name)
-  - append:
-      field: event.type
-      value: deletion
-      if: ctx.process?.name != null && ['userdel', 'groupdel'].contains(ctx.process.name)
-  - append:
-      field: event.type
-      value: change
-      if: ctx.process?.name != null && ['usermod', 'groupmod'].contains(ctx.process.name)
-  - append:
-      field: related.user
-      value: "{{{ user.name }}}"
-      allow_duplicates: false
-      if: ctx.user?.name != null && ctx.user?.name != ''
-  - append:
-      field: related.user
-      value: "{{{ user.effective.name }}}"
-      allow_duplicates: false
-      if: ctx.user?.effective?.name != null && ctx.user?.effective?.name != ''
-  - append:
-      field: related.ip
-      value: "{{{ source.ip }}}"
-      allow_duplicates: false
-      if: ctx.source?.ip != null && ctx.source?.ip != ''
-  - append:
-      field: related.hosts
-      value: "{{{ host.hostname }}}"
-      allow_duplicates: false
-      if: ctx.host?.hostname != null && ctx.host?.hostname != ''
-  - set:
-      field: ecs.version
-      value: 8.0.0
-  - remove:
-      field: event.original
-      if: "ctx?.tags == null || !(ctx.tags.contains('preserve_original_event'))"
-      ignore_failure: true
-      ignore_missing: true
 on_failure:
   - set:
       field: error.message