From ba247d48f47bf6508cfe57627f239a6c561b4af9 Mon Sep 17 00:00:00 2001 From: Kaiser <1727168348@qq.com> Date: Fri, 16 Oct 2026 13:56:51 +0800 Subject: [PATCH] fix: reject unsupported RTU exception responses --- core/protocol_engine.py | 2 + docs/modbus-rtu-exceptions.md | 19 +++++ tests/test_modbus_rtu_exceptions.py | 106 ++++++++++++++++++++++++++++ 3 files changed, 127 insertions(+) create mode 100644 docs/modbus-rtu-exceptions.md create mode 100644 tests/test_modbus_rtu_exceptions.py diff --git a/core/protocol_engine.py b/core/protocol_engine.py index a072be8..3a1fb02 100644 --- a/core/protocol_engine.py +++ b/core/protocol_engine.py @@ -243,6 +243,8 @@ def find_modbus_rtu_response(data: bytes, slave_id: int, function_code: int, continue if is_exception: code = frame[2] + if code not in MODBUS_EXCEPTION_CODES: + return False, f"Modbus RTU异常码0x{code:02X}未定义", frame if accept_exception: return True, f"设备返回Modbus异常码0x{code:02X}，按配置允许异常响应", frame return False, f"设备返回Modbus异常码0x{code:02X}", frame diff --git a/docs/modbus-rtu-exceptions.md b/docs/modbus-rtu-exceptions.md new file mode 100644 index 0000000..1bf3876 --- /dev/null +++ b/docs/modbus-rtu-exceptions.md @@ -0,0 +1,19 @@ +# Modbus RTU read exceptions / 只读异常响应 + +The configured RTU read probe validates station, function, frame length and CRC before considering an exception. `accept_exception: true` permits a defined exception response as communication evidence; it does not make an arbitrary exception byte valid or prove that a register was read successfully. + +RTU now uses the same existing exception-code contract as the TCP validators: `01 02 03 04 05 06 08 0A 0B`. A byte outside that set fails even when exceptions are enabled. The original frame remains in the connection evidence with a diagnostic identifying the unsupported code. The flag's default false behaviour and all defined-code acceptance messages remain unchanged. + +配置允许异常不等于绕过报文语义。比如`01 83 00 41 30`的CRC正确，但异常码00不在当前只读探针合同中，因此返回“需确认”；`01 83 02 C0 F1`是已有合同支持的异常，仍仅在明确允许异常时通过，并保留异常说明。它不代表寄存器读取成功、型号认证或设备健康。 + +The parser retains its existing first-matching-frame policy, including a rejected exception. Configured `any` mode can continue to a later probe; `all` mode stops at a failed probe. The change does not add serial reads/writes, retries, transport framing, discovery scans or new timeout behaviour. Normal response bytes and read-only requests remain unchanged. Legacy/vendor-specific profiles outside this contract require an explicit matching protocol; this check is not a declaration that every legacy diagnostic function uses this set. + +回归覆盖支持/不支持的异常码、CRC错误、截断、站号/功能不匹配，以及实际配置检测方法的Fake串口any/all行为与端口关闭。测试不打开真实串口，不向未知设备写入，不执行现场认证。 + +```bash +python -m pytest -q tests/test_modbus_rtu_exceptions.py +``` + +References: [Modbus Application Protocol V1.1b3 §7](https://www.modbus.org/file/secure/modbusprotocolspecification.pdf), [Serial Line V1.02 CRC/framing](https://www.modbus.org/file/secure/modbusoverserial.pdf). Prior TCP discovery work is documented separately in [modbus-discovery.md](modbus-discovery.md). + +Implementation, regression tests and documentation are AI-assisted. / 本次实现、回归及文档均AI辅助。 diff --git a/tests/test_modbus_rtu_exceptions.py b/tests/test_modbus_rtu_exceptions.py new file mode 100644 index 0000000..788ff2e --- /dev/null +++ b/tests/test_modbus_rtu_exceptions.py @@ -0,0 +1,106 @@ +"""Read-probe exception regressions; use real parsers and Fake serial, no hardware.""" +import pytest + +from core.protocol_engine import append_checksum, find_modbus_rtu_response + + +def exception_frame(code, slave=1, function=3): + # CRC is input construction, not the expected classification under test. + return append_checksum(bytes([slave, function | 0x80, code]), 'crc16_modbus') + + +@pytest.mark.parametrize('code', [0x00, 0x07, 0x09, 0xFF]) +@pytest.mark.parametrize('accept', [False, True]) +def test_unsupported_exception_cannot_pass_even_when_exceptions_allowed(code, accept): + frame = exception_frame(code) + ok, note, captured = find_modbus_rtu_response(frame, 1, 3, accept_exception=accept, expected_count=1) + assert not ok + assert captured == frame + assert '未定义' in note + + +@pytest.mark.parametrize('code', [0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x08, 0x0A, 0x0B]) +@pytest.mark.parametrize('accept', [False, True]) +def test_defined_exception_preserves_explicit_acceptance_policy(code, accept): + frame = exception_frame(code) + ok, note, captured = find_modbus_rtu_response(frame, 1, 3, accept_exception=accept, expected_count=1) + assert ok is accept + assert captured == frame + assert f'0x{code:02X}' in note + assert ('按配置允许' in note) is accept + + +@pytest.mark.parametrize('bad', [ + bytes.fromhex('01 83 02 C0'), # truncated standard exception + bytes.fromhex('01 83 02 C0 F0'), # known frame's CRC bit changed + bytes.fromhex('02 83 02 30 F1'), # different target with valid CRC + bytes.fromhex('01 84 02 C2 C1'), # different function with valid CRC +]) +def test_exception_flag_cannot_bypass_framing_identity_or_crc(bad): + assert not find_modbus_rtu_response(bad, 1, 3, accept_exception=True, expected_count=1)[0] + + +class SerialReply: + def __init__(self, replies): + self.replies = iter(replies) + self.requests = [] + self.closed = False + self.reset_count = 0 + def reset_input_buffer(self): self.reset_count += 1 + def write(self, packet): self.requests.append(packet) + def flush(self): pass + def read(self, size): + assert size == 256 + return next(self.replies) + def close(self): self.closed = True + + +def configured(monkeypatch, replies, *, accept=True, mode='any'): + from core import generic_detector + port = SerialReply(replies) + def open_fake(**kwargs): + assert kwargs == {'port':'FAKE-RTU-ONLY', 'baudrate':9600, 'bytesize':8, + 'parity':'N', 'stopbits':1., 'timeout':.02} + return port + monkeypatch.setattr(generic_detector.serial, 'Serial', open_fake) + detector = object.__new__(generic_detector.GenericDetector) + detector.connection_tests = [] + protocol = {'type':'modbus_rtu', 'slave_id':1, 'accept_exception':accept, 'match_mode':mode, + 'test_registers':[{'address':i, 'count':1} for i in range(len(replies))]} + result = detector._test_modbus_rtu_configured('FAKE-RTU-ONLY', {'timeout':.02}, protocol) + assert port.closed + assert port.reset_count == len(port.requests) == len(detector.connection_tests) + # Independent request literals: read-only function03, register0/1, quantity1. + assert [p.hex() for p in port.requests] == ['010300000001840a','010300010001d5ca'][:len(port.requests)] + return result, detector.connection_tests, port + + +def test_configured_read_does_not_mark_undefined_exception_as_responding(monkeypatch): + result, evidence, _ = configured(monkeypatch, [exception_frame(0)]) + assert not result['ok'] + assert result['status'] == '需确认' + assert '未定义' in result['summary'] + assert evidence[0]['response'] == '01 83 00 41 30' + + +def test_any_mode_can_continue_to_a_later_valid_read(monkeypatch): + read = bytes.fromhex('01 03 02 00 2A 39 9B') + result, evidence, port = configured(monkeypatch, [exception_frame(0xFF), read]) + assert result['ok'] + assert len(port.requests) == 2 + assert [x['status'] for x in evidence] == ['需确认', '有响应'] + + +def test_all_mode_stops_at_unsupported_exception(monkeypatch): + result, evidence, port = configured(monkeypatch, [exception_frame(0), exception_frame(2)], mode='all') + assert not result['ok'] + assert len(port.requests) == len(evidence) == 1 + + +@pytest.mark.parametrize('accept', [False, True]) +def test_configured_defined_exception_keeps_existing_policy_and_evidence(monkeypatch, accept): + result, evidence, _ = configured(monkeypatch, [bytes.fromhex('01 83 02 C0 F1')], accept=accept) + assert result['ok'] is accept + assert ('按配置允许' in result['summary']) is accept + assert '0x02' in result['summary'] + assert evidence[0]['response'] == '01 83 02 C0 F1'