SIEM

# SIEM 스타일 로그 분석 및 위협 탐지 - Python
import json
from datetime import datetime, timedelta
from collections import defaultdict
from typing import List, Dict, Any
import hashlib

class SimpleSIEM:
    """간단한 SIEM 상관관계 분석 엔진"""

    def __init__(self):
        self.events = []
        self.alerts = []
        self.rules = []
        # 사용자별 행동 기록 (UEBA 시뮬레이션)
        self.user_baselines = defaultdict(lambda: {
            'login_hours': set(),
            'source_ips': set(),
            'accessed_resources': set(),
        })

    def ingest_log(self, log: Dict[str, Any]):
        """로그 수집 및 정규화"""
        normalized = {
            'timestamp': datetime.fromisoformat(log.get('timestamp', datetime.now().isoformat())),
            'source': log.get('source', 'unknown'),
            'event_type': log.get('event_type', 'unknown'),
            'user': log.get('user'),
            'source_ip': log.get('source_ip'),
            'destination_ip': log.get('destination_ip'),
            'action': log.get('action'),
            'status': log.get('status'),
            'raw': json.dumps(log),
        }

        self.events.append(normalized)
        self._update_baseline(normalized)
        self._evaluate_rules(normalized)

        return normalized

    def _update_baseline(self, event: Dict):
        """사용자 행동 베이스라인 업데이트 (UEBA)"""
        if not event.get('user'):
            return

        user = event['user']
        baseline = self.user_baselines[user]

        # 로그인 시간대 기록
        if event['event_type'] == 'authentication':
            hour = event['timestamp'].hour
            baseline['login_hours'].add(hour)

        # 소스 IP 기록
        if event.get('source_ip'):
            baseline['source_ips'].add(event['source_ip'])

    def add_correlation_rule(self, rule):
        """상관관계 규칙 추가"""
        self.rules.append(rule)

    def _evaluate_rules(self, event: Dict):
        """모든 규칙 평가"""
        for rule in self.rules:
            if alert := rule(event, self.events, self.user_baselines):
                self.alerts.append(alert)
                self._notify_alert(alert)

    def _notify_alert(self, alert: Dict):
        """알림 발송 (실제로는 Slack, PagerDuty 등 연동)"""
        severity_emoji = {'critical': '🔴', 'high': '🟠', 'medium': '🟡', 'low': '🟢'}
        print(f"\n{severity_emoji.get(alert['severity'], '⚪')} ALERT: {alert['title']}")
        print(f"   Severity: {alert['severity']}")
        print(f"   Description: {alert['description']}")
        print(f"   User: {alert.get('user', 'N/A')}")
        print(f"   Source IP: {alert.get('source_ip', 'N/A')}")


# 상관관계 규칙 정의

def brute_force_rule(event, events, baselines):
    """브루트포스 공격 탐지: 5분 내 동일 IP에서 로그인 실패 5회 이상"""
    if event['event_type'] != 'authentication' or event['status'] != 'failed':
        return None

    source_ip = event.get('source_ip')
    if not source_ip:
        return None

    # 최근 5분 내 같은 IP의 로그인 실패 횟수
    window = timedelta(minutes=5)
    cutoff = event['timestamp'] - window

    failures = [
        e for e in events
        if e['event_type'] == 'authentication'
        and e['status'] == 'failed'
        and e.get('source_ip') == source_ip
        and e['timestamp'] > cutoff
    ]

    if len(failures) >= 5:
        return {
            'title': 'Brute Force Attack Detected',
            'severity': 'high',
            'description': f'{len(failures)} failed login attempts from {source_ip} in 5 minutes',
            'source_ip': source_ip,
            'rule': 'brute_force',
            'timestamp': event['timestamp'],
        }
    return None


def impossible_travel_rule(event, events, baselines):
    """불가능한 이동 탐지: 지리적으로 불가능한 시간 내 로그인"""
    if event['event_type'] != 'authentication' or event['status'] != 'success':
        return None

    user = event.get('user')
    source_ip = event.get('source_ip')
    if not user or not source_ip:
        return None

    # 최근 1시간 내 같은 사용자의 성공한 로그인
    window = timedelta(hours=1)
    cutoff = event['timestamp'] - window

    recent_logins = [
        e for e in events
        if e['event_type'] == 'authentication'
        and e['status'] == 'success'
        and e.get('user') == user
        and e['timestamp'] > cutoff
        and e.get('source_ip') != source_ip
    ]

    # 실제로는 GeoIP로 거리 계산, 여기서는 다른 IP면 의심
    if recent_logins:
        return {
            'title': 'Impossible Travel Detected',
            'severity': 'critical',
            'description': f'User {user} logged in from multiple IPs within 1 hour',
            'user': user,
            'source_ip': source_ip,
            'previous_ips': [e['source_ip'] for e in recent_logins],
            'rule': 'impossible_travel',
            'timestamp': event['timestamp'],
        }
    return None


def anomalous_hour_rule(event, events, baselines):
    """비정상 시간대 접속 탐지 (UEBA)"""
    if event['event_type'] != 'authentication' or event['status'] != 'success':
        return None

    user = event.get('user')
    if not user:
        return None

    baseline = baselines[user]
    current_hour = event['timestamp'].hour

    # 사용자의 정상 로그인 시간대가 10개 이상 기록되어 있고
    # 현재 시간이 정상 범위를 벗어나면 알림
    if len(baseline['login_hours']) >= 10 and current_hour not in baseline['login_hours']:
        return {
            'title': 'Anomalous Login Time Detected',
            'severity': 'medium',
            'description': f'User {user} logged in at unusual hour ({current_hour}:00)',
            'user': user,
            'normal_hours': sorted(baseline['login_hours']),
            'rule': 'anomalous_hour',
            'timestamp': event['timestamp'],
        }
    return None


# 사용 예시
if __name__ == '__main__':
    siem = SimpleSIEM()

    # 상관관계 규칙 등록
    siem.add_correlation_rule(brute_force_rule)
    siem.add_correlation_rule(impossible_travel_rule)
    siem.add_correlation_rule(anomalous_hour_rule)

    # 브루트포스 공격 시뮬레이션
    print("=== Simulating Brute Force Attack ===")
    for i in range(6):
        siem.ingest_log({
            'timestamp': (datetime.now() + timedelta(seconds=i*30)).isoformat(),
            'source': 'auth-server',
            'event_type': 'authentication',
            'user': f'user_{i % 3}',
            'source_ip': '192.168.1.100',
            'status': 'failed',
        })

    # 불가능한 이동 시뮬레이션
    print("\n=== Simulating Impossible Travel ===")
    siem.ingest_log({
        'timestamp': datetime.now().isoformat(),
        'source': 'auth-server',
        'event_type': 'authentication',
        'user': 'admin',
        'source_ip': '1.1.1.1',  # 한국
        'status': 'success',
    })
    siem.ingest_log({
        'timestamp': (datetime.now() + timedelta(minutes=30)).isoformat(),
        'source': 'auth-server',
        'event_type': 'authentication',
        'user': 'admin',
        'source_ip': '8.8.8.8',  # 미국 (30분 만에 이동 불가)
        'status': 'success',
    })

    print(f"\n=== Summary ===")
    print(f"Total events: {len(siem.events)}")
    print(f"Total alerts: {len(siem.alerts)}")