이상 패턴 탐지: 웹 공격

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

ATTACK_PATTERNS = {
    'sqli': re.compile(
        r'(?i)('
        r'\bUNION\s+(?:ALL\s+)?SELECT\b|'
        r'\bOR\s+1\s*=\s*1\b|'
        r'\bAND\s+1\s*=\s*1\b|'
        r"'\s*OR\s*'1'\s*=\s*'1|"
        r'\bSLEEP\s*\(\d+\)|'
        r'\bWAITFOR\s+DELAY\b|'
        r'--\s*$|'
        r"\\x27\\s*OR\\s*\\x271\\x27=\\x271"
        r')'
    ),
    'xss': re.compile(
        r'(?i)('
        r'<script[^>]*>|'
        r'javascript:[^\s]+|'
        r'on(?:click|load|error|mouseover|focus)\s*=|'
        r'<img[^>]*onerror|'
        r'<iframe[^>]*src\s*=|'
        r'\balert\s*\(|'
        r'document\.cookie|'
        r'eval\s*\('
        r')'
    ),
    'path_traversal': re.compile(
        r'(?i)('
        r'\.\.[\\/]|'
        r'%2e%2e[\\/]|'
        r'%252e%252e|'
        r'/etc/passwd|'
        r'\\windows\\system32|'
        r'C:\\Windows'
        r')'
    ),
    'command_injection': re.compile(
        r'(?i)('
        r';\s*(?:cat|ls|whoami|id|nc|wget|curl)\s|'
        r'\|\s*(?:cat|ls|whoami|id)|'
        r'\$\(.*\)|'
        r'`.*`|'
        r'&&\s*\w+'
        r')'
    ),
    'lfi_rfi': re.compile(
        r'(?i)('
        r'php://(?:filter|input)|'
        r'data://text/plain|'
        r'expect://|'
        r'file://|'
        r'(?:include|require)\s*=|'
        r'\?file=https?://'
        r')'
    ),
    'malicious_useragent': re.compile(
        r'(?i)('
        r'sqlmap|'
        r'nikto|'
        r'nmap|'
        r'masscan|'
        r'gobuster|'
        r'dirbuster|'
        r'burpsuite|'
        r'havij|'
        r'curl/[\d.]+\s+(?:scan|test)'
        r')'
    ),
}


def classify_request(line: str, parsed: dict | None = None) -> list[str]:
    """단일 로그 라인의 공격 종류 판별 — 다중 매칭 가능."""
    detected = []
    target = line  # 전체 라인에서 패턴 검색

    for attack, pattern in ATTACK_PATTERNS.items():
        if pattern.search(target):
            detected.append(attack)
    return detected

from urllib.parse import unquote_plus

def normalize_for_detection(url: str) -> str:
    """다중 인코딩까지 디코딩해서 정규식 매칭률 향상."""
    decoded = url
    for _ in range(3):  # %2527 같은 다중 인코딩
        new = unquote_plus(decoded)
        if new == decoded:
            break
        decoded = new
    return decoded.lower()


# 예: %2E%2E%2F → ../
print(normalize_for_detection('GET /api?file=%2E%2E%2Fetc%2Fpasswd HTTP/1.1'))
# 'get /api?file=../etc/passwd http/1.1'

from collections import Counter

APACHE_COMBINED = re.compile(
    r'^(?P<ip>\S+)\s+\S+\s+\S+\s+'
    r'\[(?P<time>[^\]]+)\]\s+'
    r'"(?P<method>\S+)\s+(?P<path>\S+)\s+\S+"\s+'
    r'(?P<status>\d{3})\s+\S+\s+'
    r'"(?P<referer>[^"]*)"\s+'
    r'"(?P<user_agent>[^"]*)"'
)


def scan_web_attacks(log_path: str) -> dict:
    """웹 로그에서 공격 시도 자동 분류."""
    findings = []
    attack_counter = Counter()
    ip_attacks = Counter()

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

            full_request = normalize_for_detection(
                f'{m["method"]} {m["path"]} {m["user_agent"]}'
            )
            attacks = classify_request(full_request)
            if attacks:
                findings.append({
                    'ip': m['ip'],
                    'time': m['time'],
                    'request': f'{m["method"]} {m["path"]}',
                    'user_agent': m['user_agent'][:100],
                    'attacks': attacks,
                    'status': int(m['status']),
                })
                for atk in attacks:
                    attack_counter[atk] += 1
                ip_attacks[m['ip']] += 1

    return {
        'total_findings': len(findings),
        'attack_distribution': dict(attack_counter),
        'top_attacker_ips': ip_attacks.most_common(10),
        'sample_findings': findings[:50],
    }

def severity_score(finding: dict) -> int:
    """공격의 심각도 점수 (0~100)."""
    score = 0
    weights = {
        'sqli': 30,
        'command_injection': 35,
        'lfi_rfi': 25,
        'xss': 15,
        'path_traversal': 20,
        'malicious_useragent': 10,
    }
    for atk in finding['attacks']:
        score += weights.get(atk, 5)

    # 200 응답 = 공격 성공 가능
    if finding['status'] == 200:
        score += 20

    # 404/403도 시도 자체로 의심
    if finding['status'] in (404, 403):
        score += 5

    return min(score, 100)


def filter_critical(findings: list[dict], min_score: int = 50) -> list[dict]:
    """심각도 50점 이상만 — 사람이 검토할 우선순위."""
    scored = [{'score': severity_score(f), **f} for f in findings]
    return [f for f in scored if f['score'] >= min_score]

# 정상 트래픽 중 오탐 사례:
LEGITIMATE_PATTERNS_THAT_LOOK_BAD = {
    "?search=cookies": '쇼핑몰의 정상 검색 — "cookies"가 XSS 키워드 같음',
    "/api/users/select": '경로에 SELECT가 있어도 SQLi 아님',
    "User-Agent: curl/8.0 (test from CI)": 'CI 모니터링 봇',
}

# 해결책 — 응답 코드 + 컨텍스트 결합
def is_real_attack(finding: dict) -> bool:
    """단순 키워드 매칭이 아닌, 패턴+컨텍스트로 판단."""
    # 200 응답 + sqli 키워드 → 진짜 의심
    if 'sqli' in finding['attacks'] and finding['status'] == 200:
        return True
    # 404 + path_traversal → 시도는 했지만 실패
    if 'path_traversal' in finding['attacks'] and finding['status'] != 200:
        return True  # 로그는 남기되 차단은 신중
    # 정상 검색 → 무시
    if finding['request'].startswith('/search'):
        return False
    return finding.get('score', 0) >= 70

# 권장 흐름
ATTACK_RESPONSE = {
    'block_immediately': [
        'command_injection (수술적 정확도)',
        'lfi_rfi (파일 노출 직접 위협)',
    ],
    'alert_only': [
        'sqli (오탐 가능성)',
        'xss (입력 단계만)',
        'path_traversal (정찰 단계)',
    ],
    'log_only': [
        'malicious_useragent (단순 봇)',
    ],
}