This guide is a continuation of this blog post here. The following assumes that you already have an Elasticsearch instance set up and ready to go.  This post will walk you through installing and setting up logstash for sending Cisco ASA messages to an Elasticsearch index.

As of today (6/16/2015), version 1.5.1 is the latest stable release of Logstash so I will be using 1.5.1 in my setup guide.

---

OPTIONAL

Before we begin sending data to Elasticsearch, it is probably a good idea to set up a custom template in Elasticsearch. This will set it so that specific fields are set for the correct types of data.

For example: By default, all data passed to a new index in elasticsearch is treated as a string type. You wouldn’t want this for something like the bytes field in case you wanted to add up all the bytes for a specific time window search. So you create a custom template so that the bytes field is stored as type long. This is totally optional but it is probably in your benifit to do this before you start collecting your ASA syslogs. You can read more about custom elasticsearch templates here or you can just copy mine by doing this…

Copy the text below, change the IP address at the top to the IP of your elasticsearch server and save it as template.sh on your desktop.

#!/bin/sh
curl -XPUT http://10.0.0.112:9200/_template/logstash_per_index -d '
{
    "template" : "logstash*",
    "mappings" : {
      "cisco-fw" : {
         "properties": {
            "@timestamp":{"type":"date","format":"dateOptionalTime"},
            "@version":{"type":"string", "index" : "not_analyzed"},
	    "action":{"type":"string"},
	    "bytes":{"type":"long"},
	    "cisco_message":{"type":"string"},
	    "ciscotag":{"type":"string", "index" : "not_analyzed"},
	    "connection_count":{"type":"long"},
            "connection_count_max":{"type":"long"},
	    "connection_id":{"type":"string"},
            "direction":{"type":"string"},
            "dst_interface":{"type":"string"},
	    "dst_ip":{"type":"string"},
            "dst_mapped_ip":{"type":"ip"},
	    "dst_mapped_port":{"type":"long"},
            "dst_port":{"type":"long"},
            "duration":{"type":"string"},
	    "err_dst_interface":{"type":"string"},
	    "err_dst_ip":{"type":"ip"},
	    "err_icmp_code":{"type":"string"},
	    "err_icmp_type":{"type":"string"},
	    "err_protocol":{"type":"string"},
            "err_src_interface":{"type":"string"},
            "err_src_ip":{"type":"ip"},
            "geoip":{
               "properties":{
                  "area_code":{"type":"long"},
                  "asn":{"type":"string", "index":"not_analyzed"},
                  "city_name":{"type":"string", "index":"not_analyzed"},
                  "continent_code":{"type":"string"},
                  "country_code2":{"type":"string"},
                  "country_code3":{"type":"string"},
                  "country_name":{"type":"string", "index":"not_analyzed"},
                  "dma_code":{"type":"long"},
                  "ip":{"type":"ip"},
                  "latitude":{"type":"double"},
                  "location":{"type":"geo_point"},
                  "longitude":{"type":"double"},
                  "number":{"type":"string"},
                  "postal_code":{"type":"string"},
                  "real_region_name":{"type":"string", "index":"not_analyzed"},
                  "region_name":{"type":"string", "index":"not_analyzed"},
                  "timezone":{"type":"string"}
               }
            },
            "group":{"type":"string"},
 	    "hashcode1": {"type": "string"}, 
 	    "hashcode2": {"type": "string"}, 
            "host":{"type":"string"},
            "icmp_code":{"type":"string"},
            "icmp_code_xlated":{"type":"string"},
            "icmp_seq_num":{"type":"string"},
            "icmp_type":{"type":"string"},
            "interface":{"type":"string"},
            "is_local_natted":{"type":"string"},
            "is_remote_natted":{"type":"string"},
            "message":{"type":"string"},
            "orig_dst_ip":{"type":"ip"},
            "orig_dst_port":{"type":"long"},
            "orig_protocol":{"type":"string"},
            "orig_src_ip":{"type":"ip"},
            "orig_src_port":{"type":"long"},
            "policy_id":{"type":"string"},
            "protocol":{"type":"string"},
            "reason":{"type":"string"},
            "seq_num":{"type":"long"},
            "spi":{"type":"string"},
            "src_interface":{"type":"string"},
            "src_ip":{"type":"ip"},
            "src_mapped_ip":{"type":"ip"},
            "src_mapped_port":{"type":"long"},
            "src_port":{"type":"long"},
            "src_xlated_interface":{"type":"string"},
            "src_xlated_ip":{"type":"ip"},
            "syslog_facility":{"type":"string"},
            "syslog_facility_code":{"type":"long"},
            "syslog_pri":{"type":"string"},
            "syslog_severity":{"type":"string"},
            "syslog_severity_code":{"type":"long"},
            "tags":{"type":"string"},
            "tcp_flags":{"type":"string"},
            "timestamp":{"type":"string"},
            "tunnel_type":{"type":"string"},
            "type":{"type":"string"},
            "user":{"type":"string"},
            "xlate_type":{"type":"string"}
      }
    }
  }
}'

