Logstash Ulogd example configuration

agent.conf

input {
	file {
		type => "rasp"
		path => "/etc/logstash/data/ulogd_rasp.json"
		codec => "json"
		start_position => "beginning"
		sincedb_path => "/dev/null"
	}
	file {
		type => "msi"
		path => "/etc/logstash/data/nftables"
		start_position => "beginning"
		sincedb_path => "/dev/null"
	}
}

filter {
	if [type] == "msi" {
		grok {
			match => [
				"message", "%{SYSLOGTIMESTAMP:syslog_timestamp} %{SYSLOGHOST:syslog_hostname} Rejected by nftables : %{IPTABLES}",
				"message", "%{SYSLOGTIMESTAMP:syslog_timestamp} %{SYSLOGHOST:syslog_hostname} Rejected by nftables \(ipv6\) : %{IPTABLES}"
			]
		      	add_field => [ "host", "%{syslog_hostname}" ]
			patterns_dir => "/etc/logstash/patterns"
		}
		
		date {
		      match => [ "syslog_timestamp", "MMM  d HH:mm:ss", "MMM dd HH:mm:ss" ]
	    	}
	}
}

output {
	elasticsearch {
		hosts => "elasticsearch:9200"
		index => "netfilter"
	}
}

iptables.pattern

UNSIGNED_INT [0-9]+
INTERFACE \b(\w+[-]\w+)|(\w+)\b
IPTABLES_ETHERNET IN=%{INTERFACE:in_device} OUT=%{WORD:out_device}? MAC=(?:[^\s]+)

IPTABLES_PORT_PAIR SPT=%{UNSIGNED_INT:src_port} DPT=%{UNSIGNED_INT:dst_port}

IPTABLES_TCP_FLAGS ((?<= )(CWR|ECE|URG|ACK|PSH|RST|SYN|FIN))*
IPTABLES_TCP_SEQ SEQ=%{UNSIGNED_INT:seq_seq} ACK=%{UNSIGNED_INT:seq_ack}
IPTABLES_TCP_DETAILS (?:%{IPTABLES_TCP_SEQ} )?WINDOW=%{UNSIGNED_INT:window} %{IPTABLES_TCP_FLAGS:tcp_flags} 
IPTABLES_INCOMPLETE_PACKET INCOMPLETE \[%{UNSIGNED_INT:incomplete} bytes\]

IPTABLES_UDP_DETAILS LEN=%{UNSIGNED_INT:udp_len}

IPTABLES_ICMP_EXTRA_ECHO ID=%{UNSIGNED_INT:icmp_id} SEQ=%{UNSIGNED_INT:icmp_seq}
IPTABLES_ICMP_EXTRA_PARAM PARAMETER=%{UNSIGNED_INT:icmp_parameter}
IPTABLES_ICMP_EXTRA_REDIRECT GATEWAY=%{IP:icmp_redirect}
IPTABLES_ICMP_EXTRA ( (?:%{IPTABLES_ICMP_EXTRA_ECHO}|%{IPTABLES_ICMP_EXTRA_PARAM}|%{IPTABLES_ICMP_EXTRA_REDIRECT}))*

IPTABLES_ICMP_DETAILS TYPE=%{UNSIGNED_INT:icmp_type} CODE=%{UNSIGNED_INT:icmp_code}(( %{IPTABLES_INCOMPLETE_PACKET})|%{IPTABLES_ICMP_EXTRA})

IPTABLES_PROTOCOL PROTO=(?<proto>[a-zA-Z0-9]+)

IPTABLES_IP_PAYLOAD %{IPTABLES_PROTOCOL}( %{IPTABLES_PORT_PAIR})?( (%{IPTABLES_TCP_DETAILS}|%{IPTABLES_UDP_DETAILS}|%{IPTABLES_ICMP_DETAILS}|%{IPTABLES_INCOMPLETE_PACKET}))?

IPTABLES_IP_FRAGFLAG ((?<= )(CE|DF|MF))*
IPTABLES_IP_START SRC=%{IP:src_ip} DST=%{IP:dest_ip} LEN=%{UNSIGNED_INT:length} TOS=%{BASE16NUM:tos} PREC=0x%{BASE16NUM:prec} TTL=%{UNSIGNED_INT:ttl} ID=%{UNSIGNED_INT:id}(?: %{IPTABLES_IP_FRAGFLAG:fragment_flags})?(?: FRAG: %{UNSIGNED_INT:fragment})?
IPTABLES_IP6_START SRC=%{IP:src_ip} DST=%{IP:dest_ip} LEN=%{UNSIGNED_INT:length} TC=%{UNSIGNED_INT:tc} HOPLIMIT=%{UNSIGNED_INT:hoplimit} FLOWLBL=%{UNSIGNED_INT:flowlbl}
IPTABLES_IP (?:%{IPTABLES_IP_START}|%{IPTABLES_IP6_START}) %{IPTABLES_IP_PAYLOAD}

