<h2>Intro: using a SIEM approach</h2>Effectively monitoring security across a large organization is a non-trivial task faced everyday by all sorts of organizations.The speed, scalability and flexibility of the Elastic stack can play as a great asset when trying to get visibility and proactively monitoring large amounts of data.
The traditional SIEM approach relies on normalization of the data from raw, based on a schema.
For example a failed login, be it from a Linux
<pre class="prettyprint">Nov 26 12:15:04 zeus sshd[19571]: Failed password for ciro from 10.0.4.23 port 57961 ssh2</pre>or a Windows host,
<pre class="prettyprint">Log Name: Security
Source: Microsoft-Windows-Security-Auditing
Date: 27/11/2015 2:07:33 PM
Event ID: 4625
Task Category: Logon
Level: Information
Keywords: Audit Failure
User: N/A
Computer: minerva
Description:
An account failed to log on.
Subject:
 Security ID: NULL SID
 Account Name: -
 Account Domain: -
 Logon ID: 0x0
Logon Type: 3
Account For Which Logon Failed:
 Account Name: gennaro
&lt;.....&gt;
</pre>will be indexed observing a common structured format:
<img src="https://api.contentstack.io/v2/assets/575e4d3043e9adc538715607/download?uid=bltbc40533d8f92d01e?uid=bltbc40533d8f92d01e" data-sys-asset-uid="bltc071fbb938e1f390" alt="Penguin" style="float: left;">
<code>"src_user": "ciro"</code> <code>"src_ip": "10.0.0.111"</code> <code>"auth_type": &nbsp;"ssh2"</code>
 
<img src="https://api.contentstack.io/v2/assets/575e4d309e7a83165490dbee/download?uid=blta689553056eba4ae?uid=blta689553056eba4ae" data-sys-asset-uid="blt1af8467faf7f8313" alt="Microsoft" style="float: left; margin: 0px;">
<code>“src_user”:”gennaro”</code> <code>“src_ip”:”10.0.0.118”</code> <code>“auth_type”:”3”</code>
Using a field naming convention allows to build correlation logic abstracting from which source the event originated from, be it a Windows or a Linux failed login. 
Also some tagging or categorization of the data can be performed,
<pre class="prettyprint">grok{
 match =&gt; { "message" =&gt; ["%{SSH_AUTH_1}","%{SSH_AUTH_2}"] }
 add_tag =&gt; [ "auth_success" ]
 }
</pre><pre class="prettyprint">grok{
 match =&gt; { "message" =&gt; ["%{SSH_AUTH_3}","%{SSH_AUTH_4}"] }
 add_tag =&gt; [ "auth_failure" ]
 }
</pre>where SSH_AUTH_X are our custom defined grok patterns to match success/failure events.
Using this approach, correlation logic can be applied to all the events, regardless of the datasource from which the event originated from.
Following the same approach, we will show how to use the Elastic stack to cover a basic network security use case, TCP host portscan detection, for which we'll implement alerting via email.
<h2>Implementation I: datasource</h2>When trying to detect whether a portscan against a given host on your premises was carried on , network traffic data becomes relevant.
<img src="https://api.contentstack.io/v2/assets/575e4d30d8edd48f7693798a/download?uid=blt29949ae0adc51fad?uid=blt29949ae0adc51fad" data-sys-asset-uid="blt7cefeff3687119e1" alt="implementation I.png" width="367" height="234" style="width: 367px; height: 234px; float: left; margin: 0px;">
For this use case we will want to monitor all events indicating a new TCP connection being initiated from source to target host, in short all TCP packets with SYN=1, ACK=0.
 
While we impatiently wait for <a href="https://github.com/elastic/beats/issues/670">Packetbeat Flows</a> to be released and allow more out-of-the-box network protocol level capture capabilities, we'll use tcpdump capture using the below command for the purpose of this blog:
<pre class="prettyprint">sudo tcpdump -i eth0 -n -tttt tcp[13] == 2 | nc localhost 5001
</pre>the above command will listen on the eth0 network interface of the monitored host and capture all and only the TCP packets indicating that a new TCP connection handshake was initiated, also avoiding resolving IP to hostnames for faster execution; then we pipe the results to netcat to send them to our Logstash instance for event processing, which we assume here to be running locally.
For convenience, we can launch the above command using a all time favourite linux CLI utility, <a href="http://linux.die.net/man/1/screen">screen</a>.
<pre class="prettyprint">!/bin/bash
screen -d -m /bin/bash -c 'sudo tcpdump -i eth0 -n -tttt tcp[13] == 2 | nc localhost 5001'
</pre>This is what the captured raw data looks like
<pre class="prettyprint">2016-02-09 13:51:09.625253 IP 192.168.1.105.60805 &gt; 192.168.1.1.80: Flags [S], seq 2127832187, win 29200, options [mss 1460,sackOK,TS val 259965981 ecr 0,nop,wscale 7], length 0
</pre><h2>Implementation II : event processing</h2>We'll use logstash to mangle the data and extract the information relevant to this use case, namely timestamp, src_ip and dst_port.
<pre class="prettyprint">grok{
 match =&gt; {"message" =&gt; "%{TCPD_TIMESTAMP:timestamp} IP %{IP:src_ip}\.%{INT:src_port} &gt; %{IP:dst_ip}\.%{INT:dst_port}(?&lt;payload&gt;[^$]+)"}
 add_tag =&gt; ["network","tcp_connection_started"]
 }
</pre>where TCPD_TIMESTAMP is a custom defined grok pattern to match <code>2016-02-09 13:51:09.625253</code>.
As we have extracted the information we were after (<code>timestamp</code>,<code>src_ip</code>,<code>dst_ip</code>) we can decide to trash <code>message</code> and <code>payload</code> fields:
 
<pre class="prettyprint">mutate{
 remove_field =&gt; ["message","payload"]
 }
