Elasticsearch로 대용량 로그 저장하고 검색하기 - Python SIEM 만들기 (4편)

Minseok Jeon·2025년 11월 14일
SIEMelasticsearchelkstackfilebeatkibana검색엔진데이터엔지니어링로그분석빅데이터포트폴리오

SIEM? 직접 만들어봤습니다

목록 보기
4/5
post-thumbnail

Elasticsearch Cluster

이 글은 "Python으로 나만의 SIEM 만들기" 시리즈의 4편입니다.

들어가며

"하루 1억 건의 로그를 어떻게 저장하고 검색할까요?"

일반적인 데이터베이스로는 불가능합니다.

  • MySQL: 1억 건 Full Scan → 30분 이상
  • PostgreSQL: 인덱스 있어도 수십 초
  • MongoDB: 샤딩 필요, 복잡한 운영

Elasticsearch는 다릅니다.

  • 1억 건 검색 → 1초 이내
  • 자동 샤딩 및 복제
  • RESTful API로 간편한 쿼리

실제 사례:

  • Uber: 하루 수조 건 로그 처리
  • Netflix: 100TB+ 로그 저장
  • GitHub: 코드 검색 엔진 (수억 줄)

이번 글에서는 Elasticsearch를 활용해 대용량 보안 로그를 효율적으로 저장하고 검색하는 방법을 다룹니다.

Elasticsearch 기본 개념

1. Elasticsearch란?

분산 검색 및 분석 엔진 (Distributed Search and Analytics Engine)

  • Apache Lucene 기반
  • RESTful API
  • JSON 형식 데이터
  • Near Real-Time (NRT) 검색

2. 핵심 용어

┌─────────────────────────────────────────────────┐
│              Elasticsearch Cluster              │
├─────────────────────────────────────────────────┤
│                                                 │
│  ┌───────────────────────────────────────┐     │
│  │         Index (인덱스)                 │     │  ← MySQL의 Database
│  │  "siem-logs-2025.11.11"               │     │
│  ├───────────────────────────────────────┤     │
│  │                                       │     │
│  │  ┌─────────────────────────────┐     │     │
│  │  │     Shard 0 (Primary)       │     │     │  ← 데이터 분할 단위
│  │  ├─────────────────────────────┤     │     │
│  │  │  Document 1 (로그 이벤트 1)  │     │     │  ← MySQL의 Row
│  │  │  Document 2 (로그 이벤트 2)  │     │     │
│  │  │  Document 3 (로그 이벤트 3)  │     │     │
│  │  └─────────────────────────────┘     │     │
│  │                                       │     │
│  │  ┌─────────────────────────────┐     │     │
│  │  │     Shard 1 (Primary)       │     │     │
│  │  ├─────────────────────────────┤     │     │
│  │  │  Document 4                 │     │     │
│  │  │  Document 5                 │     │     │
│  │  └─────────────────────────────┘     │     │
│  │                                       │     │
│  └───────────────────────────────────────┘     │
│                                                 │
└─────────────────────────────────────────────────┘

용어 비교:

ElasticsearchMySQL설명
ClusterDatabase Server여러 노드의 집합
NodeServer Instance단일 서버 프로세스
IndexDatabase데이터 저장 단위
DocumentRow하나의 JSON 데이터
FieldColumnJSON의 키
MappingSchema필드 타입 정의
ShardPartition데이터 분할

3. 왜 Elasticsearch가 빠른가?

역인덱스 (Inverted Index)

일반 인덱스 (Forward Index):

Document ID → 내용

Doc 1: "Brute force attack detected"
Doc 2: "SQL injection attempt"
Doc 3: "Brute force from IP 192.168.1.100"

검색: "Brute force"를 찾으려면?
→ 모든 문서를 순회 (O(n)) 😢

역인덱스 (Inverted Index):

단어 → Document ID

"brute"    → [Doc 1, Doc 3]
"force"    → [Doc 1, Doc 3]
"attack"   → [Doc 1]
"sql"      → [Doc 2]
"injection"→ [Doc 2]
"ip"       → [Doc 3]

검색: "Brute force"를 찾으려면?
→ 단어 목록에서 즉시 찾기 (O(1)) 🚀

