Custom Grok filters

[synapse303]

Hi,

Anyone know how to create custom Grok filters in SO 2? I had custom logstash filters in the older version that matched Meraki firewall logs coming in but now the default syslog parser doesn't parse the logs correctly. I'm guessing this is due to them being parsed by Elasticsearch rather than logstash now?

I've checked in /opt/so/conf/elasticsearch/ingest/ and found the syslog parser but can't figure out how to create my own. I know they need to go into /opt/so/saltstack/local/salt/elasticsearch/files/ingest/

In my old conf file I also had some logic to determine and apply Grok for different fw log types (url, flows, nat) and not sure how this would work in the new version.

Am I even on the right track?

Thanks
James

[weslambert]

The groks should be fairly similar between the Logstash config and ingest config. Can you provide an example?

[synapse303]

Thanks.

Sorry I should have made it clearer, I guess it's more a question about how I go about adding custom filtering/parsing in the new system to properly parse these syslogs.

Is this still done in logstash or somewhere else?
As I understand it, logstash now sends unfiltered to elasticsearch instead.

My old config looks like the below and sat in logstash config directory with 3000_meraki_firewall.conf so that it fit between processing and post_processing. It first matched that it was coming from the gateway then applied logic based on the type of log.

filter {
    # Match Meraki Logs
    if [host] == "gateway" {
        # Flow
        if "flows" in [message]{
                if "mac=" in [message]{
                    if "protocol=icmp" in [message]{
                            grok{
                            match => { "message" => "%{NUMBER:timestamp} %{WORD:hostname} %{WORD:type} src=%{IPV4:source_ip} dst=%{IPV4:destination_ip} mac=%{MAC:mac} protocol=%{WORD:protocol} type=%{INT:protocol_type} pattern: (%{WORD:action}|%{INT:action}) %{WORD:scope}"}
                            }
                        }
                        grok {
                        match => { "message" => "%{NUMBER:timestamp} %{WORD:hostname} %{WORD:type} src=%{IPV4:source_ip} dst=%{IPV4:destination_ip} mac=%{MAC:mac} protocol=%{WORD:protocol} sport=%{INT:source_port dport=%{INT:destination_port pattern: %{WORD:action} %{WORD:scope}"}
                        }
                    }
                if "protocol=icmp" in [message]{
                    grok {
                    match => { "message" => "%{NUMBER:timestamp} %{WORD:hostname} %{WORD:type} src=%{IPV4:source_ip} dst=%{IPV4:destination_ip} protocol=%{WORD:protocol} type=0 pattern: %{WORD:action} %{WORD:scope}"}
                        }
                }
            grok {
            match => { "message" => "%{NUMBER:timestamp} %{WORD:hostname} %{WORD:type} src=%{IPV4:source_ip} dst=%{IPV4:destination_ip} protocol=%{WORD:protocol} sport=%{INT:source_port dport=%{INT:destination_port pattern: %{WORD:action} %{WORD:scope}"}
                }
            if [action] == "1" {  mutate {gsub => ["action", "1", "deny"]}}
            if [action] == "0" {  mutate {gsub => ["action", "0", "allow"]}}
            mutate { add_field => { "host" => "meraki" }}
        }
        # Natd ip flow
		if "ip_flow" in [message]{
            grok {
            match => { "message" => "%{NUMBER:timestamp} %{WORD:type} src=%{IPV4:source_ip} dst=%{IPV4:destination_ip} protocol=%{WORD:protocol} sport=%{INT:source_port} dport=%{INT:destination_port} (translated_src_ip=%{IPV4:translated_source_ip} |translated_dst_ip=%{IPV4:translated_destination_ip} )translated_port=%{INT:translated_port}"}
                }
            mutate { add_field => { "host" => "meraki" }}
		}
        # Event
        if "events" in [message]{
            grok {
            match => { "message" => "%{NUMBER:timestamp} %{WORD:hostname} %{WORD:type} %{USERNAME:event} %{WORD:protocol}: %{GREEDYDATA:message}"}
                }
            mutate { add_field => { "host" => "meraki" }}
        }
        if "client_vpn_connect"  in [message] or "client_vpn_disconnect" in [message]{
            grok{
                match => { "message" => "%{NUMBER:timestamp} %{WORD:hostname} %{WORD:type} %{USERNAME:event} user id \'%{USERNAME:user}\' local ip %{IPV4:local_ip} connected from %{IPV4:external_ip}"}
            }
            mutate { add_field => { "host" => "meraki" }}
        }
        if "no configuration found" in [message]{
            grok{
                match => {"message" => "%{NUMBER:timestamp} %{WORD:hostname} %{WORD:type} %{GREEDYDATA:event}: no configuration found for %{IPV4:source_ip}"}
            }
            mutate { add_field => { "host" => "meraki" }}
        }
        # URL
        if "urls" in [message]{
            grok {
                match => { "message" => "%{NUMBER:timestamp} %{WORD:hostname} %{WORD:type} src=%{IPV4:source_ip}:%{INT:source_port dst=%{IPV4:destination_ip}:%{INT:destination_port mac=%{MAC:mac} request: %{WORD:method} %{URI:url}"}
                }
            mutate { add_field => { "host" => "meraki" }}
        }
        if "agent" in [message]{
            grok {
                match => { "message" => "%{NUMBER:timestamp} %{WORD:hostname} %{WORD:type} src=%{IPV4:source_ip}:%{INT:source_port dst=%{IPV4:destination_ip}:%{INT:destination_port mac=%{MAC:mac} agent=%{GREEDYDATA:agent} request: %{WORD:method} %{URI:url}"}
                 }
                 mutate { add_field => { "host" => "meraki" }}
        }
    }
}

I guess that's where I am confused about how to implement this same custom parsing/filtering in the new architecture.

Thanks!
James