implementing-ics-firewall-with-tofino

# 使用Tofino实现ICS防火墙

## 适用场景

- 在关键PLC或RTU前部署区域级防火墙保护
- 需要对工业协议（Modbus、EtherNet/IP、OPC、S7comm）进行深度包检测
- 实现具有协议感知执行的IEC 62443区域和管道边界
- 保护无法打补丁且需要补偿控制措施的遗留PLC
- 在不影响现有工业通信的情况下对控制网络区域进行分段

**不适用于**企业IT防火墙部署、IT与OT之间的周界防火墙（在DMZ处使用Palo Alto/Fortinet），或仅使用基于IP协议而无OT特定DPI需求的环境。

## 前置条件

- 带有适当许可证的Tofino Xenon设备或Tofino虚拟设备
- 用于集中策略管理的Tofino中央管理平台（CMP）
- 显示PLC/RTU位置和通信需求的网络拓扑图
- OT协议通信基线（Modbus功能码、EtherNet/IP CIP服务）
- 在网络区域之间内联部署的变更管理审批

## 工作流程

### 步骤 1：设计Tofino部署架构

```yaml
# Tofino ICS防火墙部署架构
# 使用深度包检测进行区域级保护

deployment_zones:
  zone_1_reactor_control:
    tofino_appliance: "TOFINO-XN-001"
    deployment_mode: "inline_bridge"
    protected_assets:
      - name: "PLC-REACTOR-01"
        ip: "10.10.1.10"
        vendor: "Siemens S7-1500"
        protocols: ["S7comm/102", "Profinet"]
      - name: "PLC-REACTOR-02"
        ip: "10.10.1.11"
        vendor: "Siemens S7-1500"
        protocols: ["S7comm/102", "Profinet"]
    authorized_communications:
      - source: "10.10.2.50"  # 工程师工作站
        dest: "10.10.1.0/24"
        protocols: ["S7comm"]
        access_type: "engineering"
      - source: "10.10.2.10"  # HMI服务器
        dest: "10.10.1.0/24"
        protocols: ["S7comm"]
        access_type: "operational"

  zone_2_packaging:
    tofino_appliance: "TOFINO-XN-002"
    deployment_mode: "inline_bridge"
    protected_assets:
      - name: "PLC-PACK-01"
        ip: "10.10.3.10"
        vendor: "Rockwell ControlLogix"
        protocols: ["EtherNet-IP/44818", "CIP"]
    authorized_communications:
      - source: "10.10.2.20"  # HMI
        dest: "10.10.3.0/24"
        protocols: ["EtherNet-IP"]
        access_type: "operational"

  zone_3_utilities:
    tofino_appliance: "TOFINO-XN-003"
    deployment_mode: "inline_bridge"
    protected_assets:
      - name: "RTU-BOILER-01"
        ip: "10.10.4.10"
        vendor: "Schneider M340"
        protocols: ["Modbus-TCP/502"]
    authorized_communications:
      - source: "10.10.2.30"  # SCADA服务器
        dest: "10.10.4.0/24"
        protocols: ["Modbus-TCP"]
        allowed_function_codes: [1, 2, 3, 4]  # SCADA只读
```

### 步骤 2：配置深度包检测规则

