이상 패턴 탐지: 브루트포스

# /var/log/auth.log — SSH 브루트포스
Apr 27 10:23:11 server sshd[1234]: Failed password for invalid user admin from 198.51.100.99 port 22
Apr 27 10:23:13 server sshd[1235]: Failed password for invalid user root from 198.51.100.99 port 22
Apr 27 10:23:15 server sshd[1236]: Failed password for invalid user test from 198.51.100.99 port 22
... (수백~수천 라인)

# ⚠️ 이 코드는 허가된 환경에서만 사용하세요.
import re
from datetime import datetime, timedelta
from collections import defaultdict, deque

# auth.log SSH 실패 패턴
SSH_FAILED = re.compile(
    r'^(?P<month>\w{3})\s+(?P<day>\d+)\s+'
    r'(?P<time>\d{2}:\d{2}:\d{2})\s+'
    r'\S+\s+sshd\[\d+\]:\s+'
    r'Failed password\s+'
    r'(?:for\s+(?:invalid user\s+)?(?P<user>\S+)\s+)?'
    r'from\s+(?P<ip>\S+)\s+port\s+\d+'
)


def parse_auth_time(month: str, day: str, time_str: str, year: int = 2026) -> datetime:
    """auth.log 시간을 datetime으로 (연도는 추정)."""
    return datetime.strptime(f'{year} {month} {day} {time_str}', '%Y %b %d %H:%M:%S')


def detect_bruteforce(
    log_path: str,
    threshold: int = 10,        # 5분 안에 10회 실패
    window_minutes: int = 5,
) -> dict:
    """슬라이딩 윈도우 기반 브루트포스 탐지."""
    # IP별 실패 시각 큐
    failures: dict[str, deque] = defaultdict(deque)
    detected = {}  # IP → {'count': N, 'first': time, 'last': time, 'users': set()}
    window = timedelta(minutes=window_minutes)

    with open(log_path, encoding='utf-8', errors='replace') as f:
        for line in f:
            m = SSH_FAILED.search(line)
            if not m:
                continue

            try:
                ts = parse_auth_time(m['month'], m['day'], m['time'])
            except ValueError:
                continue

            ip = m['ip']
            user = m['user'] or 'unknown'

            # 윈도우 밖의 오래된 기록 제거
            q = failures[ip]
            while q and q[0][0] < ts - window:
                q.popleft()
            q.append((ts, user))

            # 임계값 초과 → 탐지
            if len(q) >= threshold:
                if ip not in detected:
                    detected[ip] = {
                        'count': len(q),
                        'first_seen': q[0][0].isoformat(),
                        'last_seen': ts.isoformat(),
                        'users_tried': set(),
                    }
                detected[ip]['count'] = len(q)
                detected[ip]['last_seen'] = ts.isoformat()
                detected[ip]['users_tried'].update(u for _, u in q)

    # set → list로 (JSON 직렬화)
    for d in detected.values():
        d['users_tried'] = sorted(d['users_tried'])[:20]  # 최대 20개

    return detected

# 탐지된 IP를 iptables로 차단 (sudo 권한 필요)
import subprocess

def block_ip(ip: str, reason: str = 'bruteforce') -> bool:
    """iptables로 IP 차단."""
    try:
        subprocess.run([
            'sudo', 'iptables', '-A', 'INPUT',
            '-s', ip, '-j', 'DROP',
            '-m', 'comment', '--comment', f'auto-block:{reason}',
        ], check=True, timeout=5)
        log.warning(f'⛔ 차단됨: {ip} ({reason})')
        return True
    except (subprocess.CalledProcessError, subprocess.TimeoutExpired) as e:
        log.error(f'차단 실패 {ip}: {e}')
        return False


def block_detected(detected: dict, min_count: int = 50):
    """50회 이상 실패한 IP만 차단 — 오탐 방지."""
    blocked = []
    for ip, info in detected.items():
        if info['count'] >= min_count:
            if block_ip(ip, f'bruteforce-{info["count"]}'):
                blocked.append(ip)
    return blocked

import ipaddress

# 신뢰 IP는 절대 차단 안 함
WHITELIST = [
    ipaddress.ip_network('192.168.0.0/16'),  # 사내망
    ipaddress.ip_network('127.0.0.0/8'),     # 로컬
    ipaddress.ip_address('203.0.113.10'),    # 본사 외부 IP
]


def is_whitelisted(ip_str: str) -> bool:
    try:
        ip = ipaddress.ip_address(ip_str)
        for entry in WHITELIST:
            if isinstance(entry, ipaddress.IPv4Network):
                if ip in entry:
                    return True
            elif ip == entry:
                return True
    except ValueError:
        return False
    return False

import time

def tail_and_detect(log_path: str, threshold: int = 10):
    """tail -f 처럼 실시간 모니터링."""
    failures: dict[str, deque] = defaultdict(deque)
    window = timedelta(minutes=5)

    with open(log_path, encoding='utf-8', errors='replace') as f:
        f.seek(0, 2)  # 끝으로 이동
        while True:
            line = f.readline()
            if not line:
                time.sleep(0.5)
                continue

            m = SSH_FAILED.search(line)
            if not m:
                continue

            ip = m['ip']
            if is_whitelisted(ip):
                continue

            try:
                ts = parse_auth_time(m['month'], m['day'], m['time'])
            except ValueError:
                continue

            q = failures[ip]
            while q and q[0] < ts - window:
                q.popleft()
            q.append(ts)

            if len(q) == threshold:  # 임계 도달 순간만
                log.warning(f'🚨 브루트포스 감지: {ip} ({len(q)}회/5분)')
                # block_ip(ip, 'bruteforce')