파일 I/O: 로그와 설정 파일 다루기

# ⚠️ 이 코드는 허가된 환경에서만 사용하세요.

# ❌ 메모리 폭발 — 1GB 로그 파일이면 메모리 OOM
with open('access.log') as f:
    lines = f.readlines()  # 전체 메모리 적재

# ✅ 한 줄씩 스트리밍
def scan_log(path: str, pattern: str):
    """대용량 로그를 메모리 효율적으로 스캔."""
    import re
    p = re.compile(pattern)
    with open(path, encoding='utf-8', errors='replace') as f:
        for lineno, line in enumerate(f, 1):
            if p.search(line):
                yield lineno, line.rstrip()

# 사용 예
for lineno, line in scan_log('access.log', r'\b401\b'):
    print(f'{lineno}: {line}')

import csv
from collections import Counter

# 침입 탐지 시스템(IDS) 로그가 CSV 형식일 때
def top_attackers(csv_path: str, top: int = 10):
    counter = Counter()
    with open(csv_path, encoding='utf-8') as f:
        reader = csv.DictReader(f)  # 헤더 자동 인식
        for row in reader:
            if row.get('action') == 'BLOCK':
                counter[row['source_ip']] += 1
    return counter.most_common(top)

# print(top_attackers('ids_log.csv'))

import json

# 보안 점검 설정을 외부 파일로 분리
SCAN_CONFIG = {
    'targets': ['127.0.0.1', '192.168.1.0/24'],
    'ports': [22, 80, 443, 3306, 8080],
    'timeout': 2.0,
    'rate_limit': 10,
}

# 저장
with open('scan_config.json', 'w', encoding='utf-8') as f:
    json.dump(SCAN_CONFIG, f, indent=2, ensure_ascii=False)

# 읽기 + 검증
with open('scan_config.json', encoding='utf-8') as f:
    cfg = json.load(f)
    assert isinstance(cfg.get('targets'), list), 'targets는 리스트여야 함'
    assert all(isinstance(p, int) for p in cfg.get('ports', [])), '포트는 정수'

import re
from collections import Counter

# Apache/Nginx 일반 형식
LOG_PATTERN = re.compile(
    r'^(?P<ip>\S+)\s+\S+\s+\S+\s+'
    r'\[(?P<time>[^\]]+)\]\s+'
    r'"(?P<request>[^"]+)"\s+'
    r'(?P<status>\d{3})'
)

def find_attackers(log_path: str, threshold: int = 50) -> dict:
    """status 401/403/404를 너무 자주 받는 IP를 탐지."""
    failures = Counter()
    with open(log_path, encoding='utf-8', errors='replace') as f:
        for line in f:
            m = LOG_PATTERN.search(line)
            if m and m.group('status') in ('401', '403', '404'):
                failures[m.group('ip')] += 1
    return {ip: cnt for ip, cnt in failures.items() if cnt >= threshold}

# 결과를 JSON 리포트로 저장
def save_report(attackers: dict, out_path: str):
    import json, datetime
    report = {
        'generated_at': datetime.datetime.now().isoformat(),
        'total_attackers': len(attackers),
        'attackers': sorted(attackers.items(), key=lambda x: -x[1]),
    }
    with open(out_path, 'w', encoding='utf-8') as f:
        json.dump(report, f, indent=2, ensure_ascii=False)