Fix routing of unsupported MEI types

pymodbus/pdu/mei_message.py

@@ -31,6 +31,28 @@ def __init__(self, oid: int) -> None:
         self.oid = oid
 
 
+class _EncapsulatedInterfaceTransport(ModbusPDU):
+    """Base PDU for dispatching Encapsulated Interface Transport messages."""
+
+    function_code = 0x2B
+
+    @classmethod
+    def decode_sub_function_code(cls, data: bytes) -> int:
+        """Decode the MEI type."""
+        return int(data[2])
+
+    def decode(self, data: bytes) -> None:
+        """Decode the MEI type."""
+        self.sub_function_code = int(data[0])
+
+    async def datastore_update(
+        self, context: ModbusServerContext, device_id: int
+    ) -> ModbusPDU:
+        """Reject unsupported MEI types."""
+        _ = context, device_id
+        return ExceptionResponse(self.function_code, ExcCodes.ILLEGAL_FUNCTION)
+
+
 class ReadDeviceInformationRequest(ModbusPDU):
     """ReadDeviceInformationRequest."""
 
@@ -190,5 +212,5 @@ def decode(self, data: bytes) -> None:
                 ]
 
 
-DecodePDU.add_pdu(ReadDeviceInformationRequest, ReadDeviceInformationResponse)
+DecodePDU.add_pdu(_EncapsulatedInterfaceTransport, _EncapsulatedInterfaceTransport)
 DecodePDU.add_sub_pdu(ReadDeviceInformationRequest, ReadDeviceInformationResponse)

test/pdu/test_mei_messages.py

@@ -8,7 +8,8 @@
 
 import pytest
 
-from pymodbus.constants import DeviceInformation
+from pymodbus.constants import DeviceInformation, ExcCodes
+from pymodbus.pdu.decoders import DecodePDU
 from pymodbus.pdu.device import ModbusControlBlock
 from pymodbus.pdu.mei_message import (
     ReadDeviceInformationRequest,
@@ -42,6 +43,25 @@ def test_read_device_information_request_decode(self):
         assert handle.read_code == DeviceInformation.BASIC
         assert not handle.object_id
 
+    def test_device_information_mei_type_is_routed(self):
+        """Test MEI type 0x0E is routed to device identification."""
+        pdu = DecodePDU(True).decode(b"\x2b\x0e\x01\x00")
+
+        assert isinstance(pdu, ReadDeviceInformationRequest)
+
+    @pytest.mark.parametrize("mei_type", [0x00, 0x0D, 0x0F, 0xFF])
+    async def test_unsupported_mei_type_is_not_device_information(
+        self, mei_type, mock_server_context
+    ):
+        """Test unsupported MEI types are not routed to device identification."""
+        pdu = DecodePDU(True).decode(bytes([0x2B, mei_type, 0x01, 0x00]))
+
+        assert pdu
+        assert not isinstance(pdu, ReadDeviceInformationRequest)
+        assert pdu.sub_function_code == mei_type
+        response = await pdu.datastore_update(mock_server_context(), 0)
+        assert response.exception_code == ExcCodes.ILLEGAL_FUNCTION
+
     async def test_read_device_information_request(self, mock_server_context):
         """Test basic bit message encoding/decoding."""
         context = mock_server_context()