pipeline-json.yml

description: Pipeline for parsing elasticsearch audit logs in JSON format
processors:
  - json:
      field: message
      target_field: elasticsearch.audit
  - dot_expander:
      field: event.type
      path: elasticsearch.audit
  - drop:
      if: ctx.elasticsearch.audit.containsKey('type') && ctx.elasticsearch.audit.type != 'audit'
  - drop:
      if: '!ctx.elasticsearch.audit.containsKey("type") && !["rest", "transport", "ip_filter", "security_config_change"].contains(ctx.elasticsearch?.audit?.event?.type)'
  - remove:
      field: elasticsearch.audit.type
      ignore_missing: true
  - date:
      if: ctx.elasticsearch.audit['@timestamp'] != null && ctx.event.timezone != null
      field: elasticsearch.audit.@timestamp
      target_field: elasticsearch.audit.@timestamp
      formats:
        - yyyy-MM-dd'T'HH:mm:ss,SSS
        - yyyy-MM-dd'T'HH:mm:ss,SSSZ
      timezone: "{{ event.timezone }}"
      ignore_failure: true
  - remove:
      if: ctx.elasticsearch.audit['@timestamp'] == null && ctx.event.timezone != null
      field: event.timezone
  - rename:
      field: elasticsearch.audit.timestamp
      target_field: elasticsearch.audit.@timestamp
      ignore_missing: true
  - dot_expander:
      field: event.action
      path: elasticsearch.audit
  - remove:
      field: event.action
      ignore_missing: true
  - rename:
      field: elasticsearch.audit.event.action
      target_field: event.action
      ignore_missing: true
  - rename:
      field: elasticsearch.audit.event.type
      target_field: elasticsearch.audit.layer
      ignore_missing: true
  - dot_expander:
      field: origin.address
      path: elasticsearch.audit
  - grok:
      field: elasticsearch.audit.origin.address
      patterns:
        - \[%{IPORHOST:source.ip}\]:%{INT:source.port:int}
        - "%{IPORHOST:source.ip}:%{INT:source.port:int}"
      ignore_missing: true
  - remove:
      field: source.address
      ignore_missing: true
  - rename:
      field: elasticsearch.audit.origin.address
      target_field: source.address
      ignore_missing: true
  - dot_expander:
      field: url.path
      path: elasticsearch.audit
  - dot_expander:
      field: url.query
      path: elasticsearch.audit
  - set:
      if: ctx.elasticsearch.audit?.url?.query == null
      field: url.original
      value: "{{elasticsearch.audit.url.path}}"
      ignore_empty_value: true
  - set:
      if: ctx.elasticsearch.audit?.url?.path != null && ctx.elasticsearch.audit?.url?.query != null
      field: url.original
      value: "{{elasticsearch.audit.url.path}}?{{elasticsearch.audit.url.query}}"
  - remove:
      if: ctx.elasticsearch.audit?.url?.path != null
      field: elasticsearch.audit.url.path
  - remove:
      if: ctx.elasticsearch.audit?.url?.query != null
      field: elasticsearch.audit.url.query
  - dot_expander:
      field: node.id
      path: elasticsearch.audit
  - dot_expander:
      field: node.name
      path: elasticsearch.audit
  - remove:
      field: elasticsearch.node
      ignore_missing: true
  - rename:
      field: elasticsearch.audit.node
      target_field: elasticsearch.node
  - rename:
      field: elasticsearch.audit.change.disable.user.name
      target_field: user.name
      ignore_missing: true
  - rename:
      field: elasticsearch.audit.change.enable.user.name
      target_field: user.name
      ignore_missing: true
  - rename:
      field: elasticsearch.audit.delete.user.name
      target_field: user.name
      ignore_missing: true
  - rename:
      field: elasticsearch.audit.put.user.name
      target_field: user.name
      ignore_missing: true
  - rename:
      field: elasticsearch.audit.put.user.full_name
      target_field: user.full_name
      ignore_missing: true
  - rename:
      field: elasticsearch.audit.put.user.email
      target_field: user.email
      ignore_missing: true
  - remove:
      field: elasticsearch.audit.put
      ignore_missing: true
  - rename:
      field: elasticsearch.audit.invalidate.apikeys.user.name
      target_field: user.name
      ignore_missing: true
  - rename:
      field: elasticsearch.audit.invalidate.apikeys.user.realm
      target_field: elasticsearch.audit.user.realm
      ignore_missing: true
  - dot_expander:
      field: user.run_as.name
      path: elasticsearch.audit
      ignore_failure: true
  - dot_expander:
      field: user.run_as.realm
      path: elasticsearch.audit
      ignore_failure: true
  - convert:
      field: elasticsearch.audit.user.run_as.name
      target_field: user.effective.name
      type: string
      ignore_failure: true
  - dot_expander:
      field: user.name
      path: elasticsearch.audit
  - rename:
      field: elasticsearch.audit.user.name
      target_field: user.name
      ignore_missing: true
  - dot_expander:
      field: user.email
      path: elasticsearch.audit
  - dot_expander:
      field: request.method
      path: elasticsearch.audit
  - rename:
      field: elasticsearch.audit.request.method
      target_field: http.request.method
      ignore_missing: true
  - dot_expander:
      field: request.body
      path: elasticsearch.audit
  - rename:
      field: elasticsearch.audit.request.body
      target_field: http.request.body.content
      ignore_missing: true
  - dot_expander:
      field: request.id
      path: elasticsearch.audit
  - set:
      field: http.request.id
      value: '{{{elasticsearch.audit.request.id}}}'
      ignore_empty_value: true
  - dot_expander:
      field: cluster.name
      path: elasticsearch.audit
  - dot_expander:
      field: cluster.uuid
      path: elasticsearch.audit
  - rename:
      field: elasticsearch.audit.cluster.name
      target_field: elasticsearch.cluster.name
      ignore_missing: true
  - rename:
      field: elasticsearch.audit.cluster.uuid
      target_field: elasticsearch.cluster.uuid
      ignore_missing: true
  - rename:
      field: elasticsearch.audit.level
      target_field: log.level
      ignore_missing: true
  - dot_expander:
      field: trace.id
      path: elasticsearch.audit
  - rename:
      field: elasticsearch.audit.trace.id
      target_field: trace.id
      ignore_missing: true
  - remove:
      field: elasticsearch.audit.trace.id
      ignore_missing: true
  - date:
      field: elasticsearch.audit.@timestamp
      target_field: "@timestamp"
      formats:
        - ISO8601
      ignore_failure: true
on_failure:
  - set:
      field: error.message
      value: "{{ _ingest.on_failure_message }}"