交互式Modbus靶机小记


字数统计: 4.1k字   |   阅读时长≈ 19分

啥都没有，就当你啥都没看见吧

前段时间一时兴起，去做了几道HTB的硬件题，其中好几题要我和设备进行”交互“的题目让我眼前一亮：欸，为什么不自己搞一个玩玩呢？

可惜由于个人的水平限制，我只对工控协议中的Modbus协议进行过细致的了解，所以就只做了Modbus TCP的交互靶机…

Scene I：Modbus服务器，启动！

我不可能实际搞一个真的Modbus设备过来是吧，咱也没这个实力，所以就简单找了一个叫PyModbus的库去进行模拟：Welcome to PyModbus’s documentation! — PyModbus 4.0.0dev4 documentation

好消息是这个库看起来功能挺全的，坏消息是…从官方文档到官方样例都非常的…混乱…看教学视频和论坛的问答并没有让我对这个库有进一步的认识，因此我整个人是昏的，尤其是这几天我还感冒了，只能说ccb太坏了（

最后实在是没办法了，所以转头去问了D老师（DeepSeek），它也是给我抛了一个看起来很不错的代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
from pymodbus.server import StartSerialServer, StartTcpServer
from pymodbus.datastore import ModbusSequentialDataBlock, ModbusSlaveContext, ModbusServerContext
from pymodbus.transaction import ModbusRtuFramer, ModbusSocketFramer
from pymodbus.payload import BinaryPayloadBuilder, BinaryPayloadDecoder
from pymodbus.constants import Endian

def setup_server():
    # 1. 创建数据块（寄存器Slave和线圈Slave）
    # 寄存器Slave: 65536个寄存器（设备ID 0xdead）
    register_block = ModbusSequentialDataBlock.create()  # 默认全0
    register_slave = ModbusSlaveContext(
        hr=register_block,  # 保持寄存器（功能码03/06/16）
        unit=0xdead         # 设备ID
    )

    # 线圈Slave: 65536个线圈（设备ID 0xbeaf）
    coil_block = ModbusSequentialDataBlock.create()     # 默认全False
    coil_slave = ModbusSlaveContext(
        co=coil_block,      # 线圈（功能码01/05/15）
        unit=0xbeaf         # 设备ID
    )

    # 2. 创建服务器上下文（支持多Slave）
    slaves = {
        0xdead: register_slave,
        0xbeaf: coil_slave
    }
    context = ModbusServerContext(slaves=slaves, single=False)

    # 3. 启动TCP服务器（默认端口502）
    print("启动Modbus服务器...")
    StartTcpServer(
        context=context,
        framer=ModbusSocketFramer,
        address=("0.0.0.0", 502)
    )

if __name__ == "__main__":
    # 启动服务器（异步方式）
    import threading
    server_thread = threading.Thread(target=setup_server)
    server_thread.daemon = True
    server_thread.start()

    # 4. 模拟客户端写入和读取操作
    from pymodbus.client import ModbusTcpClient
    import time

    time.sleep(1)  # 等待服务器启动

    # 连接到本地服务器
    client = ModbusTcpClient('127.0.0.1', port=502)

    try:
        # --- 操作寄存器Slave (0xdead) ---
        # 写入寄存器 0x1234 = '*' 的ASCII码（42）
        print("\n写入寄存器 0x1234 = '*'")
        builder = BinaryPayloadBuilder(byteorder=Endian.BIG, wordorder=Endian.BIG)
        builder.add_string('*')  # 写入1个字符（占用1个寄存器）
        payload = builder.to_registers()
        client.write_registers(address=0x1234, values=payload, slave=0xdead)

        # 读取寄存器 0x1234
        response = client.read_holding_registers(address=0x1234, count=1, slave=0xdead)
        decoder = BinaryPayloadDecoder.fromRegisters(response.registers, byteorder=Endian.BIG, wordorder=Endian.BIG)
        print(f"读取寄存器 0x1234: {chr(decoder.decode_16bit_uint())}")

        # --- 操作线圈Slave (0xbeaf) ---
        # 写入线圈 0x5678 = True
        print("\n写入线圈 0x5678 = True")
        client.write_coil(address=0x5678, value=True, slave=0xbeaf)

        # 读取线圈 0x5678
        response = client.read_coils(address=0x5678, count=1, slave=0xbeaf)
        print(f"读取线圈 0x5678: {response.bits[0]}")

    finally:
        client.close()

