🔄 데이터 전처리 레이어 (Data Preprocessing Layer)

수집한 원시 데이터(Raw Data)를 AI 모델이 학습할 수 있는 형태로 변환하는 과정이야.

주요 전처리 작업:

1. 데이터 정제 (Data Cleaning)

• 중복 데이터 제거

• 결측값 처리 (평균값 대체, 삭제 등)

• 이상값(Outlier) 처리

2. 데이터 정규화 (Normalization)

• 스케일링: 서로 다른 범위의 값들을 동일한 스케일로 변환 (예: 0~1 사이로 정규화)

• 표준화: 평균 0, 표준편차 1로 변환

3. 특징 추출 (Feature Extraction)

원시 데이터에서 의미 있는 특징을 추출해. 예를 들어:

• 네트워크 패킷에서: 패킷 크기, 프로토콜 타입, 포트 번호, 플래그 등

• 로그 데이터에서: 이벤트 타입, 발생 시간, 사용자 ID, 접근한 리소스 등

4. 특징 선택 (Feature Selection)

모든 특징이 다 중요한 건 아니야. 중요도가 낮은 특징은 제거해서 모델의 효율성을 높여.

5. 데이터 인코딩

• 범주형 데이터를 숫자로 변환 (One-Hot Encoding, Label Encoding 등)

• 텍스트 데이터를 벡터로 변환 (TF-IDF, Word2Vec, BERT 임베딩 등)

실전 예시:

# Python 예시: 네트워크 트래픽 데이터 전처리
import pandas as pd
from sklearn.preprocessing import StandardScaler

# 데이터 로드
df = pd.read_csv('network_traffic.csv')

# 결측값 처리
df = df.dropna()

# 범주형 변수 인코딩
df = pd.get_dummies(df, columns=['protocol_type', 'service', 'flag'])

# 수치형 변수 정규화
scaler = StandardScaler()
numeric_features = ['duration', 'src_bytes', 'dst_bytes']
df[numeric_features] = scaler.fit_transform(df[numeric_features])

🖥️ 필요한 도구와 라이브러리

프로그래밍 언어

• Python: AI/ML 개발의 표준. 풍부한 라이브러리와 커뮤니티 지원

• Go: 고성능 데이터 처리 및 네트워크 프로그래밍

• Java/Scala: 대규모 분산 처리 (Hadoop, Spark)

AI/ML 프레임워크

• TensorFlow: 구글이 개발한 딥러닝 프레임워크

• PyTorch: 페이스북이 개발한 딥러닝 프레임워크, 연구용으로 인기

• Scikit-learn: 전통적인 머신러닝 알고리즘 구현

• XGBoost: 그래디언트 부스팅 알고리즘, 높은 성능

데이터 처리

• Pandas: 데이터 분석 및 조작

• NumPy: 수치 계산

• Apache Spark: 대규모 데이터 처리

보안 도구

• Scapy: 패킷 조작 및 분석

• YARA: 악성코드 패턴 매칭

• Suricata: 네트워크 침입 탐지

개발 환경 설정 예시:

# Python 가상환경 생성
python -m venv ai_security_env
source ai_security_env/bin/activate  # Linux/Mac
# ai_security_env\Scripts\activate  # Windows

# 필요한 패키지 설치
pip install tensorflow==2.13.0
pip install torch torchvision
pip install scikit-learn pandas numpy
pip install scapy pyyara
pip install kafka-python elasticsearch
pip install flask fastapi uvicorn

📝 프로젝트: AI 기반 네트워크 침입 탐지 시스템 (NIDS)

Step 1: 데이터셋 준비

NSL-KDD 데이터셋을 사용할 거야. 이건 네트워크 침입 탐지 연구에서 널리 사용되는 벤치마크 데이터셋이야.

import pandas as pd
import numpy as np
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import LabelEncoder, StandardScaler

# 데이터 로드
column_names = ['duration', 'protocol_type', 'service', 'flag', 'src_bytes',
                'dst_bytes', 'land', 'wrong_fragment', 'urgent', 'hot',
                'num_failed_logins', 'logged_in', 'num_compromised',
                'root_shell', 'su_attempted', 'num_root', 'num_file_creations',
                'num_shells', 'num_access_files', 'num_outbound_cmds',
                'is_host_login', 'is_guest_login', 'count', 'srv_count',
                'serror_rate', 'srv_serror_rate', 'rerror_rate',
                'srv_rerror_rate', 'same_srv_rate', 'diff_srv_rate',
                'srv_diff_host_rate', 'dst_host_count', 'dst_host_srv_count',
                'dst_host_same_srv_rate', 'dst_host_diff_srv_rate',
                'dst_host_same_src_port_rate', 'dst_host_srv_diff_host_rate',
                'dst_host_serror_rate', 'dst_host_srv_serror_rate',
                'dst_host_rerror_rate', 'dst_host_srv_rerror_rate', 'label']