IPTABLES %{IPTABLES_ETHERNET} %{IPTABLES_IP}

ulogd.conf

[global]
######################################################################
# GLOBAL OPTIONS
######################################################################


# logfile for status messages
logfile="/var/log/ulogd.log"

# loglevel: debug(1), info(3), notice(5), error(7) or fatal(8) (default 5)
#loglevel=3

######################################################################
# PLUGIN OPTIONS
######################################################################

plugin="/usr/lib/ulogd/ulogd_inppkt_NFLOG.so"
#plugin="/usr/lib/ulogd/ulogd_inppkt_ULOG.so"
#plugin="/usr/lib/ulogd/ulogd_inppkt_UNIXSOCK.so"
plugin="/usr/lib/ulogd/ulogd_inpflow_NFCT.so"
plugin="/usr/lib/ulogd/ulogd_filter_IFINDEX.so"
plugin="/usr/lib/ulogd/ulogd_filter_IP2STR.so"
plugin="/usr/lib/ulogd/ulogd_filter_IP2BIN.so"
#plugin="/usr/lib/ulogd/ulogd_filter_IP2HBIN.so"
plugin="/usr/lib/ulogd/ulogd_filter_PRINTPKT.so"
plugin="/usr/lib/ulogd/ulogd_filter_HWHDR.so"
plugin="/usr/lib/ulogd/ulogd_filter_PRINTFLOW.so"
#plugin="/usr/lib/ulogd/ulogd_filter_MARK.so"
plugin="/usr/lib/ulogd/ulogd_output_LOGEMU.so"
plugin="/usr/lib/ulogd/ulogd_output_SYSLOG.so"
plugin="/usr/lib/ulogd/ulogd_output_XML.so"
plugin="/usr/lib/ulogd/ulogd_output_SQLITE3.so"
plugin="/usr/lib/ulogd/ulogd_output_GPRINT.so"
#plugin="/usr/lib/ulogd/ulogd_output_NACCT.so"
#plugin="/usr/lib/ulogd/ulogd_output_PCAP.so"
#plugin="/usr/lib/ulogd/ulogd_output_PGSQL.so"
plugin="/usr/lib/ulogd/ulogd_output_MYSQL.so"
#plugin="/usr/lib/ulogd/ulogd_output_DBI.so"
plugin="/usr/lib/ulogd/ulogd_raw2packet_BASE.so"
plugin="/usr/lib/ulogd/ulogd_inpflow_NFACCT.so"
plugin="/usr/lib/ulogd/ulogd_output_GRAPHITE.so"
plugin="/usr/lib/ulogd/ulogd_output_JSON.so"

#Text based logging
stack=log1:NFLOG,base1:BASE,ifi1:IFINDEX,ip2str1:IP2STR,print1:PRINTPKT,emu1:LOGEMU
#JSON Logging
#stack=log1:NFLOG,base1:BASE,ifi1:IFINDEX,ip2str1:IP2STR,json1:JSON

[log1]
# netlink multicast group (the same as the iptables --nflog-group param)
# Group O is used by the kernel to log connection tracking invalid message
group=1
numeric_label=1
netlink_socket_buffer_size=217088
netlink_socket_buffer_maxsize=1085440
# set number of packet to queue inside kernel
#netlink_qthreshold=1
# set the delay before flushing packet in the queue inside kernel (in 10ms)
#netlink_qtimeout=100

[ulog1]
# netlink multicast group (the same as the iptables --ulog-nlgroup param)
nlgroup=1
#numeric_label=0 # optional argument

[emu1]
file="/var/log/nftables"
sync=1

[json1]
sync=1
file="/var/log/ulogd/ulogd.json"
#timestamp=0
#device="Nftables log : "
boolean_label=1

[ct1]
netlink_socket_buffer_size=217088
netlink_socket_buffer_maxsize=1085440

Ulog configuration is using raw text logging but json can be enable by uncommenting the second stack

To make logstash take the raw text logging, I had to change a little the original iptables.pattern coming from here.
A IPTABLES_IP6_START pattern has been added and an INTERFACE pattern was created to adapt the regex to interfaces containing - like those generated by docker.

My configuration for logstash includes a file input for a json log and a raw text log. The grok filter should be modified, as I have a custom text "Rejected by nftables" that I added and is coming from an nftables logging rule.