fluent-bit-filters-bro-conn.conf

# Tabular log format parser
[FILTER]
    Name parser
    Match ${observer_product}.conn
    Key_Name log
    Parser bro_conn_parser
    Preserve_Key on
    Reserve_Data on

# JSON log format parser
[FILTER]
    Name parser
    Match ${observer_product}.conn.json
    Key_Name log
    Parser bro_conn_parser_json
    Preserve_Key off
    Reserve_Data on

# Rename dots from 4-tuple fields from json
[FILTER]
    Name modify
    Match ${observer_product}.conn.json
    Rename id.orig_h orig_h
    Rename id.orig_p orig_p
    Rename id.resp_h resp_h
    Rename id.resp_p resp_p

# Prefix all regular keys ([A-z]*) with 'zeek_connection_'
# This will also catch any extended fields to move them under zeek.connection
[FILTER]
    Name lua
    Match ${observer_product}.conn*
    Script  bro_conn_parse.lua
    Call    bro_conn_prefix_all

[FILTER]
    Name modify
    Match ${observer_product}.conn
    Add zeek_log_file conn
    Add zeek_log_format tab

# These are special "_*" keys, rename them separately
[FILTER]
    Name modify
    Match ${observer_product}.conn.json
    Rename _path zeek_log_file
    Rename _write_ts zeek_write_ts
    Add zeek_log_format json

# Add/modify Connection-specific fields
[FILTER]
    Name modify
    Match ${observer_product}.conn*
    Add _event_dataset ${observer_product}.conn
    Rename zeek_connection_uid         zeek_uid
    Rename zeek_connection_orig_h      source_ip
    Rename zeek_connection_orig_p      source_port
    Rename zeek_connection_orig_pkts   source_packets
    Rename zeek_connection_resp_h      destination_ip
    Rename zeek_connection_resp_p      destination_port
    Rename zeek_connection_resp_pkts   destination_packets
    Rename zeek_connection_proto       network_transport
    Rename zeek_connection_service     network_protocol
    Rename zeek_connection_duration    _event_duration
    Rename zeek_connection_log         _log_original

@INCLUDE fluent-bit-filters-bro-conn-custom.conf

# Copy zeek.uid to event.id
[FILTER]
    Name modify
    Match ${observer_product}.conn*
    Copy zeek_uid _event_id

# Cleanup - remove keys represending IP addresses if value is "-"
[FILTER]
    Name modify
    Match ${observer_product}.conn*
    Condition Key_value_equals source_ip -
    Remove_regex source_ip

[FILTER]
    Name modify
    Match ${observer_product}.conn*
    Condition Key_value_equals destination_ip -
    Remove_regex destination_ip

# Cleanup - remove keys represending duration if value is "-"
[FILTER]
    Name modify
    Match ${observer_product}.conn*
    Condition Key_value_equals event_duration -
    Remove event_duration

# Cleanup - remove keys represending tunnel_parents if value is "-"
[FILTER]
    Name modify
    Match ${observer_product}.conn*
    Condition Key_value_equals zeek_connection_tunnel_parents -
    Remove zeek_connection_tunnel_parents

# ECS: Parse network.bytes
[FILTER]
    Name    lua
    Match   ${observer_product}.conn*
    Script  bro_conn_parse.lua
    Call    bro_conn_parse_bytes

# CUSTOM: Parse ICMP
[FILTER]
    Name    lua
    Match   ${observer_product}.conn*
    Script  bro_conn_parse.lua
    Call    bro_conn_parse_icmp

# ECS: Parse zeek_connection_conn_state
[FILTER]
    Name    lua
    Match   ${observer_product}.conn*
    Script  bro_conn_parse.lua
    Call    bro_conn_parse_state

# ECS: Parse network.direction
[FILTER]
    Name    lua
    Match   ${observer_product}.conn*
    Script  bro_conn_parse.lua
    Call    bro_conn_parse_direction

# ECS: Parse network.packets
[FILTER]
    Name    lua
    Match   ${observer_product}.conn*
    Script  bro_conn_parse.lua
    Call    bro_conn_parse_packets

# Remove no longer needed keys
[FILTER]
    Name modify
    Match ${observer_product}.conn*
    Remove zeek_connection_local_orig
    Remove zeek_connection_local_resp

# Nest JSON objects to comply with ECS notation: ICMP
[FILTER]
    Name nest
    Match ${observer_product}.conn*
    Operation nest
    Wildcard icmp_*
    Nest_under icmp
    Remove_prefix icmp_

# Nest JSON objects to comply with ECS notation: Zeek.connection
[FILTER]
    Name nest
    Match ${observer_product}.conn*
    Operation nest
    Wildcard zeek_connection_*
    Nest_under zeek_connection
    Remove_prefix zeek_connection_

# Nest JSON objects to comply with ECS notation: Zeek
[FILTER]
    Name nest
    Match ${observer_product}.conn*
    Operation nest
    Wildcard zeek_*
    Nest_under zeek
    Remove_prefix zeek_