Save it and close. Now, open up a terminal window and change to the directory where the template.sh file is located. We need to set the script as executable so we can run it. Run the following commands in a terminal window…

cd Desktop
chmod +x template.sh
./template.sh

You should get back {“acknowledged”:true}

Now back to the tutorial…

---

Download and Install Java

If you are doing this on a fresh install of Ubuntu (like me), the first thing you’re going to need to do is install Java. Logstash requires at least Java 7 to function so let’s set that up. If you already have Java on your machine, you can skip to the next section. I will be using Java 8 in this example but you can run 7 or openjdk if you wish.

Open a terminal window (ctrl+shift+t) and type…

sudo apt-add-repository ppa:webupd8team/java
sudo apt-get update
sudo apt-get install oracle-java8-installer

Once you have accepted the license agreement, Java is ready to go.

 

---

Download and Install Logstash

Open a terminal window (ctrl+shift+t) and run these commands…

wget -O - http://packages.elasticsearch.org/GPG-KEY-elasticsearch | sudo apt-key add -
echo 'deb http://packages.elasticsearch.org/logstash/1.5/debian stable main' | sudo tee /etc/apt/sources.list.d/logstash.list
sudo apt-get update
sudo apt-get install logstash

 

---

Logstash Configuration

Logstash is now installed so now we need to write a configuration file so that we can do things like specify listening port, patterns, the IP of the Elasticsearch server etc.

A Logstash configuration file is basically built of 3 parts: The input (network protocol, listening port, data type etc.), the filter (patterns, grok filters, syslog severity etc.) and the output (IP address of the elasticsearch server logstash is shipping the modified data to etc.).

First and foremost, let’s create a blank logstash configuration file. Open up your terminal window and type…

sudo vi /etc/logstash/conf.d/logstash.conf

or if you don’t know how to use VI

sudo gedit /etc/logstash/conf.d/logstash.conf

 

I am going to post my logstash configuration file below and then I will explain each part of the file and what it is doing. So, copy my code, paste it into your logstash.conf file and follow along below making changes to your file as you see fit.

logstash.conf

input {
  	udp { 
    		port => 5544
    		type => "cisco-fw"
  	}
}

