Modbus Best Practices for Security: Building an Industrial Internet of Things Protection System

Introduction: The importance of industrial network security

With industry 4.0 With the rapid development of Internet of Things technology, the boundary between industrial control systems and the Internet is increasingly blurred.Modbus As the most widely used communication protocol in the field of industrial automation, its security is directly related to the safe and stable operation of the entire industrial system. However,Modbus At the beginning of the protocol design, security factors were not taken into account, and there was a lack of authentication, encryption, and integrity verification mechanisms, which led to a lack of security Modbus The industrial system is facing severe security challenges. This article will explore in depth Modbus The security risks of the protocol provide a complete protection solution from network isolation, access control, data encryption to security monitoring, helping enterprises and developers build a reliable industrial Internet of Things security system.

1、Modbus Protocol security risk analysis

1.1 Protocol design defects

- Modbus The protocol does not include any authentication mechanism - Any client that can connect to the device can send commands - Attackers can disguise themselves as legitimate main stations for operations No encryption protection - All communication data is transmitted in plaintext - Sensitive data (such as control parameters, production data) can be eavesdropped on - Attackers can easily obtain the system's operational status No integrity verification - Unable to detect whether the data has been tampered with - Man in the middle attacks can modify control instructions - May cause equipment misoperation or damage

1.2 Common attack scenarios

Scene 1Unauthorized access

攻击者 ──▶ Modbus 设备 (端口 502)
           │
           ├── 读取敏感数据（工艺参数、生产数据）
           ├── 修改控制参数（设定点、阈值）
           └── 执行危险操作（停机、重启）

2010 Annual seismic network virus (Stuxnet) By modifying PLC Parameters caused damage to centrifuges at Iran's nuclear facilities.

Scene 2Man in the middle attack

主站 ◀── 攻击者 ──▶ 从站
       │
       ├── 窃听通信内容
       ├── 篡改控制指令
       └── 注入恶意数据

Scene 3Denial of service attack

攻击者 ──▶ 大量恶意请求 ──▶ Modbus 设备
                              │
                              ▼
                         设备过载
                              │
                              ▼
                         服务中断

1.3 Risk level assessment

2、 Network layer security protection

2.1 Network isolation strategy

Physical isolation

┌─────────────────┐     ┌─────────────────┐
│ 办公网络         │     │ 工业控制网络     │
│ (IT 网络)       │     │ (OT 网络)       │
├─────────────────┤     ├─────────────────┤
│ - 办公电脑       │     │ - PLC 设备       │
│ - 邮件服务器     │     │ - SCADA 系统     │
│ - ERP 系统       │     │ - DCS 系统       │
└─────────────────┘     └─────────────────┘
         │                       │
         └──────────┬────────────┘
                    │
            ┌───────▼───────┐
            │  工业防火墙    │
            │  (单向隔离)   │
            └───────────────┘

1. Physical separation of office network and industrial network 2. Use industrial firewalls for isolation 3. Allow only necessary communication traffic to pass through 4. Deploy unidirectional isolation gateways (data diodes)

VLAN divide

# Cisco 交换机配置示例
vlan 10
  name IT-Network
vlan 20
  name SCADA-Network
vlan 30
  name PLC-Network

interface GigabitEthernet0/1
  switchport mode access
  switchport access vlan 20
  switchport port-security
  switchport port-security maximum 2

2.2 Firewall rule configuration

iptables Configuration example

#!/bin/bash
# Modbus 安全防火墙脚本

# 清空现有规则
iptables -F
iptables -X

# 设置默认策略
iptables -P INPUT DROP
iptables -P FORWARD DROP
iptables -P OUTPUT ACCEPT

# 允许本地回环
iptables -A INPUT -i lo -j ACCEPT

# 允许已建立的连接
iptables -A INPUT -m state --state ESTABLISHED,RELATED -j ACCEPT

# 仅允许特定 IP 访问 Modbus 端口 (502)
iptables -A INPUT -p tcp -s 192.168.1.100 --dport 502 -j ACCEPT
iptables -A INPUT -p tcp -s 192.168.1.101 --dport 502 -j ACCEPT

# 允许 SSH 管理（仅限管理网段）
iptables -A INPUT -p tcp -s 192.168.100.0/24 --dport 22 -j ACCEPT

# 记录被拒绝的连接
iptables -A INPUT -p tcp --dport 502 -j LOG --log-prefix "Modbus Denied: "
iptables -A INPUT -p tcp --dport 502 -j DROP

