Elastic & Splunk를 활용한 이상징후 탐지(1)

회사 교육 프로그램의 일환으로, 멀티캠퍼스에서 진행한 ‘Elastic & Splunk를 활용한 이상징후 탐지’ 수업을 수강하게 됐다. 참고로 우리 회사는 Splunk를 활용한 보안 솔루션을 고객사에 제공하고 있고, 자연스럽게 Splunk에 대한 기술적인 이해가 필요한 상황이다.

하지만 최근엔 오픈소스 기반의 Elastic처럼 비용 부담이 적고 유연한 대안을 찾는 목소리도 늘어나고 있고, 개인적으로도 이런 오픈소스 로그 분석 시스템에 관심이 많았던 터라 마침 회사에서 장려한 교육 프로그램을 통해 수강하게 됐다.

ELK 설치부터 로그 수집, 시각화, 분석, Splunk 활용, 마지막에는 리눅스/윈도우 기반 클러스터 구성까지 실습이 포함되어 있어서 내가 원하던 “이론보다 실습 중심” 구성이라 특히 좋았다. 그래서 나중에 참고하려고, 직접 따라하며 배운 내용을 블로그에 기록해두기로 했다.

 

엘라스틱 설정이나 필터 구성 중 궁금한 부분은 아래 문서에서 자세히 확인 가능 :

🔗 https://www.elastic.co/docs

Elastic Docs | Elastic Docs

 

 

Logstash의 grok 필터는 로그 메시지를 특정 패턴으로 분리하고, 필드 단위로 나눠주는 핵심 기능이다.
이를 제대로 활용하려면 정규표현식이 필수.

🔗 정규표현식 테스트 사이트

https://regex101.com/r/c7B3tk/1

regex101: build, test, and debug regex

 

 

Elasticsearch를 공식 사이트에서 8.17.4 버전 윈도우로 받아서 설치.  
설치가 끝났으면, 이제 윈도우 서비스로 등록.

이걸 위해 `nssm`을 사용했다.  
명령어는 `nssm install` 입력하고, 실행파일 경로는 `elasticsearch.bat`로 잡아주면 된다.

 

Elasticsearch 설치가 끝났으면, 이제 Logstash도 설정해준다.
Logstash에서 데이터를 실시간으로 읽어들이려면 config 파일의 몇몇 설정을 수정해야 한다.

config 자동 반영 설정

logstash.yml에서 pipeline 설정을 조금 수정해준다.
수정 사항을 감지해서 자동으로 reload되도록 해주는 옵션이다.

이걸 true로 두면, .conf 파일을 수정한 후 Logstash를 다시 켜지 않아도 적용된다.

 

이제 apache 로그 파일을 읽을 수 있도록 .conf 파일을 작성한다.
input에 경로 설정, output은 일단 stdout으로 해두고 필터는 비워둔다.

📌 주의: 로그 파일 경로에 한글 포함되면 안 됨. 꼭 영문 경로에 위치시킬 것!

 

추가로, logstash.yml에서 ECS 호환성도 비활성화해준다.
안 건드려도 작동하긴 하지만, 실습 시엔 pipeline.ecs_compatibility: disabled로 설정하는 게 편하다.

 

이제 Logstash 실행해서 데이터 잘 들어오는지 확인해본다.

apache_sample.log 파일이 잘 읽혀서 출력되고 있다면 성공.

 

이제 filter 부분에 grok을 추가해서 로그 파싱을 해본다.
Apache 로그는 내장 패턴인 COMBINEDAPACHELOG로 매칭할 수 있다.

 

filter grok 추가 후 로그스태시가 재시작되며, 데이터가 쪼개진걸 확인할 수 있다.

 

grok + date 필터 적용

데이터가 쪼개졌다면, 이제 시간 필드도 추가한다. 시계열 차트를 그리기 위해선 timestamp 포맷을 엘라스틱이 인식할 수 있게 바꿔줘야 한다.

@timestamp 필드에 변환된 시간이 들어가게 된다.

 

엘라스틱은 아파치의 시간 포맷을 이해하지 못한다. 
"@timestamp" => 2022-05-27T12:10:24.000Z, 이런식의 포맷만 알아들을 수 있으므로 이런식으로 @timestamp에 포맷을 바꿔주고 저장을 해준 것.

 

Logstash Date 필터 공식 문서

https://www.elastic.co/guide/en/logstash/current/plugins-filters-date.html

Date filter plugin | Logstash Reference [8.17] | Elastic

 참고해서 기능을 쉽게 설정할 수 있다.

 

추가로 grok 필터 관련 공식 문서와 패턴들도 자주 참고했다

https://www.elastic.co/guide/en/logstash/current/plugins-filters-grok.html

Grok filter plugin | Logstash Reference [8.17] | Elastic

이 문서에 따르면 다음 깃에서 약 120개정도의 정규표현식 패턴들을 확인할 수 있다.

https://github.com/logstash-plugins/logstash-patterns-core/tree/main/patterns

logstash-patterns-core/patterns at main · logstash-plugins/logstash-patterns-core

하지만 정규표현식에 익숙하다면 직접 패턴 만드는 게 더 나을 때도 많다.

 

이번엔 dissect 필터 실습.
정규표현식 없이, 구분자 만으로 로그를 나눌 수 있을 때 쓰면 편하다.

간단하게 문자열을 쪼개는 예제로 테스트.

간단한 테스트를 위해 a-b:c 문자열이 들어 있는 sample.log 준비.

input에 로그 파일 경로 지정하고 stdout으로 출력 설정.

로그 스태시 실행.

 

 

dissect는 기본적으로 message를 쪼갤 때 이렇게 작성:

