"""Tofino ICS Firewall Rule Generator.
Generates Tofino firewall rules with deep packet inspection for
industrial protocols based on communication baseline analysis.
"""
import json
import sys
from datetime import datetime
from typing import Dict, List
class TofinoRuleGenerator:
"""Generates Tofino ICS firewall DPI rules."""
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 = ""):
"""Generate Modbus DPI rule."""
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 = ""):
"""Generate S7comm DPI rule."""
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 = ""):
"""Generate EtherNet/IP CIP DPI rule."""
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):
"""Add default deny rule at the end."""
self.rules.append({
"rule_id": 9999,
"protocol": "ANY",
"action": "DENY",
"source": "ANY",
"destination": "ANY",
"port": "ANY",
"description": "Default deny - block all unmatched traffic",
"log": True,
})
def generate_config(self) -> str:
"""Generate complete Tofino firewall configuration."""
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 rule summary."""
print(f"\n{'='*65}")
print("TOFINO ICS FIREWALL RULE SUMMARY")
print(f"{'='*65}")
print(f"Generated: {datetime.now().isoformat()}")
print(f"Total Rules: {len(self.rules)}")
for rule in self.rules:
action_icon = "+" if rule["action"] == "ALLOW" else "X"
print(f"\n [{action_icon}] Rule {rule['rule_id']}: {rule.get('description', '')}")
print(f" {rule['source']} -> {rule['destination']}:{rule['port']}")
print(f" Protocol: {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 - Allowed Modbus FCs: {', '.join(funcs)}")
if "allowed_operations" in dpi:
ops = [o["name"] for o in dpi["allowed_operations"]]
print(f" DPI - Allowed S7 Ops: {', '.join(ops)}")
if __name__ == "__main__":
gen = TofinoRuleGenerator()
gen.add_modbus_rule(
src="10.10.2.30",
dst="10.10.4.0/24",
allowed_funcs=[1, 2, 3, 4],
description="SCADA to utilities RTUs - read only",
)
gen.add_s7comm_rule(
src="10.10.2.50",
dst="10.10.1.0/24",
allowed_operations=["read", "write", "setup", "download", "upload"],
description="Engineering WS to reactor PLCs - full engineering access",
)
gen.add_s7comm_rule(
src="10.10.2.10",
dst="10.10.1.0/24",
allowed_operations=["read", "write", "setup"],
description="HMI to reactor PLCs - operational access only",
)
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 to packaging PLCs - operational access",
)
gen.add_default_deny()
gen.print_summary()