분산 처리 (Distributed Processing)

1억 건 로그 검색

단일 서버:
└─ 1억 건 검색 → 30분

3대 클러스터 (샤딩):
├─ Node 1: 3,333만 건 검색 → 10분
├─ Node 2: 3,333만 건 검색 → 10분
└─ Node 3: 3,334만 건 검색 → 10분
   병렬 실행 → 총 10분 (3배 빠름!)

로그 수집 파이프라인

전체 흐름

┌──────────────────┐
│  FastAPI Server  │
│  (로그 생성)      │
└────────┬─────────┘
         │
         ▼ (파일 쓰기)
┌──────────────────┐
│  /app/logs/      │
│  app.log         │
└────────┬─────────┘
         │
         ▼ (파일 감시)
┌──────────────────────────────────────┐
│  Filebeat (로그 수집기)               │
│  ┌──────────────────────────────┐   │
│  │ 1. 파일 읽기                  │   │
│  │ 2. Dissect 프로세서 (파싱)    │   │
│  │ 3. 타임스탬프 변환             │   │
│  │ 4. 필드 타입 변환              │   │
│  └──────────────────────────────┘   │
└────────┬─────────────────────────────┘
         │
         ▼ (HTTP 전송)
┌──────────────────────────────────────┐
│  Elasticsearch (로그 저장소)         │
│  ┌──────────────────────────────┐   │
│  │ 인덱스: siem-logs-YYYY.MM.DD │   │
│  │ - 역인덱스 생성               │   │
│  │ - 샤드에 분산 저장            │   │
│  └──────────────────────────────┘   │
└────────┬─────────────────────────────┘
         │
         ▼ (쿼리)
┌──────────────────┐
│  Kibana          │
│  (시각화)         │
└──────────────────┘

Filebeat 설정 상세

filebeat.yml 전체 구조