# 保存规则
iptables-save > /etc/iptables/rules.v4

Windows Firewall configuration

# PowerShell 配置示例

# 创建入站规则
New-NetFirewallRule -DisplayName "Modbus TCP Secure" `
  -Direction Inbound `
  -Protocol TCP `
  -LocalPort 502 `
  -Action Allow `
  -RemoteAddress 192.168.1.100,192.168.1.101 `
  -Profile Private

# 启用日志记录
Set-NetFirewallProfile -Name Private `
  -LogAllowed True `
  -LogBlocked True `
  -LogMaxSizeKilobytes 4096

2.3 Port security reinforcement

Modify the default port

# 不建议使用默认端口 502，改用非标准端口
# Modbus TCP 服务配置示例

# 原始配置
# ListenPort=502

# 安全配置
ListenPort=50200

Port hiding is not a true security measure and should be used in conjunction with other security mechanisms.

Port scanning protection

# 使用 fail2ban 防止端口扫描
cat > /etc/fail2ban/jail.local << EOF
[modbus-scan]
enabled  = true
port     = 502
filter   = modbus-scan
logpath  = /var/log/syslog
maxretry = 3
bantime  = 3600
EOF

3、 Application layer security protection

3.1 Implementation of authentication mechanism

JWT Token authentication

package com.example.modbus.security;

import io.jsonwebtoken.*;
import java.util.Date;

public class ModbusAuthenticator {

    private final String secretKey = "your-secret-key-at-least-32-chars";

    public String generateToken(String username, String role) {
        return Jwts.builder()
            .setSubject(username)
            .claim("role", role)
            .setIssuedAt(new Date())
            .setExpiration(new Date(System.currentTimeMillis() + 3600000))
            .signWith(SignatureAlgorithm.HS256, secretKey)
            .compact();
    }

    public boolean validateToken(String token) {
        try {
            Jwts.parser()
                .setSigningKey(secretKey)
                .parseClaimsJws(token);
            return true;
        } catch (JwtException e) {
            return false;
        }
    }

    public Claims getClaims(String token) {
        return Jwts.parser()
            .setSigningKey(secretKey)
            .parseClaimsJws(token)
            .getBody();
    }
}

API Key authentication

import hashlib
import hmac
import time

class ModbusAPIKeyAuth:
    def __init__(self, api_key, secret):
        self.api_key = api_key
        self.secret = secret

    def generate_signature(self, timestamp, method, path, body=''):
        message = f"{timestamp}{method}{path}{body}"
        signature = hmac.new(
            self.secret.encode(),
            message.encode(),
            hashlib.sha256
        ).hexdigest()
        return signature

    def get_headers(self, method, path, body=''):
        timestamp = str(int(time.time()))
        signature = self.generate_signature(timestamp, method, path, body)
        return {
            'X-API-Key': self.api_key,
            'X-Timestamp': timestamp,
            'X-Signature': signature
        }

3.2 Access control list (ACL)

Role based access control

package com.example.modbus.security;

import java.util.*;

public class ModbusAccessControl {

    enum Role {
        OPERATOR,      // 操作员：只读
        ENGINEER,      // 工程师：读写普通寄存器
        ADMINISTRATOR  // 管理员：全部权限
    }

    private final Map<Role, Set<Integer>> readPermissions = new HashMap<>();
    private final Map<Role, Set<Integer>> writePermissions = new HashMap<>();

    public ModbusAccessControl() {
        // 初始化权限
        readPermissions.put(Role.OPERATOR, Set.of(0, 1, 2, 3, 4));
        readPermissions.put(Role.ENGINEER, Set.of(0, 1, 2, 3, 4, 5, 6, 7, 8, 9));
        readPermissions.put(Role.ADMINISTRATOR, Set.of()); // 全部

        writePermissions.put(Role.OPERATOR, Set.of()); // 无写入权限
        writePermissions.put(Role.ENGINEER, Set.of(100, 101, 102));
        writePermissions.put(Role.ADMINISTRATOR, Set.of()); // 全部
    }

    public boolean canRead(Role role, int registerAddress) {
        Set<Integer> allowed = readPermissions.get(role);
        return allowed.isEmpty() || allowed.contains(registerAddress);
    }

    public boolean canWrite(Role role, int registerAddress) {
        Set<Integer> allowed = writePermissions.get(role);
        return allowed.isEmpty() || allowed.contains(registerAddress);
    }
}

IP White list

class IPWhitelist:
    def __init__(self):
        self.allowed_ips = {
            '192.168.1.100': {'read', 'write'},
            '192.168.1.101': {'read'},
            '192.168.1.102': {'read', 'write'},
        }

    def check_permission(self, ip_address, operation):
        if ip_address not in self.allowed_ips:
            return False
        return operation in self.allowed_ips[ip_address]

    def add_ip(self, ip_address, permissions):
        self.allowed_ips[ip_address] = set(permissions)

    def remove_ip(self, ip_address):
        if ip_address in self.allowed_ips:
            del self.allowed_ips[ip_address]

3.3 Data encryption transmission

TLS/SSL Encryption

package com.example.modbus.security;

import javax.net.ssl.*;
import java.io.FileInputStream;
import java.security.KeyStore;

public class SecureModbusClient {

    public SSLSocket createSecureConnection(String host, int port) 
        throws Exception {

        // 加载密钥库
        KeyStore keyStore = KeyStore.getInstance("JKS");
        keyStore.load(new FileInputStream("client.keystore"), 
                     "keystore-password".toCharArray());

        // 初始化密钥管理器
        KeyManagerFactory kmf = KeyManagerFactory.getInstance("SunX509");
        kmf.init(keyStore, "key-password".toCharArray());

        // 加载信任库
        KeyStore trustStore = KeyStore.getInstance("JKS");
        trustStore.load(new FileInputStream("truststore.jks"), 
                       "truststore-password".toCharArray());

        // 初始化信任管理器
        TrustManagerFactory tmf = TrustManagerFactory.getInstance("SunX509");
        tmf.init(trustStore);

        // 创建 SSL 上下文
        SSLContext sslContext = SSLContext.getInstance("TLSv1.3");
        sslContext.init(kmf.getKeyManagers(), tmf.getTrustManagers(), null);

        // 创建安全连接
        SSLSocketFactory factory = sslContext.getSocketFactory();
        SSLSocket socket = (SSLSocket) factory.createSocket(host, port);

        // 配置加密套件
        socket.setEnabledCipherSuites(new String[] {
            "TLS_AES_256_GCM_SHA384",
            "TLS_CHACHA20_POLY1305_SHA256"
        });

        socket.setEnabledProtocols(new String[] {"TLSv1.3"});
        socket.setUseClientMode(true);

        return socket;
    }
}

Application layer encryption

from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.backends import default_backend
import os

class ModbusDataEncryptor:
    def __init__(self, key):
        self.key = key.encode().ljust(32)[:32]  # AES-256

    def encrypt(self, plaintext):
        iv = os.urandom(16)
        cipher = Cipher(algorithms.AES(self.key), modes.CFB(iv), 
                       backend=default_backend())
        encryptor = cipher.encryptor()
        ciphertext = encryptor.update(plaintext.encode()) + encryptor.finalize()
        return iv + ciphertext

    def decrypt(self, ciphertext):
        iv = ciphertext[:16]
        actual_ciphertext = ciphertext[16:]
        cipher = Cipher(algorithms.AES(self.key), modes.CFB(iv), 
                       backend=default_backend())
        decryptor = cipher.decryptor()
        plaintext = decryptor.update(actual_ciphertext) + decryptor.finalize()
        return plaintext.decode()

4、 Security monitoring and auditing

4.1 Operation log recording

package com.example.modbus.logging;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;

public class SecurityAuditLogger {

    private static final Logger logger = LoggerFactory.getLogger(
        SecurityAuditLogger.class
    );

    private static final DateTimeFormatter formatter = 
        DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSS");

    public void logAccess(String username, String ip, String operation, 
                         int slaveId, int register, Object value) {
        String log = String.format(
            "ACCESS | time=%s | user=%s | ip=%s | op=%s | slave=%d | reg=%d | value=%s",
            LocalDateTime.now().format(formatter),
            username, ip, operation, slaveId, register, value
        );
        logger.info(log);
    }

    public void logViolation(String username, String ip, String reason) {
        String log = String.format(
            "VIOLATION | time=%s | user=%s | ip=%s | reason=%s",
            LocalDateTime.now().format(formatter),
            username, ip, reason
        );
        logger.warn(log);
    }

    public void logAttack(String ip, String attackType, String details) {
        String log = String.format(
            "ATTACK | time=%s | ip=%s | type=%s | details=%s",
            LocalDateTime.now().format(formatter),
            ip, attackType, details
        );
        logger.error(log);
    }
}

4.2 Abnormal behavior detection

import time
from collections import defaultdict, deque

class AnomalyDetector:
    def __init__(self):
        self.request_counts = defaultdict(lambda: deque(maxlen=100))
        self.thresholds = {
            'requests_per_second': 50,
            'failed_auth_per_minute': 10,
            'unique_slaves_per_minute': 20
        }

    def record_request(self, ip_address, timestamp):
        self.request_counts[ip_address].append(timestamp)

    def detect_anomaly(self, ip_address):
        now = time.time()
        requests = self.request_counts[ip_address]

        # 检测请求频率
        recent_requests = [t for t in requests if now - t < 1.0]
        if len(recent_requests) > self.thresholds['requests_per_second']:
            return 'HIGH_FREQUENCY_REQUEST'

        # 检测扫描行为
        if len(recent_requests) > 100:
            return 'POSSIBLE_SCAN'

        return None

    def get_blocked_ips(self):
        return [ip for ip in self.request_counts 
                if self.detect_anomaly(ip) is not None]

4.3 Real time monitoring dashboard

// 使用 Grafana + Prometheus 监控
// prometheus.yml 配置

global:
  scrape_interval: 15s

scrape_configs:
  - job_name: 'modbus-gateway'
    static_configs:
      - targets: ['192.168.1.50:9090']
    metrics_path: '/metrics'

// Node.js 指标导出器
const client = require('prom-client');

const modbusRequests = new client.Counter({
  name: 'modbus_requests_total',
  help: 'Total number of Modbus requests',
  labelNames: ['type', 'slave_id', 'status']
});

const modbusLatency = new client.Histogram({
  name: 'modbus_request_latency_seconds',
  help: 'Modbus request latency',
  labelNames: ['type'],
  buckets: [0.01, 0.05, 0.1, 0.5, 1.0]
});

const securityViolations = new client.Counter({
  name: 'security_violations_total',
  help: 'Total number of security violations',
  labelNames: ['type', 'ip_address']
});

5、 Safety reinforcement checklist

5.1 Network layer inspection

• [ ] Isolation between industrial and office networks
• [ ] Firewall rules have been configured and enabled
• [ ] Only open necessary ports
• [ ] use VLAN Perform logical isolation
• [ ] Deploying intrusion detection systems (IDS)
• [ ] Enable network traffic monitoring
• [ ] Regularly update network device firmware

5.2 Equipment layer inspection

• [ ] Change default password
• [ ] Disable unused services and ports
• [ ] Enable device access logs
• [ ] Configure access control list (ACL)
• [ ] Regularly backup device configuration
• [ ] Enable firmware signature verification
• [ ] Physical security measures are in place

5.3 Application layer inspection

• [ ] Implement identity authentication mechanism
• [ ] Configure role-based access control
• [ ] Enable operation audit logs
• [ ] Sensitive data encrypted storage
• [ ] Encryption transmission of communication data
• [ ] Implement abnormal behavior detection
• [ ] Regular security vulnerability scanning

5.4 Management inspection

• [ ] Develop security policies and regulations
• [ ] Regularly conduct safety training
• [ ] Establish an emergency response plan
• [ ] Regularly conduct security audits
• [ ] Maintain system and software updates
• [ ] Establish a change management process
• [ ] Regularly conduct penetration testing

6、 Practical case: Safety Modbus Gateway implementation

6.1 System architecture

┌─────────────┐     ┌──────────────────┐     ┌─────────────┐
│ Modbus 设备   │────▶│ 安全网关         │────▶│ 客户端       │
│ (从站)      │     │ (认证 + 加密 + 审计)│     │ (主站)      │
└─────────────┘     └──────────────────┘     └─────────────┘
                           │
                           ▼
                    ┌─────────────┐
                    │ 安全日志     │
                    │ 监控系统     │
                    └─────────────┘

6.2 Complete implementation code

from flask import Flask, request, jsonify
from functools import wraps
import jwt
import logging
from datetime import datetime

app = Flask(__name__)
app.config['SECRET_KEY'] = 'your-secret-key'

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('security_audit.log'),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger('modbus-security')

# IP 白名单
WHITELISTED_IPS = ['192.168.1.100', '192.168.1.101']

# 访问控制
def require_auth(f):
    @wraps(f)
    def decorated(*args, **kwargs):
        token = request.headers.get('Authorization')
        if not token:
            logger.warning(f"未授权访问尝试：{request.remote_addr}")
            return jsonify({'error': 'Missing token'}), 401

        try:
            data = jwt.decode(token, app.config['SECRET_KEY'], 
                            algorithms=['HS256'])
            request.user = data
        except jwt.InvalidTokenError:
            logger.warning(f"无效 Token：{request.remote_addr}")
            return jsonify({'error': 'Invalid token'}), 401

        if request.remote_addr not in WHITELISTED_IPS:
            logger.warning(f"IP 不在白名单：{request.remote_addr}")
            return jsonify({'error': 'IP not allowed'}), 403

        return f(*args, **kwargs)
    return decorated

@app.route('/api/modbus/read', methods=['POST'])
@require_auth
def read_register():
    data = request.json
    slave_id = data.get('slave_id')
    register = data.get('register')

    logger.info(f"READ | user={request.user['username']} | "
               f"ip={request.remote_addr} | slave={slave_id} | "
               f"register={register}")

    # 实际 Modbus 读取逻辑
    # value = modbus_client.read_register(slave_id, register)

    return jsonify({'status': 'success', 'value': 0})

@app.route('/api/modbus/write', methods=['POST'])
@require_auth
def write_register():
    data = request.json
    slave_id = data.get('slave_id')
    register = data.get('register')
    value = data.get('value')

    # 检查写权限
    if request.user['role'] == 'operator':
        logger.warning(f"写操作被拒绝：user={request.user['username']}")
        return jsonify({'error': 'Insufficient permissions'}), 403

    logger.info(f"WRITE | user={request.user['username']} | "
               f"ip={request.remote_addr} | slave={slave_id} | "
               f"register={register} | value={value}")

    # 实际 Modbus 写入逻辑
    # modbus_client.write_register(slave_id, register, value)

    return jsonify({'status': 'success'})

if __name__ == '__main__':
    # 使用 HTTPS
    app.run(
        host='0.0.0.0',
        port=502,
        ssl_context=('cert.pem', 'key.pem')
    )

7、 Compliance and standards

7.1 Relevant safety standards

7.2 Key points of compliance inspection

1. Risk assessmentRegularly conduct security risk assessments
2. Access controlImplement the principle of minimum authority
3. Security auditReserve at least 6 Monthly operation logs
4. Event responseEstablish a security incident response process
5. Continuous monitoringReal time monitoring of network and device status
6. Regular testingConduct penetration testing at least once a year

8、 Summary and outlook

8.1 Core principles of safety protection

1. Defense in depthMulti layer protection, not relying on a single security measure
2. Minimum permissionGrant only necessary access permissions
3. Default rejectionBy default, all access is prohibited and only authorized traffic is allowed
4. Continuous monitoringReal time monitoring and auditing of all operations
5. Timely responseQuickly detect and respond to security incidents

8.2 Future development trends

1. Zero trust architectureDo not trust any internal or external networks
2. AI Drive safetyUsing machine learning to detect abnormal behavior
3. Blockchain auditingUsing blockchain to ensure that logs are tamper proof
4. Quantum encryptionAddressing the security challenges posed by quantum computing
5. Automated responseAutomatically isolate and repair security threats

8.3 Actionable recommendations

- Change all default passwords - Configure firewall rules - Enable operation logging - Perform security vulnerability scanning Short term plan (1-3 Months): - Implement network isolation - Deploy identity authentication system - Establish a security monitoring system - Develop security strategies long-range planning (6-12 Months): - Certified for safety compliance - Establish a security operation center (SOC) - Implement a zero trust architecture - Regularly conduct red blue confrontation exercises

Appendix: Recommended Security Tools

A.1 Scanning tools

• Nmap: Network scanning and port detection
• ModbusScan: special-purpose Modbus Device scanning
• Wireshark: Network protocol analysis
• Metasploit: Penetration testing framework

A.2 Monitoring tools

• Snort: Intrusion detection system
• OSSEC: Host intrusion detection
• Grafana + Prometheus: Monitoring dashboard
• ELK Stack: Log analysis platform

A.3 Reinforcement tools

• Lynis: Security audit tools
• OpenSCAP: Compliance inspection
• Fail2ban: Intrusion prevention
• UFW/iptables: Firewall management

: Modbus Security, industrial network security, access control, data encryption, security auditing,IEC 62443Intrusion detection Word count: About 12,000 character