看起来很美好吧？但是实际运行就是一个又一个的Bug：这里多参数了，那里又类型不对，这个类还已经找不到了…总之还是一片混乱，但是相比官方文档已经好上不知道多少了，感觉好一些Python库的官方文档就是各种混乱，不敢想象前人是怎么吃得下官方文档这坨的（以至于我觉得看源码实现都比看文档好上不知道多少倍）

稍微提一些踩到的坑：（使用的是截至本文编写日期pyModbus库的最新稳定版3.9.2）

  • 设备ID不是通过ModbusSlaveContext的unit参数进行传参的，此时这个类已经没有这个参数了，设备ID是在第24行的字典进行传入的，其ID为字典的key
  • 个人学艺不精：Modbus设备ID只能是1~255，不能是2字节
  • 启动TCP服务器的时候，使用的Framer不是代码里的pymodbus.transaction.ModbusSocketFramer（没记错的话好像位置也不在这里），而应该是pymodbus.FramerType.SOCKET
  • 解析读取寄存器的返回数据的时候，如果使用decoder.decode_16bit_uint()，会将这个直接解析成UTF-8，所以最好还是使用decoder.decode_string(size=string_size)好一些

我不清楚有没有更多的了，总而言之最后的成品如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
from pymodbus.server import StartSerialServer, StartTcpServer
from pymodbus.datastore import ModbusSequentialDataBlock, ModbusSlaveContext, ModbusServerContext
from pymodbus import FramerType
from pymodbus.payload import BinaryPayloadBuilder, BinaryPayloadDecoder
from pymodbus.constants import Endian

def setup_server():
    # 1. 创建数据块（寄存器Slave和线圈Slave）
    # 寄存器Slave: 65536个寄存器（设备ID 0x25）
    register_block = ModbusSequentialDataBlock.create()  # 默认全0
    register_slave = ModbusSlaveContext(
        hr=register_block,  # 保持寄存器（功能码03/06/16）
    )

    # 线圈Slave: 65536个线圈（设备ID 0x38）
    coil_block = ModbusSequentialDataBlock.create()     # 默认全False
    coil_slave = ModbusSlaveContext(
        co=coil_block,      # 线圈（功能码01/05/15）
    )

    # 2. 创建服务器上下文（支持多Slave）
    slaves = {
        0x25: register_slave,
        0x38: coil_slave
    }
    context = ModbusServerContext(slaves=slaves, single=False)

    # 3. 启动TCP服务器（默认端口502）
    print("启动Modbus服务器...")
    StartTcpServer(
        context=context,
        framer=FramerType.SOCKET,
        address=("0.0.0.0", 502)
    )

if __name__ == "__main__":
    # 启动服务器（异步方式）
    import threading
    server_thread = threading.Thread(target=setup_server)
    server_thread.daemon = True
    server_thread.start()

    # 4. 模拟客户端写入和读取操作
    from pymodbus.client import ModbusTcpClient
    import time

    time.sleep(1)  # 等待服务器启动

    # 连接到本地服务器
    client = ModbusTcpClient('127.0.0.1', port=502)

    try:
        # --- 操作寄存器Slave (0x25) ---
        # 写入寄存器 0x1234 = 'H' 的ASCII码
        print("\n写入寄存器 0x1234 = 'He'")
        builder = BinaryPayloadBuilder(byteorder=Endian.BIG, wordorder=Endian.BIG)
        builder.add_string('He')  # 写入1个字符（占用1个寄存器）
        payload = builder.to_registers()
        client.write_registers(address=0x1234, values=payload, slave=0x25)

        # 读取寄存器 0x1234
        response = client.read_holding_registers(address=0x1234, count=2, slave=0x25)
        decoder = BinaryPayloadDecoder.fromRegisters(response.registers, byteorder=Endian.BIG, wordorder=Endian.BIG)
        print(f"读取寄存器 0x1234: {decoder.decode_string(size=2)}")

        # --- 操作线圈Slave (0x38) ---
        # 写入线圈 0x5678 = True
        print("\n写入线圈 0x5678 = True")
        client.write_coil(address=0x5678, value=True, slave=0x38)

        # 读取线圈 0x5678
        response = client.read_coils(address=0x5678, count=1, slave=0x38)
        print(f"读取线圈 0x5678: {response.bits[0]}")
        
        response2 = client.read_coils(address=0x5679, count=1, slave=0x38)
        print(f"读取线圈 0x5679: {response2.bits[0]}")

    except:
        print("Error Occurred.")
    # finally:
    #     time.sleep(1000000000)
    #     client.close()

