Python部署工控安全风险评估系统

1. 工控安全基础理论

1.1 风险评估概念

• 风险定义：风险 = 事件发生的可能性 × 事件的影响

• 影响分析（Impact Analysis）：评估特定事件（如设备故障、网络攻击）对工控系统的关键资产（设备、数据、流程）的破坏程度，通常从以下维度分析：

  • 机密性（Confidentiality）：数据是否被泄露。

  • 完整性（Integrity）：数据或设备是否被篡改。

  • 可用性（Availability）：系统是否能够正常运行。

• 工控系统特点：实时性、设备互联性、高可靠性要求。

1.2 风险评估模型

• 定性模型：基于专家经验划分风险等级（如高、中、低）。

• 定量模型：通过数学公式计算风险值（如 风险值 = 可能性 × 影响值）。

• 混合模型：结合定性和定量方法，例如：

  • 使用 层次分析法（AHP） 确定权重。

  • 通过 风险矩阵（Risk Matrix） 映射风险等级。

2. 技术部分

2.1 系统架构设计

1. 数据采集层：
   - 实时采集设备状态（传感器数据、设备日志）。
   - 网络流量监控（流量协议、源/目标IP、数据包特征）。
2. 数据处理层：
   - 数据清洗（去噪、格式标准化）。
   - 特征提取（如异常流量特征、设备故障特征）。
3. 分析引擎：
   - 影响分析模块：基于规则或机器学习模型评估事件影响。
   - 风险评估模块：计算风险值并分类风险等级。
4. 可视化与报告：
   - 生成风险评估报告。
   - 实时仪表盘展示风险状态。

2.1.1 系统架构

• 数据采集模块：负责从工控系统中收集数据，如传感器数据、设备状态、网络流量等。

• 影响分析模块：分析不同事件对工控系统的影响，如设备故障、网络攻击等。

• 风险评估模块：基于影响分析的结果，评估风险等级。

• 报告生成模块：生成风险评估报告，供决策者参考。

2.1.2 数据模型

• 设备状态：设备ID、状态（正常/故障）、时间戳等。

• 网络流量：源IP、目标IP、协议、数据包大小、时间戳等。

• 事件：事件类型（如设备故障、网络攻击）、影响等级、时间戳等。

2.2 关键技术

• 数据采集技术：

  • OPC UA：工业协议，用于设备数据采集。

  • Snort：网络流量监控工具。

• 影响分析技术：

  • 规则引擎：基于预定义规则（如“设备故障导致产线停工”）。

  • 图计算：分析设备依赖关系的影响传播（如设备A故障影响设备B）。

• 风险评估技术：

  • 风险矩阵：将可能性和影响映射到风险等级。

  • 模糊逻辑：处理不确定性风险因素。

3. 代码实现

3.1 数据采集与预处理

import pandas as pd
from datetime import datetime

# 模拟设备状态数据
def collect_device_data():
    data = {
        "device_id": [1, 2, 3],
        "status": ["normal", "fault", "normal"],
        "timestamp": [datetime.now().isoformat() for _ in range(3)]
    }
    return pd.DataFrame(data)

# 模拟网络流量数据
def collect_network_data():
    data = {
        "src_ip": ["192.168.1.1", "192.168.1.2"],
        "dst_ip": ["192.168.1.3", "192.168.1.4"],
        "protocol": ["TCP", "UDP"],
        "packet_size": [120, 1500]  # 1500可能为异常大包
    }
    return pd.DataFrame(data)

# 数据合并
device_df = collect_device_data()
network_df = collect_network_data()

3.2 影响分析模块

class ImpactAnalyzer:
    def __init__(self):
        # 定义影响规则：设备故障影响产线可用性
        self.rules = {
            "device_fault": {"availability": 0.8},  # 可用性下降80%
            "large_packet": {"integrity": 0.5}       # 完整性风险增加50%
        }

    def analyze_device(self, device_df):
        faults = device_df[device_df["status"] == "fault"]
        if not faults.empty:
            return self.rules["device_fault"]
        return {}

    def analyze_network(self, network_df):
        large_packets = network_df[network_df["packet_size"] > 1000]
        if not large_packets.empty:
            return self.rules["large_packet"]
        return {}

# 执行分析
analyzer = ImpactAnalyzer()
device_impact = analyzer.analyze_device(device_df)
network_impact = analyzer.analyze_network(network_df)
print("设备影响:", device_impact)
print("网络影响:", network_impact)

3.3 风险评估模块

class RiskEvaluator:
    def __init__(self):
        # 定义风险矩阵：可能性 × 影响
        self.risk_matrix = {
            (0.1, 0.3): "低",
            (0.4, 0.6): "中",
            (0.7, 1.0): "高"
        }

    def evaluate(self, probability, impact):
        for (min_prob, max_prob), risk_level in self.risk_matrix.items():
            if min_prob <= probability <= max_prob and impact >= min_prob:
                return risk_level
        return "未知"

# 假设事件可能性为0.5，影响值为0.8
evaluator = RiskEvaluator()
risk_level = evaluator.evaluate(probability=0.5, impact=0.8)
print("风险等级:", risk_level)  # 输出: 高

3.4 报告生成模块

def generate_report(device_impact, network_impact, risk_level):
    report = {
        "timestamp": datetime.now().isoformat(),
        "device_impact": device_impact,
        "network_impact": network_impact,
        "risk_level": risk_level
    }
    # 保存为JSON或导出为PDF
    import json
    with open("risk_report.json", "w") as f:
        json.dump(report, f, indent=2)
    return report

report = generate_report(device_impact, network_impact, risk_level)
print("报告已生成:", report)

4. 系统扩展

1. 实时性优化：使用 Kafka 或 RabbitMQ 实现流数据处理。

2. 机器学习集成：用 PyTorch 或 Scikit-learn 训练异常检测模型。

3. 可视化界面：通过 Dash 或 Streamlit 构建实时监控仪表盘。

总结

• 理论：风险 = 可能性 × 影响，工控系统需关注CIA三性。

• 技术：规则引擎、风险矩阵、数据采集与特征分析。

• 代码：通过Python实现数据采集、影响分析、风险评估和报告生成。

此框架可根据实际工控场景扩展规则库和算法复杂度（例如加入设备依赖关系图谱）