SIEM에서 탐지 규칙이 아무리 좋아도 들어온 로그가 비어 있거나 겹쳐 있으면 결과를 믿기 어렵습니다. 이 글에서는 일부러 문제를 심은 20줄짜리 로그를 만들고, 작은 Python 스크립트로 유실, 중복, 파싱 실패, 시간 어긋남을 하나씩 찾아냅니다. 로그 품질은 보통 완전성, 정확성, 적시성, 일관성 네 가지 지표로 나눠 봅니다. 이 글은 완전성, 정확성, 적시성 세 가지를 다루고, 필드 이름 표준화 같은 일관성 점검은 '이어서 해 볼 것'에서 방향만 소개합니다. 모든 데이터는 직접 만든 가짜 로그이고, 외부 시스템에는 아무것도 보내지 않습니다.
준비물
| 항목 | 내용 |
|---|---|
| Python | 3.12 (표준 라이브러리만 사용) |
| 작업 폴더 | 새 빈 폴더 하나 |
| 예상 시간 | 15분 |
Elasticsearch나 Logstash는 필요하지 않습니다. 같은 점검을 SIEM 안에서 하는 방향은 '이어서 해 볼 것'에 적었습니다.
점검 기준 정하기
점검마다 무엇을 어떤 문제로 보는지 먼저 정해 둡니다.
| 점검 | 지표 | 이 글의 판정 방법 |
|---|---|---|
| 유실 | 완전성 | 호스트별 일련번호(seq)에 빈 번호가 있는지 |
| 파싱 실패 | 정확성 | 정해진 형식에 맞지 않는 줄이 있는지 |
| 중복 | 정확성 | 완전히 같은 줄이 두 번 이상 있는지 |
| 시간 어긋남 | 적시성 | 수집 시각과 이벤트 시각의 차이가 5초를 넘거나 음수인지 |
일련번호는 로그를 보내는 쪽이 붙여 주어야 쓸 수 있습니다. 번호가 없는 소스는 단위 시간당 건수를 소스 쪽과 SIEM 쪽에서 각각 세어 비교하는 방식으로 바꿔야 합니다.
1단계. 문제가 섞인 샘플 로그 만들기
web01, web02 두 호스트가 이벤트를 10건씩 보냈다고 가정합니다. 한 줄은 이벤트 시각 수집 시각 호스트 seq=번호 요청 형식입니다.
아래 스크립트가 넣는 문제는 네 가지입니다.
- 유실: web01의 4번, 5번 이벤트가 도착하지 않음
- 중복: web01의 8번 줄이 두 번 들어옴
- 시간 어긋남: web02의 7번 이벤트에 시계가 느린 호스트의 시각(40초 이전)이 찍힘
- 파싱 실패: 형식이 깨진 줄 하나
# make_sample.py
from datetime import datetime, timedelta, timezone
base = datetime(2026, 10, 1, 9, 0, 0, tzinfo=timezone.utc)
lines = []
for host in ("web01", "web02"):
for seq in range(1, 11):
if host == "web01" and seq in (4, 5):
continue # 유실: 4, 5번이 도착하지 않음
t = base + timedelta(seconds=seq)
ingested = t + timedelta(seconds=1)
if host == "web02" and seq == 7:
t = t - timedelta(seconds=40) # 시계가 40초 느린 호스트의 이벤트
line = f"{t:%Y-%m-%dT%H:%M:%SZ} {ingested:%Y-%m-%dT%H:%M:%SZ} {host} seq={seq} GET /index 200"
lines.append(line)
if host == "web01" and seq == 8:
lines.append(line) # 중복: 같은 줄이 두 번
lines.append("garbled line without fields") # 파싱 실패
open("sample.log", "w", encoding="utf-8").write("\n".join(lines) + "\n")
print(len(lines), "lines written")
python make_sample.py
기대 출력은 다음과 같습니다.
20 lines written
2단계. 점검 스크립트 작성하기
줄마다 정규식으로 필드를 꺼내고, 위 표의 기준을 그대로 코드로 옮긴 스크립트입니다. 중복 줄은 건수 계산에서 한 번만 세어야 유실 판정이 흔들리지 않으므로, 중복으로 판정한 줄은 이후 점검에서 뺍니다.
# check_quality.py
import re
import sys
from collections import defaultdict
from datetime import datetime
MAX_DELAY = 5 # 초. 이벤트 시각과 수집 시각의 허용 차이
PATTERN = re.compile(r"^(\S+) (\S+) (\S+) seq=(\d+) ")
def parse_time(text):
return datetime.strptime(text, "%Y-%m-%dT%H:%M:%SZ")
total = 0
parse_failed = []
seen = {}
duplicates = []
by_host = defaultdict(list)
delay_problems = []
for number, raw in enumerate(open(sys.argv[1], encoding="utf-8"), 1):
line = raw.rstrip("\n")
total += 1
m = PATTERN.match(line)
if not m:
parse_failed.append(number)
continue
event, ingested, host, seq = parse_time(m[1]), parse_time(m[2]), m[3], int(m[4])
if line in seen:
duplicates.append((number, seen[line]))
continue
seen[line] = number
by_host[host].append((seq, event))
delay = (ingested - event).total_seconds()
if delay < 0 or delay > MAX_DELAY:
delay_problems.append((host, seq, delay))
print(f"전체 {total}줄, 파싱 실패 {len(parse_failed)}줄 (줄 번호 {parse_failed})")
print(f"중복 {len(duplicates)}건 (줄 {duplicates[0][0]} 은 줄 {duplicates[0][1]} 과 같음)" if duplicates else "중복 0건")
for host, items in sorted(by_host.items()):
seqs = sorted(s for s, _ in items)
missing = [n for n in range(seqs[0], seqs[-1] + 1) if n not in set(seqs)]
print(f"{host}: 받은 {len(seqs)}건, 번호 {seqs[0]}~{seqs[-1]}, 유실 {missing}")
for host, seq, delay in delay_problems:
print(f"시간 이상: {host} seq={seq} 수집 시각과 이벤트 시각 차이 {delay:+.0f}초")
for host, items in sorted(by_host.items()):
for (s1, t1), (s2, t2) in zip(items, items[1:]):
if t2 < t1:
print(f"순서 역전: {host} seq={s1} 다음 seq={s2} 의 이벤트 시각이 {int((t1 - t2).total_seconds())}초 더 이른 시각")
3단계. 실행하고 결과 읽기
python check_quality.py sample.log
기대 출력은 다음과 같습니다.
전체 20줄, 파싱 실패 1줄 (줄 번호 [20])
중복 1건 (줄 7 은 줄 6 과 같음)
web01: 받은 8건, 번호 1~10, 유실 [4, 5]
web02: 받은 10건, 번호 1~10, 유실 []
시간 이상: web02 seq=7 수집 시각과 이벤트 시각 차이 +41초
순서 역전: web02 seq=6 다음 seq=7 의 이벤트 시각이 39초 더 이른 시각
심어 둔 네 가지가 모두 잡혔습니다. web02 7번의 차이가 40초가 아니라 41초로 나온 것은 수집 시각이 원래 이벤트 시각보다 1초 뒤로 찍혀 있기 때문입니다. 이벤트 시각이 실제보다 과거로 찍히면 같은 호스트의 앞뒤 이벤트와 순서가 뒤바뀌어(출력의 마지막 두 줄), 공격 순서를 시간으로 재구성하는 분석이 틀어집니다.
자주 막히는 곳
UnicodeEncodeError: 'charmap' codec can't encode characters in position 0-1: character maps to <undefined>
콘솔 코드 페이지나 출력 파일의 인코딩이 한글을 지원하지 않을 때(예: cp1252) 한글 출력에서 납니다. 한글을 지원하는 환경에서는 나지 않습니다. python -X utf8 check_quality.py sample.log 로 실행하면 해결됩니다.
IndexError: list index out of range
파일 경로를 빼고 python check_quality.py 만 실행하면 sys.argv[1] 에서 납니다. 경로가 틀리면 FileNotFoundError: [Errno 2] No such file or directory: 'nofile.log' 가 나옵니다. 작업 폴더를 확인합니다.
ValueError: time data ... does not match format
시간 형식이 2026-10-01T09:00:01Z 와 다른 로그를 넣었을 때 납니다. 소스마다 시각 형식이 다르면 parse_time 을 소스별로 나누거나, 수집 단계에서 UTC로 맞춘 뒤 점검합니다.
유실이 0건으로 나오는데 로그가 모자란 것 같을 때 이 스크립트는 가장 작은 번호부터 가장 큰 번호 사이의 빈 번호만 봅니다. 끝부분에서 끊긴 유실, 예를 들어 web01의 9번과 10번이 통째로 없는 경우는 잡히지 않습니다. 소스 쪽이 알려 주는 마지막 번호와 비교해야 합니다.
이어서 해 볼 것
- Elasticsearch로 옮길 때는 호스트별로 시간 구간 단위 건수 집계(date_histogram)를 소스 쪽 건수와 비교해 유실을 봅니다. 일련번호가 있으면 호스트별 최댓값과 최솟값의 차이에 1을 더한 값을 문서 수와 비교합니다.
- 중복은 호스트, 이벤트 시각, 요청을 묶어 건수를 세는 집계로 찾고, 시간 어긋남은 수집 시각 필드에서 이벤트 시각 필드를 뺀 값의 범위로 찾습니다. Logstash에서는 파이프라인에서 수집 시각 필드를 직접 추가해 이 값을 계산할 수 있게 합니다. 이 글에서는 실행하지 않았으므로 쿼리는 방향만 적었습니다.
- 중복 판정을 줄 전체 비교에서
호스트 + 이벤트 시각 + 요청해시 비교로 바꿉니다. 수집 시각만 다른 재전송 줄도 중복으로 잡을 수 있습니다. - 모든 소스가 같은 필드 이름을 쓰는지 확인하는 일관성 점검을 추가합니다. 예를 들어 출발지 IP가 어떤 소스에서는
src_ip, 다른 소스에서는source.ip로 들어오면 소스를 가로지르는 검색이 한쪽만 찾습니다.
확인 범위
위 두 스크립트는 Python 3.12.10에서 직접 실행했고, 기대 출력은 실제 실행 결과입니다. Elasticsearch, Logstash 같은 SIEM 구성요소에서의 점검은 이 글에서 실행하지 않았습니다.