VyOS를 운용해 가는 가운데,/var/log/message를 자주 확인하면, SSH 브루트포스를 피해를 받고 있는 것을 알았다. 따라서, 이번은 대상의 VyOS로부터 Syslog를 감시 서버(Fluent, Elasticsearch, Kibana)에 송신해, 서버상에서 가시화하는 것을 목표로 하고 싶다.

덧붙여 지난번 의 GeoIP를 이용하므로, 그쪽의 구축이 아직의 사람은 기사를 참조해 주세요.

다음 사이트를 참고로 작성했습니다. (감사합니다.)

fluentd와 Kibana의 SSH 액세스 소스 맵

소개

이번에는 VyOS의 아래와 같은 syslog는 parse하고 상세하게 시각화하고, 다른 syslog에 관해서는 표시하는 것처럼 느껴집니다.

Aug 27 14:52:15 vyos sshd[20938]: pam_unix(sshd:auth): authentication failure; logname= uid=0 euid=0 tty=ssh ruser= rhost=xxx.xxx.xxx.xxx  user=???
Aug 27 14:52:24 vyos sshd[20938]: PAM 2 more authentication failures; logname= uid=0 euid=0 tty=ssh ruser= rhost=xxx.xxx.xxx.xxx  user=???

Syslog 모니터링 서버 구축

소개 이번에 새롭게 사용할 Fluend 플러그인을 설치합니다.
그건 그렇고, syslog를받는 플러그인은 기본적으로 들어있었습니다.

받은 syslog를 구문 분석하기 위한 플러그인인 fluent-plugin-parser 설치

$ sudo /opt/td-agent/embedded/bin/fluent-gem install fluent-plugin-parser

수신 된 syslog를 다중 분석 (일반적으로 하나)하는 플러그인 인 fluent-plugin-multi-format-parser 설치

$ sudo /opt/td-agent/embedded/bin/fluent-gem install fluent-plugin-multi-format-parser

Fluentd conf 파일에 Syslog 수신 및 parser (분석) 설정 작성

$ sudo vim /etc/td-agent/td-agent.conf
--省略--

<source>
  type syslog
  port 42185
  format none
  tag raw.syslog.event
</source>

<match raw.syslog.**>
  @type parser
  format multi_format
  remove_prefix raw
  add_prefix geo
  <pattern>
    format /^(?<time>[^ ]* [^ ]* [^ ]*) (?<host>[^ ]*) (?<ident>[a-zA-Z0-9_\/\.\-]*)(?:\[(?<pid>[0-9]+)\]): (?<message>.*); logname=(?<logname>[^ ]*) uid=(?<uid>[^ ]*) euid=(?<euid>[^ ]*) tty=(?<tty>[^ ]*) ruser=(?<ruser>[^ ]*) rhost=(?<rhost>[^ ]*)  user=(?<user>[^ ]*)$/
    time_format %b %d %H:%M:%S
  </pattern>
  <pattern>
    format syslog
  </pattern>
  #format syslog
  key_name message
</match>

<match geo.syslog.**>
  type geoip
  geoip_lookup_key rhost
  <record>
    geoip_country ${country_code['rhost']}
    geoip_city ${city['rhost']}
    geoip_lat ${latitude['rhost']}
    geoip_lon ${longitude['rhost']}
    geoip_pin ${latitude["rhost"]},${longitude["rhost"]}
  </record>
  remove_tag_prefix geo.
  add_tag_prefix es.
  skip_adding_null_record
</match>

<match es.syslog.**>
  type elasticsearch
  host localhost
  port 9200
  type_name syslog
  logstash_format true
  logstash_prefix syslog
  logstash_dateformat %Y%m%d
  flush_interval 10s
</match>

간단히 설명하면,
부분에서 syslog를 수신하는 설정(대기 포트 번호:42185)을 써.
부분적으로, 수신한 syslog로 ssh의 log에 매치하는 것(정규 표현을 이용)과 그 외의 log를 해석하고 있습니다.
덧붙여서 정규 표현의 디버그는 ぅ tp : // / ぅ 엔츠 ぁ r. 어리석은 p. 이 m 가 최강입니다.
부분적으로, ssh의 log에 있어서의 rhost(리모트 호스트의 IP 주소)를 geoid에 의한 해석을 해, 위치 정보를 해석하고 있습니다. 다른 syslog는 건너뜁니다.
부분적으로 이러한 데이터를 elasticsearch로 전송하는 느낌입니다.