</pre>Next we send these events to Elasticsearch index <code>logstash-tcpdump-%{+YYYY.MM.dd}</code>
<pre class="prettyprint"> elasticsearch {
 hosts =&gt; "es-server:9200"
 index =&gt; "logstash-tcpdump-%{+YYYY.MM.dd}"
 user =&gt; "logstash"
 password =&gt; "verysecretpassword"
 ssl =&gt; true
 cacert =&gt; "/path/to/cacert.pem"
 }
</pre><h2>Implementation III: searching for a portscan</h2>We're now at the stage where events are coming into Elasticsearch and we want to be automatically alerted when our monitored host will receive (or launch!) a portscan.
<img src="https://api.contentstack.io/v2/assets/575e4d313dc542cb38c09ead/download?uid=blt58b7e482dc763b72?uid=blt58b7e482dc763b72" style="max-width:100%;" data-sys-asset-uid="bltf0d23aa74ab8927b" alt="Thinking person">
This is what our indexed event looks like:
<pre class="prettyprint">{
 "@version": "1",
 "@timestamp": "2016-02-08T00:56:58.407Z",
 "host": "127.0.0.1",
 "port": 41433,
 "type": "tcpdump",
 "timestamp": "2016-02-08 01:56:58.407625",
 "src_ip": "192.168.1.105",
 "src_port": "55203",
 "dst_ip": "192.168.1.1",
 "dst_port": "80"
}
</pre>We can define a TCP host portscan as a large amount of connections attempted within a short amount of time between a source and a target host, where the target port is always changing from connection to connection. How would this translate to an elasticsearch query?
<pre class="prettyprint">GET logstash-tcpdump-*/_search
{
 "size": 0,
 "query": {
 "bool": {
 "must": [
 {
 "match": {
 "tags": "tcp_connection_started"
 }
 },
 {
 "range": {
 "@timestamp": {
 "gte": "now-30s"
 }
 }
 }
 ]
 }
 },
 "aggs": {
 "by_src_ip": {
 "terms": {
 "field": "src_ip"
 },
 "aggs": {
 "by_target_ip": {
 "terms": {
 "field": "dst_ip",
 "order": {
 "unique_port_count": "desc"
 }
 },
 "aggs": {
 "unique_port_count": {
 "cardinality": {
 "field": "dst_port"
 }
 }
 }
 }
 }
 }
 }
}
</pre>We leverage here a killer feature of Elasticsearch: aggregations. Specifically <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/search-aggregations-bucket-terms-aggregation.html">terms</a>&nbsp;and <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/search-aggregations-metrics-cardinality-aggregation.html">cardinality</a>&nbsp;aggregations.
Note we're purely interested in aggregated results, hence setting <code>size:0</code>. The response we receive looks like:
<pre class="prettyprint">{
 "took": 9,
 "timed_out": false,
 "_shards": {
 "total": 24,
 "successful": 24,
 "failed": 0
 },
 "hits": {
 "total": 46,
 "max_score": 0,
 "hits": []
 },
 "aggregations": {
 "by_src_ip": {
 "doc_count_error_upper_bound": 0,
 "sum_other_doc_count": 0,
 "buckets": [
 {
 "key": "192.168.1.17",
 "doc_count": 44,
 "by_target_ip": {
 "doc_count_error_upper_bound": 0,
 "sum_other_doc_count": 0,
 "buckets": [
 {
 "key": "192.168.1.105",
 "doc_count": 44,
 "unique_port_count": {
 "value": 41
 }
 }
 ]
 }
 },
 {
 "key": "192.168.1.105",
 "doc_count": 2,
 "by_target_ip": {
 "doc_count_error_upper_bound": 0,
 "sum_other_doc_count": 0,
 "buckets": [
 {
 "key": "192.168.1.10",
 "doc_count": 1,
 "unique_port_count": {
 "value": 1
 }
 },
 {
 "key": "192.168.1.32",
 "doc_count": 1,
 "unique_port_count": {
 "value": 1
 }
 }
 ]
 }
 }
 ]
 }
 }
}
</pre>From the above we can infer that host 192.168.1.17 has initiated 41 different TCP connections against host 192.168.1.105 which seems suspicious: 192.168.1.17 is our attacker.
Also host 192.168.1.105 has initiated 2 TCP connections against hosts 192.168.1.10 and 192.168.1.32, which seems legitimate.
Next we'll see how we can use Watcher to automatically receive an email when an event like this happens.
<h2>Implementation IV: alert me!</h2><a href="/what-is/elasticsearch-alerting">Watcher</a>&nbsp;is our friend here, all we need to do is to <a href="https://www.elastic.co/guide/en/watcher/current/email-services.html">configure</a>&nbsp;a service email account, then define a new Watch and define how to act when a portscan is detected.
First we define a schedule, how often should the Watch be executed:
<pre class="prettyprint">"trigger": {
 "schedule": {
 "interval": "10s"
 }
 }