# 1. 입력 설정 (어디서 로그를 읽을까?)
filebeat.inputs:
  - type: log
    enabled: true
    paths:
      - /var/log/mini_siem/*.log

    # 2. 프로세서 (로그를 어떻게 파싱할까?)
    processors:
      - dissect:
          tokenizer: "%{timestamp} [%{log_level}] [EVENT] %{event_type} | IP=%{source_ip} | Severity=%{severity} | Threat=%{is_threat}"
          field: "message"
          target_prefix: "siem"
          ignore_failure: true

      - timestamp:
          field: siem.timestamp
          layouts:
            - '2006-01-02 15:04:05,000'
          ignore_failure: true

      - convert:
          fields:
            - {from: "siem.is_threat", type: "boolean"}
          ignore_failure: true

# 3. 출력 설정 (어디로 보낼까?)
output.elasticsearch:
  hosts: ["elasticsearch:9200"]
  index: "siem-logs-%{+yyyy.MM.dd}"

# 4. 인덱스 템플릿 설정
setup.ilm.enabled: false
setup.template.name: "siem-logs"
setup.template.pattern: "siem-logs-*"

Dissect 프로세서 상세 분석

원본 로그

2025-11-11 10:30:00,123 [INFO] [EVENT] login_failed | IP=192.168.1.100 | Severity=medium | Threat=True

Dissect 토크나이저

tokenizer: "%{timestamp} [%{log_level}] [EVENT] %{event_type} | IP=%{source_ip} | Severity=%{severity} | Threat=%{is_threat}"

토크나이저 분석:

%{timestamp}       → "2025-11-11 10:30:00,123"
[%{log_level}]     → "[INFO]" → "INFO"
[EVENT]            → 리터럴 (매칭만)
%{event_type}      → "login_failed"
IP=%{source_ip}    → "IP=192.168.1.100" → "192.168.1.100"
Severity=%{severity} → "Severity=medium" → "medium"
Threat=%{is_threat} → "Threat=True" → "True"

파싱 결과

{
  "message": "2025-11-11 10:30:00,123 [INFO] [EVENT] login_failed | IP=192.168.1.100 | Severity=medium | Threat=True",
  "siem": {
    "timestamp": "2025-11-11 10:30:00,123",
    "log_level": "INFO",
    "event_type": "login_failed",
    "source_ip": "192.168.1.100",
    "severity": "medium",
    "is_threat": "True"
  }
}

프로세서 체인

1. Dissect (파싱)

- dissect:
    tokenizer: "..."
    field: "message"           # 입력 필드
    target_prefix: "siem"      # 출력 필드 접두사
    ignore_failure: true       # 파싱 실패 시 무시

ignore_failure의 중요성:

로그 형식 A: "2025-11-11 [INFO] [EVENT] ..."  ✅ 파싱 성공
로그 형식 B: "2025-11-11 [WARNING] THREAT..."  ❌ 파싱 실패

ignore_failure: true  → 형식 B도 계속 처리 (다음 프로세서로)
ignore_failure: false → 형식 B에서 중단 (로그 유실!)

여러 패턴 처리:

processors:
  # 패턴 1: [EVENT] 형식
  - dissect:
      tokenizer: "%{timestamp} [%{log_level}] [EVENT] ..."
      ignore_failure: true

  # 패턴 2: THREAT DETECTED 형식
  - dissect:
      tokenizer: "%{timestamp} [%{log_level}] %{?emoji} THREAT DETECTED: %{threat_details}"
      ignore_failure: true

→ 두 패턴 모두 시도, 하나만 성공하면 OK!

2. Timestamp (타임스탬프 변환)

- timestamp:
    field: siem.timestamp            # 소스 필드
    layouts:
      - '2006-01-02 15:04:05,000'    # Go 시간 형식
    ignore_failure: true

Go 시간 형식 해석:

2006-01-02 15:04:05,000
│    │  │  │  │  │  └─ 밀리초 (000)
│    │  │  │  │  └─── 초 (05)
│    │  │  │  └────── 분 (04)
│    │  │  └───────── 시 (15 = 3PM)
│    │  └──────────── 일 (02)
│    └─────────────── 월 (01)
└──────────────────── 년 (2006)

변환 전후:

// 변환 전
{
  "siem": {
    "timestamp": "2025-11-11 10:30:00,123"  // 문자열
  }
}

// 변환 후
{
  "@timestamp": "2025-11-11T10:30:00.123Z",  // ISO 8601 형식
  "siem": {
    "timestamp": "2025-11-11 10:30:00,123"
  }
}

3. Convert (타입 변환)

- convert:
    fields:
      - {from: "siem.is_threat", type: "boolean"}
    ignore_failure: true

변환 규칙:

"True"  → true   (boolean)
"true"  → true
"1"     → true
"False" → false
"false" → false
"0"     → false

타입 변환의 중요성:

// ❌ 타입 변환 안 함
{
  "siem": {
    "is_threat": "True"  // 문자열!
  }
}

// Elasticsearch 쿼리
GET /siem-logs/_search
{
  "query": {
    "term": {
      "siem.is_threat": true  // 매칭 실패! (문자열 vs 불린)
    }
  }
}

// ✅ 타입 변환 함
{
  "siem": {
    "is_threat": true  // 불린!
  }
}

// Elasticsearch 쿼리
GET /siem-logs/_search
{
  "query": {
    "term": {
      "siem.is_threat": true  // 매칭 성공!
    }
  }
}

Elasticsearch 인덱스 설계

일별 인덱스 전략

인덱스 명명 규칙:

siem-logs-2025.11.11
siem-logs-2025.11.12
siem-logs-2025.11.13
...

장점:

  1. 빠른 삭제

    # 30일 이전 로그 삭제
DELETE /siem-logs-2025.10.12
# 인덱스 전체 삭제 (수 초 내 완료!)

# vs. 일반 DB
DELETE FROM logs WHERE date < '2025-10-12';
# 수백만 건 삭제 (수십 분 소요)

  2. 시간 기반 검색 최적화

    # 특정 날짜만 검색
GET /siem-logs-2025.11.11/_search
# 해당 날짜 데이터만 검색 (빠름!)

# 범위 검색
GET /siem-logs-2025.11.*/_search
# 2025년 11월 전체 검색

  3. 샤드 크기 관리

    단일 인덱스 (1년치):
