2050-netflow.conf

# SOF-ELK® Configuration File
# (C)2023 Lewes Technology Consulting, LLC
#
# This file contains filters, transforms, and enrichments for NetFlow records

filter {
  # tag with any known uniquely-identifying characteristics are present
  if [type] == "archive-netflow" and [raw][logName] {
    mutate {
      add_tag => [ "gcp-vpcflow" ]
    }
  }

  if [type] == "archive-netflow" and "gcp-vpcflow" not in [tags] {
    # drop the header row and NODATA rows from Amazon VPC Flow entries
    if [message] == "version account-id interface-id srcaddr dstaddr srcport dstport protocol packets bytes start end action log-status"  or [message] =~ "NODATA" {
      drop { }
    }
    grok {
      patterns_dir => [ "/usr/local/sof-elk/grok-patterns" ]
      # nfdump2sof-elk.sh
      match => [ "message", "(?:%{IPORHOST:exporter})?(?:%{SPACE})?%{NONNEGINT:destination_as:int}%{SPACE}%{NONNEGINT:destination_mask:int}%{SPACE}%{NONNEGINT:engine_type:int}/%{NONNEGINT:engine_id:int}%{SPACE}%{TIMESTAMP_NETFLOW:flow_start_raw}%{SPACE}%{NONNEGINT:flow_records}%{SPACE}%{NONNEGINT}%{SPACE}%{NONNEGINT:in_bytes}%{SPACE}%{NONNEGINT:in_packets}%{SPACE}%{NONNEGINT:input_snmp:int}%{SPACE}%{IP:destination_ip}%{SPACE}%{IP:next_hop_ip}%{SPACE}%{IP:source_ip}%{SPACE}(%{ICMP_TYPECODE}|%{NONNEGINT:destination_port:int})%{SPACE}%{NONNEGINT:source_port:int}%{SPACE}%{TIMESTAMP_NETFLOW:flow_end_raw}%{SPACE}%{NONNEGINT:output_snmp:int}%{SPACE}%{INT:protocol:int}%{SPACE}%{NONNEGINT}%{SPACE}%{NONNEGINT}%{SPACE}%{NONNEGINT:source_as:int}%{SPACE}%{NONNEGINT:source_mask:int}%{SPACE}%{NONNEGINT:source_tos:int}%{SPACE}%{NOTSPACE:tcp_flags}%{SPACE}%{NONNEGINT:version:int}" ]

      # aws-vpcflow2sof-elk.sh
      match => [ "message", "%{INT:vpcflow_version} %{INT:vpcflow_account_id} %{NOTSPACE:vpcflow_interface_id} %{IP:source_ip} %{IP:destination_ip} %{INT:source_port} %{INT:destination_port} %{INT:protocol:int} %{INT:in_packets} %{INT:in_bytes} %{INT:flow_start_raw} %{INT:flow_end_raw} %{WORD:vpcflow_action} %{WORD:vpcflow_logstatus}" ]
      # AWS VPC Flow v5
      match => [ "message", "%{INT:vpcflow_account_id} %{WORD:vpcflow_action} %{NOTSPACE:vpcflow_azid} %{INT:in_bytes} %{IP:destination_ip} %{INT:destination_port} %{INT:flow_end_raw} %{WORD:vpcflow_direction} %{NOTSPACE:vpcflow_instance_id} %{NOTSPACE:vpcflow_interface_id} %{WORD:vpcflow_logstatus} %{INT:in_packets} %{NOTSPACE:vpcflow_destination_aws_service} %{IP:original_destination_ip} %{NOTSPACE:vpcflow_source_aws_service} %{IP:original_source_ip} %{INT:protocol:int} %{NOTSPACE:vpcflow_region} %{IP:source_ip} %{INT:source_port} %{INT:flow_start_raw} %{NOTSPACE:vpcflow_vpc_id} %{INT:vpcflow_version} %{WORD:vpcflow_type} %{NOTSPACE:vpcflow_traffic_path} %{INT:tcp_flags_int} %{NOTSPACE:vpcflow_subnet_id} %{NOTSPACE:vpcflow_sublocation_type} %{NOTSPACE:vpcflow_sublocation_id}" ]

      # azure-vpcflow2sof-elk.sh
      match => [ "message", "^%{GUID:exporter_guid},%{COMMONMAC:exporter_mac},%{POSINT:vpcflow_version},%{DATA:flow_rule},%{DATA:source},%{WORD:state},%{POSINT:flow_start_raw},%{POSINT:flow_end_raw},%{IP:source_ip},%{INT:source_port},%{IP:destination_ip},%{INT:destination_port},%{INT:protocol:int},%{INT:out_bytes},%{INT:out_packets},%{INT:in_bytes},%{INT:in_packets},%{INT:direction},%{WORD:traffic_decision}"]

      remove_field => [ "message", "MSECOND" ]
    }

  # Google Compute Platform VPC Flow Logs in JSON format
  # reference: https://cloud.google.com/vpc/docs/flow-logs#VpcDetails
  } else if [type] == "archive-netflow" and "gcp-vpcflow" in [tags] {
    mutate {
      rename => {
        "[raw][jsonPayload][start_time]" => "flow_start_raw"
        "[raw][jsonPayload][end_time]" => "flow_end_raw"
        "[raw][jsonPayload][connection][src_ip]" => "source_ip"
        "[raw][jsonPayload][connection][dest_ip]" => "destination_ip"
        "[raw][jsonPayload][connection][src_port]" => "source_port"
        "[raw][jsonPayload][connection][dest_port]" => "destination_port"
        "[raw][jsonPayload][connection][protocol]" => "protocol"
        "[raw][jsonPayload][bytes_sent]" => "out_bytes"
        "[raw][jsonPayload][packets_sent]" => "out_packets"
        "[raw][jsonPayload][src_instance][project_id]" => "source_instance_project_id"
        "[raw][jsonPayload][src_instance][region]" => "source_region"
        "[raw][jsonPayload][src_instance][vm_name]" => "source_vm_name"
        "[raw][jsonPayload][src_instance][zone]" => "source_zone"
        "[raw][jsonPayload][dest_instance][project_id]" => "destination_instance_project_id"
        "[raw][jsonPayload][dest_instance][region]" => "destination_region"
        "[raw][jsonPayload][dest_instance][vm_name]" => "destination_vm_name"
        "[raw][jsonPayload][dest_instance][zone]" => "destination_zone"
        "[raw][jsonPayload][src_vpc][project_id]" => "source_vpc_project_id"
        "[raw][jsonPayload][src_vpc][subnetwork_name]" => "source_vpc_subnetwork_name"
        "[raw][jsonPayload][src_vpc][vpc_name]" => "source_vpc_name"
        "[raw][jsonPayload][dest_vpc][project_id]" => "destination_vpc_project_id"
        "[raw][jsonPayload][dest_vpc][subnetwork_name]" => "destination_vpc_subnetwork_name"
        "[raw][jsonPayload][dest_vpc][vpc_name]" => "destination_vpc_name"
        "[raw][jsonPayload][src_location][asn]" => "source_as"
        "[raw][jsonPayload][dest_location][asn]" => "destination_as"
        "[raw][logName]" => "log_name"
        "[raw][resource][labels][location]" => "resource_location"
        "[raw][resource][project_id]" => "project_id"
        "[raw][resource][subnetwork_id]" => "subnetwork_id"
        "[raw][resource][subnetwork_name]" => "subnetwork_name"
        "[raw][resource][type]" => "resource_type"
      }
    }
    if [raw][jsonPayload][reporter] == "SRC" {
      mutate {
        add_field => { "exporter" => "%{source_ip}" }
      }
    } else if [raw][jsonPayload][reporter] == "DEST" {
      mutate {
        add_field => { "exporter" => "%{destination_ip}" }
      }
    }

    # remove remaining fields
    mutate {
      remove_field => [ "raw" ]
    }
  }

  if [type] == "archive-netflow" {
    if ![exporter] {
      mutate {
        replace => { "exporter" => "0.0.0.0" }
      }
    }

    # use the flow start timestamp for the event timestamp
    ##  2014-05-11 16:25:11.841 (archived from nfdump)
    date {
      match => [ "flow_start_raw", "YYYY-MM-dd'THH:mm:ss.SSS'Z", "YYYY-MM-dd HH:mm:ss.SSS", "ISO8601", "UNIX" ]
    }

    ## unify timestamp formats between live and archive
    date {
      match => [ "flow_start_raw", "YYYY-MM-dd'THH:mm:ss.SSS'Z", "YYYY-MM-dd HH:mm:ss.SSS", "ISO8601", "UNIX" ]
      target => "flow_start"
    }
    date {
      match => [ "flow_end_raw", "YYYY-MM-dd'THH:mm:ss.SSS'Z", "YYYY-MM-dd HH:mm:ss.SSS", "ISO8601", "UNIX" ]
      target => "flow_end"
    }
    mutate {
      remove_field => [ "flow_start_raw", "flow_end_raw" ]
    }
  }

  if [type] == "netflow" {
    # remove the [nf] superfield, to better allow ES dynamic mappings
    # this method originally found here: https://discuss.elastic.co/t/netflow-codec-output-to-the-root-of-the-message/61954
    json_encode {
      source => "nf"
    }
    json {
      source => "nf"
      remove_field => [ "nf" ]
    }

    # convert duration to seconds if it exists
    if [flowDurationMilliseconds] {
      ruby {
        code => "event.set('flow_duration', event.get('flowDurationMilliseconds').to_f / 1000)"
      }
    }
    if [flowDurationMicroseconds] {
      ruby {
        code => "event.set('flow_duration', event_get('flowDurationMicroseconds').to_f / 1000000)"
      }
    }

    # rename fields as needed for better field naming hygeine
    mutate {
      rename => {
        "host" => "exporter"
        "src_tos" => "source_tos"
        "l4_src_port" => "source_port"
        "ipv4_src_addr" => "source_ip"
        "src_mask" => "source_mask"
        "l4_dst_port" => "destination_port"
        "ipv4_dst_addr" => "destination_ip"
        "dst_mask" => "destination_mask"
        "ipv4_next_hop" => "next_hop_ip"
        "src_as" => "source_as"
        "dst_as" => "destination_as"
        "bgp_ipv4_next_hop" => "bgp_next_hop_ip"
        "mul_dst_pkts" => "mul_destination_packets"
        "mul_dst_bytes" => "mul_destination_bytes"
        "in_pkts" => "in_packets"
        "out_pkts" => "out_packets"
        "source_ip_v6" => "source_ip"
        "destination_ip_v6" => "destination_ip"
        "ipv6_src_addr" => "source_ip"
        "ipv6_dst_addr" => "destination_ip"
        "ipv6_src_mask" => "source_mask"
        "ipv6_dst_mask" => "destination_mask"
        "mpls_top_label_ip_addr" => "mpls_top_label_ip"
        "ipv4_ident" => "ident"
        "dst_tos" => "destination_tos"
        "in_src_mac" => "in_source_mac"
        "out_dst_mac" => "out_destination_mac"
        "src_vlan" => "source_vlan"
        "dst_vlan" => "destination_vlan"
        "ipv6_next_hop" => "next_hop_ip"
        "bgp_ipv6_next_hop" => "bgp_next_hop_ip"
        "ipv6_option_headers" => "option_headers"
        "in_dst_mac" => "in_destination_mac"
        "out_src_mac" => "out_source_mac"
        "flowStartSeconds" => "flow_start"
        "flowEndSeconds" => "flow_end"
        "flow_start_msec" => "flow_start"
        "flow_end_msec" => "flow_end"
        "flowStartMicroseconds" => "flow_start"
        "flowEndMicroseconds" => "flow_end"
        "flowStartNanoseconds" => "flow_start"
        "flowEndNanoseconds" => "flow_end"
        "observedFlowTotalCount" => "flow_records"
        "sourceIPv6Prefix" => "source_prefix"
        "icmp_type_ipv6" => "icmp_type"
        "icmp_code_ipv6" => "icmp_code"
        "udp_src_port" => "source_port"
        "udp_dst_port" => "destination_port"
        "tcp_src_port" => "source_port"
        "tcp_dst_port" => "destination_port"
        "tcpDestinationPort" => "destination_port"
        "xlate_src_addr_ipv4" => "xlate_source_ip"
        "xlate_dst_addr_ipv4" => "xlate_destination_ip"
        "xlate_src_addr_ipv6" => "xlate_source_ip"
        "xlate_dst_addr_ipv6" => "xlate_destination_ip"
      }
      convert => {
        "tcp_flags" => "integer"
      }
    }

    # use the flow start timestamp for the event timestamp
    ##  2014-10-14T00:41:56.999Z (live netflow v5)
    ## 1555359537557 (live netflow v9)
    date {
      match => [ "flow_start", "YYYY-MM-dd'THH:mm:ss.SSS'Z", "YYYY-MM-dd HH:mm:ss.SSS", "ISO8601" , "UNIX_MS" ]
    }

    # Set up the first and last switched date values
    date {
      match => [ "flow_start", "YYYY-MM-dd'THH:mm:ss.SSS'Z", "YYYY-MM-dd HH:mm:ss.SSS", "ISO8601" , "UNIX_MS" ]
      target => "flow_start"
    }
    date {
      match => [ "flow_end", "YYYY-MM-dd'THH:mm:ss.SSS'Z", "YYYY-MM-dd HH:mm:ss.SSS", "ISO8601" , "UNIX_MS" ]
      target => "flow_end"
    }
  }

  if [type] == "netflow" or [type] == "archive-netflow" {
    # Calculate the duration from flow_start and flow_end
    if [flow_start] and [flow_end] and ![flow_duration] {
      ruby {
        init => "require 'date'"
        code => "event.set('flow_duration', event.get('flow_end').to_f - event.get('flow_start').to_f)"
        add_tag => [ "calculated_duration" ]
      }
    }
    # if not TCP, no need for the tcp_flags field
    if [protocol] != 6 {
      mutate {
        remove_field => [ "tcp_flags" ]
      }

    } else if [tcp_flags] {
      # convert integer TCP flag value to an array, flag string, and hex byte
      # convert text TCP flag string to an array and hex byte
      ruby {
        path => "/usr/local/sof-elk/supporting-scripts/tcp_flags_to_array.rb"
        script_params => { "source_field" => "tcp_flags" }
      }
    }

    # replace the numerical protocol number with a text equivalent
    translate {
      dictionary_path => "/usr/local/sof-elk/lib/dictionaries/ip_proto_int2netflow.yaml"
      source => "protocol"
      target => "aprotocol"
    }

    # populate additional fields or modify as needed for other flow sources
    if ![flow_records] or [flow_records] == "0" {
      mutate {
        add_field => { "flow_records" => "1" }
      }
    }
    if ![missed_bytes] {
      mutate {
        add_field => { "missed_bytes" => "0" }
      }
    }
    if ![in_bytes] {
      mutate {
        add_field => { "in_bytes" => "0" }
      }
    }
    if ![in_packets] {
      mutate {
        add_field => { "in_packets" => "0" }
      }
    }
    if ![out_bytes] {
      mutate {
        add_field => { "out_bytes" => "0" }
      }
    }
    if ![out_packets] {
      mutate {
        add_field => { "out_packets" => "0" }
      }
    }

    # convert types
    mutate {
      convert => {
        "in_bytes" => "integer"
        "out_bytes" => "integer"
        "flow_records" => "integer"
        "in_packets" => "integer"
        "out_packets" => "integer"
        "missed_bytes" => "integer"
      }
      add_field => { "path" => "NetFlow from %{exporter}" }
    }
  }

  if [type] == "archive-netflow" {
    if [protocol] == "1" {
      mutate {
        replace => { "source_port" => "%{icmp_type}" }
      }
      mutate {
        replace => { "destination_port" => "%{icmp_code}" }
      }
      mutate {
        remove_field => [ "icmp_type", "icmp_code" ]
      }
    }
  }
}