df = pd.read_csv('KDDTrain+.txt', names=column_names)

# 레이블 간소화 (정상 vs 공격)
df['label'] = df['label'].apply(lambda x: 0 if x == 'normal' else 1)

print(f"데이터셋 크기: {df.shape}")
print(f"정상 트래픽: {(df['label']==0).sum()}, 공격 트래픽: {(df['label']==1).sum()}")

Step 2: 데이터 전처리

# 범주형 변수 인코딩
categorical_columns = ['protocol_type', 'service', 'flag']
label_encoders = {}

for col in categorical_columns:
    le = LabelEncoder()
    df[col] = le.fit_transform(df[col])
    label_encoders[col] = le

# 특징과 레이블 분리
X = df.drop('label', axis=1)
y = df['label']

# 학습/테스트 데이터 분할
X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42, stratify=y
)

# 정규화
scaler = StandardScaler()
X_train_scaled = scaler.fit_transform(X_train)
X_test_scaled = scaler.transform(X_test)

print(f"학습 데이터: {X_train_scaled.shape}")
print(f"테스트 데이터: {X_test_scaled.shape}")

Step 3: 모델 학습 - 랜덤 포레스트

from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import classification_report, confusion_matrix, accuracy_score

# 모델 생성 및 학습
rf_model = RandomForestClassifier(
    n_estimators=100,
    max_depth=20,
    random_state=42,
    n_jobs=-1
)

print("모델 학습 중...")
rf_model.fit(X_train_scaled, y_train)

# 예측
y_pred = rf_model.predict(X_test_scaled)

# 평가
print("\n=== 랜덤 포레스트 성능 ===")
print(f"정확도: {accuracy_score(y_test, y_pred):.4f}")
print("\n분류 리포트:")
print(classification_report(y_test, y_pred, target_names=['Normal', 'Attack']))
print("\n혼동 행렬:")
print(confusion_matrix(y_test, y_pred))

Step 4: 딥러닝 모델 - LSTM

import tensorflow as tf
from tensorflow import keras
from tensorflow.keras import layers

# 데이터 reshape (LSTM은 3D 입력 필요: [samples, timesteps, features])
X_train_lstm = X_train_scaled.reshape((X_train_scaled.shape[0], 1, X_train_scaled.shape[1]))
X_test_lstm = X_test_scaled.reshape((X_test_scaled.shape[0], 1, X_test_scaled.shape[1]))

# LSTM 모델 구축
model = keras.Sequential([
    layers.LSTM(128, input_shape=(1, X_train_scaled.shape[1]), return_sequences=True),
    layers.Dropout(0.3),
    layers.LSTM(64),
    layers.Dropout(0.3),
    layers.Dense(32, activation='relu'),
    layers.Dropout(0.2),
    layers.Dense(1, activation='sigmoid')
])

model.compile(
    optimizer='adam',
    loss='binary_crossentropy',
    metrics=['accuracy', keras.metrics.Precision(), keras.metrics.Recall()]
)

print(model.summary())

# 모델 학습
history = model.fit(
    X_train_lstm, y_train,
    epochs=20,
    batch_size=128,
    validation_split=0.2,
    verbose=1
)

# 평가
loss, accuracy, precision, recall = model.evaluate(X_test_lstm, y_test)
print(f"\n=== LSTM 모델 성능 ===")
print(f"정확도: {accuracy:.4f}")
print(f"정밀도: {precision:.4f}")
print(f"재현율: {recall:.4f}")
print(f"F1 Score: {2 * (precision * recall) / (precision + recall):.4f}")

Step 5: 실시간 탐지 시스템 구현

import scapy.all as scapy
from collections import deque
import time