后面的sleep是为了让我用mbpoll进行实际的查看，看看是不是真的能直接通过Modbus查看对应地址的数据，最后确认是OK的

只能说还得是D老师，这些注释一下就解决了我70%的问题，我现在也简单理解了这个库的使用了

OK，临时小测，代码里面的ModbusSlaveContext中有4个参数：di，ir，co，hr，请问它们分别代表什么？

各位大可去搜官方文档，反正是依托细碎，我最后是看源码注释才搞懂其实就是Discrete，Input Registers，Coils，Holding Registers四种数据类型，非常傻逼有没有

Scene II：服务器数据的预处理

既然我们已经基本搞定了Modbus这一部分了，那么我们就可以开始进行一个题的出了，那么要不从一个最简单的题目来吧：让做题者直接和Modbus服务器进行通信？

那么如果是这样，我们的要求也就能分成这两部分了：

  • 启动Modbus服务器并持续开放监听，同时其端口要暴露给Docker
  • 启动服务器后创建一个Client进行数据的预处理，比如将flag塞进特定地址的寄存器中

看起来很简单，但是问题就在于Modbus服务器的StartTcpServer会进行堵塞，因此两个任务不能在一个脚本内同时执行，如果使用线程的话可能会导致Python脚本执行完毕后线程也被关闭，就像前面的代码一样

上面这一段我瞎说的，我不是很会线程这方面的知识，非常欢迎大伙指出问题并纠正

除此之外，我个人常使用的是xinetd进行多线程管理，但是xinetd是连接的时候才执行，因此不是很符合预处理的情况，因此最后决定是抛弃了xinetd，将server和预处理的client分开，其中server在后台持续运行，client执行后就直接关闭，代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
# server.py
from pymodbus.server import StartTcpServer
from pymodbus.datastore import ModbusSlaveContext, ModbusServerContext
from pymodbus.datastore import ModbusSequentialDataBlock

def run_server():
    register_block = ModbusSequentialDataBlock.create()
    register_slave = ModbusSlaveContext(hr=register_block)
    context = ModbusServerContext(slaves={0x14: register_slave}, single=False)
    StartTcpServer(context=context, address=("0.0.0.0", 502))

if __name__ == "__main__":
    run_server()


# client.py
from pymodbus.client import ModbusTcpClient
from pymodbus.payload import BinaryPayloadBuilder
from pymodbus.constants import Endian

client = ModbusTcpClient('127.0.0.1', port=502)
try:
    builder = BinaryPayloadBuilder(byteorder=Endian.BIG, wordorder=Endian.BIG)
    builder.add_string('Aurora{testflag}')
    client.write_registers(address=0x1145, values=builder.to_registers(), slave=0x14)
    print("Flag initialized.")
finally:
    client.close()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
# Dockerfile
FROM python:3.10.12-slim-bullseye

# apt更换镜像源，并更新软件包列表信息
RUN  sed -i s@/archive.ubuntu.com/@/mirrors.aliyun.com/@g /etc/apt/sources.list
RUN apt-get update 

RUN pip install pymodbus

# 部署程序
COPY ./server.py /app/server.py
COPY ./client.py /app/init_flag.py
COPY ./init_script.sh /app/init_script.sh

CMD ["bash","/app/init_script.sh"]

EXPOSE 502
1
2
3
4
5
6
7
8
9
10
11
12
13
14
#!/bin/bash
# init_script.sh

# 直接启动Modbus服务器（确保初始化脚本能连接）
python /app/server.py &

