ini 使用ELK堆栈（Elasticsearch + Logstash + Kibana）进行脱机SSHD日志分析

Posted 2021-05-24

### 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/](http://localhost:5601/)
and start analyzing the data.

Here are few [sample screenshots](http://imgur.com/a/FiJ0K).

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

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 { }
}