└─ 10TB → 샤드 크기 초과 → 성능 저하

일별 인덱스:
├─ 2025.11.11: 30GB ✅
├─ 2025.11.12: 28GB ✅
└─ 2025.11.13: 32GB ✅

매핑 (Mapping) 설계

자동 매핑 vs 명시적 매핑:

// ❌ 자동 매핑 (권장하지 않음)
// Elasticsearch가 첫 데이터로 타입 추론
{
  "siem": {
    "source_ip": "192.168.1.100"  // → text (검색용)
  }
}
// 문제: IP는 keyword여야 함!

// ✅ 명시적 매핑 (권장)
PUT /siem-logs-2025.11.11
{
  "mappings": {
    "properties": {
      "@timestamp": {
        "type": "date"
      },
      "siem": {
        "properties": {
          "timestamp": {
            "type": "date"
          },
          "log_level": {
            "type": "keyword"  // 정확한 매칭
          },
          "event_type": {
            "type": "keyword"
          },
          "source_ip": {
            "type": "ip"  // IP 전용 타입
          },
          "severity": {
            "type": "keyword"
          },
          "is_threat": {
            "type": "boolean"
          },
          "threat_details": {
            "type": "text",  // 전문 검색
            "fields": {
              "keyword": {  // 정렬/집계용
                "type": "keyword"
              }
            }
          }
        }
      }
    }
  }
}

필드 타입 상세

Elasticsearch 타입설명예시검색 방법
keyword정확한 매칭"login_failed"term 쿼리
text전문 검색"Brute force attack"match 쿼리
date날짜/시간"2025-11-11T10:30:00Z"range 쿼리
booleantrue/falsetrueterm 쿼리
ipIP 주소"192.168.1.100"CIDR 쿼리
integer정수5range 쿼리
float실수3.14range 쿼리

keyword vs text 차이:

// keyword (정확한 매칭)
{
  "event_type": "login_failed"
}

GET /_search
{
  "query": {
    "term": {
      "event_type": "login_failed"  // ✅ 매칭
    }
  }
}

{
  "query": {
    "term": {
      "event_type": "login"  // ❌ 매칭 안 됨 (부분 매칭 불가)
    }
  }
}

// text (전문 검색)
{
  "threat_details": "Brute force attack detected from IP"
}

// 자동으로 토큰화됨:
// ["brute", "force", "attack", "detected", "from", "ip"]

GET /_search
{
  "query": {
    "match": {
      "threat_details": "brute"  // ✅ 매칭
    }
  }
}

{
  "query": {
    "match": {
      "threat_details": "attack"  // ✅ 매칭 (부분 매칭 가능!)
    }
  }
}

인덱스 템플릿

자동 매핑 적용:

PUT /_index_template/siem-logs-template
{
  "index_patterns": ["siem-logs-*"],  // 패턴 매칭
  "template": {
    "settings": {
      "number_of_shards": 1,     // 샤드 수 (노드 수에 따라 조정)
      "number_of_replicas": 1,   // 복제본 수 (가용성)
      "refresh_interval": "5s"   // 검색 가능 시점 (실시간성)
    },
    "mappings": {
      "properties": {
        "@timestamp": {"type": "date"},
        "siem": {
          "properties": {
            "timestamp": {"type": "date"},
            "log_level": {"type": "keyword"},
            "event_type": {"type": "keyword"},
            "source_ip": {"type": "ip"},
            "severity": {"type": "keyword"},
            "is_threat": {"type": "boolean"},
            "threat_details": {
              "type": "text",
              "fields": {
                "keyword": {"type": "keyword"}
              }
            }
          }
        }
      }
    }
  }
}

효과:

# 새 인덱스 자동 생성 시 템플릿 적용
POST /siem-logs-2025.11.14/_doc
{
  "siem": {
    "source_ip": "192.168.1.100"
  }
}

# 자동으로 ip 타입으로 매핑됨! ✅

Elasticsearch 쿼리 (Query DSL)

1. 기본 검색

// 모든 위협 로그 조회
GET /siem-logs-*/_search
{
  "query": {
    "term": {
      "siem.is_threat": true
    }
  }
}