class RealTimeNIDS:
    def __init__(self, model, scaler, threshold=0.5):
        self.model = model
        self.scaler = scaler
        self.threshold = threshold
        self.packet_buffer = deque(maxlen=100)
        
    def extract_features(self, packet):
        """패킷에서 특징 추출"""
        features = {}
        
        # 기본 특징
        features['protocol_type'] = packet.proto if hasattr(packet, 'proto') else 0
        features['src_bytes'] = len(packet) if packet else 0
        
        # TCP 특징
        if packet.haslayer(scapy.TCP):
            features['flag'] = packet[scapy.TCP].flags
            features['src_port'] = packet[scapy.TCP].sport
            features['dst_port'] = packet[scapy.TCP].dport
        else:
            features['flag'] = 0
            features['src_port'] = 0
            features['dst_port'] = 0
            
        # 더 많은 특징 추출...
        # (실제로는 NSL-KDD의 41개 특징을 모두 추출해야 함)
        
        return features
    
    def predict(self, features):
        """위협 예측"""
        # 특징을 모델 입력 형태로 변환
        feature_vector = self.prepare_features(features)
        feature_scaled = self.scaler.transform([feature_vector])
        
        # 예측
        prediction = self.model.predict(feature_scaled)
        probability = prediction[0]
        
        return probability > self.threshold, probability
    
    def packet_callback(self, packet):
        """패킷 캡처 콜백 함수"""
        try:
            # 특징 추출
            features = self.extract_features(packet)
            
            # 예측
            is_attack, probability = self.predict(features)
            
            if is_attack:
                print(f"[경고] 의심스러운 트래픽 탐지!")
                print(f"  확률: {probability:.4f}")
                print(f"  출발지: {packet[scapy.IP].src if packet.haslayer(scapy.IP) else 'Unknown'}")
                print(f"  목적지: {packet[scapy.IP].dst if packet.haslayer(scapy.IP) else 'Unknown'}")
                print(f"  시간: {time.strftime('%Y-%m-%d %H:%M:%S')}")
                print("-" * 50)
                
                # 여기서 자동 대응 액션 수행 가능
                # self.block_ip(packet[scapy.IP].src)
                # self.send_alert(features, probability)
                
        except Exception as e:
            print(f"패킷 처리 오류: {e}")
    
    def start_monitoring(self, interface='eth0'):
        """실시간 모니터링 시작"""
        print(f"[시작] {interface}에서 네트워크 모니터링 시작...")
        print("Ctrl+C를 눌러 중지")
        
        try:
            scapy.sniff(iface=interface, prn=self.packet_callback, store=False)
        except KeyboardInterrupt:
            print("\n[종료] 모니터링 중지")

# 사용 예시
nids = RealTimeNIDS(model=rf_model, scaler=scaler)
# nids.start_monitoring(interface='eth0')  # 실제 실행 시 주석 해제

🦠 CNN 기반 악성코드 이미지 분류

악성코드의 바이너리를 이미지로 변환해서 CNN으로 분류하는 방법이야. 정말 신기하지? 🖼️

Step 1: 바이너리를 이미지로 변환

import numpy as np
from PIL import Image
import os

def binary_to_image(binary_path, width=256):
    """바이너리 파일을 그레이스케일 이미지로 변환"""
    with open(binary_path, 'rb') as f:
        binary_data = f.read()
    
    # 바이트를 0-255 값으로 변환
    byte_array = np.frombuffer(binary_data, dtype=np.uint8)
    
    # 이미지 크기 계산
    size = len(byte_array)
    height = size // width
    
    # 남는 바이트는 버림
    byte_array = byte_array[:width * height]
    
    # 2D 배열로 reshape
    image_array = byte_array.reshape((height, width))
    
    # PIL 이미지로 변환
    image = Image.fromarray(image_array, mode='L')
    
    return image

# 사용 예시
# malware_image = binary_to_image('malware_sample.exe')
# malware_image.save('malware_visualization.png')

Step 2: CNN 모델 구축

from tensorflow.keras import layers, models
from tensorflow.keras.preprocessing.image import ImageDataGenerator

def build_malware_cnn(input_shape=(256, 256, 1), num_classes=2):
    """악성코드 분류용 CNN 모델"""
    model = models.Sequential([
        # 첫 번째 컨볼루션 블록
        layers.Conv2D(32, (3, 3), activation='relu', input_shape=input_shape),
        layers.BatchNormalization(),
        layers.MaxPooling2D((2, 2)),
        layers.Dropout(0.25),
        
        # 두 번째 컨볼루션 블록
        layers.Conv2D(64, (3, 3), activation='relu'),
        layers.BatchNormalization(),
        layers.MaxPooling2D((2, 2)),
        layers.Dropout(0.25),
        
        # 세 번째 컨볼루션 블록
        layers.Conv2D(128, (3, 3), activation='relu'),
        layers.BatchNormalization(),
        layers.MaxPooling2D((2, 2)),
        layers.Dropout(0.25),
        
        # 완전 연결 레이어
        layers.Flatten(),
        layers.Dense(256, activation='relu'),
        layers.BatchNormalization(),
        layers.Dropout(0.5),
        layers.Dense(num_classes, activation='softmax')
    ])
    
    model.compile(
        optimizer='adam',
        loss='categorical_crossentropy',
        metrics=['accuracy']
    )
    
    return model

# 모델 생성
malware_model = build_malware_cnn()
print(malware_model.summary())

