data_dir: /var/lib/vector

sources:
  syslog_udp:
    type: socket
    address: 0.0.0.0:514
    mode: udp
    max_length: 65507
    decoding:
      codec: bytes
  syslog_tcp:
    type: socket
    address: 0.0.0.0:514
    mode: tcp
    decoding:
      codec: bytes
    framing:
      method: newline_delimited
  # Vector has no released IPFIX decoder. Archive datagrams with a site tag.
  netflow_ronkonkoma:
    type: socket
    address: 0.0.0.0:2055
    mode: udp
    max_length: 65507
    decoding:
      codec: bytes
  netflow_locust:
    type: socket
    address: 0.0.0.0:2056
    mode: udp
    max_length: 65507
    decoding:
      codec: bytes

transforms:
  parse_syslog:
    type: remap
    inputs: [syslog_udp, syslog_tcp]
    source: |-
      raw = to_string(.message) ?? encode_json(.)
      src_ip = to_string(.host) ?? ""
      site = "unknown"
      if starts_with(src_ip, "10.10.") {
        site = "ronkonkoma"
      }
      if starts_with(src_ip, "10.30.") {
        site = "locust"
      }

      format = "other"
      if contains(raw, "CEF:") {
        format = "cef"
      } else if contains(raw, "SRC=") && contains(raw, "DST=") {
        format = "iptables"
      }

      src = null
      dst = null
      proto = null
      action = null
      hostname = to_string(.hostname) ?? ""

      if format == "iptables" {
        src_m, err = parse_regex(raw, r'SRC=(?P<v>[0-9.]+)')
        if err == null { src = src_m.v }
        dst_m, err = parse_regex(raw, r'DST=(?P<v>[0-9.]+)')
        if err == null { dst = dst_m.v }
        proto_m, err = parse_regex(raw, r'PROTO=(?P<v>[A-Za-z0-9]+)')
        if err == null { proto = proto_m.v }
        if contains(raw, "DROP") || contains(raw, "REJECT") {
          action = "deny"
        } else if contains(raw, "ACCEPT") {
          action = "allow"
        }
      }

      if format == "cef" {
        src_m, err = parse_regex(raw, r'(?:src|sourceAddress)=(?P<v>[0-9.]+)')
        if err == null { src = src_m.v }
        dst_m, err = parse_regex(raw, r'(?:dst|destinationAddress)=(?P<v>[0-9.]+)')
        if err == null { dst = dst_m.v }
        proto_m, err = parse_regex(raw, r'proto=(?P<v>[A-Za-z0-9]+)')
        if err == null { proto = proto_m.v }
        if contains(upcase(raw), "BLOCK") || contains(upcase(raw), "DENY") || contains(upcase(raw), "DROP") {
          action = "deny"
        }
        host_m, err = parse_regex(raw, r'UNIFIhost=(?P<v>[^ ]+)')
        if err == null { hostname = host_m.v }
      }

      . = {
        "timestamp": format_timestamp!(now(), "%Y-%m-%dT%H:%M:%SZ"),
        "site": site,
        "format": format,
        "hostname": hostname,
        "src": src,
        "dst": dst,
        "proto": proto,
        "action": action,
        "raw": raw
      }

  parse_netflow_ronkonkoma:
    type: remap
    inputs: [netflow_ronkonkoma]
    source: |-
      payload = to_string(.message) ?? encode_json(.)
      . = {
        "timestamp": format_timestamp!(now(), "%Y-%m-%dT%H:%M:%SZ"),
        "site": "ronkonkoma",
        "format": "netflow",
        "hostname": "",
        "src": null,
        "dst": null,
        "proto": "ipfix",
        "action": null,
        "raw": encode_base64(payload) ?? payload
      }

  parse_netflow_locust:
    type: remap
    inputs: [netflow_locust]
    source: |-
      payload = to_string(.message) ?? encode_json(.)
      . = {
        "timestamp": format_timestamp!(now(), "%Y-%m-%dT%H:%M:%SZ"),
        "site": "locust",
        "format": "netflow",
        "hostname": "",
        "src": null,
        "dst": null,
        "proto": "ipfix",
        "action": null,
        "raw": encode_base64(payload) ?? payload
      }

sinks:
  firehose:
    type: aws_kinesis_firehose
    inputs: [parse_syslog, parse_netflow_ronkonkoma, parse_netflow_locust]
    region: ${aws_region}
    stream_name: ${firehose_stream}
    encoding:
      codec: json
    request:
      timeout_secs: 30