케세라세라: Elasticsearch 활용(DNS Event Log 분석

2017년 6월 25일 일요일

Elasticsearch 활용(DNS Event Log 분석 - 2nd)

데이터 정리가 문제지, 정리된 데이터의 분석은 쉽다
- 데이터 분석이 쉬워지는 정규표현식(459페이지)

엘라스틱도 마찬가지. 3006, 5156 이벤트 로그를 분석에 편리한 형태로 정리해주는 Logstash 설정은 다음과 같다.

input { 
 beats { 
  port => 5044 
  } 
 }

filter {
 if [event_id] == 3006 {
  if [event_data][QueryName] in ["localhost","wpad","isatap","lg","LG"] or [event_data][QueryType] == "28" {
   drop { }
  }

  mutate { 
   replace => {
    "type" => "dns_event"
   }
  }

  grok { 
   match => {
    "[event_data][QueryName]" => "([^.]+\.)*?(?<url>[^.]+\.((ac|co|go|ne|nm|or|pe|re)\.)?[^.]+\.?$)"
   }
  }
 }

 if [event_id] == 5156 {
  if [event_data][DestAddress] == "172.20.10.1" and [event_data][DestPort] != "53" or [event_data][SourcePort] == "1900" or [event_data][DestPort] == "1900" or [event_data][SourceAddress] == [event_data][DestAddress] or [event_data][SourceAddress] =~".*:.*"  or [event_data][SourceAddress] == "192.168.56.1" {
   drop { }
  }

  mutate { 
   replace => {
    "type" => "network_event"
   }
  }

  grok { 
   match => {
    "[event_data][Application]" => "(?<proc_path>.+\\).+"
   }
  }

  grok {
   match => {
    "[event_data][Application]" => "(.+\\)?(?<proc_name>.+)" 
   }
  }

  if [event_data][Protocol] == "6" {
   mutate {
    replace => { "[event_data][Protocol]" => "TCP" }
   }
  }

  if [event_data][Protocol] == "17" {
   mutate {
    replace => { "[event_data][Protocol]" => "UDP" }
   }
  }

  if [event_data][Direction] == "%%14593" {
   mutate {
    replace => { "[event_data][Direction]" => "OUT" }
   }
  }

  if [event_data][Direction] == "%%14592" {
   mutate {
    replace => { "[event_data][Direction]" => "IN" }
   }
  }

  if [event_data][SourceAddress] =~ "(\d+\.){3}\d+" {
   geoip {
    source => "[event_data][SourceAddress]"
    target => "src"
   }
  }

  if [event_data][DestAddress] =~ "(\d+\.){3}\d+" {
   geoip {
    source => "[event_data][DestAddress]"
    target => "dest"
   }
  }

  mutate { 
   add_field => {
    "sPort" => "%{[event_data][SourcePort]}"
    "dPort" => "%{[event_data][DestPort]}"
   }
   convert => {
    "sPort" => "float"
    "dPort" => "float"
   }
  }
 }
}

output {
 elasticsearch { 
  hosts => [ "localhost:9200" ] 
  index => "win_event"
 }
  stdout {
   codec => rubydebug 
 }
}

불필요한 데이터를 제거하고, 가독성을 높이는 과정에서 filter 영역이 좀 복잡해졌지만, 대부분 grok이나 mutate 필터를 이용해서 분석을 원하는 특정 데이터 영역을 추출하거나, 읽기 편하게 바꾸는 과정. 마지막 부분에 추가한 geoip 필터는 IP의 지리 정보를 추가해준다.