description: Pipeline for AWS VPC Flow Logs

processors:
  - set:
      field: event.ingested
      value: '{{_ingest.timestamp}}'

  # Convert Unix epoch to timestamp
  - date:
      field: "aws.vpcflow.end"
      target_field: "@timestamp"
      ignore_failure: true
      formats:
        - UNIX
  - date:
      field: "aws.vpcflow.start"
      target_field: "event.start"
      ignore_failure: true
      formats:
        - UNIX
  - date:
      field: "aws.vpcflow.end"
      target_field: "event.end"
      ignore_failure: true
      formats:
        - UNIX
  - remove:
      field: ["aws.vpcflow.start", "aws.vpcflow.end"]
      ignore_missing: true

  - script:
      lang: painless
      ignore_failure: true
      if: ctx?.aws != null
      source: >-
        void handleMap(Map map) {
          for (def x : map.values()) {
            if (x instanceof Map) {
                handleMap(x);
            } else if (x instanceof List) {
                handleList(x);
            }
          }
          map.values().removeIf(v -> v instanceof String && v == "-");
        }
        void handleList(List list) {
          for (def x : list) {
              if (x instanceof Map) {
                  handleMap(x);
              } else if (x instanceof List) {
                  handleList(x);
              }
          }
        }
        handleMap(ctx.aws);

  # IP Geolocation Lookup
  - geoip:
      field: source.ip
      target_field: source.geo
      ignore_missing: true
  - geoip:
      field: destination.ip
      target_field: destination.geo
      ignore_missing: true

  # IP Autonomous System (AS) Lookup
  - geoip:
      database_file: GeoLite2-ASN.mmdb
      field: source.ip
      target_field: source.as
      properties:
        - asn
        - organization_name
      ignore_missing: true
  - geoip:
      database_file: GeoLite2-ASN.mmdb
      field: destination.ip
      target_field: destination.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
  - rename:
      field: destination.as.asn
      target_field: destination.as.number
      ignore_missing: true
  - rename:
      field: destination.as.organization_name
      target_field: destination.as.organization.name
      ignore_missing: true
  - rename:
      field: message
      target_field: event.original
      ignore_missing: true

  # Generate related.ip field
  - append:
      if: ctx.source?.ip != null && ctx.destination?.ip != null
      field: related.ip
      value: ["{{source.ip}}", "{{destination.ip}}"]

  - set:
      field: cloud.provider
      value: aws

  - set:
      field: cloud.account.id
      value: "{{aws.vpcflow.account_id}}"
      ignore_empty_value: true

  - set:
      if: "ctx.aws?.vpcflow?.instance_id != null && ctx.aws.vpcflow.instance_id != '-'"
      field: cloud.instance.id
      value: "{{aws.vpcflow.instance_id}}"
      ignore_empty_value: true

  - set:
      field: event.kind
      value: event

  - script:
      lang: painless
      ignore_failure: true
      if: "ctx.aws?.vpcflow?.tcp_flags != null"
      source: |
        if (ctx.aws.vpcflow.tcp_flags_array == null) {
          ArrayList al = new ArrayList();
          ctx.aws.vpcflow.put("tcp_flags_array", al);
        }

        def flags = Integer.parseUnsignedInt(ctx.aws.vpcflow.tcp_flags);

        if ((flags & 0x01) != 0) {
          ctx.aws.vpcflow.tcp_flags_array.add('fin');
        }
        if ((flags & 0x02) != 0) {
          ctx.aws.vpcflow.tcp_flags_array.add('syn');
        }
        if ((flags & 0x04) != 0) {
          ctx.aws.vpcflow.tcp_flags_array.add('rst');
        }
        if ((flags & 0x08) != 0) {
          ctx.aws.vpcflow.tcp_flags_array.add('psh');
        }
        if ((flags & 0x10) != 0) {
          ctx.aws.vpcflow.tcp_flags_array.add('ack');
        }
        if ((flags & 0x20) != 0) {
          ctx.aws.vpcflow.tcp_flags_array.add('urg');
        }

on_failure:
  - set:
      field: "error.message"
      value: "{{ _ingest.on_failure_message }}"
