This is an automated email from the ASF dual-hosted git repository.
sruehl pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/plc4x.git
The following commit(s) were added to refs/heads/develop by this push:
new c771277ea2 fix: robust Modbus Receive with desync detection and resync
ability (#2362)
c771277ea2 is described below
commit c771277ea27cf75be902609c6b4cc55dc9c33de9
Author: Shaun <[email protected]>
AuthorDate: Thu May 28 00:16:24 2026 -0700
fix: robust Modbus Receive with desync detection and resync ability (#2362)
* fix: harden Modbus Receive against truncated/extended frames
- retry parsing with full buffered data when initial MBAP length is wrong
- treat EOF as incomplete data and avoid dropping partial frames
- keep discarding truly unparsable packets with diagnostic logging
* fix: watch for and discard trailing CRCs from misbehaving gateways
fix: attempt to resynchronize the read stream if desynchronized
* fix: additional sanity checks on the MBAP
fix: io.EOF is a trap, we checked fragmentation above
* fix: deal with TCP keep-alive padding that leaks from the kernel
* refactor: attempting to simply logic while still covering desync handling
* refactor: more robust consistency checks
* fix: Final consistency check case should discard all available bytes
* refactor: reduce log spam
* fix: handleDesync more robustly handles the padding leak issue
* fix: keep the last 5 bytes to avoid breaking fragmentation
---
plc4go/internal/modbus/MessageCodec.go | 378 +++++++++++++++++++++++++++++----
plc4go/spi/utils/base64Stringer.go | 36 ++++
2 files changed, 377 insertions(+), 37 deletions(-)
diff --git a/plc4go/internal/modbus/MessageCodec.go
b/plc4go/internal/modbus/MessageCodec.go
index 5aa6456a5a..1c388e7987 100644
--- a/plc4go/internal/modbus/MessageCodec.go
+++ b/plc4go/internal/modbus/MessageCodec.go
@@ -7,7 +7,7 @@
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
- * https://www.apache.org/licenses/LICENSE-2.0
+ * https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
@@ -21,7 +21,8 @@ package modbus
import (
"context"
- "encoding/base64"
+ "fmt"
+ "io"
"github.com/rs/zerolog"
@@ -31,6 +32,7 @@ import (
"github.com/apache/plc4x/plc4go/spi/errors"
"github.com/apache/plc4x/plc4go/spi/options"
"github.com/apache/plc4x/plc4go/spi/transports"
+ "github.com/apache/plc4x/plc4go/spi/utils"
)
//go:generate go tool plc4xGenerator -type=MessageCodec
@@ -88,6 +90,7 @@ func (m *MessageCodec) Receive(ctx context.Context)
(spi.Message, error) {
return nil, errors.New("Transport instance not connected")
}
+ // 1. Fill the buffer
if err := ti.FillBuffer(ctx, func(pos uint, currentByte byte, reader
transports.ExtendedReader) bool {
m.log.Trace().Uint("pos", pos).Uint8("currentByte",
currentByte).Msg("filling")
numBytesAvailable, err := ti.GetNumBytesAvailableInBuffer()
@@ -95,50 +98,351 @@ func (m *MessageCodec) Receive(ctx context.Context)
(spi.Message, error) {
m.log.Debug().Err(err).Msg("error getting available
bytes")
return false
}
- m.log.Trace().Uint32("numBytesAvailable",
numBytesAvailable).Msg("check available bytes < 6")
return numBytesAvailable < 6
}); err != nil {
- m.log.Debug().Err(err).Msg("error filling buffer")
+ if err != io.EOF {
+ m.log.Debug().Err(err).Msg("error filling buffer")
+ }
+ // Fall through on errors, we might have enough data...
}
- // We need at least 6 bytes in order to know how big the packet is in
total
- if num, err := ti.GetNumBytesAvailableInBuffer(); (err == nil) && (num
>= 6) {
- m.log.Debug().Uint32("num", num).Msg("we got num readable
bytes")
- data, err := ti.PeekReadableBytes(ctx, 6)
- if err != nil {
- m.log.Warn().Err(err).Msg("error peeking")
- // TODO: Possibly clean up ...
+ // 2. Check buffer status
+ numBytesAvail, err := ti.GetNumBytesAvailableInBuffer()
+ if err != nil {
+ // Yield if we can't check buffer
+ if err == io.EOF {
return nil, nil
}
- // Get the size of the entire packet
- packetSize := (uint32(data[4]) << 8) + uint32(data[5]) + 6
- if num < packetSize {
- m.log.Debug().
- Uint32("num", num).
- Uint32("packetSize", packetSize).Msg("Not
enough bytes. Got: num Need: packetSize")
- return nil, nil
+ return nil, fmt.Errorf("error getting buffer length")
+ }
+
+ // Need at least 6 bytes for MBAP header
+ if numBytesAvail < 6 {
+ return nil, nil
+ }
+
+ // Re-peek the header at the current head
+ header, err := ti.PeekReadableBytes(ctx, 6)
+ if err != nil {
+ m.log.Warn().Err(err).Msg("error peeking header")
+ return nil, nil
+ }
+
+ //
-------------------------------------------------------------------------
+ // MBAP SANITY CHECKS
+ //
-------------------------------------------------------------------------
+
+ // We now peek up to 9 bytes (or whatever is available) to perform the
full Consistency Check.
+ // IMPORTANT: If we have 8 bytes (Header + Unit + Func), we MUST pass
all 8 bytes
+ // so that checkPacketConsistency can validate the Function Code and
Length limits.
+ // If we only pass 6, it will skip the Function check and potentially
accept huge lengths.
+ var checkBytes []byte
+ peekLen := uint32(9)
+ if numBytesAvail < 9 {
+ peekLen = numBytesAvail
+ }
+ checkBytes, _ = ti.PeekReadableBytes(ctx, peekLen)
+
+ // Perform Consistency Check
+ if !m.checkPacketConsistency(checkBytes) {
+ return m.handleDesync(ctx, "Sanity Check Failed",
map[string]interface{}{
+ "data": utils.Base64Stringer(checkBytes),
+ })
+ }
+
+ //
-------------------------------------------------------------------------
+ // PARSING
+ //
-------------------------------------------------------------------------
+
+ // Length field is big endian encoded WORD
+ payloadLength := (uint32(header[4]) << 8) + uint32(header[5])
+ packetSize := payloadLength + 6
+
+ // Yield on TCP fragmentation
+ if numBytesAvail < packetSize {
+ // Wait for more data (standard TCP fragmentation handling)
+ var peekedBytes []byte
+ if m.log.Debug().Enabled() {
+ peekedBytes, _ = ti.PeekReadableBytes(ctx,
numBytesAvail)
}
- data, err = ti.Read(ctx, packetSize)
- if err != nil {
- // TODO: Possibly clean up ...
- return nil, nil
+ m.log.Debug().
+ Stringer("dataFragment",
utils.Base64Stringer(peekedBytes)).
+ Uint32("num", numBytesAvail).
+ Uint32("packetSize", packetSize).Msgf("Received
fragment. Got: %d Need: %d. Waiting for more data...", numBytesAvail,
packetSize)
+ return nil, nil
+ }
+
+ // Read the entire frame
+ frameSlice, err := ti.PeekReadableBytes(ctx, packetSize)
+ if err != nil {
+ m.log.Warn().Err(err).Msg("Error peeking frame slice")
+ return nil, nil
+ }
+
+ // Parse the frame
+ ctxForModel := options.GetLoggerContextForModel(ctx, m.log,
options.WithPassLoggerToModel(m.passLogToModel))
+ tcpAdu, err := model.ModbusADUParse[model.ModbusTcpADU](ctxForModel,
frameSlice, model.DriverType_MODBUS_TCP, true)
+ if err != nil {
+ // Parser wasn't happy at packetSize, if there is more
available, try parsing all of it
+ if numBytesAvail > packetSize {
+ extendedSlice, extendedErr := ti.PeekReadableBytes(ctx,
numBytesAvail)
+ if extendedErr == nil {
+ // Try parsing it all
+ extendedAdu, extendedParseErr :=
model.ModbusADUParse[model.ModbusTcpADU](ctxForModel, extendedSlice,
model.DriverType_MODBUS_TCP, true)
+ if extendedParseErr == nil {
+ // Parse succeded...
+ // NOTE: MBAP length errors are rare in
well-behaved devices, but they happen when
+ // firmware miscomputes the length
field (or hard-codes to 6), mixes RTU framing
+ // with TCP transport, or when
intermediate gateways/proxies truncate or merge
+ // frames. Duplicate reads and
out-of-order slicing in buggy drivers can also leave
+ // stale data that makes the reported
length disagree with the actual payload.
+ // ... looking at you old Moxa and
Lantronix "transparent" serial-to-tcp gateways.
+
+ // What is the actual ADU size parsed?
+ actualSize :=
uint32(extendedAdu.GetLengthInBytes(ctxForModel))
+ if actualSize > numBytesAvail {
+ return nil, nil
+ }
+
+ m.log.Info().
+ Uint32("reportedSize",
packetSize).
+ Uint32("actualSize",
actualSize).
+ Stringer("extendedData",
utils.Base64Stringer(extendedSlice)).
+ Stringer("consumedData",
utils.Base64Stringer(extendedSlice[:actualSize])).
+ Msg("MBAP had wrong/hardcoded
length. Consumed extended frame with corrected size")
+ if _, consumeErr := ti.Read(ctx,
actualSize); consumeErr != nil {
+
m.log.Debug().Err(consumeErr).Msg("error consuming extended frame")
+ return nil, nil
+ }
+ return extendedAdu, nil
+ }
+ }
}
- ctxForModel := options.GetLoggerContextForModel(ctx, m.log,
options.WithPassLoggerToModel(m.passLogToModel))
- tcpAdu, err :=
model.ModbusADUParse[model.ModbusTcpADU](ctxForModel, data,
model.DriverType_MODBUS_TCP, true)
- if err != nil {
- dataStr := base64.StdEncoding.EncodeToString(data)
- m.log.Warn().Err(err).
- Str("data", dataStr). // Max PDU size is 253
bytes, and catching parse errors for inspection is important
- Uint32("packetSize", packetSize).
- Msg("error parsing")
- // TODO: Possibly clean up ...
- return nil, nil
+
+ // Seems unparsable - log and discard
+ // NOTE: If checkPacketConsistency PASSED, but Parse FAILED, it
means the header
+ // structure is valid (Length, Proto, Func), but the Content is
bad.
+ // We MUST discard this frame to advance the stream.
+ // We DO NOT call handleDesync here because we don't want to
scan ahead;
+ // we just want to eat the bad packet and try the next one.
+ m.log.Warn().Err(err).
+ Stringer("data", utils.Base64Stringer(frameSlice)).
+ Uint32("packetSize", packetSize).
+ Msg("Error parsing frame. Discarding invalid frame.")
+
+ // Discard the unparsable frame from the buffer
+ if _, discardErr := ti.Read(ctx, packetSize); discardErr != nil
{
+ m.log.Debug().Err(discardErr).Uint32("packetSize",
packetSize).Msg("error discarding unparsable frame")
+ return nil, discardErr
+ }
+
+ // Yield
+ return nil, nil
+ }
+
+ //
-------------------------------------------------------------------------
+ // SUCCESS
+ //
-------------------------------------------------------------------------
+ if _, consumeErr := ti.Read(ctx, packetSize); consumeErr != nil {
+ m.log.Debug().Err(consumeErr).Msg("error consuming parsed
frame")
+ return nil, nil
+ }
+
+ return tcpAdu, nil
+}
+
+// checkPacketConsistency validates a candidate Modbus TCP packet against the
protocol spec.
+// It enforces strict relationship rules between the MBAP Header Length and
the PDU content (Function Code & Byte Count).
+// Returns true if the packet looks valid, false if it is definitely garbage.
+// NOTE: When new functions are added to the mspec, this needs to be updated
to reflect their structure.
+func (m *MessageCodec) checkPacketConsistency(data []byte) bool {
+ // 1. Check Protocol ID immediately (Need 4 bytes)
+ if len(data) < 4 {
+ return false
+ }
+ // Modbus TCP Protocol ID must be 0
+ if data[2] != 0x00 || data[3] != 0x00 {
+ return false
+ }
+
+ // 2. Check Header Length (Need 6 bytes)
+ if len(data) < 6 {
+ // Not enough data to check length, but Protocol ID was OK.
+ // We shouldn't reject yet if we just don't have the bytes.
+ // However, this function is typically called with a slice that
*should* contain the header.
+ return false
+ }
+
+ length := (uint32(data[4]) << 8) + uint32(data[5])
+ if length < 2 {
+ return false
+ }
+
+ // 3. Check Function Code (Need 8 bytes: 6 Header + 1 Unit + 1 Func)
+ if len(data) < 8 {
+ // We have a valid ProtoID and Length, but not enough data to
check content.
+ // Return true (benefit of the doubt) because we can't prove
it's bad yet.
+ return true
+ }
+
+ fc := data[7]
+ if fc == 0 {
+ return false
+ }
+
+ // VALIDATION: Check Internal Consistency of Response Structure
+ // NOTE: These rules are primarily for Modbus TCP RESPONSES (Client
Mode).
+ // If acting as a Server (receiving Requests), some rules (like Write
Multiple) would be different.
+
+ if fc >= 0x80 {
+ // --- EXCEPTION ---
+ // Structure: [UnitID] [FuncCode+0x80] [ExceptionCode]
+ // Length must be exactly 3.
+ return length == 3
+ }
+
+ // --- STANDARD FUNCTIONS ---
+ switch fc {
+ case 0x01, 0x02, 0x03, 0x04, 0x17, 0x14, 0x15:
+ // Variable Length Responses
+ // Rule: Header Length == ByteCount + 3
+ // We need 9 bytes to see the ByteCount at offset 8.
+ if len(data) >= 9 {
+ byteCount := uint32(data[8])
+ if length != byteCount+3 {
+ return false
+ }
+ }
+ // Implicit Max Check: Max ByteCount 255 -> Max Length 258.
+ if length > 258 {
+ return false
+ }
+
+ case 0x05, 0x06, 0x0F, 0x10:
+ // Fixed Length Responses (Write Single, Write Multi)
+ // Structure: [UnitID] [FuncCode] [AddrHi] [AddrLo] [ValHi]
[ValLo]
+ // Length must be exactly 6.
+ if length != 6 {
+ // SERVER MODE CAVEAT:
+ // If we are a Server receiving a Write Multiple
Request (FC 15/16),
+ // the length will be > 6. If we strictly return false
here, we break Server mode.
+ // As a heuristic, if length is > 6 for these codes, we
treat it as potentially valid
+ // (assuming it's a Request) to be safe, unless it's
huge.
+ if length > 260 {
+ return false
+ }
+ // Ideally we would enforce Request structure (Len =
ByteCount + 7),
+ // but that requires more bytes than we might have
peeked.
+ return true
+ }
+
+ case 0x16:
+ // Mask Write Register
+ // Length must be exactly 8.
+ if length != 8 {
+ return false
+ }
+
+ default:
+ // Other/Custom Functions.
+ // Max PDU size is 253 -> Length 254. Allow margin.
+ if length > 260 {
+ return false
+ }
+ }
+
+ return true
+}
+
+// handleDesync handles stream realignment when an invalid header is detected
at the head.
+// It strictly scans the available buffer for a valid MBAP header using
checkPacketConsistency.
+// If one is found, it realigns the stream.
+// If NO valid header is found in the *entire* buffer, it treats the
connection as dead.
+func (m *MessageCodec) handleDesync(ctx context.Context, reason string, fields
map[string]interface{}) (spi.Message, error) {
+ ti := m.GetTransportInstance()
+
+ // Get total available bytes
+ numBytesAvail, err := ti.GetNumBytesAvailableInBuffer()
+ if err != nil {
+ return nil, err // Yield if we can't check buffer
+ }
+
+ // Create a logger with context
+ fields["reason"] = reason
+ fields["bytesAvail"] = numBytesAvail
+ opLog := m.log.With().
+ Interface("desyncContext", fields).
+ Logger()
+
+ // Log on the way out to minimize log spam
+ dispMsg := ""
+ dispLevel := zerolog.WarnLevel
+ defer func() {
+ opLog.WithLevel(dispLevel).Msg("Desync detected at stream head:
" + dispMsg)
+ }()
+
+ // CASE 1: Small Buffer (< 10 bytes).
+ // We don't have enough data to scan for a full header+function (need
~8-9 bytes min).
+ // We can't definitively say the stream is dead, so we just Discard 1
and Yield.
+ if numBytesAvail < 10 {
+ dispMsg = "Small buffer - Discard 1"
+ dispLevel = zerolog.DebugLevel
+ if _, err := ti.Read(ctx, 1); err != nil {
+ opLog.Debug().Err(err).Msg("Error reading byte during
discard")
+ return nil, err // Return error to kill connection if
we can't consume
}
- return tcpAdu, nil
- } else if err != nil {
- m.log.Warn().Err(err).Msg("Got error reading")
return nil, nil
}
- // TODO: maybe we return here a not enough error error
- return nil, nil
+
+ // CASE 2: Scan for Recovery.
+ // We peek everything we have.
+ allBytes, err := ti.PeekReadableBytes(ctx, numBytesAvail)
+ if err != nil {
+ return nil, fmt.Errorf("error peeking bytes during desync
recovery")
+ }
+
+ // Scan Loop: Start at offset 1 (since offset 0 is known bad).
+ // We verify candidates up to the end of the buffer.
+ // We stop when we don't have enough bytes left to even check the
Protocol ID (4 bytes).
+ // ... This deals with any gateway forwarding RTU CRC16 extra bytes or
leaky kernels that
+ // shove TCP keep-alive Ethernet II frame padding bytes into the stream.
+ limit := uint32(0)
+ if numBytesAvail >= 4 {
+ limit = numBytesAvail - 4
+ }
+
+ for i := uint32(1); i <= limit; i++ {
+ // Use our robust consistency check on the slice starting at i
+ if m.checkPacketConsistency(allBytes[i:]) {
+ dispMsg = fmt.Sprintf("Found MBAP candidate at +%d", i)
+ dispLevel = zerolog.DebugLevel
+
+ // Discard 'i' bytes to align the stream to this
candidate
+ if _, err := ti.Read(ctx, i); err != nil {
+ opLog.Debug().Err(err).Msg("Error discarding
garbage during recovery")
+ return nil, err // Return error to kill
connection
+ }
+
+ // Yield. Next Receive() will pick up this valid header.
+ return nil, nil
+ }
+ }
+
+ // CASE 3: We scanned the entire available buffer and found NO valid
candidates.
+ // We want to keep the last 5 bytes, as they might be the start of a
header
+ // (TransID[2] + ProtoID[2] + Len[1 here, 1 missing]) that completes in
the
+ // next TCP packet of at least 1 byte.
+ bytesToDiscard := numBytesAvail
+ if numBytesAvail > 5 {
+ bytesToDiscard = numBytesAvail - 5
+ }
+
+ dispMsg = fmt.Sprintf("No MBAP candidate found. Discarding (%d) bytes,
keeping 5-byte tail", bytesToDiscard)
+ dispLevel = zerolog.InfoLevel
+ if _, err := ti.Read(ctx, bytesToDiscard); err != nil {
+ opLog.Debug().Err(err).Msg("Error discarding garbage during
recovery")
+ return nil, err // Return error to kill connection
+ }
+
+ return nil, fmt.Errorf("stream desynchronized: discarded %d bytes of
garbage", bytesToDiscard)
}
diff --git a/plc4go/spi/utils/base64Stringer.go
b/plc4go/spi/utils/base64Stringer.go
new file mode 100644
index 0000000000..098421debb
--- /dev/null
+++ b/plc4go/spi/utils/base64Stringer.go
@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package utils
+
+import (
+ "encoding/base64"
+)
+
+// Base64Stringer is a helper to print byte slices as base64 encoded strings
+// and can be used in structured logging, so that String() is only called when
+// the log level requires it.
+type Base64Stringer []byte
+
+func (b Base64Stringer) String() string {
+ if b == nil {
+ return "<nil>"
+ }
+ return base64.StdEncoding.EncodeToString(b)
+}