# 等待服务就绪
sleep 5

# 执行初始化
python /app/init_flag.py

# 保持容器运行
wait

肯定是写的不够好的，但是能动就行，还要啥自行车啊

Scene III：交互拓展和端口转发

目前我们只是进行了一些简单的数据的预处理，但是假如我们设置一些更复杂的操作呢？比如说配合梯形图进行操作？那我们需要一个验证端吧？但是因为CTFd好像只能用FRP穿透1个端口，因此最后是考虑将验证端和预处理放在同一个脚本中，不过到时候可能还得给一个说明书之类的东西

为了保证验证端输出的数据不会影响到正常的Modbus通信（比如做题者直接使用mbpoll和靶机进行通信），因此靶机设置的是不会进行引导式输出的，你只需要往里面传入数据，然后验证端会自动判断你的输入是Modbus数据还是检验设备寄存器/线圈状态的请求，我这里是设置了一个status的命令，你如果只传入这个字符串就会返回当前的设备状态，而如果符合条件则会返回flag

当然，如果是Modbus数据，验证端就会将数据转发给服务器，并将响应数据再转发回请求侧，至于分辨的话就很简单的用了长度+特定位置的值进行判断，一个简单的样例代码如下（当然大部分是AI搓的就是了）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
import socket
from threading import Thread
from pymodbus.client import ModbusTcpClient
from pymodbus.payload import BinaryPayloadBuilder, BinaryPayloadDecoder
from pymodbus.constants import Endian

# 配置参数
MODBUS_SERVER_HOST = '127.0.0.1'
MODBUS_SERVER_PORT = 502
PROXY_PORT = 504         # 代理监听端口
STATUS_REGISTER = 0x1145 # 状态寄存器地址
STATUS_SLAVE = 0x14      # 设备ID
FLAG = "Aurora{testflag}"

def handle_client(client_socket):
    try:
        while True:
            data = client_socket.recv(1024)
            if not data:
                return

            # 判断是否为Modbus请求（检查事务ID和协议标识符）
            if len(data) >= 6 and data[2:4] == b'\x00\x00':
                # 转发到Modbus服务器
                with ModbusTcpClient(MODBUS_SERVER_HOST, port=MODBUS_SERVER_PORT) as mb_client:
                    mb_client.socket.send(data)
                    response = mb_client.socket.recv(1024)
                    client_socket.send(response)
            elif data.strip().lower() == b'status':
                # 检查状态寄存器值
                with ModbusTcpClient(MODBUS_SERVER_HOST, port=MODBUS_SERVER_PORT) as mb_client:
                    result = mb_client.read_holding_registers(address=STATUS_REGISTER, count=1, slave=STATUS_SLAVE)
                    if result.isError():
                        resp = "Status check failed"
                    else:
                        decoder = BinaryPayloadDecoder.fromRegisters(result.registers, byteorder=Endian.BIG, wordorder=Endian.BIG)
                        value = decoder.decode_string(size=1)
                        if value == b'B':
                            resp = f"Status: OK\nFlag: {FLAG}\n"
                        else:
                            resp = f"Status: Invalid (Current Value: {value})\n"
                client_socket.send(resp.encode())
            else:
                client_socket.send(b"Unknown request data")
    except Exception as e:
        print(f"Client error: {e}")
    finally:
        client_socket.close()

def run_proxy_server():
    proxy = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    proxy.bind(('0.0.0.0', PROXY_PORT))
    proxy.listen(5)
    print(f"[*] Listening on port {PROXY_PORT}")

    while True:
        client_sock, addr = proxy.accept()
        print(f"[*] Accepted connection from {addr[0]}:{addr[1]}")
        handler = Thread(target=handle_client, args=(client_sock,))
        handler.start()

if __name__ == "__main__":
    # 初始化Flag（仅首次运行）
    try:
        with ModbusTcpClient(MODBUS_SERVER_HOST, port=MODBUS_SERVER_PORT) as client:
            builder = BinaryPayloadBuilder(byteorder=Endian.BIG, wordorder=Endian.BIG)
            builder.add_string('A')
            client.write_registers(address=STATUS_REGISTER, values=builder.to_registers(), slave=STATUS_SLAVE)
            print("[*] Initialized flag register")
    except Exception as e:
        print(f"Initialization failed: {e}")

    # 启动代理服务器
    run_proxy_server()