filter에 dissect 추가.정규표현식 없이 구분자만으로 필드 나누기.
message 전체를 하나의 필드(field)로 파싱함.

 

 

dissect로 클라이언트 IP만 따로 뽑아보기 위해 log 형식에 맞춰 %{clientip} - - %{} 패턴 작성.

dissect로 필드가 잘 나뉘는 걸 확인.

 

참고 : dissect 옵션 관련 공식 문서. 지금 쓴건 여기서 mapping 참고

https://www.elastic.co/guide/en/logstash/current/plugins-filters-dissect.html

Dissect filter plugin | Logstash Reference [8.17] | Elastic

 

 

dissect의 mapping 옵션 활용해 전체 로그 구조화.
method, uri, status, bytes, referer, agent까지 필드로 분리.

 

 

필요 없는 필드 제거.
host, path, @version, @timestamp 필드는 remove_field로 제거.
시계열 차트 작성을 위해 timestamp 필드를 date 필터로 재설정.

 

 

데이터가 많아지면 성능 저하 우려가 있을 수 있기 때문에 stdout은 주석처리 후, sample 로그파일도 원본 apache.log로 변경

stdout을 주석처리 했으므로 더이상 화면에 안뿌리고 elasticsearch에 저장중.

 

키바나에서 확인한 인덱스 목록. logstash-날짜 형식으로 인덱스가 잘 저장되어 있는 걸 확인할 수 있다.

 

index pattern에 logstash 입력 후 저장.
이걸 통해 Kibana에서 인덱스를 조회할 수 있는 데이터 뷰(Data View)를 만드는 작업.

 

discover로 가서 데이터 잘 가져왔나 확인.

 

apache.log파일에서 gg를 눌러 맨 끝 라인으로 가서, 총 데이터 수를 비교해 본다.

이 수치와 Discover에 표시된 document 수가 비슷해야 로그 수집이 잘 된 것.

 

시계열 차트 생성.  X축에 Date Histogram, timestamp.

 

상태코드(status) 필드로 split series 추가. 404, 200 등 어떤 응답 코드가 얼마나 발생했는지 쉽게 확인 가능. → "404가 가장 많이 발생했구나" 처럼 로그 패턴 파악을 좀 더 자세하게 할 수 있다.

 

내가 만든 차트들을 save를 눌러 라이브러리에 저장하고, 대시보드에 추가해서 언제든지 볼 수 있다.

 

Legacy - Data table 사용 예 : 404 중에서도 이러한 패턴들이 많이 발생했구나를 알 수 있다.

 

 

통계데이터의 특징 : 데이터를 쪼개면 쪼갤 수록 그에 대한 정보를 더 자세히 알 수 있다.

이제 다시 테스트 모드 시작. 로그파일을 다시 샘플 파일로 바꿔주고, elasticsearch 부분은 주석처리. stdout은 주석 해제하여 콘솔창에서 볼 수 있도록 해준다.

 

이제 단순히 자르는 걸 넘어서 조건문을 활용해 상황에 따라 다르게 처리해보자. Logstash는 조건문과 연산자, 정규표현식도 지원하기 때문에 유연한 파싱이 가능하다.
아래 공식 문서를 참고:

https://www.elastic.co/guide/en/logstash/current/event-dependent-configuration.html

Accessing event data and fields | Logstash Reference [8.17] | Elastic

 

dissect와 date사이에 다음과 같은 if 조건문을 넣어 url과 변수를 쪼개기

 

 

grok 추가

 

File 정보 추출

 

이번엔 한번 더 쪼개서 파일의 확장자를 추출해보자 

 

ruby 추가

 

param의 문자열 길이값을 측정해서 paramlen필드에 저장

 

event명령어를 어떻게 쓰면 좋은지 나와있는 가이드 :

https://www.elastic.co/guide/en/logstash/current/plugins-filters-ruby.html

Ruby filter plugin | Logstash Reference [8.17] | Elastic

 

루비 필터 안에 쓸 수 있는 명령어들 참고 사이트 :

https://ruby-doc.org/3.4.1/String.html

class String - RDoc Documentation

 

이제 마찬가지로 엘라스틱 서치에 저장 모드 on

stdout은 주석해주고 elasticsearch 주석 해제

 

date 추가

 

원본 로그 파일로 수정

 

apachelog-2022가 생성된 걸 확인 가능.

이제 이걸로 키바나에서 검색하고 차트 만드는 작업들을 할 수 있다.

 

이번에는 일부러 에러를 내고, 데이터 시각화를 해보는 작업을 해보겠다.

에러 메시지의 가독성이 지독히도 떨어짐..

 

줄바꿈을 추가해서 잘 볼 수 있게 했다. 꼭 화려하고 멋진 차트들만이 데이터 시각화는 아니다. 이렇게 로그를 읽기 쉽게 만드는 것도 일종의 데이터 시각화라 볼 수 있다.

 

 

다음으로, iis로그를 이용한 새로운 conf 파일을 만들어서 실습 ㄱㄱ

 

실제 로그에 #이 포함되어 있을 수도 있으니, 다음과 같이 바꾸고 타임스탬프도 추가하여 저장.

 

 

사실 엘라스틱 서치를 한 2년전에 맛보기만 해봤던 경험이 있는데, 어설프게 알고있던 ELK 스택을 이번에 하나씩 실습하면서 개념들이 연결되는 느낌이었다. dissect 필터는 정규표현식보다 직관적이라 편했고, 콘솔창 로그를 정리해서 보는 것도 일종의 시각화란 게 새삼스러웠다. 아직 grok이나 조건문 같은건 손에 잘 안 익지만, 계속 써보면서 감을 익혀야겠다..!