2. 복합 조건 (Bool Query)

// Critical 위협 중 특정 IP만
GET /siem-logs-*/_search
{
  "query": {
    "bool": {
      "must": [                          // AND 조건
        {"term": {"siem.severity": "critical"}},
        {"term": {"siem.is_threat": true}}
      ],
      "filter": [                        // 필터 (스코어 계산 안 함)
        {"term": {"siem.source_ip": "192.168.1.100"}}
      ]
    }
  }
}

bool 쿼리 조건:

조건의미스코어 영향
mustAND (반드시 매칭)✅ 영향
must_notNOT (매칭 안 됨)❌ 영향 없음
shouldOR (하나라도 매칭)✅ 영향
filterAND (반드시 매칭)❌ 영향 없음

3. 시간 범위 검색

// 최근 1시간 위협
GET /siem-logs-*/_search
{
  "query": {
    "bool": {
      "must": [
        {"term": {"siem.is_threat": true}}
      ],
      "filter": [
        {
          "range": {
            "@timestamp": {
              "gte": "now-1h",  // Greater Than or Equal
              "lte": "now"      // Less Than or Equal
            }
          }
        }
      ]
    }
  }
}

// 특정 날짜 범위
{
  "range": {
    "@timestamp": {
      "gte": "2025-11-01T00:00:00",
      "lte": "2025-11-30T23:59:59",
      "format": "yyyy-MM-dd'T'HH:mm:ss"
    }
  }
}

4. IP 범위 검색 (CIDR)

// 192.168.1.0/24 네트워크에서 발생한 공격
GET /siem-logs-*/_search
{
  "query": {
    "bool": {
      "must": [
        {"term": {"siem.is_threat": true}}
      ],
      "filter": [
        {
          "term": {
            "siem.source_ip": "192.168.1.0/24"
          }
        }
      ]
    }
  }
}

5. 전문 검색 (Full-Text Search)

// "SQL Injection" 포함된 위협 찾기
GET /siem-logs-*/_search
{
  "query": {
    "match": {
      "siem.threat_details": "SQL Injection"
    }
  }
}

// 여러 단어 모두 포함 (AND)
{
  "query": {
    "match": {
      "siem.threat_details": {
        "query": "brute force attack",
        "operator": "and"
      }
    }
  }
}

// 정규식 검색
{
  "query": {
    "regexp": {
      "siem.threat_details": ".*injection.*"
    }
  }
}

6. 집계 (Aggregation)

// 이벤트 타입별 통계
GET /siem-logs-*/_search
{
  "size": 0,  // 문서는 반환하지 않고 집계만
  "aggs": {
    "by_event_type": {
      "terms": {
        "field": "siem.event_type",
        "size": 10
      }
    }
  }
}

// 응답:
{
  "aggregations": {
    "by_event_type": {
      "buckets": [
        {"key": "login_failed", "doc_count": 1523},
        {"key": "sql_injection", "doc_count": 234},
        {"key": "privilege_escalation", "doc_count": 89}
      ]
    }
  }
}

심화 집계:

// 심각도별 + 이벤트 타입별 (중첩 집계)
GET /siem-logs-*/_search
{
  "size": 0,
  "aggs": {
    "by_severity": {
      "terms": {
        "field": "siem.severity"
      },
      "aggs": {
        "by_event_type": {
          "terms": {
            "field": "siem.event_type"
          }
        }
      }
    }
  }
}

// 시간별 추세 (히스토그램)
{
  "aggs": {
    "threats_over_time": {
      "date_histogram": {
        "field": "@timestamp",
        "calendar_interval": "1h"  // 1시간 단위
      },
      "aggs": {
        "threat_count": {
          "filter": {
            "term": {"siem.is_threat": true}
          }
        }
      }
    }
  }
}

7. 상위 N개 조회

// 상위 10개 공격 IP
GET /siem-logs-*/_search
{
  "size": 0,
  "query": {
    "term": {"siem.is_threat": true}
  },
  "aggs": {
    "top_attack_ips": {
      "terms": {
        "field": "siem.source_ip",
        "size": 10,
        "order": {"_count": "desc"}
      }
    }
  }
}