filter {
	grok {
    		match => ["message", "%{CISCO_TAGGED_SYSLOG} %{GREEDYDATA:cisco_message}"]
  	}

  	# Extract fields from the each of the detailed message types
  	# The patterns provided below are included in core of LogStash 1.4.2.
	grok {
		match => [
  			"cisco_message", "%{CISCOFW106001}",
  			"cisco_message", "%{CISCOFW106006_106007_106010}",
  			"cisco_message", "%{CISCOFW106014}",
  			"cisco_message", "%{CISCOFW106015}",
  			"cisco_message", "%{CISCOFW106021}",
  			"cisco_message", "%{CISCOFW106023}",
  			"cisco_message", "%{CISCOFW106100}",
  			"cisco_message", "%{CISCOFW110002}",
  			"cisco_message", "%{CISCOFW302010}",
  			"cisco_message", "%{CISCOFW302013_302014_302015_302016}",
  			"cisco_message", "%{CISCOFW302020_302021}",
  			"cisco_message", "%{CISCOFW305011}",
  			"cisco_message", "%{CISCOFW313001_313004_313008}",
  			"cisco_message", "%{CISCOFW313005}",
  			"cisco_message", "%{CISCOFW402117}",
  			"cisco_message", "%{CISCOFW402119}",
  			"cisco_message", "%{CISCOFW419001}",
  			"cisco_message", "%{CISCOFW419002}",
  			"cisco_message", "%{CISCOFW500004}",
  			"cisco_message", "%{CISCOFW602303_602304}",
  			"cisco_message", "%{CISCOFW710001_710002_710003_710005_710006}",
  			"cisco_message", "%{CISCOFW713172}",
  			"cisco_message", "%{CISCOFW733100}"
		]
	}

  	# Parse the syslog severity and facility
  	syslog_pri { }

	geoip {
      		add_tag => [ "GeoIP" ]
      		database => "/opt/logstash/databases/GeoLiteCity.dat"
      		source => "src_ip"
    	}

	if [geoip][city_name]      == "" { mutate { remove_field => "[geoip][city_name]" } }
    	if [geoip][continent_code] == "" { mutate { remove_field => "[geoip][continent_code]" } }
    	if [geoip][country_code2]  == "" { mutate { remove_field => "[geoip][country_code2]" } }
    	if [geoip][country_code3]  == "" { mutate { remove_field => "[geoip][country_code3]" } }
    	if [geoip][country_name]   == "" { mutate { remove_field => "[geoip][country_name]" } }
    	if [geoip][latitude]       == "" { mutate { remove_field => "[geoip][latitude]" } }
    	if [geoip][longitude]      == "" { mutate { remove_field => "[geoip][longitude]" } }
    	if [geoip][postal_code]    == "" { mutate { remove_field => "[geoip][postal_code]" } }
    	if [geoip][region_name]    == "" { mutate { remove_field => "[geoip][region_name]" } }
    	if [geoip][time_zone]      == "" { mutate { remove_field => "[geoip][time_zone]" } }

	# Gets the source IP whois information from the GeoIPASNum.dat flat file database
	geoip {
      		add_tag => [ "Whois" ]
      		database => "/opt/logstash/databases/GeoIPASNum.dat"
      		source => "src_ip"
    	}

 	# Parse the date
  	date {
    		match => ["timestamp",
      			"MMM dd HH:mm:ss",
      			"MMM  d HH:mm:ss",
      			"MMM dd yyyy HH:mm:ss",
      			"MMM  d yyyy HH:mm:ss"
    		]
  	}
}

output {
	stdout { 
		codec => json
	}

	elasticsearch {
    		host => "10.0.0.133"
    		flush_size => 1
	}
}

End of logstash.conf

Don’t freak out! I will walk you through my configuration file and explain what each section is doing. Like I said before, a logstash configuration file is made up of 3 parts: the input, the filter and the output. So let’s walk through it, shall we?

Note: All of my examples will be using this as an example ASA syslog message.

<182>May 07 2015 13:26:42: %ASA-6-302014: Teardown TCP connection 48809467 for outside:124.35.68.19/46505 to inside:10.10.10.32/443 duration 0:00:00 bytes 300 TCP FINs

 

The Input

input {
  udp { 
    port => 5544
    type => "cisco-fw"
  }
}

This tells logstash the protocol (UDP) and what port to listen (5544). You can make it any port you want but you just need to set it in your ASA firewall like so…

or for you CLI people…

logging enable
logging timestamp
logging buffer-size 40960
logging trap informational
logging facility 22
logging host inside 10.0.0.133 17/5544

You can learn more on how to set this up by checking out my ASA syslog tutorial here

The Filter

This is the filter section. This is where most of the work will be done in logstash. The is where you can do things like use grok patterns to split data into fields based off the message and other neat little features for manipulating your data. My filter section has 5 different parts: grok, syslog_pri, geoip, mutate and date. Let’s look at them individually and show what exactly they do.

Grok

