Manual Packet Manipulation for Coverage Testing

[scully-jmccarthy]

I am using this library for implementing coverage tests of a Modbus product. While writing the coverage tests for coils illegal data value, I attempted to use the trace_packet handler so I could mangle the bytes to an invalid number outside of the range 1-2000 to validate that my device returns the 0x03, Illegal data value, exception code.

The issue is that the API still checks even when using this lower level handler to inject other data.

    def verifyCount(self, max_count: int, count: int = -1) -> None:
        """Validate API supplied count."""
        if count == -1:
            count = self.count
        if not 1 <= count <= max_count:
>           raise ValueError(f"1 < count {count} < {max_count} !")
E           ValueError: 1 < count 0 < 2000 !

Is there a recommended method to implement this type of testing with the library, or will I have to go deeper and mangle between the pdu and serial layer?

[janiversen]

Your handler do not return anything, so it is not injecting anything.

The trace_packet gets a byte stream as it comes from the serial driver, and are expected to return a byte stream, identical or modified. Please remember that by using a serial connection you make life a bit more difficult for yourself, because you need to regenerate the CRC.

We use trace_packet to check the framer layer, and trace_pdu to check the pdu layer...since you want to check an application that I assume uses pymodbus, then trace_pdu would be the logical choice, and it is far easier to work with than trace_packet especially for serial communication.

you are wrong about the layers:

client -> API -> transaction -> pdu (trace_pdu) -> framer -> transport (trace_packet).

[janiversen]

If your product is not using pymodbus, then please explain your setup, because then I am lost.

[scully-jmccarthy]

I am using this library for implementing coverage tests of a Modbus product

I have an existing Modbus product. I am using the pymodbus library to implement a portion of the testing of said device. The product does not run pymodbus. Pymodbus is the library being used for unit testing of my product.

I attempted to use the trace_packet handler so I could mangle the bytes to an invalid number outside of the range 1-2000 to validate that my device returns the 0x03, Illegal data value, exception code.

From the purpose of unit testing my device, I am trying to validate that an improper read coils operation from a client is handled correctly by my server. This case, sending read coils with count 0 coils, which should result in an illegal data value exception response. The top level API does not allow for this, as you mentioned:

client -> API -> transaction -> pdu (trace_pdu) -> framer -> transport (trace_packet).

I do understand the layering, and that is why I tried to use the trace_packet callback which is allegedly at the transport level, which to me would mean that there should be 0 "application" level validation.

The above is the trace of the error reported, not the code. I was trying to use the trace_packet to inject different data, which is supported by the library:
https://pymodbus.readthedocs.io/en/latest/_sources/source/client.rst.txt

As far as relevant code snippets:

    client = ModbusClient.ModbusSerialClient(
        serial_port,
        framer=FramerType.RTU,
        timeout=1,
        retries=0,
        baudrate=baud,
        bytesize=8,
        parity=parity,
        stopbits=stop,
        trace_packet=coil_qty_mangle,
    )
    
    client.connect()

    # Test reading invalid coil count
    try:
        result = client.read_coils(address=0, count=10, device_id=slave_addr)
    except ModbusException as exc:
        print(f"Received library exception {exc}")
        client.close()
        assert 0
    client.close()

and

def coil_qty_mangle(cb_bool, cb_bytes):
    cb_bytes[5] = 0

This works, in that I am successfully mangling the payload; however, the API catches this improper value, so doesn't let it go to the wire, the error I reported above. trace_packet is supposed to be at the transport layer, which means there should be no care about the Modbus application level, unless this library does not follow the OSI model.

[janiversen]

I do not understand why do you mean the API catches this improper value ???? The API is called by your app, then passed down the layers until trace_packet is called so it cannot "catch" anything. When pymodbus receives a packet it calls trace_packet and passes up the layers until the API passes the result to your app, so the API cannot "catch" anything.

if you add a trace_packet method, you of course should only inject the modified data when sending !! sounds a bit like you are injecting on read. And be aware if you do it at this level you have to add a new CRC.

I really believe it would be a lot easier to make your test with the trace_pdu, but I leave that to you.

There are no OSI model for modbus, that is at quite a different level.

[scully-jmccarthy]

There is an OSI model for Modbus, even in their official documentation, so I'm not sure what you are trying to say there. The OSI model fits all communication protocols, just some may or may not implement all levels.

Going back to how this library is structured:
client -> API -> transaction -> pdu (trace_pdu) -> framer -> transport (trace_packet).