Scene IV：开始瞎78出题！

现在该有的基础设施都已经OK了，接下来就是自己瞎设计PLC梯形图然后搓代码了！（实际上出这种题最复杂的点就是设计梯形图和根据梯形图完成逻辑实现，只能说当时还是太天真了）

这里献个丑，放一下本人给学校内部靶场供的题目（此处仅提供client.py代码，其它附件如有需要请自行联系我，不过质量肯定不高所以我觉得就没必要联系我了www）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
import socket
from threading import Thread
from pymodbus.client import ModbusTcpClient
from pymodbus.payload import BinaryPayloadBuilder, BinaryPayloadDecoder
from pymodbus.constants import Endian
import random

# 配置参数
MODBUS_SERVER_HOST = '127.0.0.1'
MODBUS_SERVER_PORT = 502
PROXY_PORT = 504         # 代理监听端口
STATUS_SLAVE = 0x12      # 设备ID
FUEL_SLAVE = 0x34
STATUS_COILS = {"Auto Mode":0x0721, "Manual-Initialize":0x1337,"Manual Mode":0x1145,"Use Backup":0x3137, "Main Power":0x7777,"Backup Power":0x8888,"Fuel System":0x7eaf,"Power System":0x8100,"Propulsion System":0x1919} # 状态寄存器地址
current_status = [True, False, False, False, False, False, False, False]
INTENTIONAL_STATUS = [False, True, True, True, False, True, True, True]
FUEL_POINTER = 0x3000
FUEL_DEVICE_START = 0xea57
fuel_system = False
FLAG = "Aurora{testflag}"

def handle_logic(data:bytes):
    device_id = data[6]
    function_id = data[7]
    global fuel_system
    if device_id not in [STATUS_SLAVE, FUEL_SLAVE] or function_id not in [5,6,15,16,22,23]:
        return
    with ModbusTcpClient(MODBUS_SERVER_HOST, port=MODBUS_SERVER_PORT) as mb_client:
        if device_id == 0x12:
            status = []
            for value in STATUS_COILS.values():
                if value == 0x7eaf:
                    continue
                coil_status = mb_client.read_coils(address=value, count=1,slave=STATUS_SLAVE)
                status.append(coil_status.bits[0])
            print(status)
            if status[0] and status[1] and not status[2]:
                status[0] = False
                status[2] = True
                mb_client.write_coil(address=STATUS_COILS["Auto Mode"], value=False, slave=STATUS_SLAVE)
                mb_client.write_coil(address=STATUS_COILS["Manual Mode"], value=True, slave=STATUS_SLAVE)
            if not status[1]:
                status[0] = True
                status[2] = False
                mb_client.write_coil(address=STATUS_COILS["Auto Mode"], value=True, slave=STATUS_SLAVE)
                mb_client.write_coil(address=STATUS_COILS["Manual Mode"], value=False, slave=STATUS_SLAVE)
            if status[2] and status[3] and not status[5]:
                status[5] = True
                mb_client.write_coil(address=STATUS_COILS["Backup Power"], value=True, slave=STATUS_SLAVE)
            if not status[2]:
                status[3] = False
                mb_client.write_coil(address=STATUS_COILS["Use Backup"], value=False, slave=STATUS_SLAVE)
            if status[4] or status[5]:
                status[6] = True
                mb_client.write_coil(address=STATUS_COILS["Power System"], value=True, slave=STATUS_SLAVE)
            print(status)
            for i in range(8):
                current_status[i] = status[i]
            
        elif device_id == 0x34:
            result = mb_client.read_holding_registers(address=FUEL_POINTER, count=1, slave=FUEL_SLAVE)
            decoder = BinaryPayloadDecoder.fromRegisters(result.registers, byteorder=Endian.BIG, wordorder=Endian.BIG)
            fuel_device_id = decoder.decode_16bit_int()
            print(fuel_device_id)
            result = mb_client.read_holding_registers(address=FUEL_DEVICE_START+fuel_device_id-1, count=1, slave=FUEL_SLAVE)
            decoder = BinaryPayloadDecoder.fromRegisters(result.registers, byteorder=Endian.BIG, wordorder=Endian.BIG)
            fuel_amount = decoder.decode_16bit_int()
            print(fuel_amount)
            if fuel_amount >= 75:
                fuel_system = True
                mb_client.write_coil(address=STATUS_COILS["Fuel System"], value=True, slave=STATUS_SLAVE)
            else:
                fuel_system = False
                mb_client.write_coil(address=STATUS_COILS["Fuel System"], value=False, slave=STATUS_SLAVE)
        if fuel_system and current_status[6]:
            current_status[7] = True
            mb_client.write_coil(address=STATUS_COILS["Propulsion System"], value=True, slave=STATUS_SLAVE)

