Elasticsearch Logstash Kibana (ELK) 로 웹서버에 로거 연동

작년 2학기 학교 <웹보안 및 실습> 수업 때 "악성 pdf 탐지 웹 구현" 프로젝트를 진행했었는데, 잊기 전에 elk 로거 관련해서 기록을 남겨보고자 한다. 임시 글로 썩고 있었는데 이제야 올린다니 반성해야겠다.

우선 우리의 악성 pdf 탐지 웹은 다음과 같은 원리로 구상되었다.

1. 준비된 bulk 정상 pdf / 악성 pdf 들을 각각 hidost 라이브러리를 이용해 구조 분석
2. hidost 분석 결과 저장 후 randomforest 이용해 모델 학습
3. 새로운 pdf 를 웹에 업로드
4. 해당 pdf 구조를 hidost 로 다시 분석
5. 모델에 넣어 결과 출력
6. elk로 결과 로그 생성

elk 결과 로그를 붙이기로 한 이유는, 웹을 사용하는 유저들 중 비정상적인 활동을 하는 유저의 로그를 수집하기 위해서이다. 예를 들면 악성 pdf를 대량으로 검사하는 유저가 있다면 높은 확률로 해킹을 준비하는 악의적인 유저일 수 있겠다는 생각에서 출발했다.
아쉬운 점은 elk로 pdf 구조 분석도 했으면 좋았을텐데, 프로젝트 기간 내에 생각을 못했다. 악성 pdf에서 자바스크립트 구조가 더 잘 발견될 것 같은데, 이를 분석하는 데에 elk를 추후에 사용해봐야겠다.

엄청나게 긴 서두를 뒤로 하고, 이제 본격적인 내용으로 들어가보자 !!

ELK란?

ELK는 Elastic search, Logstash, Kibana 의 약자이다.

우선 Logstash는 데이터를 수집한 후, Elasticsearch 와 같은 stash(저장 공간)로 전송하는 데이터 처리 파이프 라인이다. Elasticsearch는 JSON 기반의 분산형 RESTful 검색 및 분석 엔진을 말하며 Kibana는 데이터를 시각화하는 도구이다.

프로젝트 구조

시작하기 전, 코드의 출처는 https://woolbro.tistory.com/88 !!

글에서 필요한 주요 내용들만 가져왔고, 전체 구조는 https://github.com/julia-ing/MalPdf-Detection 여기서 확인 가능합니닷 :)

MalPdf-Detection
├── docker-elk 
    ├── docker-compose.yml 
    ├── elasticsearch 
    	├── config
        	└── elasticsearch.yml
        └── Dockerfile
    ├── kibana
    	├── config
        	└── kibana.yml
        └── Dockerfile
    └── logstash 
    	├── config
        	└── logstash.yml
        ├── pipeline
        	└── logstash.conf
        └── Dockerfile
├── logger 
	└── elk_logger.py
├── libsvm_to_csv.py
├── predict.py
├── app.py
└── ML.py

docker-elk 폴더로 진입하고
docker-compose up --build -d 명령어를 입력하면 도커가 정상적으로 빌드될 것이다.

도커 관련 파일들 관련해서는 docker-elk 폴더 안에 docker-compose.yml 파일을 보면 네트워크와 흐름을 쉽게 파악할 수 있다.

실질적으로 로그를 붙이고 분석하고 확인하는 과정을 보자면,

logstash 에서 로그를 받는다.

1. logger > elk_logger.py

import logging, logstash

log_format = logging.Formatter('\n[%(levelname)s|%(name)s|%(filename)s:%(lineno)s] %(asctime)s > %(message)s')


def create_logger(logger_name):
    logger = logging.getLogger(logger_name)
    if len(logger.handlers) > 0:  # 로거가 이미 존재하는 경우
        return logger 

    logger.setLevel(logging.INFO)
    logger.addHandler(logstash.TCPLogstashHandler('localhost', 5000, version=1))  # 

    return logger