Geoip (geoip_pin)을 사용하고 있으므로, 수신하는 syslog의 Index에 "type": "geo_point"의 포맷을 추가합시다.

$ curl -XPUT http://localhost:9200/_template/syslog -d '
{
  "template" : "syslog-*",
  "mappings" : {
    "syslog" : {
      "properties" : {
        "@timestamp" : {
          "type" : "date",
          "format" : "dateOptionalTime"
        },
        "geoip_pin" : {
          "type" : "geo_point"
        }
      }
   }
  }
}'

이상, fluend(td-agent)를 재기동한다

$ /etc/init.d/td-agent restart

제대로 작동하는지 확인하기 위해 확인합시다.

$ tailf /var/log/td-agent/td-agent.log

이것으로 서버측의 준비는 OK입니다만, syslog는 Index가 건간 모여 가므로 정기적으로 삭제하도록 해 둡시다.

이전 에 사용한 curator를 사용해 갑니다. 10일 전의 Index는 클로즈, 20일전의 Index는 삭제해 가는 설정은 이하입니다.

$ vim ~/.curator/close_delete_indices.yml

---
# Remember, leave a key empty if there is no value.  None will be a string,
# not a Python "NoneType"
#
# Also remember that all examples have 'disable_action' set to True.  If you
# want to use this action as a template, be sure to set this to False after
# copying it.
actions:
  1:
    action: delete_indices
    description: >-
      Delete indices older than 20 days (based on index name), for syslog-
      prefixed indices. Ignore the error if the filter does not result in an
      actionable list of indices (ignore_empty_list) and exit cleanly.
    options:
      ignore_empty_list: True
      disable_action: False
    filters:
    - filtertype: pattern
      kind: prefix
      value: syslog-
    - filtertype: age
      source: name
      direction: older
      timestring: '%Y%m%d'
      unit: days
      unit_count: 20
  2:
    action: close
    description: >-
      Close indices older than 10 days (based on index name), for syslog-
      prefixed indices.
    options:
      ignore_empty_list: True
      delete_aliases: False
      disable_action: False
    filters:
    - filtertype: pattern
      kind: prefix
      value: syslog-
    - filtertype: age
      source: name
      direction: older
      timestring: '%Y%m%d'
      unit: days
      unit_count: 10

crontab에 다음을 등록합시다.

0 12 * * 1 curator ~/.curator/close_delete_indices.yml

VyOS 설정

VyOS는 다음과 같은 약간의 설정입니다. facility,level의 지식은 각자 깊게 봐 주세요.

# 設定モードに入る
configure

# syslogサーバの指定、facility(送るlogの選択)、level(送るlogの重要度的な)
set system host 192.168.1.100:42185 facility all level notice

# 変更のコミットおよび設定保存
commit
save

동작 확인

색인이 제대로 작성되었는지 확인

$ curl -XGET 'localhost:9200/_cat/indices?v&pretty'

syslog Index가 제대로 구성되었는지 확인

$ curl "http://localhost:9200/syslog-yyyymmdd/_search?pretty"

감시 화면 만들기

먼저 syslog Index를 등록합니다.



이것으로 Discover를 열면, syslog의 감시(ssh의 log는 geoip등의 정보를 부가)한 것이 보인다고 생각합니다.



다음으로 Visualize를 만드는 것입니다.
시작하기 전에 데이터 테이블을 만듭니다.
보기 어렵지만, 다음과 같은 설정입니다.



작성하면 ssh 액세스했을 때 지정한 사용자, 소스 IP 주소, 소스 국가, 그 횟수를 표에 시각화할 수 있습니다.

히트 맵은 바삭바삭과 같은 느낌입니다.



이것으로 완료됩니다. 정규 표현을 이용하거나, parse의 방법을 공부하는 것으로, 다양한 로그 감시를 할 수 있을 것 같네요.