흙수저의 엔지니어링

logstash 로그 중

Received an event that has a different character encoding than you configured.

expected_charset => "UTF-8"

로그가 있다면

input 설정 내용에 아래의 charset을 추가하여 설정해주세요.

codec => plain { charset => "CP949" }

OS가 centos에서 ubuntu로 변경되어 ubuntu에서도 logstash 설정하는 내용을 남깁니다.

설명을 위해서 fortigate log를 파싱하기 위한 logstash 설정 입니다.

내용 확인 후 활용하시면 좋을 것 같아요.

logstash 설정 파일 생성하기

logmgmt@syslogserver:~$ sudo cat /etc/logstash/conf.d/logstash.conf
input {
        file {
                path => "/var/log/rsyslog/192.168.000.000/*.log"
                start_position => "beginning"
                tags => ["fortigate"]
        }
}

filter {
        if "fortigate" in [tags] {
                grok {
                        patterns_dir => ["/etc/logstash/pattern.d"]
                                match => { "message" => [ "%{FORTILOG} %{GREEDYDATA:sub_message}" ] }
                }
                kv {
                        source => "sub_message"
                        value_split => "="
                }
                mutate { remove_field => [ "message" ] }
                mutate { remove_field => [ "sub_message" ] }
                if "wan" in [srcintfrole] {
                        geoip {
                                source => "srcip"
                                target => "geoip_src"
                        }
                }
                if [sentbyte] != "" and [rcvdbyte] != "" {
                        bytes {
                                source => "rcvdbyte"
                                target => "receivedbyte"
                        }
                        bytes {
                                source => "sentbyte"
                                target => "sentedbyte"
                        }
                }
                mutate {
                        convert => {
                                "rcvdpkt" => "integer"
                                "sentpkt" => "integer"
                                "proto" => "integer"
                                "srcserver" => "integer"
                                "sessionid" => "integer"
                                "duration" => "integer"
                                "policyid" => "integer"
                                "HOUR" => "integer"
                                "MINUTE" => "integer"
                                "SECOND" => "integer"
                        }
                }
        }
}

output {
        if "fortigate" in [tags] {
                if "traffic" in [type] {
                        elasticsearch {
                                hosts => "http://192.168.000.000:9200"
                                index => "logstash-fortigate-traffic-index-%{+YYYY.MM.dd}"
                        }
                }
                else if "event" in [type] {
                        elasticsearch {
                                hosts => "http://192.168.000.000:9200"
                                index => "logstash-fortigate-event-index-%{+YYYY.MM.dd}"
                        }
                }
                else if "utm" in [type] {
                        elasticsearch {
                                hosts => "http://192.168.000.000:9200"
                                index => "logstash-fortigate-utm-index-%{+YYYY.MM.dd}"
                        }
                }
        }
}
logmgmt@syslogserver:~$

fortigate 패턴 파일 생성하기

logmgmt@syslogserver:~$ sudo cat /etc/logstash/pattern.d/fortigate.pattern
####################################
###Fortinet Syslog Pattern Types:###
####################################

FORTILOG (?<timestamp>^\w+\s+\d+\s+\d+\:\d+\:\d+)\s(?<log_send_dev>(?:%{IP}|\_gateway))
logmgmt@syslogserver:~$

logstash-filter-bytes 필터 설치하기

logmgmt@syslogserver:~$ sudo /usr/share/logstash/bin/logstash-plugin install logstash-filter-bytes
Using bundled JDK: /usr/share/logstash/jdk
Validating logstash-filter-bytes
Resolving mixin dependencies
Installing logstash-filter-bytes
Installation successful
logmgmt@syslogserver:~$

fortigate log 파싱 부분이 추가되었습니다.

fortigate 로그 파싱할때에는 logstash-filter-bytes를 사용하였습니다.

이유는 logstash에서 로그를 파싱하면 모든 자료는 기본적으로 type이 text입니다.

마찬가지로 숫자도 text로 인식합니다.

그래서 데이터 전송량도 type에 맞게 bytes라는 필터를 추가로 설치하였습니다.

