De-duplicate pipeline steps in system.auth module (#41229)

(cherry picked from commit f69b501)

# Conflicts:
#	filebeat/module/system/auth/ingest/entrypoint.yml
#	filebeat/module/system/auth/ingest/journald.yml
#	filebeat/module/system/auth/manifest.yml

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

@@ -0,0 +1,18 @@
+description: Entrypoint Pipeline for system/auth Filebeat module
+processors:
+  - set:
+      field: event.ingested
+      copy_from: _ingest.timestamp
+  - script:
+      source: |
+              if(ctx?.journald != null){
+                  ctx['auth_pipeline'] = '{< IngestPipeline "journald" >}';
+                  return;
+                }
+              ctx['auth_pipeline'] = '{< IngestPipeline "files" >}';
+              return;
+  - pipeline:
+      name: "{{ auth_pipeline }}"
+  - remove:
+      ignore_failure: true
+      field: "auth_pipeline"

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

@@ -0,0 +1,30 @@
+description: Journald Pipeline for system/auth Filebeat module
+processors:
+  - rename:
+      field: "journald.process.name"
+      target_field: process.name
+  - rename:
+      field: message
+      target_field: _temp.message
+  - pipeline:
+      description: Grok specific auth messages.
+      name: '{< IngestPipeline "grok-auth-messages" >}'
+      ignore_failure: true
+  - rename:
+      field: _temp.message
+      target_field: message
+  - pipeline:
+      name: "{< IngestPipeline "common" >}"
+  - remove:
+      description: Remove the extra fields added by the Journald input
+      ignore_missing: true
+      field:
+        - journald
+        - process.thread
+        - syslog
+        - systemd
+        - message_id
+on_failure:
+  - set:
+      field: error.message
+      value: '{{{ _ingest.on_failure_message }}}'