# 데이터 증강 (Data Augmentation)
train_datagen = ImageDataGenerator(
    rescale=1./255,
    rotation_range=10,
    width_shift_range=0.1,
    height_shift_range=0.1,
    horizontal_flip=True,
    validation_split=0.2
)

# 학습 (실제 데이터가 있다면)
# history = malware_model.fit(
#     train_generator,
#     epochs=50,
#     validation_data=validation_generator
# )

Step 3: 정적 분석 + 동적 분석 결합

import pefile
import hashlib

class MalwareAnalyzer:
    def __init__(self, cnn_model):
        self.cnn_model = cnn_model
        
    def static_analysis(self, file_path):
        """정적 분석: PE 파일 특징 추출"""
        features = {}
        
        try:
            pe = pefile.PE(file_path)
            
            # 기본 정보
            features['file_size'] = os.path.getsize(file_path)
            features['num_sections'] = len(pe.sections)
            features['num_imports'] = len(pe.DIRECTORY_ENTRY_IMPORT) if hasattr(pe, 'DIRECTORY_ENTRY_IMPORT') else 0
            
            # 엔트로피 계산 (암호화/패킹 여부 판단)
            with open(file_path, 'rb') as f:
                data = f.read()
            features['entropy'] = self.calculate_entropy(data)
            
            # 의심스러운 섹션 이름
            suspicious_sections = ['.text', '.data', '.rsrc']
            for section in pe.sections:
                section_name = section.Name.decode().rstrip('\x00')
                if section_name not in suspicious_sections:
                    features['has_suspicious_section'] = 1
                    break
            else:
                features['has_suspicious_section'] = 0
            
            # 임포트 함수 분석
            suspicious_imports = ['CreateRemoteThread', 'WriteProcessMemory', 'VirtualAllocEx']
            features['suspicious_import_count'] = 0
            
            if hasattr(pe, 'DIRECTORY_ENTRY_IMPORT'):
                for entry in pe.DIRECTORY_ENTRY_IMPORT:
                    for imp in entry.imports:
                        if imp.name and imp.name.decode() in suspicious_imports:
                            features['suspicious_import_count'] += 1
            
        except Exception as e:
            print(f"정적 분석 오류: {e}")
            return None
        
        return features
    
    def calculate_entropy(self, data):
        """섀넌 엔트로피 계산"""
        if not data:
            return 0
        
        entropy = 0
        for x in range(256):
            p_x = float(data.count(bytes([x]))) / len(data)
            if p_x > 0:
                entropy += - p_x * np.log2(p_x)
        
        return entropy
    
    def image_analysis(self, file_path):
        """이미지 기반 분석 (CNN)"""
        # 바이너리를 이미지로 변환
        image = binary_to_image(file_path)
        image = image.resize((256, 256))
        
        # 모델 입력 형태로 변환
        image_array = np.array(image) / 255.0
        image_array = image_array.reshape((1, 256, 256, 1))
        
        # 예측
        prediction = self.cnn_model.predict(image_array)
        
        return prediction
    
    def analyze(self, file_path):
        """종합 분석"""
        print(f"[분석 시작] {file_path}")
        
        # 정적 분석
        static_features = self.static_analysis(file_path)
        if static_features:
            print("\n=== 정적 분석 결과 ===")
            for key, value in static_features.items():
                print(f"{key}: {value}")
        
        # 이미지 기반 분석
        image_prediction = self.image_analysis(file_path)
        print(f"\n=== CNN 분석 결과 ===")
        print(f"악성 확률: {image_prediction[0][1]:.4f}")
        
        # 종합 판단
        is_malicious = False
        confidence = 0.0
        
        if static_features:
            # 정적 분석 점수 계산
            static_score = 0
            if static_features.get('entropy', 0) > 7.0:  # 높은 엔트로피 = 패킹/암호화
                static_score += 0.3
            if static_features.get('has_suspicious_section', 0) == 1:
                static_score += 0.2
            if static_features.get('suspicious_import_count', 0) > 0:
                static_score += 0.2 * static_features['suspicious_import_count']
            
            # CNN 점수와 결합
            cnn_score = image_prediction[0][1]
            final_score = (static_score * 0.4) + (cnn_score * 0.6)
            
            is_malicious = final_score > 0.5
            confidence = final_score
        
        print(f"\n=== 최종 판정 ===")
        print(f"악성 여부: {'악성' if is_malicious else '정상'}")
        print(f"신뢰도: {confidence:.4f}")
        
        return is_malicious, confidence

# 사용 예시
# analyzer = MalwareAnalyzer(cnn_model=malware_model)
# analyzer.analyze('suspicious_file.exe')