logstash-filter-bytes의 설치는 아래의 작성 글을 참고하시기 바랍니다.

https://dirt-spoon.tistory.com/80

input {
        file {
                path => "/var/log/rsyslog/192.168.10.2/*.log"
                start_position => "beginning"
                tags => ["ap1"]
        }
        file {
                path => "/var/log/rsyslog/192.168.10.3/*.log"
                start_position => "beginning"
                tags => ["ap2"]
        }
        file {
                path => "/var/log/rsyslog/192.168.10.4/*.log"
                start_position => "beginning"
                tags => ["ap3"]
        }
        file {
                path => "/var/log/rsyslog/192.168.10.5/*.log"
                start_position => "beginning"
                tags => ["ap4"]
        }
        file {
                path => "/var/log/rsyslog/192.168.0.14/*.log"
                start_position => "beginning"
                tags => ["fortigate"]
        }
}

filter {
        if "ap1" in [tags] or "ap2" in [tags] or "ap3" in [tags] or "ap4" in [tags]  {
                grok {
                        patterns_dir => ["/etc/logstash/pattern.d"]
                        match => {
                                "message" => [
                                        "%{SYSLOGTIMESTAMP:access_time} %{IPORHOST:ip_or_host} %{IPORHOST:process}\[%{BASE10NUM:process_id}\]\: %{GREEDYDATA:sub_message}",
                                        "%{SYSLOGTIMESTAMP:access_time} %{IPORHOST:ip_or_host} \[%{BASE10NUM:process_id}\]\: %{GREEDYDATA:sub_message}",
                                        "%{SYSLOGTIMESTAMP:access_time} %{IPORHOST:ip_or_host} %{WORD:process}\: User\: %{GREEDYDATA:user} last logged %{GREEDYDATA:access_result} in %{GREEDYDATA:access_day}\#%{NUMBER:deauthentication_reason_code}, to %{IP:destination_ip}, from %{IP:source_ip} using %{WORD:access_method}",
                                        "%{SYSLOGTIMESTAMP:access_time} %{IPORHOST:ip_or_host} %{WORD:process}\: User\: %{GREEDYDATA:reason} in %{GREEDYDATA:access_day}\#%{NUMBER:deauthentication_reason_code}, to %{IP:destination_ip}, from %{IP:source_ip} using %{WORD:access_method}"
                                ]
                        }
                }
                mutate { remove_field => [ "message" ] }
        }
        else if "fortigate" in [tags] {
                grok {
                        patterns_dir => ["/etc/logstash/pattern.d"]
                        match => { "message" => [ "%{FORTILOG} %{GREEDYDATA:sub_message}" ] }
                }
                kv {
                        source => "sub_message"
                        value_split => "="
                }
                mutate { remove_field => [ "message" ] }
                mutate { remove_field => [ "sub_message" ] }
                if "wan" in [srcintfrole] {
                        geoip {
                                source => "srcip"
                                target => "geoip_src"
                        }
                }
                if [sentbyte] != "" and [rcvdbyte] != "" {
                        bytes {
                                source => "rcvdbyte"
                                target => "receivedbyte"
                        }
                        bytes {
                                source => "sentbyte"
                                target => "sentedbyte"
                        }
                }
                mutate {
                        convert => {
                                "rcvdpkt" => "integer"
                                "sentpkt" => "integer"
                                "proto" => "integer"
                                "srcserver" => "integer"
                                "sessionid" => "integer"
                                "duration" => "integer"
                                "policyid" => "integer"
                                "HOUR" => "integer"
                                "MINUTE" => "integer"
                                "SECOND" => "integer"
                        }
                }
        }
}