I am TRYING to intentionally mangle the packet that goes on the wire to intentionally transmit a noncompliant packet, so that I can validate that my product returns the expected error code. This is a simple error handling testing exercise. I was hoping to use this library since it claims to give access at the frame level, but I have found that this is not true. That is WHY I am accessing this via trace_packet, rather than trace_pdu.

if you add a trace_packet method, you of course should only inject the modified data when sending !! sounds a bit like you are injecting on read. And be aware if you do it at this level you have to add a new CRC.

I am well aware that I need to update the CRC, but the problem is even though I am at the what I believed would be the interface to the transport layer, it appears that the library is still checking for Modbus Application layer things, despite a Modbus Frame being defined in the Transport Layer.

[janiversen]

Your picture points perfectly why there is no OSI model, you find modbus at level 1 but only with the serial protocol.

level 3 missed the modbus network specific part.
level 4 missed the transport specifics (not included in the serial protocol)
level 5 missed the how sessions are handled in modbus
level 6 can be discussed but it would normally be the API.

You did not answer how the API catches the changed packet, I cannot see how that is possible.

Pymodbus do NOT have any modbus specific code below that point, please have a look at the code:

def pdu_send(self, pdu: ModbusPDU, addr: tuple | None = None) -> None:
        """Build byte stream and send."""
        if not self.is_server:
            self.request_dev_id = pdu.dev_id
            self.request_transaction_id = pdu.transaction_id
        packet = self.framer.buildFrame(self.trace_pdu(True, pdu))
        if self.is_sync and self.comm_params.handle_local_echo:
            self.sent_buffer = packet
        self.low_level_send(self.trace_packet(True, packet), addr=addr)

Did you look at our examples, we provide one that even though it is for a server it works identical.

[scully-jmccarthy]

Your picture points perfectly why there is no OSI model, you find modbus at level 1 but only with the serial protocol.

level 3 missed the modbus network specific part.
level 4 missed the transport specifics (not included in the serial protocol)
level 5 missed the how sessions are handled in modbus
level 6 can be discussed but it would normally be the API.

You realize that "my picture" is the official documentation for Modbus that explicitly talks about how most of Modbus is at the Application layer, layer 7, and for serial layer 1 and layer 2.

Just because there is no layer 3 nor layer 4 in Modbus RTU unlike Modbus TCP, does not mean there "is no OSI model". Regardless of understanding the OSI model or not, that is irrelevant to the issue at hand here.

You did not answer how the API catches the changed packet, I cannot see how that is possible.

This is where I do not think you are actually reading what I am writing, and instead debating about what the definition of the OSI model is. Perhaps you should submit a ticket to the Modbus Organization to tell them that their documentation incorrectly talks about the Modbus OSI model: https://www.modbus.org/contact-us

I include the snippets of the code above, where I am using the trace_packet level callback to intentionally mangle the message.

    client = ModbusClient.ModbusSerialClient(
        serial_port,
        framer=FramerType.RTU,
        timeout=1,
        retries=0,
        baudrate=baud,
        bytesize=8,
        parity=parity,
        stopbits=stop,
        trace_packet=coil_qty_mangle,
    )
    
    client.connect()

    # Test reading invalid coil count
    try:
        result = client.read_coils(address=0, count=10, device_id=slave_addr)
    except ModbusException as exc:
        print(f"Received library exception {exc}")
        client.close()
        assert 0
    client.close()

def coil_qty_mangle(cb_bool, cb_bytes):
    cb_bytes[5] = 0

When I try to do this, at runtime I get the following error from your library:

    def verifyCount(self, max_count: int, count: int = -1) -> None:
        """Validate API supplied count."""
        if count == -1:
            count = self.count
        if not 1 <= count <= max_count:
>           raise ValueError(f"1 < count {count} < {max_count} !")
E           ValueError: 1 < count 0 < 2000 !

This suggests to me that you are incorrectly validating the PDU contents at the Modbus Frame level, before sending it on the wire which is incorrect to the Modbus specification. Since I have an oscilloscope on the data lines I can confirm that this error is happening on the write side of the pymodbus library, not the read side. After reviewing your unit tests, I do not see any indication of real world testing, only mocks for the serial interface, so I am assuming that this functionality has never been really testing, and why you may assume the low level handlers work the way you think they should. For my purposes, the answer to my original question is clear, that no this library does not support what I need. This is fine, as a Modbus library doesn't necessarily need to handle forced invalid packets, but the whole point of my issue and question was whether or not that was intended or not since the documentation seemed to suggest it should allow for it.

