AI red teamer （人工智能红队）系列26 – 人工智能信息安全应用 – 数据预处理

1. 数据预处理概述

数据预处理是将原始数据转换成适合机器学习算法使用格式的关键步骤。

1.1 数据预处理的重要性

• 提高模型准确性：清洁、结构化的数据能显著提升模型性能
• 减少训练时间：预处理后的数据能加速模型训练过程
• 增强模型鲁棒性：处理异常值和噪声提高模型的稳定性
• 确保数据一致性：统一的数据格式避免模型训练中的错误
• 优化特征质量：有效的特征工程提升模型的预测能力

1.2 数据预处理的主要技术

数据预处理包含以下关键技术：

数据清洗（Data Cleaning）

• 处理缺失值：识别和填补或删除缺失数据
• 删除重复数据：识别和处理重复记录
• 平滑噪声数据：检测和处理异常值和噪声

数据转换（Data Transformation）

• 数据标准化：将数据缩放到统一范围
• 数据归一化：调整数据分布
• 编码转换：将分类数据转换为数值形式
• 特征缩放：调整特征的数值范围

数据集成（Data Integration）

• 多源数据合并：整合来自不同来源的数据
• 数据聚合：按需求汇总数据
• 模式匹配：统一不同数据源的模式

数据格式化（Data Formatting）

• 数据类型转换：确保正确的数据类型
• 数据结构重塑：调整数据的组织形式
• 时间戳标准化：统一时间格式

2. 安全数据特有的预处理挑战

2.1 网络安全数据的特点

高维度和稀疏性

• 特征维度高：网络日志可能包含数百个特征
• 稀疏数据：许多特征值为零或缺失
• 类别不平衡：正常流量远多于恶意流量

时间相关性

• 时序依赖：安全事件具有时间上的相关性
• 概念漂移：攻击模式随时间变化
• 实时性要求：需要快速处理和响应

多源异构性

• 格式多样：来自不同设备和系统的数据格式不同
• 质量差异：不同来源的数据质量参差不齐
• 语义差异：同一概念在不同系统中表示方式不同

3. 识别和验证无效值

3.1 检查无效的IP地址

import re

def is_valid_ip(ip):
    """验证IP地址格式是否正确"""
    pattern = re.compile(r'^((25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.){3}(25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)$')
    return bool(pattern.match(ip))

# 检查无效的IP地址
invalid_ips = data[~data['source_ip'].astype(str).apply(is_valid_ip)]
print(invalid_ips)

3.2 检查无效端口号

def is_valid_port(port):
    """验证端口号是否在有效范围内（0-65535）"""
    try:
        port = int(port)
        return 0 <= port <= 65535
    except ValueError:
        return False

# 检查无效的端口号
invalid_ports = data[~data['destination_port'].apply(is_valid_port)]
print(invalid_ports)

3.3 检查无效的协议值

# 定义有效的协议列表
valid_protocols = ['TCP', 'TLS', 'SSH', 'POP3', 'DNS', 'HTTPS', 'SMTP', 'FTP', 'UDP', 'HTTP']

# 检查无效的协议值
invalid_protocols = data[~data['protocol'].isin(valid_protocols)]
print(invalid_protocols)

3.4 检查无效的传输字节

def is_valid_bytes(bytes):
    """验证传输字节数是否为非负数"""
    try:
        bytes = int(bytes)
        return bytes >= 0
    except ValueError:
        return False

# 检查无效的传输字节
invalid_bytes = data[~data['bytes_transferred'].apply(is_valid_bytes)]
print(invalid_bytes)

3.5 检查无效的威胁级别

def is_valid_threat_level(threat_level):
    """验证威胁级别是否在有效范围内（0-2）"""
    try:
        threat_level = int(threat_level)
        return 0 <= threat_level <= 2
    except ValueError:
        return False

# 检查无效的威胁级别
invalid_threat_levels = data[~data['threat_level'].apply(is_valid_threat_level)]
print(invalid_threat_levels)

4. 数据清洗策略

4.1 删除策略