output {
        if "ap1" in [tags] {
                elasticsearch {
                        hosts => "http://192.168.0.17:9200"
                        index => "logstash-ap1-index-%{+YYYY.MM.dd}"
                }
        }
        else if "ap2" in [tags] {
                elasticsearch {
                        hosts => "http://192.168.0.17:9200"
                        index => "logstash-ap2-index-%{+YYYY.MM.dd}"
                }
        }
        else if "ap3" in [tags] {
                elasticsearch {
                        hosts => "http://192.168.0.17:9200"
                        index => "logstash-ap3-index-%{+YYYY.MM.dd}"
                }
        }
        else if "ap4" in [tags] {
                elasticsearch {
                        hosts => "http://192.168.0.17:9200"
                        index => "logstash-ap4-index-%{+YYYY.MM.dd}"
                }
        }
        else if "fortigate" in [tags] {
                if "traffic" in [LOG_TYPE] {
                        elasticsearch {
                                hosts => "http://192.168.0.17:9200"
                                index => "logstash-fortigate-traffic-index-%{+YYYY.MM.dd}"
                        }
                }
                else if "event" in [LOG_TYPE] {
                        elasticsearch {
                                hosts => "http://192.168.0.17:9200"
                                index => "logstash-fortigate-event-index-%{+YYYY.MM.dd}"
                        }
                }
                else if "utm" in [LOG_TYPE] {
                        elasticsearch {
                                hosts => "http://192.168.0.17:9200"
                                index => "logstash-fortigate-utm-index-%{+YYYY.MM.dd}"
                        }
                }
        }
}

유입되는 ap 로그 중 grok 패턴을 기존보다 상세히 하여 ap에 로그인한 것과 로그인 실패한 것을 구분

elastalert rule example 참조

- https://dirt-spoon.tistory.com/24

- https://dirt-spoon.tistory.com/25

grok의 pattern_dir에 있는 pattern은 제가 작성했던 pattern을 그대로 사용하시면 됩니다.

grok 패턴 설명 1

grok 패턴 설명 2

grok 패턴 설명 3

grok 패턴 설명 4

input {
        file {
                path => "/var/log/rsyslog/192.168.10.2/*.log"
                start_position => "beginning"
                tags => ["ap1"]
        }
        file {
                path => "/var/log/rsyslog/192.168.10.3/*.log"
                start_position => "beginning"
                tags => ["ap2"]
        }
        file {
                path => "/var/log/rsyslog/192.168.10.4/*.log"
                start_position => "beginning"
                tags => ["ap3"]
        }
        file {
                path => "/var/log/rsyslog/192.168.10.5/*.log"
                start_position => "beginning"
                tags => ["ap4"]
        }
}

filter {
        if "ap1" in [tags] or "ap2" in [tags] or "ap3" in [tags] or "ap4" in [tags]  {
                grok {
                        patterns_dir => ["/etc/logstash/pattern.d"]
                                match => {
                                        "message" => [
                                                "%{SYSLOGTIMESTAMP:access_time} %{IPORHOST:ip_or_host} %{IPORHOST:process}\[%{BASE10NUM:process_id}\]\: %{GREEDYDATA:sub_message}",
                                                "%{SYSLOGTIMESTAMP:access_time} %{IPORHOST:ip_or_host} \[%{BASE10NUM:process_id}\]\: %{GREEDYDATA:sub_message}",
                                                "%{SYSLOGTIMESTAMP:access_time} %{IPORHOST:ip_or_host} %{WORD:process}\: User\: %{GREEDYDATA:user} last logged %{GREEDYDATA:access_result} in %{GREEDYDATA:access_day}\#%{NUMBER:deauthentication_reason_code}, to %{IP:destination_ip}, from %{IP:source_ip} using %{WORD:access_method}",
                                                "%{SYSLOGTIMESTAMP:access_time} %{IPORHOST:ip_or_host} %{WORD:process}\: User\: %{GREEDYDATA:reason} in %{GREEDYDATA:access_day}\#%{NUMBER:deauthentication_reason_code}, to %{IP:destination_ip}, from %{IP:source_ip} using %{WORD:access_method}"
                                        ]
                                }
                }
                mutate { remove_field => [ "message" ] }
        }
}