성능 최적화

1. 샤드 설계

샤드 수 결정:

적정 샤드 크기: 20-50GB

일일 로그 량: 100GB
→ number_of_shards: 3 (각 샤드 ~33GB)

일일 로그 량: 10GB
→ number_of_shards: 1 (단일 샤드로 충분)

과다 샤딩의 문제:

❌ number_of_shards: 100 (10GB 인덱스)
→ 각 샤드: 100MB
→ 오버헤드 증가, 성능 저하

✅ number_of_shards: 1 (10GB 인덱스)
→ 단일 샤드: 10GB
→ 효율적

2. 복제본 (Replica)

{
  "settings": {
    "number_of_replicas": 1  // 프로덕션 권장
  }
}

복제본 효과:

  1. 가용성: 노드 장애 시에도 서비스 지속
  2. 검색 성능: 복제본도 검색에 참여 (부하 분산)

단점:

  • 저장 공간 2배 소비

3. 리프레시 간격

{
  "settings": {
    "refresh_interval": "5s"  // 기본값: 1s
  }
}

refresh_interval 의미:

1초마다 refresh → 새 데이터가 검색 가능해짐

refresh_interval: 1s  → 1초 대기 (실시간성 높음)
refresh_interval: 5s  → 5초 대기 (색인 성능 5배 향상)
refresh_interval: -1  → 자동 refresh 비활성화 (대량 색인 시)

대량 색인 시 최적화:

# 1. refresh 비활성화
PUT /siem-logs-2025.11.11/_settings
{
  "refresh_interval": "-1"
}

# 2. 대량 데이터 색인
POST /_bulk
...

# 3. 수동 refresh
POST /siem-logs-2025.11.11/_refresh

# 4. refresh 재활성화
PUT /siem-logs-2025.11.11/_settings
{
  "refresh_interval": "5s"
}

4. 벌크 API (Bulk API)

// ❌ 나쁜 예: 개별 색인 (느림)
POST /siem-logs-2025.11.11/_doc
{"siem": {"event_type": "login_failed", ...}}

POST /siem-logs-2025.11.11/_doc
{"siem": {"event_type": "sql_injection", ...}}
// 각 요청마다 HTTP 오버헤드 발생

// ✅ 좋은 예: 벌크 색인 (빠름)
POST /_bulk
{"index": {"_index": "siem-logs-2025.11.11"}}
{"siem": {"event_type": "login_failed", ...}}
{"index": {"_index": "siem-logs-2025.11.11"}}
{"siem": {"event_type": "sql_injection", ...}}
// 한 번의 HTTP 요청으로 다수 문서 색인

// 성능: 100배 이상 빠름!

5. 필드 데이터 캐싱

{
  "mappings": {
    "properties": {
      "siem.source_ip": {
        "type": "ip",
        "eager_global_ordinals": true  // 집계 성능 향상
      }
    }
  }
}

Kibana 대시보드 구성

1. Index Pattern 생성

Management → Index Patterns → Create Index Pattern

Step 1: Index pattern name
  siem-logs-*

Step 2: Time field
  @timestamp

→ Create index pattern

2. Discover (로그 탐색)

필터 추가:

siem.is_threat: true
siem.severity: critical
siem.source_ip: 192.168.1.100

시간 범위 선택:

Last 15 minutes
Last 1 hour
Last 24 hours
Last 7 days
Custom (절대 시간)

3. Visualize (시각화)

1) Line Chart - 시간별 위협 추이

Visualization Type: Line
Metrics:
  Y-axis: Count
Buckets:
  X-axis: Date Histogram
    Field: @timestamp
    Interval: 1 hour
  Split Series:
    Field: siem.severity

2) Pie Chart - 이벤트 타입별 분포

Visualization Type: Pie
Metrics:
  Slice Size: Count
Buckets:
  Split Slices:
    Aggregation: Terms
    Field: siem.event_type
    Size: 10

3) Data Table - 상위 공격 IP

Visualization Type: Data Table
Metrics:
  Count
Buckets:
  Split Rows:
    Aggregation: Terms
    Field: siem.source_ip
    Order By: metric: Count
    Order: Descending
    Size: 10