2. libsvm_to_csv.py

import os
import pandas as pd
from logger import elk_logger
from predict import predict


def print_result(index):
    f = open('../test-data.libsvm', 'r')

    lines = f.readlines()
    result = os.popen("cat ../features.nppf | wc -l")
    result = result.read()

    for line in lines:
        minus_list = [-1] * (int(result) - 1)

        l = line.split(' #')[0]
        l = l.split(' ')
        file_name = line.split(' #')[1].strip('\n')

        for i in range(len(l)):
            if i != 0:
                key = int(l[i].split(':')[0])
                val = float(l[i].split(':')[1])
                minus_list[key - 1] = val

        result_string = str(minus_list).strip('[]')

        tmp = [[]]
        for i in range(len(result_string.split(','))):
            tmp[0].append(result_string.split(',')[i])

        final = pd.DataFrame(tmp)
        prediction = predict(final)
        print(prediction)

        if prediction == "['B']" or prediction == ['B']:
            logger = elk_logger.create_logger('elk-test-logger')
            logger.info('prediction result: benign pdf')
            return "Benign pdf with high probability"
        elif prediction == "['M']" or prediction == ['M']:
            logger = elk_logger.create_logger('elk-test-logger')
            logger.info('prediction result: malicious pdf')
            return "Malicious pdf with high probablity"
        else:
            return "Failed to predict the result"

윗부분 코드들은 데이터를 처리하고, 모델에 넣어 해당 pdf가 악성일 확률이 높은지 아닌지 실제로 예측하는 코드이다.
if prediction ~~ 여기부터 로그를 붙이는 코드인데, 정상 pdf 일 확률이 높다면 info 에 benign 내용을 담아주고 / 악성 pdf 일 확률이 높다면 info 에 malicious 내용을 포함시킨다.

elasticsearch 에서 로그를 분석한다.

docker-elk >> logstash >> pipeline >> logstash.conf 설정

input {
	tcp {
		port => 5000
	}
}

output {
	elasticsearch {
		hosts => "elasticsearch:9200"
		user => "elasticsearch"
		password => "password"
		index => "elk-logger"
	}
}

로그스태시에서 수집한 로그를 엘라스틱 서치에 보내 분석하려면 둘을 연결해야하는데, 이를 위해 conf 파일에서 hosts에 "elasticsearch:9200"을 붙여줘야 한다.
키바나에 로그인할 유저 이름과 비밀번호를 설정해준다(이 값은 자유롭게 바꿔도 됨).
추가적으로 인덱스를 설정해주면 "elk-logger"라는 이름으로 로그가 수집된다.

kibana 에서 분석 결과를 시각화한다.

테스트할 때 bulk pdf 를 넣었어서 391번이나 되어있다 😂

이건 악성 pdf 일 확률이 높은 pdf 의 결과인데, message에 아까 설정한 메모가 잘 입력된 걸 확인할 수 있었다.

추가 - 웹 동작 요약 사진

구현한 웹을 요약한 사진 두 장이다.
파일을 여러개 선택해서 업로드하면 -> hidost에서 구조를 분석한 후 -> 머신러닝 모델에서 예측해 -> 각각 결과를 출력해준다.

느낀점

이번 실습 프로젝트를 하면서 정말 많이 성장했음을 느꼈다. 한 웹을 구현하기 위해 보안 + 머신러닝 + 백엔드 + 프론트 + elk .. 이 많은 내용들을 공부하고 커버한 것 만으로 큰 성공인 것 같다. elk 코드 관련해서는 프로젝트 시간 관계상 구글링을 통해 많은 내용들을 참고했지만, 로그를 붙이고 확인하는 과정을 직접 이해하고 경험해보며 다음에는 꼭 내가 만들고 싶은 서비스에 맞게 elk 를 적용해보고 싶다는 생각이 들었다.
아무튼 좋은 경험이었던 것이 분명하다 !!