</pre>Next, define what query search_type to run, on what indices and document types:
<pre class="prettyprint">"input": {
 "search": {
 "request": {
 "search_type": "query_then_fetch",
 "indices": [
 "logstash-tcpdump-*"
 ],
 "types": [
 "tcpdump"
 ],
 "body": { #&lt;insert query discussed in previous paragraph here&gt;
}
</pre>Now specify what condition would trigger the watch:
<pre class="prettyprint">"condition": {
 "script": {
 "inline": "for (int i = 0; i &lt; ctx.payload.aggregations.by_src_ip.buckets.size(); i++) {for (int j = 0; j &lt; ctx.payload.aggregations.by_src_ip.buckets[i].by_target_ip.buckets.size(); j++) {if (ctx.payload.aggregations.by_src_ip.buckets[i].by_target_ip.buckets[j].unique_port_count.value &gt; threshold) return true;};};return false;",
 "params": {
 "threshold": 50
 }
 }
 }
</pre>The above groovy script will scan our aggregated results and look for a <code>unique_port_count</code> bucket where the <code>cardinality</code> is greater than 50; so putting within context, if a host has established within 30 seconds timerange, more than 50 connection each using a different port against another host, we will call this a portscan.
Last, what action should our Watch perform once its conditions are met? Send a nice email to warn us!
<pre class="prettyprint">"actions": {
 "email_administrator": {
 "transform": {
 "script": {
 "inline": "def target='';def attacker='';def body='';for (int i = 0; i &lt; ctx.payload.aggregations.by_src_ip.buckets.size(); i++) {for (int j = 0; j &lt; ctx.payload.aggregations.by_src_ip.buckets[i].by_target_ip.buckets.size(); j++) {if (ctx.payload.aggregations.by_src_ip.buckets[i].by_target_ip.buckets[j].unique_port_count.value &gt; threshold) {target=ctx.payload.aggregations.by_src_ip.buckets[i].by_target_ip.buckets[j].key;attacker=ctx.payload.aggregations.by_src_ip.buckets[i].key;body='Detected portscan from ['+attacker+'] to ['+target+']. '+ctx.payload.aggregations.by_src_ip.buckets[i].by_target_ip.buckets[j].unique_port_count.value+ ' unique ports scanned.'; return [ body : body ];};};};",
 "params": {
 "threshold": 50
 }
 }
 },
 "email": {
 "profile": "standard",
 "attach_data": true,
 "priority": "high",
 "to": [
 "info@elastic.co"
 ],
 "subject": "[Security Alert] - Port scan detected",
 "body": "{{ctx.payload.body}}"
 }
 }
 }
</pre>What we do here is scanning again through the results to pick the attacker and target hosts, plus the count of how many unique ports were scanned.
The resulting watch then becomes:
<pre class="prettyprint">PUT _watcher/watch/port_scan_watch
{
 "trigger": {
 "schedule": {
 "interval": "10s"
 }
 },
 "input": {
 "search": {
 "request": {
 "search_type": "query_then_fetch",
 "indices": [
 "logstash-tcpdump-*"
 ],
 "types": [
 "tcpdump"
 ],
 "body": {
 "size": 0,
 "query": {
 "bool": {
 "must": [
 {
 "match": {
 "tags": "tcp_connection_started"
 }
 },
 {
 "range": {
 "@timestamp": {
 "gte": "now-30s"
 }
 }
 }
 ]
 }
 },
 "aggs": {
 "by_src_ip": {
 "terms": {
 "field": "src_ip"
 },
 "aggs": {
 "by_target_ip": {
 "terms": {
 "field": "dst_ip",
 "order": {
 "unique_port_count": "desc"
 }
 },
 "aggs": {
 "unique_port_count": {
 "cardinality": {
 "field": "dst_port"
 }
 }
 }
 }
 }
 }
 }
 }
 }
 }
 },
 "condition": {
 "script": {
 "inline": "for (int i = 0; i &lt; ctx.payload.aggregations.by_src_ip.buckets.size(); i++) {for (int j = 0; j &lt; ctx.payload.aggregations.by_src_ip.buckets[i].by_target_ip.buckets.size(); j++) {if (ctx.payload.aggregations.by_src_ip.buckets[i].by_target_ip.buckets[j].unique_port_count.value &gt; threshold) return true;};};return false;",
 "params": {
 "threshold": 50
 }
 }
 },
 "throttle_period": "30s",
 "actions": {
 "email_administrator": {
 "transform": {
 "script": {
 "inline": "def target='';def attacker='';def body='';for (int i = 0; i &lt; ctx.payload.aggregations.by_src_ip.buckets.size(); i++) {for (int j = 0; j &lt; ctx.payload.aggregations.by_src_ip.buckets[i].by_target_ip.buckets.size(); j++) {if (ctx.payload.aggregations.by_src_ip.buckets[i].by_target_ip.buckets[j].unique_port_count.value &gt; threshold) {target=ctx.payload.aggregations.by_src_ip.buckets[i].by_target_ip.buckets[j].key;attacker=ctx.payload.aggregations.by_src_ip.buckets[i].key;body='Detected portscan from ['+attacker+'] to ['+target+']. '+ctx.payload.aggregations.by_src_ip.buckets[i].by_target_ip.buckets[j].unique_port_count.value+ ' unique ports scanned.'; return [ body : body ];};};};",
 "params": {
 "threshold": 50
 }
 }
 },
 "email": {
 "profile": "standard",
 "attach_data": true,
 "priority": "high",
 "to": [
 "antonio@elastic.co"
 ],
 "subject": "[Security Alert] - Port scan detected",
 "body": "{{ctx.payload.body}}"
 }
 }
 }
}
</pre><h2>Testing our setup: you got mail!</h2>Now on to seeing some action, let's login to a host that has connectivity towards our monitored host (in this example 192.168.1.105) and launch a port scan against it:
<pre class="prettyprint">Elastic-MacBook-Air:~ user$ nmap 192.168.1.105 -p1-500
Starting Nmap 6.47 ( http://nmap.org ) at 2016-02-09 15:38 CET
Nmap scan report for w530 (192.168.1.105)
Host is up (0.0078s latency).
Not shown: 495 closed ports
PORT STATE SERVICE
22/tcp open ssh
80/tcp open http
139/tcp open netbios-ssn
389/tcp open ldap
445/tcp open microsoft-ds
</pre>Explicitly looking to probe privileged ports from 1 to 500. 
A few seconds later, we receive an email:
<img src="https://api.contentstack.io/v2/assets/575e4d303dc542cb38c09e6d/download?uid=bltb85771d49534ed62?uid=bltb85771d49534ed62" data-sys-asset-uid="blt6bd7ff3ddb5d2b40" alt="security-alert-port-scan-detected-email.png" style="“max-width:100%;&quot;">
Et voila! The alert was triggered and intended watch action was performed.
Note that we could have multiple detections from different hosts, however for the purpose of this blog post we limit ourselves to detecting and reporting only the first one in the list.
As a side node, if you like NMap, take a look at this <a href="
https://www.elastic.co/blog/using-nmap-logstash-to-gain-insight-into-your-network">blog post</a> to see all the awesome things you can do using <a href="https://www.elastic.co/guide/en/logstash/2.2/plugins-codecs-nmap.html">logstash-codec-nmap</a>.
This is just an example of how to leverage the Elastic stack for performing security monitoring, creativity is the only limit.
Happy alerting!

fingers-typing-on-kyeboard-dark.jpg

Elasticsearch and SIEM: implementing host portscan detection

Find in this bug hunt post-mortem how VMMap saved the day by uncovering missing memory allocations in Logstash on Windows

blog-banner-generic-teal.png

logstash-jruby-windows-vmmap-memory-leak.png

Post-mortem: Memory Leak Hunting on Windows

<html><head></head><body>We&apos;re approaching that time of the year when a day will actually be longer than a normal day, by just one second.That is what the scientific community refers to as the &#x201C;Leap Second.&#x201D;A Leap Second is a one-second adjustment that is occasionally applied to Coordinated Universal Time (UTC) in order to keep its time of day close to the mean solar time, or UT1. The next scheduled occurrence of leap second will be on June 30, 2015 23:59:60 UTC.While the leap second is certainly an interesting subject for discussions among scientists, it has proven to be a big headache for an IT infrastructure.The most recent occurrence of the leap second in 2012 disrupted operations for many major companies, bringing down services by causing high cpu usage on production servers.How does that affect my Elasticsearch deployment? The main problem actually sits in Linux kernel layer. Most Linux vendors have published guides explaining what you need to do (if anything) to test and prepare for the upcoming leap second:<ul><li dir="ltr"><a href="https://access.redhat.com/articles/15145">RHEL/CentOS</a></li><li dir="ltr"><a href="https://bugs.launchpad.net/ubuntu/+source/linux/+bug/1020285">Ubuntu</a></li><li dir="ltr"><a href="https://www.suse.com/support/kb/doc.php?id=7016150">SLES/OpenSUSE</a></li></ul>Please refer to your distribution&apos;s support channels if it is not listed above and for additional details. Elastic recommends you follow your distribution&apos;s recommended guidelines for handling this issue in order to avoid any further complications that may arise from unofficial or ad-hoc fixes.Additionally, some of the major cloud providers have also published their own information which you should read if you are running on one of their platforms:<ul><li dir="ltr"><a href="https://aws.amazon.com/blogs/aws/look-before-you-leap-the-coming-leap-second-and-aws/">Amazon</a></li><li dir="ltr"><a href="http://googlecloudplatform.blogspot.com.au/2015/05/Got-a-second-A-leap-second-that-is-Be-ready-for-June-30th.html">Google Cloud Platform</a></li></ul> </body></html>

Elasticsearch and the 2015 Leap Second

<html><head></head><body>
	Among the countless possible use cases where ELK can help save the day, displaying security-relevant data is certainly a very interesting one.&#xA0;In this blog post, using a virtual machine sitting on the cloud, we&apos;re going to show how to quickly set up a clustered instance of Elasticsearch to visualise firewall and honeypot datasources, namely IPtables and KippoSSH, focusing on the ELK-relevant configuration bits.


	KippoSSH is a medium interaction honeypot capable of recording plenty of information about the attacker, including interactive TTY sessions recordings; for the purpose of this blog post, we&apos;ll leave that latter piece of info aside, and focus on making sense of some brute force data.
Starting from the live raw data, we have logs containing:

<ul>
	<li>Denied TCP and UDP connection;</li>
	<li>Brute force SSH login attacks;</li>
</ul>

	These look like:


	#iptables

<pre class="prettyprint">elk@debianVM:~$ tail -3 /var/log/kern.log
Sep 24 13:52:06 debianVM kernel: iptables denied: IN=eth0 OUT= MAC=f2:3c:91:73:6c:71:84:78:ac:0d:8f:41:08:00 SRC=184.95.32.162 DST=85.159.211.137 LEN=48 TOS=0x00 PREC=0x00 TTL=110 ID=6886 PROTO=TCP SPT=37377 DPT=3306
Sep 24 13:52:44 debianVM kernel: iptables denied: IN=eth0 OUT= MAC=f2:3c:91:73:6c:71:84:78:ac:0d:8f:41:08:00 SRC=122.143.247.8 DST=85.159.211.137 LEN=40 TOS=0x00 PREC=0x00 TTL=107 ID=54661 PROTO=TCP SPT=6000 DPT=1433
Sep 24 13:55:22 debianVM kernel: iptables denied: IN=eth0 OUT= MAC=f2:3c:91:73:6c:71:84:78:ac:0d:8f:41:08:00 SRC=50.63.43.162 DST=85.159.211.137 LEN=44 TOS=0x00 PREC=0x00 TTL=57 ID=55765 PROTO=TCP SPT=80 DPT=5568
</pre>

	#kippo SSH

<pre class="prettyprint">elk@debianVM:~$ egrep &apos;New connection&apos; -A10 /opt/kippo/kippo-master/log/kippo.log | tail -11
2014-09-24 15:00:51+0100 [kippo.core.honeypot.HoneyPotSSHFactory] New connection: 221.146.74.146:49609 (85.159.211.137:2222) [session: 660]
2014-09-24 15:00:51+0100 [HoneyPotTransport,660,221.146.74.146] Remote SSH version: SSH-2.0-libssh2_1.4.1
2014-09-24 15:00:51+0100 [HoneyPotTransport,660,221.146.74.146] kex alg, key alg: diffie-hellman-group1-sha1 ssh-rsa
2014-09-24 15:00:51+0100 [HoneyPotTransport,660,221.146.74.146] outgoing: aes128-ctr hmac-sha1 none
2014-09-24 15:00:51+0100 [HoneyPotTransport,660,221.146.74.146] incoming: aes128-ctr hmac-sha1 none
2014-09-24 15:00:52+0100 [HoneyPotTransport,660,221.146.74.146] NEW KEYS
2014-09-24 15:00:52+0100 [HoneyPotTransport,660,221.146.74.146] starting service ssh-userauth
2014-09-24 15:00:53+0100 [SSHService ssh-userauth on HoneyPotTransport,660,221.146.74.146] root trying auth password
2014-09-24 15:00:53+0100 [SSHService ssh-userauth on HoneyPotTransport,660,221.146.74.146] login attempt [root/123456] failed
2014-09-24 15:00:54+0100 [-] root failed auth password
2014-09-24 15:00:54+0100 [-] unauthorized login:
</pre>

	Lots of interesting information that can be extracted from the above data.

<h2>Event collection and processing: Logstash</h2>

	Logstash is an awesome piece of software and the first layer of the ELK stack, where the journey of an event begins.


	Starting from the raw data, we want to be able to:

<ol>
	<li>identify relevant events by matching chosen patterns;</li>
	<li>extract relevant tokens to leverage them in our searches and dashboards later;</li>
	<li>send them to our chosen destination (elasticsearch cluster in this case) to perform analytics and investigations;</li>
</ol>

	So we want to extract fields of interest like source IP, destination port, target usernames and passwords to name the obvious ones.


	For sake of brevity, we will go specifically after only two log entries here, one from IPTables:

<pre class="prettyprint">Sep 24 15:42:03 debianVM kernel: iptables denied: IN=eth0 OUT= MAC=f2:3c:91:73:6c:71:84:78:ac:0d:8f:41:08:00 SRC=184.95.32.162 DST=85.159.211.137 LEN=48 TOS=0x00 PREC=0x00 TTL=110 ID=6886 PROTO=TCP SPT=37377 DPT=3306
</pre>

	and one from KippoSSH:

<pre class="prettyprint">2014-09-24 15:00:53+0100 [SSHService ssh-userauth on HoneyPotTransport,660,221.146.74.146] login attempt [root/123456] failed
</pre>

	We&apos;ll need to tell Logstash where the logs are located and what to do with them, below is our logstash config file ($LOGSTASH_HOME/config/logstash.conf). Notice the three different sections <a title="Input,Filter,Output" href="http://logstash.net/docs/1.4.2/life-of-an-event" target="_blank">Input, Filter, Output</a>:

<pre class="prettyprint">#input section, what data do we want to collect 
input { 
 file { 
 type =&gt; &quot;linux-syslog&quot; 
 path =&gt; &quot;/var/log/kern.log&quot; 
 } 
 file { 
 type =&gt; &quot;honey-kippo&quot; 
 path =&gt; &quot;/opt/kippo/kippo-master/log/kippo.log&quot; 
 } 
} 
#filter section, what to do with the data, process it , enrich it... 
filter { 
 if [type] == &quot;linux-syslog&quot; {
 grok { 
 #for linux-syslog type events use Grok definition below
 #to match the messages and extract fields of interest 
 match =&gt; [ &quot;message&quot;, &quot;%{IPTABLES_DENIED}&quot;] 
 } 
 date { 
 #use the field timestamp to match event time and 
 #populate @timestamp field (used by Elasticsearch) 
 match =&gt; [ &quot;timestamp&quot;, &quot;MMM dd HH:mm:ss&quot;] 
 timezone =&gt; &quot;Europe/London&quot; 
 } 
 } else if [type] == &quot;honey-kippo&quot; { 
 grok { 
 #like above, but using three separate 
 #Grok definitions (see Grok definitions later) 
 match =&gt; [ &quot;message&quot;, &quot;%{KIPPO_TIMESTAMP}\+\d+\s%{KIPPO_BODY_SSHSERVICE}\s%{KIPPO_MSG_LOGIN}&quot; ] 
 } 
 date { 
 match =&gt; [ &quot;timestamp&quot;, &quot;YYYY-MM-dd HH:mm:ss&quot; ] 
 timezone =&gt; &quot;Europe/London&quot; } 
 }
 } 
 geoip { 
 #enrich both event types with geo fields based on the 
 #src_ip field for analytics and drawing pretty maps 
 source =&gt; &quot;src_ip&quot; 
 } 
}
#output section, where do we send the data 
output { 
 #events failing to match Grok definitions will be 
 #automatically tagged with &apos;_grokparsefailure&apos; 
 #in this case we want to send only events where 
 #field extraction will be happening correctly 
 if &quot;_grokparsefailure&quot; not in [tags] { 
 if [type] == &quot;linux-syslog&quot; { 
 elasticsearch { 
 embedded =&gt; false 
 cluster =&gt; &quot;joinus&quot; 
 host =&gt; &quot;127.0.0.1&quot; 
 bind_host =&gt; &quot;127.0.0.1&quot; 
 index =&gt; &quot;logstash-os&quot; 
 index_type =&gt; &quot;linux-syslog&quot; 
 } 
 } else if [type] == &quot;honey-kippo&quot; { 
 elasticsearch { 
 embedded =&gt; false 
 cluster =&gt; &quot;joinus&quot; 
 host =&gt; &quot;127.0.0.1&quot; 
 bind_host =&gt; &quot;127.0.0.1&quot; 
 index =&gt; &quot;logstash-honey&quot; 
 index_type =&gt; &quot;kippo&quot; 
 }
 } 
 } else { 
 #let&apos;s print to logstash standard output 
 #events not captured by our Grok definitions 
 stdout { 
 codec =&gt; rubydebug 
 } 
 } 
}
</pre>

	Notice in the above output section that the index names both start with logstash-*.


	This is in order to leverage logstash-* <a title="index template mappings" href="https://github.com/elasticsearch/logstash/blob/v1.3.1/lib/logstash/outputs/elasticsearch/elasticsearch-template.json" target="_blank">index template mapping</a>, which will allow us to make use of both an &apos;analyzed&apos; and &apos;<a title="not_analyzed" href="/guide/en/elasticsearch/guide/current/mapping-intro.html" target="_blank">not_analyzed</a>&apos; versions of the fields, in order to be able to correctly draw our dashboards.See more on this aspect <a href="/blog/logstash-1-3-1-released/" title="here" target="_blank">here</a>


	Now let&apos;s take a closer look at the Grok definitions we will be using.


	We can get our raw data recognised and parsed by adding the 4 lines below to a pattern file in $LOGSTASH_HOME/patterns:

<pre class="prettyprint">IPTABLES_DENIED %{SYSLOGTIMESTAMP:timestamp} %{HOSTNAME:_host} kernel: iptables denied: IN=(?&lt;in&gt;eth0) OUT= MAC=(?&lt;mac_addr&gt;\\S+) SRC=%{IP:src_ip} DST=%{IP:dst_ip} LEN=\\d+ TOS=0x\\d+ PREC=0x\\d+ TTL=\\d+ ID=\\d+(?:\\sDF)? PROTO=(?&lt;proto&gt;\\S+) SPT=(?&lt;src_port&gt;\\d+) DPT=(?&lt;dst_port&gt;\\d+)(?:\\sWINDOW=\\d+)?(?:\\sRES=0x\\d+)?(?:\\s[ACKSYNFIRT]{3})+(?:\\sURGP=\\d)?
KIPPO_TIMESTAMP (?&lt;timestamp&gt;\\d{4}-\\d{2}-\\d{2} \\d{2}:\\d{2}:\\d{2})
KIPPO_BODY_SSHSERVICE \\[SSHService ssh-userauth on HoneyPotTransport,\\d+,%{IP:src_ip&gt;}\\]
KIPPO_MSG_LOGIN login\\sattempt\\s\\[(?&lt;src_user&gt;\\S+)/(?&lt;src_pwd&gt;\\S+)\\]\\s(?&lt;outcome&gt;succeeded|failed)
</pre>

	Here we are just telling Grok to be aware of these new definitions that we have referenced in Logstash main config file earlier.&#xA0;Grok will match the event types we have configured and use the regular expression above to extract our fields of interest.


	You can see that some definitions are based on already existing patterns - for e.g. IPTABLES_DENIED reuses Grok patterns SYSLOGTIMESTAMP, HOSTNAME and IP. This is one of the key strengths of Grok, aimed at making a better use of your time, other than writing and re-writing regex after regex.


	Using good online regex tools can help speed up on this task. <a title="Grok Debugger" href="https://grokdebug.herokuapp.com/" target="_blank">Grok Debugger</a> is a great tool you can rely on to construct and debug Grok patterns. Some other resources worth mentioning for pure regex testing are <a title="regex101.com" href="http://regex101.com/" target="_blank">regex101.com</a>, <a title="regextester.com" href="http://regextester.com/" target="_blank">regextester.com</a>.


	See Grok Debugger in action below:


	<img src="https://api.contentstack.io/v2/assets/575e4d970342dfd738265fbd/download?uid=bltfabd4db12b6c7810?uid=bltfabd4db12b6c7810" style="max-width:100%;">


	<img src="https://api.contentstack.io/v2/assets/575e4d97d8edd48f76938b40/download?uid=blt84966b845c24ee06?uid=blt84966b845c24ee06" style="max-width:100%;">


	<img src="https://api.contentstack.io/v2/assets/575e4d9758208ba076e2ade5/download?uid=blte297155d9831386e?uid=blte297155d9831386e" style="max-width:100%;">


	<img src="https://api.contentstack.io/v2/assets/575e4d9743e9adc53871679a/download?uid=bltc22f8ac8a5499355?uid=bltc22f8ac8a5499355" style="max-width:100%;">


	Now we&apos;re good to index as much data as we like (and our hardware can handle), so we will move to the next layer on the ELK stack.

<h2>Store the data and make it searchable: Elasticsearch</h2>

	Setting up an Elasticsearch cluster is straightforward. In this demo we setup 2 elasticsearch nodes on a single host. You could also do this in production, if you have enough capacity on a single machine in some specific scenarios. See <a href="/guide/en/elasticsearch/guide/current/hardware.html" target="_blank">here</a> for more details.


	We&#xA0;could leave default settings and get started by:

<ol>
	<li>extract the elasticsearch tar archive</li>
	<li>launch each instance as a daemon</li>
</ol>

	and these instance would just talk to each other like good old friends using multicast zen discovery and form a cluster with no configuration needed!


	As we&apos;d like to have a bit more control over the cluster behaviour we proceed to amend some defaults for each of the 2 nodes.For example we set node 1 to use:

<pre class="prettyprint">#Set a name for each node, for readability and manageability
node.name: &quot;node-1&quot;
#Disable multicast discovery, you never know what happens in a network you don&apos;t own
discovery.zen.ping.multicast.enabled: false
#Configure an initial list of master nodes in the cluster, we know who we are
discovery.zen.ping.unicast.hosts: [&quot;localhost:9301&quot;]
#Set a cluster name, elasticsearch is a nice name though, we like here to set our own
cluster.name: joinus
</pre>

	And that is it!


	Cluster &apos;joinus&apos; is ready to accept data and make it searchable, resilient and all the magic Elasticsearch will do for us.


	Let&apos;s just say &quot;hi,how are you?&quot; to our cluster

<pre class="prettyprint">elk@debianVM:~$ curl -XGET &apos;localhost:9200/_cluster/health?pretty&apos;
{
&quot;cluster_name&quot; : &quot;joinus&quot;,
&quot;status&quot; : &quot;green&quot;,
&quot;timed_out&quot; : false,
&quot;number_of_nodes&quot; : 2,
&quot;number_of_data_nodes&quot; : 2,
&quot;active_primary_shards&quot; : 100,
&quot;active_shards&quot; : 200,
&quot;relocating_shards&quot; : 0,
&quot;initializing_shards&quot; : 0,
&quot;unassigned_shards&quot; : 0
}
</pre>

	You know, looks good!


	We can now create two indexes in our Elasticsearch cluster by issuing:

<pre class="prettyprint">$ curl -XPUT &apos;localhost:9200/logstash-os?pretty&apos;
</pre>

	and

<pre class="prettyprint">$ curl -XPUT &apos;localhost:9200/logstash-honey?pretty&apos;
</pre>

	For each of these requests, we will receive the below answer, saying index creation was successful:

<pre class="prettyprint">{
&quot;acknowledged&quot; : true
}
</pre>

	We could also tell Elasticsearch explicitly how to interpret our fields (is this an integer or a date or a string), but for this simple demo we&apos;re happy to let Elasticsearch automatically determine the field types.

<h2>Open my eyes, show me what you&apos;ve got: Kibana</h2>

	While we impatiently await <a title="Kibana 4" href="/blog/kibana-4-beta-1-1-pointy-needles-blunted/" target="_blank">Kibana 4</a> to go GA let&apos;s use Kibana 3 to plot some data.


	We have now left this running for a while to allow the bad guys to feed us with some events and make our dashboards nice and pretty.


	Once Kibana is <a title="installed" href="/downloads/kibana">installed</a> we are greeted with


	<img src="https://api.contentstack.io/v2/assets/575e4d979e7a83165490ed43/download?uid=blt05f10f79e01de52d?uid=blt05f10f79e01de52d" style="max-width:100%;">


	If we go and have a look at the sample dashboard, we see


	<img src="https://api.contentstack.io/v2/assets/575e4d989e7a83165490ed65/download?uid=bltc6bf501fde5a5eb0?uid=bltc6bf501fde5a5eb0" style="max-width:100%;">


	Notice our document types are showing up in &apos;Document Types&apos; panel. Contextually on the left handside we also have a list of available searchable field:


	<img src="https://api.contentstack.io/v2/assets/575e4d973dc542cb38c0b06d/download?uid=blt5e39b8dec932c99a?uid=blt5e39b8dec932c99a">


	Now that we have validated the data looks good, we can go ahead and start building our first dashboard from scratch!


	Let&apos;s go one step back, and choose option 3, &apos;Blank Dashboard&apos;, and let&apos;s set a title for our new dashboard


	<img src="https://api.contentstack.io/v2/assets/575e4d9858208ba076e2ae03/download?uid=blt1a2408e0ce6abc61?uid=blt1a2408e0ce6abc61" style="max-width:100%;">


	And point it towards the index &apos;logstash-os&apos; where we are storing IPTables events (by default &apos;_all&apos; indexes are set to be queried)


	<img src="https://api.contentstack.io/v2/assets/575e4d9843e9adc5387167bb/download?uid=blt743e0e9b099ed190?uid=blt743e0e9b099ed190" style="max-width:100%;">


	Then let&apos;s add a row to our empty &apos;Denied Connections&apos; dashboard.


	<img src="https://api.contentstack.io/v2/assets/575e4d9843e9adc5387167c9/download?uid=blt7e6e46e641ed8040?uid=blt7e6e46e641ed8040" style="max-width:100%;">


	Finally let&apos;s add our first panel and choose panel type &apos;Terms&apos; to show some nice aggregations


	<img src="https://api.contentstack.io/v2/assets/575e4d980342dfd738265ff4/download?uid=blt7565765237eb8555?uid=blt7565765237eb8555" style="max-width:100%;">


	Notice that as you type your first char in Field form, Kibana will show you all the possible field matching the string as you type it. We have also here available fields named with extension .raw , this is because we are leveraging logstash index mapping template for having &apos;not_analyzed&apos; fields.


	<img src="https://api.contentstack.io/v2/assets/575e4d99d8edd48f76938b8b/download?uid=bltd7c1343cdd6803fa?uid=bltd7c1343cdd6803fa" style="max-width:100%;">


	We choose now &apos;src_ip&apos; for this panel and let&apos;s set up some more options, we want to see the top 20 IPs, sorted by count, using bars as the visualization format.


	<img src="https://api.contentstack.io/v2/assets/575e4d999e7a83165490ed81/download?uid=blta4371bc96a48a4c9?uid=blta4371bc96a48a4c9" style="max-width:100%;">


	Just hit &apos;Save&apos; Et voil&#xE0;&#xA0;!


	<img src="https://api.contentstack.io/v2/assets/575e4d999985d5897637970f/download?uid=blt9650ca011cfe0d2a?uid=blt9650ca011cfe0d2a" style="max-width:100%;">


	You know, Kibana rocks! 
	Reiterate the above and have fun!


	We have now plenty of data to leverage, manipulate and visualise from many different angles.


	We can ask all the questions we want, just add a few queries to slice your data as you please, each panel can display all or a selection of the queries.


	<img src="https://api.contentstack.io/v2/assets/575e4d993dc542cb38c0b0cf/download?uid=blt9512e1d025d62bdd?uid=blt9512e1d025d62bdd" style="max-width:100%;">


	Then we can plot beautiful charts giving us lots of insight on the bad guys trying to access our host:


	<img src="https://api.contentstack.io/v2/assets/575e4d99d8edd48f76938b96/download?uid=blt98a5c7d9623f8e2a?uid=blt98a5c7d9623f8e2a" style="max-width:100%;">


	<img src="https://api.contentstack.io/v2/assets/575e4d9958208ba076e2ae4f/download?uid=blte15b15c39a24a2ec?uid=blte15b15c39a24a2ec" style="max-width:100%;">


	<img src="https://api.contentstack.io/v2/assets/575e4d999e7a83165490ed98/download?uid=bltdf231a27076b0d1e?uid=bltdf231a27076b0d1e" style="max-width:100%;">


	Likewise, for our KippoSSH honeypot data


	<img src="https://api.contentstack.io/v2/assets/575e4d9912798d8e38200311/download?uid=blta9f5df0a7e2fe2d7?uid=blta9f5df0a7e2fe2d7" style="max-width:100%;">


	<img src="https://api.contentstack.io/v2/assets/575e4d9a9985d5897637973d/download?uid=bltf87b044ff735fdef?uid=bltf87b044ff735fdef" style="max-width:100%;">


	You can click on any dashboard, drilldown and start your investigations: just ask the questions, ELK will give you the answers - easy peasy!&quot;
</body></html>

Use ELK to Visualise Security Data: IPTables and KippoSSH Honeypot

Start free trial

Blog Footer CTA

Sign up for Elastic Cloud free trial

Software Engineer

Antonio Bonuccelli

By submitting you acknowledge that you've read and agree to our <a href="/legal/elastic-cloud-account-terms" target="_self">Terms of Service</a>, and you agree to receive the <a href="/legal/privacy-statement#how-we-use-the-information" target="_self">selected communications</a> at the contact details you provided above. See <a href="/legal/privacy-statement/" target="_self">Elastic’s Privacy Statement</a> for more details or to opt-out at any time.

Communication Preferences - ONLY for /communication-preferences

提交即表示您确认您已阅读并同意我们的<a href="/cn/legal/elastic-cloud-account-terms" target="_self">服务条款</a>，并且您同意通过您提供的上述联系方式接收<a href="/cn/legal/privacy-statement#how-we-use-the-information" target="_self">所选的通信</a>。如需了解更多信息，或想随时取消订阅，请参阅 <a href="/cn/legal/privacy-statement/" target="_self">Elastic 的隐私声明</a>。

By submitting you acknowledge that you've read and agree to our <a href="/legal/serverless-preview-terms" target="_blank">Terms of Service</a>, and that Elastic may <a href="/legal/privacy-statement#how-we-use-the-information" target="_blank">contact you</a> about our related products and services, using the details you provide above. See <a href="/legal/privacy-statement/" target="_blank">Elastic’s Privacy Statement</a> for more details or to opt-out at any time.

Serverless GDPR Text

提交即表示您确认您已阅读并同意我们的<a href="/cn/legal/serverless-preview-terms" target="_self">服务条款</a>，并且 Elastic 可以使用您提供的上述详细信息与您联系，以向您介绍我们相关的<a href="/cn/legal/privacy-statement#how-we-use-the-information" target="_self">产品和服务</a>。如需了解更多信息，或想随时取消订阅，请参阅 <a href="/cn/legal/privacy-statement/" target="_self">Elastic 的隐私声明</a>。

By submitting you agree to <a href="/agreements/global/cloud-services-monthly" rel="nofollow">Elastic Cloud Terms of Service</a>. Your personal data will be processed in accordance with <a href="/legal/privacy-statement">Elastic's Privacy Statement</a>.

Cloud Trial GDPR Text

提交即表示您同意 <a href="/cn/agreements/global/cloud-services-monthly" rel="nofollow">Elastic Cloud 服务条款</a>。我们将会按照 <a href="/cn/legal/privacy-statement">Elastic 的隐私声明</a>处理您的个人数据。

By submitting you acknowledge that you've read and agree to our <a href="/legal/terms-of-use" target="_blank">Terms of Service</a>, and that Elastic may <a href="/legal/privacy-statement#how-we-use-the-information" target="_blank">contact you</a> about our related products and services, using the details you provide above. See <a href="/legal/privacy-statement/" target="_blank">Elastic’s Privacy Statement</a> for more details or to opt-out at any time.

Newsletter GDPR Text

提交即表示您确认您已阅读并同意我们的<a href="/cn/legal/terms-of-use" target="_self">服务条款</a>，并且 Elastic 可以使用您提供的上述详细信息与您联系，以向您介绍我们相关的<a href="/cn/legal/privacy-statement#how-we-use-the-information" target="_self">产品和服务</a>。如需了解更多信息，或想随时取消订阅，请参阅 <a href="/cn/legal/privacy-statement/" target="_self">Elastic 的隐私声明</a>。

Explore similar demos

Overview

Speakers

Close

See more insights

The content on this page is not available in the selected language. As Elastic grows globally, we continue to support content in multiple languages.

The content on this page is not available in the selected language.

本页内容尚不支持所选语言。Elastic 正在不断努力，以实现对多种语言内容的支持。感谢您在此期间给予的耐心与陪伴！

Author

Articles by

Learn more

Watch now (no PT)

Watch now

See all top stories

All (no PT translation)

Contact information

Press Release

Share on Reddit

Share this story

Share by Email

Thank you for your interest!

Contact Info

Follow us on Youtube

Load More

Share on LinkedIn

Follow us on LinkedIn

Search Integrations

All Solutions

Video for

Follow us on Twitter

See when this webinar starts in my time zone

查看本次网络研讨会在我的时区的开始时间

Register to Watch

Register now

See All Posts

Can't make it? Register and we'll send you the recording. You'll also receive an email with related content

无法实时参加？注册后我们会给您发送录像。您还将收到相关内容的电子邮件。

Author

Articles by Antonio Bonuccelli

Software Engineer, Elastic

Elasticsearch and SIEM: implementing host portscan detection

Post-mortem: Memory Leak Hunting on Windows

Elasticsearch and the 2015 Leap Second

Use ELK to Visualise Security Data: IPTables and KippoSSH Honeypot