output {
        if "ap1" in [tags] {
                elasticsearch {
                        hosts => ["http://192.168.0.17:9200"]
                        index => "logstash-ap1-index-%{+YYYY.MM.dd}"
                }
        }
        else if "ap2" in [tags] {
                elasticsearch {
                        hosts => ["http://192.168.0.17:9200"]
                        index => "logstash-ap2-index-%{+YYYY.MM.dd}"
                }
        }
        else if "ap3" in [tags] {
                elasticsearch {
                        hosts => ["http://192.168.0.17:9200"]
                        index => "logstash-ap3-index-%{+YYYY.MM.dd}"
                }
        }
        else if "ap4" in [tags] {
                elasticsearch {
                        hosts => ["http://192.168.0.17:9200"]
                        index => "logstash-ap4-index-%{+YYYY.MM.dd}"
                }
        }
}

저는 syslog 서버를 기준으로 구축을 한다고 밝혔습니다.

그래서 ap에서 받은 syslog를 기준으로 설명을 진행하겠습니다.

rsyslog의 설정을 제가 한 대로 따라하셨다면, 해당 폴더에 syslog 를 수신한 폴더와 내부에 파일이 보일 겁니다.

해당 폴더에 들어가면 수신날자.log 형태의 파일을 확인할 수 있습니다.

kibana에서 보고 싶은 파일의 위치를 확인 하신 후 logstash의 설정을 합니다.

우선 먼저 어떻게 파일이 수신되는지 확인을 합니다.

저는 logstash의 로그 파싱 설정을 /etc/logstash/conf.d 폴더에 logstash.conf 로 작성하였습니다.

[root@tmplogsvr 192.168.10.2]# cat /etc/logstash/conf.d/logstash.conf
input { # logstash에서 파일을 받아들이겠다는 선언
        file { # 어떤 형태의 입력 type을 받겠다는 선언
                path => "/var/log/rsyslog/192.168.10.2/*.log" # 받아 들일 로그의 위치 선언
                start_position => "beginning" # 파일을 읽는 방식으로 차후 자세한 설명
                tags => ["ap1"] # 해당 경로의 파일인 경우 tag에 "ap1"을 표시
        }
}

output { #logstash에서 파일을 내보내겠다는 선언
        if "ap1" in [tags] { # 로그 중 tags에 "ap1"이 표시되어 있다면
                elasticsearch { # 로그를 elasticsearch에 보내겠다는 선언
                        hosts => ["http://192.168.0.17:9200"] # elasticsearch url 정보
                        index => "logstash-ap1-index-%{+YYYY.MM.dd}" # elasticsearch의 index 입력 방식
                }
        }
}

위의 설정대로 하신 후 logstash를 재시작 합니다.

이제 kibana에 접속을 합니다.

정상적으로 출력이 됐다면, logstash.conf 파일을 아래와 같이 수정

input {
  file {
    path => "/var/log/rsyslog/192.168.10.2/*.log"
    start_position => "beginning"
    tags => ["ap1"]
  }
}

filter {
  if "ap1" in [tags] { # tags에 "ap1"이 있다면
    grok { # grok 적용해라
      patterns_dir => ["/etc/logstash/patterns.d"] # grok 적용 전 패턴 등록
      match => { "message" => [ "%{SYSLOGTIMESTAMP:access_time} %{IPORHOST:ip} %{IPORHOST:process}\[%{BASE10NUM:process_id}\]\: %{GREEDYDATA:sub_message}" ] }
    }
    mutate { remove_field => [ "message" ] } # message 필드 삭제
  }
}

output {
        if "ap1" in [tags] {
                elasticsearch {
                        hosts => ["http://192.168.0.17:9200"]
                        index => "logstash-ap1-index-%{+YYYY.MM.dd}"
                }
        }
}

수정 후 logstash 재시작 수행

이 후 kibana에서 로그 확인 시 아래와 같이 변경된 것을 볼 수 있습니다.

설치할 logstash 방식 선택

해당 매뉴얼은 centos에서 설치하여 yum을 선택하였습니다.

https://www.elastic.co/kr/downloads/logstash

yum을 이용한 repository 추가

설치 시 elasticsearch의 버전과 동일한 버전을 설치하세요.

yum을 이용한 logstash 설치

logstash 시스템 등록

logstash 시스템 시작 / 중지