I will handle my coverage testing manually instead of using this library.

[janiversen]

Your trace_packet function changes data both for sending and receiving, you need to check the bool flag.

[scully-jmccarthy]

So, I think there was a large misunderstanding of my original post. The output that I had shown in that post, was not my handler, it was the error message that I got when running:

pymodbus/pymodbus/pdu/pdu.py

Lines 47 to 52 in ac88891

	def verifyCount(self, max_count: int, count: int = -1) -> None:
	"""Validate API supplied count."""
	if count == -1:
	count = self.count
	if not 1 <= count <= max_count:
	raise ValueError(f"1 < count {count} < {max_count} !")

My subsequent confusion was from you telling me the code was wrong, when it was not my code, it was the library code giving an error.

I haven't been able to figure out why, but when my trace_packet handler had no byte string, that somehow causes the coil PDU error in my original message. This is improper use of the library, so it is not a library problem. After inspection of the source, I have no clue how it is happening on my machine.

Regardless, in case anyone in the future wants to use this library to inject faults for testing Modbus devices, below is an example of validating read coil invalid count exception behavior:

def coil_qty_mangle(cb_bool, cb_bytes):
    if cb_bool == True: # TX
        return b'\x01\x01\x00\x00\x00\x00\x3c\x0a'
    return cb_bytes

@pytest.mark.parametrize(
        "baud, parity, stop",
        [
            pytest.param(9600, "E", 1, marks=pytest.mark.smoke),
            (9600, "N", 2),
            (19200, "E", 1),
            (19200, "N", 2),
            (38400, "E", 1),
            (38400, "N", 2),
            (115200, "E", 1),
            (115200, "N", 2),
        ],
    )
def test_invalid_qty_coil_read(rig: fixture.Rig, baud, parity, stop):
    serial_port = rig.get_port()
    slave_addr= rig.get_slave_addr()
    
    client = ModbusClient.ModbusSerialClient(
        serial_port,
        framer=FramerType.RTU,
        timeout=1,
        retries=0,
        baudrate=baud,
        bytesize=8,
        parity=parity,
        stopbits=stop,
        trace_packet=coil_qty_mangle,
    )
    
    client.connect()

    # Test reading invalid coil count
    try:
        result = client.read_coils(address=0, count=2, device_id=slave_addr)
    except ModbusException as exc:
        print(f"Received library exception {exc}")
        client.close()
        assert 0
    client.close()

    if result.isError():
        print(f"Received exception from device ({result})")

    assert result.isError() == True
    assert result.dev_id == slave_addr
    assert result.function_code == 0x81
    assert result.exception_code == 3

If I have a minute to spare, I will submit a PR to add a little more information to the docs for how the trace functions should be used, so source inspection to reverse engineer it is not required to understand how to use them.

Additionally, would you be open to me submitting a PR with something like:

pymodbus/pymodbus/transaction/transaction.py

Lines 203 to 211 in ac88891

	def pdu_send(self, pdu: ModbusPDU, addr: tuple | None = None) -> None:
	"""Build byte stream and send."""
	if not self.is_server:
	self.request_dev_id = pdu.dev_id
	self.request_transaction_id = pdu.transaction_id
	packet = self.framer.buildFrame(self.trace_pdu(True, pdu))
	if self.is_sync and self.comm_params.handle_local_echo:
	self.sent_buffer = packet
	self.low_level_send(self.trace_packet(True, packet), addr=addr)

packet = self.trace_packet(True, packet)
if packet == None:
    Log.warning("WARNING: trace_packet handler returned no buffer, no bytes sent")
else:
    self.low_level_send(self.trace_packet(True, packet), addr=addr)

To mitigate this confusion in the future? I would also submit a comparable one for trace_pdu.

[janiversen]

We are always open to PRs !

A better documentation have been on my list for very long time, so that is more than welcome.

The signature of the trace_packet, clearly defines that only bytes are allowed return value, so it really should not be needed.

However I am not against such a check, because it does not hurt...but then if should be done for all trace functions, and in a separate PR from the trace documentation.

If you had used the PDU trace, you could easily have tested with a lot of different values, because at that level you deal with registers/values and do not care about CRC. The trace_packet level is really only meant to deal with malformed frames etc.

[janiversen]

Closing as there are no pymodbus bug.

PRs are welcome.