Using ELK stack (Elasticsearch + Logstash + Kibana) for offline SSHD log analysis

00-elk-stack-sshd-log-analysis.md

Using ELK stack for offline SSHD log analysis

To start Elasticsearch + Logstash + Kibana execute:

docker-compose up

The container images will be downloaded from docker hub at first run.

Next, import the log file data to logstash

nc localhost 5000 < /var/log/auth.log

The logstash configuration file logstash-auth.conf contains match rules for parsing SSHD login records (both failed and successful) from syslog file into logstash events. Events will contain username, source IP address and geographical location for each login attempt.

Then connect to Kibana at http://localhost:5601/ and start analyzing the data.

Here are few sample screenshots.

docker-compose.yml

logstash:
  image: logstash:latest
  command: logstash -f /config/logstash-auth.conf
  ports:
    - "5000:5000"
  links:
    - elasticsearch
  volumes:
    - ./config:/config/


elasticsearch:
  image: elasticsearch:latest
  ports:
    - "9200:9200"


kibana:
  image: kibana:latest
  ports:
    - "5601:5601"
  links:
    - elasticsearch

logstash-auth.conf

input {
  tcp {
    	port => 5000
  }
}


filter {
  grok {
    add_tag => [ "valid" ]

    # Example log entries for both failed and successful logins:
    #
    # Aug  9 09:13:25 vmubu01 sshd[5761]: Failed password for root from 218.87.111.109 port 45712 ssh2
    # Aug  9 09:13:31 vmubu01 sshd[5761]: message repeated 2 times: [ Failed password for root from 218.87.111.109 port 45712 ssh2]
    # Aug 14 17:25:47 vmubu01 sshd[22101]: Failed password for invalid user test from 115.68.23.130 port 43092 ssh2
    # Aug 16 13:47:44 vmubu01 sshd[730]: Accepted publickey for username from 192.168.1.225 port 38783 ssh2: RSA 01:02:03:04:05:06:07:08:09:0a:0b:0c:0d:0e:0f:10
    # Aug 16 13:47:57 vmubu01 sshd[816]: Accepted password for username from 192.168.1.225 port 38786 ssh2
    match => [
      "message", "%{SYSLOGTIMESTAMP:syslog_date} %{SYSLOGHOST:syslog_host} %{DATA:syslog_program}(?:\[%{POSINT}\])?: %{WORD:login} password for %{USERNAME:username} from %{IP:ip} %{GREEDYDATA}",
      "message", "%{SYSLOGTIMESTAMP:syslog_date} %{SYSLOGHOST:syslog_host} %{DATA:syslog_program}(?:\[%{POSINT}\])?: message repeated 2 times: \[ %{WORD:login} password for %{USERNAME:username} from %{IP:ip} %{GREEDYDATA}",
      "message", "%{SYSLOGTIMESTAMP:syslog_date} %{SYSLOGHOST:syslog_host} %{DATA:syslog_program}(?:\[%{POSINT}\])?: %{WORD:login} password for invalid user %{USERNAME:username} from %{IP:ip} %{GREEDYDATA}",
      "message", "%{SYSLOGTIMESTAMP:syslog_date} %{SYSLOGHOST:syslog_host} %{DATA:syslog_program}(?:\[%{POSINT}\])?: %{WORD:login} %{WORD:auth_method} for %{USERNAME:username} from %{IP:ip} %{GREEDYDATA}"
    ]
  }

  if "valid" not in [tags] {
    drop { }
  }

  mutate {
    remove_tag => [ "valid" ]
    lowercase => [ "login" ]
  }
  

  date {
    match => [ "syslog_date", "MMM  d HH:mm:ss", "MMM dd HH:mm:ss", "ISO8601" ]
    timezone => "Europe/Helsinki"
  }

  geoip {
    source => "ip"
  }
}


output {
  elasticsearch { }
}