# Parse firewall logs collected by logstash through syslog **URL:** \ **Category:** Logstash\ **Created:** [May 22, 2018, 12:06pm UTC](https://discuss.elastic.co/t/parse-firewall-logs-collected-by-logstash-through-syslog/132812 "2018-05-22T12:06:59Z")\ **Posts on this page:** 13\ **Page:** 1
**Author:** ![Dora](https://avatars.discourse-cdn.com/v4/letter/d/b9e5f3/32.png) [@Dora](https://discuss.elastic.co/u/Dora)\ **Post date:** [May 22, 2018, 12:06pm UTC](https://discuss.elastic.co/t/parse-firewall-logs-collected-by-logstash-through-syslog/132812/1 "2018-05-22T12:06:59Z")
Hello, I am using ELK (version 6.2.4), and I would like to collect firewall logs (Fortigate) from another SIEM, so I followed the following steps: - I configured the other SIEM to forward these logs to ELK via the UDP protocol: port 514 in payload format - I checked if the logs are received on the network interface with tcpdump, they are received - I created a conf file (syslog.conf) as follows: **input {** \*\* udp {\*\* \*\* port =\> 514\*\* \*\* type =\> syslog\*\* \*\* }\*\* **filter {** \*\* if [type] == "syslog" {\*\* \*\* grok {\*\* \*\* match =\> { "message" =\> "%{SYSLOGTIMESTAMP:syslog\_timestamp} %{SYSLOGHOST:syslog\_hostname} %{DATA:syslog\_program}(?:[%{POSINT:syslog\_pid}])?: %{GREEDYDATA:syslog\_message}" }\*\* \*\* add\_field =\> ["received\_at", "%{@timestamp}"]\*\* \*\* add\_field =\> ["received\_from", "%{host}"]\*\* \*\* }\*\* \*\* date {\*\* \*\* match =\> ["syslog\_timestamp", "MMM d HH:mm:ss", "MMM dd HH:mm:ss"]\*\* \*\* }\*\* \*\* }\*\* **}** **output {** \*\* elasticsearch {\*\* \*\* hosts =\> ["localhost:9200"]\*\* \*\* user =\> "user"\*\* \*\* password =\> "password"\*\* \*\* index =\> "fortigate-%{+YYYY.MM.dd}"\*\* \*\* }\*\* **}** But the index is not created and i have the following error: [logstash.agent] Failed to execute action {:action=\>LogStash::PipelineAction::Create/pipeline\_id:fortigate, :exception=\>"LogStash::ConfigurationError", :message=\>"Expected one of #, =\> at line 8, column 6 (byte 71) after input {\n udp {\n port =\> 514\n type =\> syslog\n }\n\nfilter {\n if ", :backtrace=\>["/usr/share/logstash/logstash-core/lib/logstash/compiler.rb:42:in `compile_imperative'", "/usr/share/logstash/logstash-core/lib/logstash/compiler.rb:50:in`compile\_graph'", "/usr/share/logstash/logstash-core/lib/logstash/compiler.rb:12:in `block in compile_sources'", "org/jruby/RubyArray.java:2486:in`map'", "/usr/share/logstash/logstash-core/lib/logstash/compiler.rb:11:in `compile_sources'", "/usr/share/logstash/logstash-core/lib/logstash/pipeline.rb:51:in`initialize'", "/usr/share/logstash/logstash-core/lib/logstash/pipeline.rb:169:in `initialize'", "/usr/share/logstash/logstash-core/lib/logstash/pipeline_action/create.rb:40:in`execute'", "/usr/share/logstash/logstash-core/lib/logstash/agent.rb:315:in `block in converge_state'", "/usr/share/logstash/logstash-core/lib/logstash/agent.rb:141:in`with\_pipelines'", "/usr/share/logstash/logstash-core/lib/logstash/agent.rb:312:in `block in converge_state'", "org/jruby/RubyArray.java:1734:in`each'", "/usr/share/logstash/logstash-core/lib/logstash/agent.rb:299:in `converge_state'", "/usr/share/logstash/logstash-core/lib/logstash/agent.rb:166:in`block in converge\_state\_and\_update'", "/usr/share/logstash/logstash-core/lib/logstash/agent.rb:141:in `with_pipelines'", "/usr/share/logstash/logstash-core/lib/logstash/agent.rb:164:in`converge\_state\_and\_update'", "/usr/share/logstash/logstash-core/lib/logstash/agent.rb:90:in `execute'", "/usr/share/logstash/logstash-core/lib/logstash/runner.rb:348:in`block in execute'", "/usr/share/logstash/vendor/bundle/jruby/2.3.0/gems/stud-0.0.23/lib/stud/task.rb:24:in `block in initialize'"]} - I would like to know please how to parse these logs and be able to create the index, and if there are other ways to do it. For information, I already collect the netflow with elastiflow and windows logs with winlogbeat whose pipelines are different. ---
**Author:** ![magnusbaeck](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/magnusbaeck/32/44943_2.png) [@magnusbaeck](https://discuss.elastic.co/u/magnusbaeck)\ **Post date:** [May 24, 2018, 8:42pm UTC](https://discuss.elastic.co/t/parse-firewall-logs-collected-by-logstash-through-syslog/132812/2 "2018-05-24T20:42:46Z")
You're not closing the input block with `}`. ---
**Author:** ![Dora](https://avatars.discourse-cdn.com/v4/letter/d/b9e5f3/32.png) [@Dora](https://discuss.elastic.co/u/Dora)\ **Post date:** [May 27, 2018, 3:07pm UTC](https://discuss.elastic.co/t/parse-firewall-logs-collected-by-logstash-through-syslog/132812/3 "2018-05-27T15:07:58Z")
Thank you, I'm going to change that and try again. I will keep you informed of the result. ---
**Author:** ![Dora](https://avatars.discourse-cdn.com/v4/letter/d/b9e5f3/32.png) [@Dora](https://discuss.elastic.co/u/Dora)\ **Post date:** [May 28, 2018, 10:51am UTC](https://discuss.elastic.co/t/parse-firewall-logs-collected-by-logstash-through-syslog/132812/4 "2018-05-28T10:51:55Z")
Hello, Everything seems ok now but this time I have the following error: [INFO][logstash.pipeline] Pipeline started successfully {:pipeline\_id=\>"firewall", :thread=\>"#\"} [INFO][logstash.agent] Pipelines running {:count=\>4, :pipelines=\>[".monitoring-logstash", "beats", "elastiflow", "firewall"]} [INFO][logstash.inputs.udp] Starting UDP listener {:address=\>"0.0.0.0:514"} [WARN][logstash.inputs.udp] UDP listener died {:exception=\>#\, :backtrace=\>["org/jruby/ext/socket/RubyUDPSocket.java:200:in `bind'", "/usr/share/logstash/vendor/bundle/jruby/2.3.0/gems/logstash-input-udp-3.3.3/lib/logstash/inputs/udp.rb:102:in`udp\_listener'", "/usr/share/logstash/vendor/bundle/jruby/2.3.0/gems/logstash-input-udp-3.3.3/lib/logstash/inputs/udp.rb:58:in `run'", "/usr/share/logstash/logstash-core/lib/logstash/pipeline.rb:514:in`inputworker'", "/usr/share/logstash/logstash-core/lib/logstash/pipeline.rb:507:in `block in start\_input'"]} Thank you for your help ---
**Author:** ![magnusbaeck](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/magnusbaeck/32/44943_2.png) [@magnusbaeck](https://discuss.elastic.co/u/magnusbaeck)\ **Post date:** [May 28, 2018, 11:51am UTC](https://discuss.elastic.co/t/parse-firewall-logs-collected-by-logstash-through-syslog/132812/5 "2018-05-28T11:51:49Z")
Unless you run Logstash as root it can't listen on port 514. You can however use iptables to redirect that port to a port that Logstash _can_ listen on. Does it work if you pick another port (\>1024)? ---
**Author:** ![Dora](https://avatars.discourse-cdn.com/v4/letter/d/b9e5f3/32.png) [@Dora](https://discuss.elastic.co/u/Dora)\ **Post date:** [May 28, 2018, 1:25pm UTC](https://discuss.elastic.co/t/parse-firewall-logs-collected-by-logstash-through-syslog/132812/6 "2018-05-28T13:25:28Z")
Thank you very much, I actually redirected port 514 to port 5044, it worked and I was able to view the firewall logs on kibana. However, I would like to know please how to parse them with logstash, I receive them as a raw message (type, @timestamp, message) I tried to parse them with "Grok debbuger" and I managed to do it with that: time=%{TIME:time} devname=%{HOST:hostname} devid=%{HOST:devid} logid=%{NUMBER:logid} type=%{WORD:type} subtype=%{WORD:subtype} level=%{WORD:level} vd=%{WORD:vdom} srcip=%{IP:srcip} srcport=%{NUMBER:srcport} srcintf="%{HOST:srcintf}" dstip=%{IP:dstip} dstport=%{NUMBER:dstport} dstintf="%{WORD:dstintf}" sessionid=%{NUMBER:sessionid} proto=%{NUMBER:proto} action=%{WORD:action} policyid=%{DATA:policyid} dstcountry="%{WORD:dstcountry}" srccountry="%{WORD:srccountry}" trandisp=%{WORD:trandisp} service="%{WORD:service}" duration=%{NUMBER,duration} sentbyte=%{INT:sentbyte} rcvdbyte=%{INT:rcvdbyte} sentpkt=%{INT:sentpkt} rcvdpkt=%{INT:rcvdpkt} vpn="%{WORD:vpn}" vpntype=%{GREEDYDATA:vpntype} shapersentname="%{WORD:shapersentname}" shaperdropsentbyte=%{INT:shaperdropsentbyte} thank you in advance ---
**Author:** ![magnusbaeck](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/magnusbaeck/32/44943_2.png) [@magnusbaeck](https://discuss.elastic.co/u/magnusbaeck)\ **Post date:** [May 28, 2018, 1:40pm UTC](https://discuss.elastic.co/t/parse-firewall-logs-collected-by-logstash-through-syslog/132812/7 "2018-05-28T13:40:52Z")
Use a kv filter instead. ---
**Author:** ![Dora](https://avatars.discourse-cdn.com/v4/letter/d/b9e5f3/32.png) [@Dora](https://discuss.elastic.co/u/Dora)\ **Post date:** [May 28, 2018, 2:39pm UTC](https://discuss.elastic.co/t/parse-firewall-logs-collected-by-logstash-through-syslog/132812/8 "2018-05-28T14:39:17Z")
I modified the filter section as follows: filter { if [type] == "syslog" { grok { match =\> { "message" =\> "%{SYSLOGTIMESTAMP:syslog\_timestamp} %{SYSLOGHOST:syslog\_hostname} %{DATA:syslog\_program}(?:[%{POSINT:syslog\_pid}])?: %{GREEDYDATA:syslog\_message}" } add\_field =\> ["received\_at", "%{@timestamp}"] add\_field =\> ["received\_from", "%{host}"] } date { match =\> ["syslog\_timestamp", "MMM d HH:mm:ss", "MMM dd HH:mm:ss"] } } kv {} } it could identify some fields and add them but it doesn't parse all the message ---
**Author:** ![magnusbaeck](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/magnusbaeck/32/44943_2.png) [@magnusbaeck](https://discuss.elastic.co/u/magnusbaeck)\ **Post date:** [May 28, 2018, 4:51pm UTC](https://discuss.elastic.co/t/parse-firewall-logs-collected-by-logstash-through-syslog/132812/9 "2018-05-28T16:51:26Z")
What does a processed message look like then? Use a `stdout { codec => rubydebug }` output. ---
**Author:** ![Dora](https://avatars.discourse-cdn.com/v4/letter/d/b9e5f3/32.png) [@Dora](https://discuss.elastic.co/u/Dora)\ **Post date:** [June 4, 2018, 12:05pm UTC](https://discuss.elastic.co/t/parse-firewall-logs-collected-by-logstash-through-syslog/132812/10 "2018-06-04T12:05:09Z")
Hello, In fact, I have to explain a little more. So, the structure of the logs that I receive is not identical everywhere. I get the raw firewall logs of: PaloAlto and Fortigate and I also get the WAF logs: F5 and Deny All. and I created only one configuration file that I named "firewall.conf" to receive all that. That's why there are messages that are partially parsed and others not with kv. I show below an extract of each one's logs to see what his structure looks like. What interests me most is the firewall logs **PaloAlto** : the fields are separated by | |usrName=|SourceUser=|DestinationUser=|Application=not-applicable| **Foritgate** : the fields are separated by spaces date=2018-06-04 time=12:49:09 devname=FG-B thank you in advance ---
**Author:** ![Dora](https://avatars.discourse-cdn.com/v4/letter/d/b9e5f3/32.png) [@Dora](https://discuss.elastic.co/u/Dora)\ **Post date:** [June 4, 2018, 12:13pm UTC](https://discuss.elastic.co/t/parse-firewall-logs-collected-by-logstash-through-syslog/132812/11 "2018-06-04T12:13:24Z")
I think I need to create a configuration file for each one, but I don't know how logstash will know that it's fortigate logs and not Paloalto ones knowing that they are all sent to port 5044, to parse them differently. Do I have to specify the name of the device on the configuration file? ---
**Author:** ![w0rk3r](https://avatars.discourse-cdn.com/v4/letter/w/e495f1/32.png) [@w0rk3r](https://discuss.elastic.co/u/w0rk3r)\ **Post date:** [June 15, 2018, 2:55pm UTC](https://discuss.elastic.co/t/parse-firewall-logs-collected-by-logstash-through-syslog/132812/12 "2018-06-15T14:55:26Z")
Hey dora, here palo alto logs are in csv format, and to specify on which port we are going to receive the palo alto logs, u can use the following: input { syslog { timezone =\> "Brazil/East" port =\> "5044" type =\> "syslog" } } for the fortigate, ossec has built-in decoders for them... i think wazuh could be useful on your enviroment. ---
**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\ **Post date:** [July 13, 2018, 3:06pm UTC](https://discuss.elastic.co/t/parse-firewall-logs-collected-by-logstash-through-syslog/132812/13 "2018-07-13T15:06:53Z")
This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.