```python
#!/usr/bin/env python3
"""Tofino ICS防火墙规则生成器。

根据通信基线分析，生成带有工业协议
深度包检测的Tofino防火墙规则。
"""

import json
import sys
from datetime import datetime
from typing import Dict, List


class TofinoRuleGenerator:
    """生成Tofino ICS防火墙DPI规则。"""

    def __init__(self):
        self.rules = []
        self.rule_id = 1000

    def add_modbus_rule(self, src: str, dst: str, allowed_funcs: List[int],
                        allowed_registers: List[dict] = None, description: str = ""):
        """生成Modbus DPI规则。"""
        func_names = {
            1: "read_coils", 2: "read_discrete_inputs",
            3: "read_holding_registers", 4: "read_input_registers",
            5: "write_single_coil", 6: "write_single_register",
            15: "write_multiple_coils", 16: "write_multiple_registers",
        }

        rule = {
            "rule_id": self.rule_id,
            "protocol": "Modbus-TCP",
            "action": "ALLOW",
            "source": src,
            "destination": dst,
            "port": 502,
            "dpi_policy": {
                "allowed_function_codes": [
                    {"code": fc, "name": func_names.get(fc, f"FC{fc}")}
                    for fc in allowed_funcs
                ],
                "blocked_function_codes": [
                    fc for fc in range(1, 128) if fc not in allowed_funcs
                ],
            },
            "description": description,
            "log": True,
        }

        if allowed_registers:
            rule["dpi_policy"]["allowed_register_ranges"] = allowed_registers

        self.rules.append(rule)
        self.rule_id += 1
        return rule

    def add_s7comm_rule(self, src: str, dst: str, allowed_operations: List[str],
                        description: str = ""):
        """生成S7comm DPI规则。"""
        operation_map = {
            "read": {"function": 0x04, "name": "Read Variable"},
            "write": {"function": 0x05, "name": "Write Variable"},
            "setup": {"function": 0xF0, "name": "Setup Communication"},
            "download": {"function": 0x1A, "name": "Request Download"},
            "upload": {"function": 0x1D, "name": "Start Upload"},
            "cpu_stop": {"function": 0x29, "name": "PLC Stop"},
            "cpu_start": {"function": 0x28, "name": "PI Service (Start)"},
        }

        rule = {
            "rule_id": self.rule_id,
            "protocol": "S7comm",
            "action": "ALLOW",
            "source": src,
            "destination": dst,
            "port": 102,
            "dpi_policy": {
                "allowed_operations": [
                    operation_map[op] for op in allowed_operations if op in operation_map
                ],
                "block_cpu_stop": "cpu_stop" not in allowed_operations,
                "block_program_download": "download" not in allowed_operations,
            },
            "description": description,
            "log": True,
        }

        self.rules.append(rule)
        self.rule_id += 1
        return rule

    def add_ethernet_ip_rule(self, src: str, dst: str, allowed_services: List[str],
                              description: str = ""):
        """生成EtherNet/IP CIP DPI规则。"""
        rule = {
            "rule_id": self.rule_id,
            "protocol": "EtherNet-IP",
            "action": "ALLOW",
            "source": src,
            "destination": dst,
            "port": 44818,
            "dpi_policy": {
                "allowed_cip_services": allowed_services,
                "block_firmware_flash": True,
                "block_program_download": "program_download" not in allowed_services,
            },
            "description": description,
            "log": True,
        }

        self.rules.append(rule)
        self.rule_id += 1
        return rule

    def add_default_deny(self):
        """在末尾添加默认拒绝规则。"""
        self.rules.append({
            "rule_id": 9999,
            "protocol": "ANY",
            "action": "DENY",
            "source": "ANY",
            "destination": "ANY",
            "port": "ANY",
            "description": "默认拒绝 - 阻止所有不匹配的流量",
            "log": True,
        })

    def generate_config(self) -> str:
        """生成完整的Tofino防火墙配置。"""
        config = {
            "tofino_configuration": {
                "generated": datetime.now().isoformat(),
                "appliance_model": "Tofino Xenon",
                "firmware_version": "4.2",
                "mode": "inline_bridge",
                "failsafe": "fail_open",
                "rules": self.rules,
            }
        }
        return json.dumps(config, indent=2)

    def print_summary(self):
        """打印规则摘要。"""
        print(f"\n{'='*65}")
        print("TOFINO ICS防火墙规则摘要")
        print(f"{'='*65}")
        print(f"生成时间: {datetime.now().isoformat()}")
        print(f"规则总数: {len(self.rules)}")

        for rule in self.rules:
            action_icon = "+" if rule["action"] == "ALLOW" else "X"
            print(f"\n  [{action_icon}] 规则 {rule['rule_id']}: {rule.get('description', '')}")
            print(f"      {rule['source']} -> {rule['destination']}:{rule['port']}")
            print(f"      协议: {rule['protocol']}")
            if "dpi_policy" in rule:
                dpi = rule["dpi_policy"]
                if "allowed_function_codes" in dpi:
                    funcs = [f["name"] for f in dpi["allowed_function_codes"]]
                    print(f"      DPI - 允许的Modbus功能码: {', '.join(funcs)}")
                if "allowed_operations" in dpi:
                    ops = [o["name"] for o in dpi["allowed_operations"]]
                    print(f"      DPI - 允许的S7操作: {', '.join(ops)}")


if __name__ == "__main__":
    gen = TofinoRuleGenerator()

    # SCADA服务器到Modbus RTU：只读
    gen.add_modbus_rule(
        src="10.10.2.30",
        dst="10.10.4.0/24",
        allowed_funcs=[1, 2, 3, 4],
        description="SCADA到公用RTU - 只读",
    )

    # 工程师工作站到西门子PLC：完全访问
    gen.add_s7comm_rule(
        src="10.10.2.50",
        dst="10.10.1.0/24",
        allowed_operations=["read", "write", "setup", "download", "upload"],
        description="工程师工作站到反应器PLC - 完全工程访问",
    )

    # HMI到西门子PLC：仅读写（无程序下载）
    gen.add_s7comm_rule(
        src="10.10.2.10",
        dst="10.10.1.0/24",
        allowed_operations=["read", "write", "setup"],
        description="HMI到反应器PLC - 仅操作访问",
    )

    # HMI到Rockwell PLC：操作访问
    gen.add_ethernet_ip_rule(
        src="10.10.2.20",
        dst="10.10.3.0/24",
        allowed_services=["read_tag", "write_tag", "get_attribute"],
        description="HMI到包装PLC - 操作访问",
    )

    gen.add_default_deny()
    gen.print_summary()
```

## 核心概念

| 术语 | 定义 |
|------|------|
| Tofino Xenon | Belden/Hirschmann工业防火墙设备，具备OT协议深度包检测能力 |
| 深度包检测（DPI） | 检查标头之外的报文载荷内容，对工业协议操作强制执行细粒度规则 |
| 内联桥接模式（Inline Bridge Mode） | 透明部署模式，防火墙位于网络段之间，无需更改IP地址 |
| 失败开放（Fail-Open） | 安全模式，设备故障时防火墙放行所有流量，维持过程可用性 |
| 可加载安全模块（LSM） | Tofino插件模块，为Modbus、EtherNet/IP、OPC或其他协议提供特定协议DPI |
| 中央管理平台（CMP） | Tofino集中管理服务器，用于跨多个Tofino设备部署和管理策略 |

## 输出格式

```
TOFINO部署报告
===========================
日期: YYYY-MM-DD
已部署设备数: [数量]

各设备摘要:
  [设备ID]:
    模式: 内联桥接
    故障安全: 失败开放
    受保护资产: [数量]
    规则: [数量]
    DPI协议: [列表]

规则摘要:
  允许规则: [数量]
  拒绝规则: [数量]
  DPI强制规则: [数量]

监控:
  阻止的数据包（24小时）: [数量]
  DPI违规（24小时）: [数量]
```