---
description: Pipeline for AWS VPC Flow Logs

processors:
  - set:
      field: event.ingested
      value: '{{_ingest.timestamp}}'
  - set:
      field: ecs.version
      value: '1.12.0'
  - rename:
      field: message
      target_field: event.original
      ignore_missing: true
  - set:
      field: event.type
      value: flow
  - set:
      field: event.category
      value: network_traffic
  - drop:
      if: 'ctx.event?.original.startsWith("version") || ctx.event?.original.startsWith("instance-id")'
  - script:
      lang: painless
      if: ctx.event?.original != null
      source: >-
        ctx._temp_ = new HashMap();
        ctx._temp_.message_token_count = ctx.event?.original.splitOnToken(" ").length;
  - dissect:
      field: event.original
      pattern: '%{aws.vpcflow.version} %{aws.vpcflow.account_id} %{aws.vpcflow.interface_id} %{aws.vpcflow.srcaddr} %{aws.vpcflow.dstaddr} %{aws.vpcflow.srcport} %{aws.vpcflow.dstport} %{aws.vpcflow.protocol} %{aws.vpcflow.packets} %{aws.vpcflow.bytes} %{aws.vpcflow.start} %{aws.vpcflow.end} %{aws.vpcflow.action} %{aws.vpcflow.log_status}'
      if: ctx?._temp_?.message_token_count == 14
  - dissect:
      field: event.original
      pattern: '%{aws.vpcflow.instance_id} %{aws.vpcflow.interface_id} %{aws.vpcflow.srcaddr} %{aws.vpcflow.dstaddr} %{aws.vpcflow.pkt_srcaddr} %{aws.vpcflow.pkt_dstaddr}'
      if: ctx?._temp_?.message_token_count == 6
  - dissect:
      field: event.original
      pattern: '%{aws.vpcflow.version} %{aws.vpcflow.interface_id} %{aws.vpcflow.account_id} %{aws.vpcflow.vpc_id} %{aws.vpcflow.subnet_id} %{aws.vpcflow.instance_id} %{aws.vpcflow.srcaddr} %{aws.vpcflow.dstaddr} %{aws.vpcflow.srcport} %{aws.vpcflow.dstport} %{aws.vpcflow.protocol} %{aws.vpcflow.tcp_flags} %{aws.vpcflow.type} %{aws.vpcflow.pkt_srcaddr} %{aws.vpcflow.pkt_dstaddr} %{aws.vpcflow.action} %{aws.vpcflow.log_status}'
      if: ctx?._temp_?.message_token_count == 17
  - dissect:
      field: event.original
      pattern: '%{aws.vpcflow.version} %{aws.vpcflow.vpc_id} %{aws.vpcflow.subnet_id} %{aws.vpcflow.instance_id} %{aws.vpcflow.interface_id} %{aws.vpcflow.account_id} %{aws.vpcflow.type} %{aws.vpcflow.srcaddr} %{aws.vpcflow.dstaddr} %{aws.vpcflow.srcport} %{aws.vpcflow.dstport} %{aws.vpcflow.pkt_srcaddr} %{aws.vpcflow.pkt_dstaddr} %{aws.vpcflow.protocol} %{aws.vpcflow.bytes} %{aws.vpcflow.packets} %{aws.vpcflow.start} %{aws.vpcflow.end} %{aws.vpcflow.action} %{aws.vpcflow.tcp_flags} %{aws.vpcflow.log_status}'
      if: ctx?._temp_?.message_token_count == 21

  # 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);
  - set:
      field: event.outcome
      value: allow
      if: ctx.aws?.vpcflow?.action == "ACCEPT"
  - set:
      field: event.outcome
      value: deny
      if: ctx.aws?.vpcflow?.action == "REJECT"
  - rename:
      field: aws.vpcflow.srcaddr
      target_field: source.address
      ignore_missing: true
  - set:
      field: source.ip
      copy_from: source.address
      if: ctx.source?.address != null
  - convert:
      field: aws.vpcflow.srcport
      target_field: source.port
      type: integer
      ignore_missing: true
  - rename:
      field: aws.vpcflow.dstaddr
      target_field: destination.address
      ignore_missing: true
  - set:
      field: destination.ip
      copy_from: destination.address
      if: ctx.destination?.address != null
  - convert:
      field: aws.vpcflow.dstport
      target_field: destination.port
      type: integer
      ignore_missing: true
  - rename:
      field: aws.vpcflow.protocol
      target_field: network.iana_number
      ignore_missing: true
  - convert:
      field: aws.vpcflow.packets
      target_field: source.packets
      type: long
      ignore_missing: true
  - convert:
      field: aws.vpcflow.bytes
      target_field: source.bytes
      type: long
      ignore_missing: true
  - set:
      field: network.bytes
      copy_from: source.bytes
      if: ctx.source?.bytes != null
  - set:
      field: network.packets
      copy_from: source.packets
      if: ctx.source?.packets != null
  - set:
      field: network.type
      value: ipv4
      if: 'ctx.source?.ip != null && ctx.source?.ip.contains(".")'
  - set:
      field: network.type
      value: ipv6
      if: 'ctx.source?.ip != null && ctx.source?.ip.contains(":")'
  - set:
      field: network.transport
      value: tcp
      if: ctx.network?.iana_number == "6"
  - set:
      field: network.transport
      value: udp
      if: ctx.network?.iana_number == "17"
  - community_id:
      target_field: network.community_id
      ignore_failure: true
  # 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
  # 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:
      if: ctx.aws?.vpcflow?.account_id != null
      field: cloud.account.id
      value: '{{aws.vpcflow.account_id}}'
  - set:
      if: 'ctx?.aws?.vpcflow?.instance_id != null && ctx.aws.vpcflow.instance_id != "-"'
      field: cloud.instance.id
      value: '{{aws.vpcflow.instance_id}}'
  - 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');
        }
  - remove:
      field:
        - _temp_
        - aws.vpcflow.srcaddr
        - aws.vpcflow.srcport
        - aws.vpcflow.dstaddr
        - aws.vpcflow.dstport
        - aws.vpcflow.bytes
        - aws.vpcflow.packets
        - aws.vpcflow.protocol
      ignore_missing: true
  - 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'
      value: '{{ _ingest.on_failure_message }}'