最直接的方法是完全删除无效条目。这可以确保剩余的数据集干净，不包含任何可能的误导性信息。

# 使用ignore参数处理可能重叠的索引
data = data.drop(invalid_ips.index, errors='ignore') 
data = data.drop(invalid_ports.index, errors='ignore')
data = data.drop(invalid_protocols.index, errors='ignore')
data = data.drop(invalid_bytes.index, errors='ignore')
data = data.drop(invalid_threat_levels.index, errors='ignore')

print(data.describe(include='all'))

当数据准确性至关重要，且部分数据点的丢失不会对整体分析造成重大影响时，通常首选此方法。然而，这种方法并非总是可行，尤其是在数据集较小或无效条目占数据很大一部分的情况下。

4.2 修复和插值策略

有时可以清理或将无效条目转换为有效且可用的数据，而不是丢弃它们。这种方法旨在从数据集中保留尽可能多的信息。

首先，将所有无效或损坏的条目转换为NaN，实现统一的缺失值表示：

import pandas as pd
import numpy as np
import re

df = pd.read_csv('demo_dataset.csv')

# 定义无效值标识
invalid_ips = ['INVALID_IP', 'MISSING_IP']
invalid_ports = ['STRING_PORT', 'UNUSED_PORT']
invalid_bytes = ['NON_NUMERIC', 'NEGATIVE']
invalid_threat = ['?']

# 将无效值替换为NaN
df.replace(invalid_ips + invalid_ports + invalid_bytes + invalid_threat, np.nan, inplace=True)

# 将数值列转换为正确的数据类型
df['destination_port'] = pd.to_numeric(df['destination_port'], errors='coerce')
df['bytes_transferred'] = pd.to_numeric(df['bytes_transferred'], errors='coerce')
df['threat_level'] = pd.to_numeric(df['threat_level'], errors='coerce')

def is_valid_ip(ip):
    """验证IP地址并返回有效值或NaN"""
    pattern = re.compile(r'^((25[0-5]|2[0-4][0-9]|[01]?\d?\d)\.){3}(25[0-5]|2[0-4]\d|[01]?\d?\d)$')
    if pd.isna(ip) or not pattern.match(str(ip)):
        return np.nan
    return ip

df['source_ip'] = df['source_ip'].apply(is_valid_ip)

简单插补方法

对于数值列使用统计方法，对于分类列使用最频繁值：

from sklearn.impute import SimpleImputer

# 定义数值列和分类列
numeric_cols = ['destination_port', 'bytes_transferred', 'threat_level']
categorical_cols = ['protocol']

# 使用中位数插补数值列
num_imputer = SimpleImputer(strategy='median')
df[numeric_cols] = num_imputer.fit_transform(df[numeric_cols])

# 使用最频繁值插补分类列
cat_imputer = SimpleImputer(strategy='most_frequent')
df[categorical_cols] = cat_imputer.fit_transform(df[categorical_cols])

高级插补方法

对于更复杂的场景，可以使用考虑特征间关系的高级技术：

from sklearn.impute import KNNImputer

# 使用K近邻插补，考虑特征间的关系
knn_imputer = KNNImputer(n_neighbors=5)
df[numeric_cols] = knn_imputer.fit_transform(df[numeric_cols])

领域知识应用

清理和插补后，应用领域特定的知识进行最终验证和修正：

# 验证协议值的有效性
valid_protocols = ['TCP', 'TLS', 'SSH', 'POP3', 'DNS', 'HTTPS', 'SMTP', 'FTP', 'UDP', 'HTTP']
df.loc[~df['protocol'].isin(valid_protocols), 'protocol'] = df['protocol'].mode()[0]

# 为缺失的IP地址分配默认值
df['source_ip'] = df['source_ip'].fillna('0.0.0.0')

# 确保端口号在有效范围内
df['destination_port'] = df['destination_port'].clip(lower=0, upper=65535)

通过这种分层次的处理方法，我们能够最大化保留数据信息的同时确保数据质量，为后续的机器学习模型训练提供可靠的基础。