AI 사이버 방어 전략 수립 툴킷

# pip install scikit-learn nltk
import re
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.ensemble import RandomForestClassifier

# 1. AI 피싱 탐지 특징 추출
def extract_features(email_text):
    features = {}
    features['urgency_score'] = len(re.findall(r'urgent|immediate|now|click|verify', email_text, re.I))
    features['link_count'] = len(re.findall(r'http[s]?://', email_text))
    features['personalization'] = len(re.findall(r'\bname\b|\byou\b|\byour\b', email_text, re.I))
    features['grammar_errors'] = len(re.findall(r'\b(teh|recieve|seperate)\b', email_text, re.I))
    return features

# 2. NLP 기반 의도 분석 (AI 생성 텍스트 탐지)
def detect_ai_phishing(emails, labels):
    vectorizer = TfidfVectorizer(max_features=1000, ngram_range=(1,3))
    X = vectorizer.fit_transform(emails)
    clf = RandomForestClassifier(n_estimators=100)
    clf.fit(X, labels)
    return vectorizer, clf

# 3. 실시간 탐지
def scan_email(email_text, vectorizer, model):
    X = vectorizer.transform([email_text])
    prob = model.predict_proba(X)[0]
    return {'phishing_prob': round(prob[1],3), 'is_phishing': prob[1] > 0.85}

# pip install numpy pandas scikit-learn
import numpy as np
from sklearn.ensemble import IsolationForest

class NetworkAnomalyDetector:
    def __init__(self, contamination=0.05):
        self.model = IsolationForest(
            contamination=contamination,
            random_state=42,
            n_estimators=200
        )
        self.baseline = None

    def fit_baseline(self, normal_logs):
        """정상 네트워크 트래픽으로 기준선 학습"""
        self.baseline = normal_logs
        self.model.fit(normal_logs)

    def detect(self, new_logs):
        """에이전틱 AI 공격 탐지"""
        preds = self.model.predict(new_logs)
        scores = self.model.score_samples(new_logs)
        return {
            'anomalies': np.where(preds == -1)[0],
            'risk_scores': -scores
        }

    def detect_lateral_movement(self, logs, time_window=300):
        """횡적 이동 탐지: 동일 계정으로 다중 서버 접근"""
        # 구현: 시간 윈도우 내 동일 계정의 서버 간 이동 패턴 분석
        pass

import json, re
from collections import Counter

class LogAnalyzer:
    def __init__(self):
        self.ioc_patterns = {
            'suspicious_ips': [],
            'failed_logins': Counter(),
            'privilege_escalation': []
        }

    def parse_auth_log(self, log_line):
        """인증 로그 파싱: brute force, 권한 상승 탐지"""
        if 'Failed password' in log_line:
            ip = re.search(r'from\s+(\S+)', log_line)
            if ip:
                self.ioc_patterns['failed_logins'][ip.group(1)] += 1

        if 'sudo' in log_line and 'COMMAND=' in log_line:
            self.ioc_patterns['privilege_escalation'].append(log_line)

    def generate_alert(self, threshold=5):
        """임계값 초과 시 알림 생성"""
        alerts = []
        for ip, count in self.ioc_patterns['failed_logins'].items():
            if count >= threshold:
                alerts.append({
                    'severity': 'HIGH',
                    'type': 'Brute Force',
                    'source_ip': ip,
                    'failed_attempts': count,
                    'action': 'Block IP & Force MFA'
                })
        return alerts

    def export_ioc(self, filepath):
        """IOC 리스트 내보내기 (방화벽 연동)"""
        with open(filepath, 'w') as f:
            json.dump(self.ioc_patterns, f, indent=2)

import requests
from urllib.parse import urljoin

class WebVulnScanner:
    def __init__(self, target_url):
        self.target = target_url
        self.payloads = {
            'sql_injection': ["' OR '1'='1", "admin'--"],
            'xss': ['<script>alert(1)</script>', '"><img src=x onerror=alert(1)>'],
            'path_traversal': ['../../../etc/passwd', '..%2f..%2fetc%2fpasswd']
        }

    def scan_form(self, form_url, params):
        """AI 생성 코드의 보안 결함 자동 탐지"""
        findings = []
        for vuln_type, payloads in self.payloads.items():
            for payload in payloads:
                test_params = {k: payload for k in params}
                try:
                    r = requests.post(form_url, data=test_params, timeout=5)
                    if self._confirm_vuln(r, vuln_type):
                        findings.append({
                            'type': vuln_type,
                            'payload': payload,
                            'url': form_url
                        })
                except Exception as e:
                    continue
        return findings

    def _confirm_vuln(self, response, vuln_type):
        """응답 분석으로 취약점 확인"""
        indicators = {
            'sql_injection': ['sql syntax', 'mysql_fetch', 'odbc_exec'],
            'xss': ['<script>alert(1)</script>'],
            'path_traversal': ['root:x:', 'bin/bash']
        }
        text = response.text.lower()
        return any(ind in text for ind in indicators.get(vuln_type, []))