killvxk 使用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部署架构
# 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：配置深度包检测规则
#!/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小时）: [数量]