ELK 활용 사례 / 엘라스틱서치 활용사례
전체 서비스 평일 평균 엘라스틱서치 클러스터로 쌓이는 전체 보안 이벤트 로그는 아래와 같은 상황이다.
날짜별 Document 수는 3억2천~3억4천건 정도... 가끔 불특정 다수 해외 IP에서 지속적으로 스캔성 공격 시도를 할때는
늘어나는 경우도 있다.
사람이 일일이 이렇게 방대한 로그를 다 들여다보기란 바닷가 백사장에서 바늘 찾기다.
어떻게 하면 좀더 효율적으로 보안 위협 로그를 선별 할수 있을까?
현재 구성은 IDC 전체 구간 각 서비스별 보안장비 -> arcsight -> ELK 스택(elasticsearch) 이다.
요약하자면, 모든 보안 이벤트 로그를 엘라스틱서치로 받고 있다.
여유가 있다면 xpack 을 유료로 구매하여 Watcher (알림 통보)를 사용하면 된다.
방대한 보안 이벤트중 오픈 소스 "elastalert" 를 사용하여 보안 위협에 대하여 자동으로 알려주는
elasticsearch 활용 사례에 대하여 포스팅한다.
elastalert 설치 과정은 생략...
type에 대한 설명도 생략....
인터넷 서치하면 많이 나온다.
알려진 C&C 서버로의 통신 탐지
- type 은 blacklist
- blacklist 파일 내용과 매칭되면 알람을 통보해준다.
- c2-list.txt 내용은 아래와 같이 IP를 적어주면된다.
| name: "C2 서버 통신 탐지" type: blacklist use_local_time: false use_strftime_index: true index: xxxx-%Y.%m.%d blacklist: - "!file /root/.../elastalert/list/c2-list.txt" # 특정 source IP 및 특정 목적지 Port 는 제외 filter: - query: query_string: query: "NOT sourceAddress:(59.x.x.x OR 1.x.x.x OR 10.20.x.x) AND NOT destinationPort:(25 OR 443 OR 3389)" compare_key: "destinationAddress" alert_text: "Match Count: {0} \n{1} {2} {3} {4} {5} {6} {7} {8} {9}" alert_text_type: "alert_text_only" alert_text_args: ["num_matches", "@timestamp", "deviceHostName", "deviceAction", "source.country_code2", "sourceAddress", "sourcePort", "destination.country_code2", "destinationAddress", "destinationPort"] # 아래는 알림에 포함하고 싶은 field만 정의 #include: ["deviceHostName", "deviceAction", "source.country_code2", "sourceAddress", "sourcePort", "destination.country_code2", "destinationAddress", "destinationPort"] alert: - slack slack: slack_webhook_url: "https://hooks.slack.com/services/T9YJY3HV0/BBBH341PT/...................................................." email: - "user1@xxxxxxxx.com" - "user2@xxxxxxxx.com" |
탐지 Notification
slack
스캔성 공격 탐지 IP
- type 은 frequency
- 조건은 1분에 1000건 이상 해외 IP중에 accept 된 source IP 탐지후 알람
| name: "1분 동안 1000회 이상 접속 Source IP" type: frequency num_events: 1000 timeframe: minutes: 1 use_strftime_index: true index: xxxx-%Y.%m.%d # 해외 IP중 accept 된것만 그리고 icmp는 제외 filter: - query: query_string: query: "NOT source.country_code2:KR AND NOT deviceAction:(deny OR drop OR Block) AND _exists_:source.country_code2 AND NOT destinationPort:2048" top_count_keys: - sourceAddress #top_count_number: 2 realert: minutes: 5 query_key: "sourceAddress" ################ Alert ################# alert_text: "Match Count: {0} \n{1} {2} {3} {4} {5} {6} {7} {8} {9}" alert_text_type: "alert_text_only" alert_text_args: ["num_matches", "@timestamp", "deviceHostName", "deviceAction", "source.country_code2", "sourceAddress", "sourcePort", "destination.country_code2", "destinationAddress", "destinationPort"] alert: - slack slack: slack_webhook_url: "https://hooks.slack.com/services/T9YJY3HV0/BBBH341PT........................................" email: - "example01@mail.com" - "example02@mail.com" |
탐지 알람 화면
slack
보안 장비 장애 알람 (특정 이벤트 발생시)
- type은 frequency (다른 type도 가능)
- 조건 특정 필드에 critical 이라고 매칭 되면 알람 통보
| name: "보안장비 이슈 발생" type: frequency use_strftime_index: true index: xxxx-%Y.%m.%d num_events: 1 timeframe: minutes: 2 # critical 문자열이 매칭되면 알람 통보 filter: - query: query_string: query: "(level:critical OR pri:critical) AND NOT deviceAction:update" alert_text: "Match Count: {0} \n{1} {2} {3} {4} {5} {6} {7}" alert_text_type: "alert_text_only" alert_text_args: ["num_matches", "@timestamp", "deviceHostName", "deviceAction", "pri", "level", "hbdn_reason", "reason"] alert: - slack slack: slack_webhook_url: "https://hooks.slack.com/services/T9YJY3HV0/BBBH34..........................................................." email: - "example1@mail.com" - "example2@mail.com" - "example3@mail.com" |
탐지 알람 화면
slack
elastalert 시작은 각 OS 별로 service에 등록하여 실행하는 방법도 있지만
각 rule별로 백그라운드로 실행하면 여러가지 장점이 있다.
예: c2서버 통신 접속 탐지 시작 스크립트
cat elastalert-c2
#/bin/bash
DIR="/root/.../elastalert"
NAMEs="c2"
CONFs="$NAMEs.yaml"
RULEs="$NAMEs.rule.yaml"
LOGs="$NAMEs.log"
nohup /usr/local/bin/elastalert --verbose --start NOW --config $DIR/etc/$CONFs --rule $DIR/rules/$RULEs > $DIR/log/$LOGs 2>&1 &
sleep 2
ps aux | grep $NAMEs.rule.yaml | grep -v "grep"
elastalert 프로세스가 아래와 같이 실행된다.
ElasticSearch 활용 사례
elasticsearch 으로 하고 있는것들... 최초 개발이유 1. 통합 로그 검색 시스템이 필요 (장애나 이슈 발생시 연관된 수많은 Node(서버,스위치,스토리지 등등) 의 Log를 일일이 접속해서 들여다본다는
rootkey.tistory.com