4) Heatmap - 시간대별 공격 분포

Visualization Type: Heatmap
Metrics:
  Count
Buckets:
  X-axis:
    Aggregation: Date Histogram
    Field: @timestamp
    Interval: 1 hour
  Y-axis:
    Aggregation: Terms
    Field: siem.event_type

4. Dashboard 생성

Dashboard → Create Dashboard → Add Visualizations

레이아웃:
┌─────────────────────────────────────────┐
│  시간별 위협 추이 (Line Chart)           │
├──────────────────┬──────────────────────┤
│ 이벤트 타입 분포  │  상위 공격 IP         │
│ (Pie Chart)      │  (Data Table)        │
├──────────────────┴──────────────────────┤
│  시간대별 공격 분포 (Heatmap)            │
└─────────────────────────────────────────┘

5. Alert 설정 (Kibana Alerting)

Stack Management → Alerting → Create Rule

Rule Type: Elasticsearch Query
Index: siem-logs-*
Query:
  {
    "query": {
      "bool": {
        "must": [
          {"term": {"siem.is_threat": true}},
          {"term": {"siem.severity": "critical"}}
        ],
        "filter": [
          {"range": {"@timestamp": {"gte": "now-5m"}}}
        ]
      }
    }
  }

Threshold: count > 0
Action: Send Email / Slack / Webhook

실전 예제

Python에서 Elasticsearch 조회

from elasticsearch import Elasticsearch

# Elasticsearch 클라이언트 생성
es = Elasticsearch(
    ["http://localhost:9200"],
    basic_auth=("elastic", "password")
)

# 1. 최근 1시간 Critical 위협 조회
response = es.search(
    index="siem-logs-*",
    body={
        "query": {
            "bool": {
                "must": [
                    {"term": {"siem.is_threat": True}},
                    {"term": {"siem.severity": "critical"}}
                ],
                "filter": [
                    {"range": {"@timestamp": {"gte": "now-1h"}}}
                ]
            }
        },
        "sort": [
            {"@timestamp": {"order": "desc"}}
        ],
        "size": 100
    }
)

# 결과 출력
for hit in response['hits']['hits']:
    log = hit['_source']
    print(f"[{log['@timestamp']}] {log['siem']['event_type']} from {log['siem']['source_ip']}")

# 2. 상위 10개 공격 IP 집계
agg_response = es.search(
    index="siem-logs-*",
    body={
        "size": 0,
        "query": {
            "term": {"siem.is_threat": True}
        },
        "aggs": {
            "top_ips": {
                "terms": {
                    "field": "siem.source_ip",
                    "size": 10
                }
            }
        }
    }
)

for bucket in agg_response['aggregations']['top_ips']['buckets']:
    print(f"IP: {bucket['key']}, Count: {bucket['doc_count']}")

마치며

핵심 요약

  1. Elasticsearch = 속도 + 확장성

    • 역인덱스로 1초 내 검색
    • 샤딩으로 페타바이트 저장

  2. Filebeat = 안정적 로그 수집

    • Dissect 프로세서로 파싱
    • At-least-once 보장

  3. Kibana = 강력한 시각화

    • 드래그 앤 드롭 대시보드
    • 실시간 알림

다음 편 예고

5편: 보안 설계 원칙을 코드로 구현하기

  • Defense in Depth 실전 적용
  • Fail-Safe Defaults 예제
  • Least Privilege 구현
  • OWASP Top 10 대응 코드

참고 자료

프로젝트 정보

  • GitHub: mini-siem-log-monitoring
  • 시리즈: Python SIEM 만들기 (4/5편)
  • 코드 위치: filebeat/filebeat.yml, docker-compose.yml

질문이나 피드백은 댓글로 남겨주세요!

💡 도움이 되셨다면 GitHub Star와 좋아요 부탁드립니다!
💬 다음 편에서 만나요!

profile
Minseok Jeon
이전 포스트

FastAPI로 실시간 보안 이벤트 처리하기 - Python SIEM 만들기 (3편)

다음 포스트

보안 설계 원칙을 코드로 구현하기 - Python SIEM 만들기 (5편)

0개의 댓글