def check_status(client_socket):
    # 检查状态线圈值
    resp = f"Auto Mode: {current_status[0]}, Manual Mode: {current_status[2]}, Main Power: {current_status[4]}, Backup Power: {current_status[5]}, Power System: {current_status[6]}, Fuel System: {fuel_system}, Propulsion System: {current_status[7]}\n"
    if current_status[7] == True:
        if current_status == INTENTIONAL_STATUS:
            resp = resp + f"Congratulations, your rocket is now able to be launched!\nHere's your flag: {FLAG}\n"
        else:
            resp = resp + f"I don't know how you did it, but your rocket seems to be able to be launched...? Good luck taking your journey back...\nHere's your flag: {FLAG}\n"
    client_socket.send(resp.encode())

def handle_client(client_socket):
    try:
        while True:
            data = client_socket.recv(1024)
            if not data:
                return

            # 判断是否为Modbus请求（检查事务ID和协议标识符）
            if len(data) >= 6 and data[2:4] == b'\x00\x00':
                # 转发到Modbus服务器
                with ModbusTcpClient(MODBUS_SERVER_HOST, port=MODBUS_SERVER_PORT) as mb_client:
                    mb_client.socket.send(data)
                    response = mb_client.socket.recv(1024)
                    client_socket.send(response)
                    handle_logic(data)
            elif data.strip().lower() == b'status':
                check_status(client_socket)
            else:
                client_socket.send(b"Unknown request data\n")
    except Exception as e:
        print(f"Client error: {e}")
    finally:
        client_socket.close()

def run_proxy_server():
    proxy = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    proxy.bind(('0.0.0.0', PROXY_PORT))
    proxy.listen(5)
    print(f"[*] Listening on port {PROXY_PORT}")

    while True:
        client_sock, addr = proxy.accept()
        print(f"[*] Accepted connection from {addr[0]}:{addr[1]}")
        handler = Thread(target=handle_client, args=(client_sock,))
        handler.start()

if __name__ == "__main__":
    # 初始化Flag（仅首次运行）
    try:
        with ModbusTcpClient(MODBUS_SERVER_HOST, port=MODBUS_SERVER_PORT) as client:
            selected_device = random.randint(0,3)
            for i in range(4):
                if i != selected_device:
                    amount = random.randint(1,74)
                else:
                    amount = random.randint(75,100)
                builder = BinaryPayloadBuilder(byteorder=Endian.BIG, wordorder=Endian.BIG)
                builder.add_16bit_int(amount)
                # print(builder.to_registers())
                client.write_registers(address=FUEL_DEVICE_START+i, values=builder.to_registers(), slave=FUEL_SLAVE)
            print("[*] Initialized fuel registers")
            client.write_coil(address=STATUS_COILS["Auto Mode"], value=True, slave=STATUS_SLAVE)
            print("[*] Initialized status coils")
    except Exception as e:
        print(f"Initialization failed: {e}")

    # 启动代理服务器
    run_proxy_server()

为了防止泄题，本人对代码的部分数据进行了改动，不过基础逻辑大致相同，应该还是能正常运行的

至于Repo我就不放了，现在想想也没啥大不了的是吧，各位感兴趣的可以根据自己的想法去搓，我个人很多东西没有考虑进去，所以也只是抛砖引玉

扫一扫，分享到微信

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia-plus根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

一个喜欢睡觉的fw~~~