如何为Pymodbus的StartTcpServer设置IP访问限制？

Pymodbus本身没有直接提供IP白名单/访问列表的配置选项，但可以通过自定义服务器的连接处理逻辑来实现这个功能。以下是针对不同后端的实现方案：

方案1：基于同步TCP服务器（默认后端） #

如果你的代码使用的是Pymodbus默认的同步TCP服务器，可以继承ModbusTcpServer类，重写连接处理方法添加IP校验：

from pymodbus.server.sync import ModbusTcpServer
from pymodbus.datastore import ModbusSlaveContext, ModbusSequentialDataBlock, ModbusServerContext

# 定义允许访问的IP白名单
ALLOWED_IPS = {"192.168.1.100", "127.0.0.1"}

class RestrictedModbusTcpServer(ModbusTcpServer):
    def handle_accepted(self, sock, addr):
        client_ip = addr[0]
        # 检查客户端IP是否在白名单内
        if client_ip not in ALLOWED_IPS:
            print(f"拒绝来自未授权IP {client_ip} 的连接")
            sock.close()
            return
        # 允许连接，调用父类的处理逻辑
        super().handle_accepted(sock, addr)

# 初始化数据存储
store = ModbusSlaveContext(hr=ModbusSequentialDataBlock(0, [0]*100))
context = ModbusServerContext(slaves=store, single=True)

debug = True
port = 502

if debug:
    print('pymobus in debug')
    from pymodbus import pymodbus_apply_logging_config
    pymodbus_apply_logging_config("DEBUG")

# 启动自定义的受限服务器
try:
    print(f'Starting ModbusTCP server at port {port}.')
    server = RestrictedModbusTcpServer(context=context, address=("0.0.0.0", port))
    server.serve_forever()
except KeyboardInterrupt:
    server.shutdown()

方案2：基于AsyncIO后端（推荐，适用于新版Pymodbus） #

如果使用Pymodbus的AsyncIO后端，需要自定义ModbusTcpProtocol来实现IP过滤：

from pymodbus.server.async_io import ModbusTcpServer, ModbusTcpProtocol
from pymodbus.datastore import ModbusSlaveContext, ModbusSequentialDataBlock, ModbusServerContext
import asyncio

ALLOWED_IPS = {"192.168.1.100", "127.0.0.1"}

class RestrictedModbusTcpProtocol(ModbusTcpProtocol):
    def connection_made(self, transport):
        client_ip = transport.get_extra_info('peername')[0]
        if client_ip not in ALLOWED_IPS:
            print(f"拒绝来自未授权IP {client_ip} 的连接")
            transport.close()
            return
        super().connection_made(transport)

# 初始化数据存储
store = ModbusSlaveContext(hr=ModbusSequentialDataBlock(0, [0]*100))
context = ModbusServerContext(slaves=store, single=True)

debug = True
port = 502

if debug:
    print('pymobus in debug')
    from pymodbus import pymodbus_apply_logging_config
    pymodbus_apply_logging_config("DEBUG")

# 启动AsyncIO服务器
async def run_server():
    print(f'Starting ModbusTCP server at port {port}.')
    server = await ModbusTcpServer.from_context(
        context,
        protocol_class=RestrictedModbusTcpProtocol,
        address=("0.0.0.0", port)
    )
    await server.serve_forever()

try:
    asyncio.run(run_server())
except KeyboardInterrupt:
    pass

关键说明 #

• 可根据需求修改ALLOWED_IPS集合，添加允许访问的客户端IP地址
• 未授权IP的连接会被直接关闭，服务器不会处理其任何请求
• 如果需要IP段匹配这类复杂规则，可以用ipaddress模块扩展校验逻辑，示例：

  import ipaddress
  def is_allowed(ip):
      return ipaddress.ip_address(ip) in ipaddress.ip_network("192.168.1.0/24")
  

  之后在连接校验时调用该函数替代直接的集合判断即可。

内容的提问来源于stack exchange，提问作者kpankhurst