filter {
	grok {
    		match => ["message", "%{CISCO_TAGGED_SYSLOG} %{GREEDYDATA:cisco_message}"]
  	...
}

This section is where you use patterns to match various elements of your log messages and extract them into fields. This specific pattern is using the CISCO_TAGGED_SYSLOG definition in the firewalls pattern built into logstash and splitting it into sections.

The CISCO_TAGGED_SYSLOG looks like this…

CISCO_TAGGED_SYSLOG ^<%{POSINT:syslog_pri}>%{CISCOTIMESTAMP:timestamp}( %{SYSLOGHOST:sysloghost})?: %%{CISCOTAG:ciscotag}:

and, based off our example ASA syslog May 07 2015 13:26:42: %ASA-6-302014: Teardown TCP connection 48809467 for outside:124.35.68.19/46505 to inside:10.10.10.32/443 duration 0:00:00 bytes 300 TCP FINs, it pulls out this into these fields…

syslog_pri = 182
timestamp = May 07 2015 13:26:42
sysloghost =
ciscotag = ASA-6-302014

and the rest of the message of Teardown TCP connection 48809467 for outside:124.35.68.19/46505 to inside:10.10.10.32/443 duration 0:00:00 bytes 300 TCP FINs is just is just set as GREEDYDATA:cisco_message, which will be used later in the next grok filter below.

grok {
    match => [
    	"cisco_message", "%{CISCOFW106001}",
    	....
    	"cisco_message", "%{CISCOFW733100}"
    ]
}

Now, using the GREEDYDATA:cisco_message from the previous grok filter, we are going to use the same firewall patterns file built into logstash and match the message type based off the message. The filter goes through all the patterns until it finds a match and then splits the contents of the message into fields. So, based off our GREEDYDATA:cisco_message of Teardown TCP connection 48809467 for outside:124.35.68.19/46505 to inside:10.10.10.32/443 duration 0:00:00 bytes 300 TCP FINs from our example syslog, it will try to match it against the patterns file. Our specific example matches the CISCOFW302013_302014_302015_302016 line in the patterns file, which reads…

CISCOFW302013_302014_302015_302016 %{CISCO_ACTION:action}(?: %{CISCO_DIRECTION:direction})? %{WORD:protocol} connection %{INT:connection_id} for %{DATA:src_interface}:%{IP:src_ip}/%{INT:src_port}( \(%{IP:src_mapped_ip}/%{INT:src_mapped_port}\))?(\(%{DATA:src_fwuser}\))? to %{DATA:dst_interface}:%{IP:dst_ip}/%{INT:dst_port}( \(%{IP:dst_mapped_ip}/%{INT:dst_mapped_port}\))?(\(%{DATA:dst_fwuser}\))?( duration %{TIME:duration} bytes %{INT:bytes})?(?: %{CISCO_REASON:reason})?( \(%{DATA:user}\))?

…and puts the data into these fields…

action = Teardown
protocol = TCP
connection_id = 48809467
src_interface = outside
src_ip = 124.35.68.19
src_port = 46505
dst_interface = inside
dst_ip = 10.10.10.32
dst_port = 443
duration = 0 =00 =00
bytes = 300
reason = TCP FINs

 

Syslog_pri

syslog_pri { }

This section takes the POSINT syslog_pri from the first grok filter and gets the facility and severity level of the syslog message. Our example had a syslog_pri number of 182 and logstash can determine that the message is an informational based message from the local6 facility. I was able to get that by referencing this chart and finding which column and row 182 fell under.

             emergency   alert   critical   error   warning   notice   info   debug
 kernel              0       1          2       3         4        5      6       7
 user                8       9         10      11        12       13     14      15
 mail               16      17         18      19        20       21     22      23
 system             24      25         26      27        28       29     30      31
 security           32      33         34      35        36       37     38      39
 syslog             40      41         42      43        44       45     46      47
 lpd                48      49         50      51        52       53     54      55
 nntp               56      57         58      59        60       61     62      63
 uucp               64      65         66      67        68       69     70      71
 time               72      73         74      75        76       77     78      79
 security           80      81         82      83        84       85     86      87
 ftpd               88      89         90      91        92       93     94      95
 ntpd               96      97         98      99       100      101    102     103
 logaudit          104     105        106     107       108      109    110     111
 logalert          112     113        114     115       116      117    118     119
 clock             120     121        122     123       124      125    126     127
 local0            128     129        130     131       132      133    134     135
 local1            136     137        138     139       140      141    142     143
 local2            144     145        146     147       148      149    150     151
 local3            152     153        154     155       156      157    158     159
 local4            160     161        162     163       164      165    166     167
 local5            168     169        170     171       172      173    174     175
 local6            176     177        178     179       180      181    182     183
 local7            184     185        186     187       188      189    190     191

Pretty cool stuff.

---

OPTIONAL: GeoIP

geoip {
    	add_tag => [ "GeoIP" ]
    	database => "/opt/logstash/databases/GeoLiteCity.dat"
    	source => "src_ip"
}

This part is completely optional and up to you if you want to set it up. What it does is it uses the source ip address in the syslog and it gets location data based of a flat file database. This flat file database will be used by logstash to get the location of the IP addresses hitting the firewall so you can get turn the source IP address of 124.35.68.19 into this…

"geoip": {
      "ip": "124.35.68.19",
      "country_code2": "US",
      "country_code3": "USA",
      "country_name": "United States",
      "continent_code": "NA",
      "region_name": "NJ",
      "city_name": "Edison",
      "postal_code": "08820",
      "latitude": 40.57669999999999,
      "longitude": -74.3674,
      "dma_code": 501,
      "area_code": 732,
      "timezone": "America/New_York",
      "real_region_name": "New Jersey",
      "location": [
        -74.3674,
        40.57669999999999
      ]
}

If this doesn’t interest you, remove the geoip section from the config file and skip to the next section. If this does interest you, follow the steps below.

Open a terminal window and type…

cd ~
wget http://geolite.maxmind.com/download/geoip/database/GeoLiteCity.dat.gz
sudo mkdir /opt/logstash/databases
gunzip GeoLiteCity.dat.gz
sudo mv ~/GeoLiteCity.dat /opt/logstash/databases/

and that is it for the first GeoIP tag for location data.

This is another optional GeoIP filter…

geoip {
    	add_tag => [ "Whois" ]
    	database => "/opt/logstash/databases/GeoIPASNum.dat"
    	source => "src_ip"
}

Just like the GeoIP location database, the GeoIP ASN database uses the source IP address, but instead of returning location information, it returns the ASN information. So essentially it turns 124.35.68.19 into…

"number": "AS6128",
"asn": "Cablevision Systems Corp."

So you can see ISP names in your elasticsearch searches. Also very cool and highly recommended for your setup. If this does not interest you for your setup, remove the geoip section from your configuration file and skip to the next section. If this does interest you, follow the steps below…

Open a terminal window and type…

cd ~
wget http://download.maxmind.com/download/geoip/database/asnum/GeoIPASNum.dat.gz
gunzip GeoIPASNum.dat.gz
sudo mv ~/GeoIPASNum.dat /opt/logstash/databases/

and that’s it with the GeoIP section.
Mutate

if [geoip][city_name]      == "" { mutate { remove_field => "[geoip][city_name]" } }

This part is fairly straight forward. Basically what is happening is that if the entry in the GeoIPASNum.dat file for a particular IP address has a country but does not have a city, it will remove that field before it is inserted into elasticsearch. That is basically it with the mutate section of my config file.

Date

date {
    match => ["timestamp",
    	"MMM dd HH:mm:ss",
    	"MMM  d HH:mm:ss",
    	"MMM dd yyyy HH:mm:ss",
    	"MMM  d yyyy HH:mm:ss"
    ]
}

This part is also pretty straight forward. It takes the timestamp value from the first grok filter and sets it as the timestamp when putting it into elasticsearch. So basically, instead of the timestamp being set as the time when logstash received the message, the timestamp is set as when the event was triggered on the firewall based off the firewalls clock and hopefully the firewall is configured to use NTP so that all your devices clocks are synchronized.

Output

output {
	stdout { 
		codec => json
	}

	elasticsearch {
		host => "10.0.0.133"
		flush_size => 1
	}
}

This section is how the final result is displayed/sent to various things. My output section has 2 parts: stdout and elasticsearch.

Stdout

stdout { 
	codec => json
}

Stdout is optional but I have it in there so I can see if everything is working properly through the terminal window like so

Obviously you do not need this part to run but I like to have it in there for debugging purposes.

Elasticsearch

elasticsearch {
    	host => "10.0.0.133"
    	flush_size => 1
}

Elasticsearch is where you specify the IP address of your elasticsearch server and that is pretty much it.

---

So now that you have your logstash.conf file set up, you can now run logstash for your ASA firewall. Save your config file and type in this into a terminal window…

/opt/logstash/bin/logstash -f /etc/logstash/conf.d/logstash.conf

That’s it, you’re finished.

Note: If you have not turned on syslog messages on your ASA firewall, read my other blog post here.