Backport robustness improvements from python-s7comm (#580)

Backports concrete robustness improvements from nikteliy/python-s7comm
to the native Python S7 implementation:

- 210+ S7 protocol error codes from Wireshark's S7 dissector covering
  USERDATA parameter errors (0xD0xx), protocol errors (0x8xxx), and
  resource errors. parse_response() now raises descriptive S7ProtocolError
  on header errors, and USERDATA response parsing logs warnings.

- Stale packet detection with retry: new validate_pdu_reference()
  validates response sequence numbers. New _send_receive() helper wraps
  the send/receive/parse pattern with automatic retry on stale packets
  (up to 3 retries). Refactored ~15 call sites in client.py to use it.

- Automatic PDU splitting for read_area()/write_area(): requests
  exceeding the negotiated PDU length are automatically chunked,
  preventing PLC rejections on large reads (e.g. db_get() on big DBs).

- TPDUSize enum (ISO 8073) with configurable COTP negotiation: replaces
  hardcoded 0x0A with a proper IntEnum (128-8192 bytes).

- MAX_VARS=20 limit on read_multi_vars()/write_multi_vars(): raises
  ValueError when exceeding the S7 protocol limit.

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

Author: gijzelaerr

CLAUDE.md AGENTS.mdCLAUDE.md renamed to AGENTS.md

@@ -8,11 +8,28 @@
 import socket
 import struct
 import logging
+from enum import IntEnum
 from typing import Optional, Type
 from types import TracebackType
 
 from .error import S7ConnectionError, S7TimeoutError
 
+
+class TPDUSize(IntEnum):
+    """TPDU sizes per ISO 8073 / RFC 905.
+
+    The value is the exponent: actual size = 2^value bytes.
+    """
+
+    S_128 = 0x07
+    S_256 = 0x08
+    S_512 = 0x09
+    S_1024 = 0x0A
+    S_2048 = 0x0B
+    S_4096 = 0x0C
+    S_8192 = 0x0D
+
+
 logger = logging.getLogger(__name__)
 
 
@@ -44,7 +61,14 @@ class ISOTCPConnection:
     COTP_PARAM_CALLING_TSAP = 0xC1
     COTP_PARAM_CALLED_TSAP = 0xC2
 
-    def __init__(self, host: str, port: int = 102, local_tsap: int = 0x0100, remote_tsap: int = 0x0102):
+    def __init__(
+        self,
+        host: str,
+        port: int = 102,
+        local_tsap: int = 0x0100,
+        remote_tsap: int = 0x0102,
+        tpdu_size: TPDUSize = TPDUSize.S_1024,
+    ):
         """
         Initialize ISO TCP connection.
 
@@ -53,11 +77,13 @@ def __init__(self, host: str, port: int = 102, local_tsap: int = 0x0100, remote_
             port: TCP port (default 102 for S7)
             local_tsap: Local Transport Service Access Point
             remote_tsap: Remote Transport Service Access Point
+            tpdu_size: TPDU size to request during COTP negotiation
         """
         self.host = host
         self.port = port
         self.local_tsap = local_tsap
         self.remote_tsap = remote_tsap
+        self.tpdu_size = tpdu_size
         self.socket: Optional[socket.socket] = None
         self.connected = False
         self.pdu_size = 240  # Default PDU size, negotiated during connection
@@ -241,8 +267,8 @@ def _build_cotp_cr(self) -> bytes:
         calling_tsap = struct.pack(">BBH", self.COTP_PARAM_CALLING_TSAP, tsap_length, self.local_tsap)
         # Called TSAP (remote)
         called_tsap = struct.pack(">BBH", self.COTP_PARAM_CALLED_TSAP, tsap_length, self.remote_tsap)
-        # PDU Size parameter (ISO 8073 code: 0x0A = 1024 bytes)
-        pdu_size_param = struct.pack(">BBB", self.COTP_PARAM_PDU_SIZE, 1, 0x0A)
+        # PDU Size parameter (ISO 8073 code, e.g. 0x0A = 1024 bytes)
+        pdu_size_param = struct.pack(">BBB", self.COTP_PARAM_PDU_SIZE, 1, self.tpdu_size)
 
         parameters = calling_tsap + called_tsap + pdu_size_param
 

snap7/client.py

@@ -14,6 +14,12 @@ def _assert_never(value: NoReturn) -> NoReturn:
     raise AssertionError(f"Unhandled value: {value!r}")
 
 
+def _validate_bit_addr(bit_addr: int) -> None:
+    """Validate that a bit address is in the valid range 0-7."""
+    if not 0 <= bit_addr <= 7:
+        raise ValueError(f"Bit address must be 0-7, got {bit_addr}")
+
+
 class S7Area(IntEnum):
     """S7 memory area identifiers."""
 
@@ -123,12 +129,12 @@ def decode_s7_data(data: bytes, word_len: S7WordLen, count: int) -> List[Union[b
                 values.append(bool(byte_val))
                 offset += 1
 
-            elif word_len in [S7WordLen.BYTE, S7WordLen.CHAR]:
+            elif word_len == S7WordLen.BYTE or word_len == S7WordLen.CHAR:
                 # 8-bit values
                 values.append(data[offset])
                 offset += 1
 
-            elif word_len in [S7WordLen.WORD, S7WordLen.COUNTER, S7WordLen.TIMER]:
+            elif word_len == S7WordLen.WORD or word_len == S7WordLen.COUNTER or word_len == S7WordLen.TIMER:
                 # 16-bit unsigned values (big-endian)
                 value = struct.unpack(">H", data[offset : offset + 2])[0]
                 values.append(value)
@@ -158,6 +164,9 @@ def decode_s7_data(data: bytes, word_len: S7WordLen, count: int) -> List[Union[b
                 values.append(value)
                 offset += 4
 
+            else:
+                _assert_never(word_len)
+
         return values
 
     @staticmethod
@@ -174,11 +183,11 @@ def encode_s7_data(values: Sequence[Union[bool, int, float]], word_len: S7WordLe
                 # Single bit to byte
                 data.append(0x01 if value else 0x00)
 
-            elif word_len in [S7WordLen.BYTE, S7WordLen.CHAR]:
+            elif word_len == S7WordLen.BYTE or word_len == S7WordLen.CHAR:
                 # 8-bit values
                 data.append(int(value) & 0xFF)
 
-            elif word_len in [S7WordLen.WORD, S7WordLen.COUNTER, S7WordLen.TIMER]:
+            elif word_len == S7WordLen.WORD or word_len == S7WordLen.COUNTER or word_len == S7WordLen.TIMER:
                 # 16-bit unsigned values (big-endian)
                 data.extend(struct.pack(">H", int(value) & 0xFFFF))
 
@@ -199,7 +208,7 @@ def encode_s7_data(values: Sequence[Union[bool, int, float]], word_len: S7WordLe
                 data.extend(struct.pack(">f", float(value)))
 
             else:
-                _assert_never(word_len)  # type: ignore[arg-type]
+                _assert_never(word_len)
 
         return bytes(data)
 
@@ -224,10 +233,8 @@ def parse_address(address_str: str) -> Tuple[S7Area, int, int]:
                 # Bit address: DBX10.5
                 if "." in addr_part:
                     byte_addr, bit_addr = addr_part[3:].split(".")
-                    bit_val = int(bit_addr)
-                    if not 0 <= bit_val <= 7:
-                        raise ValueError(f"Bit address must be 0-7, got {bit_val}")
-                    offset = int(byte_addr) * 8 + bit_val
+                    _validate_bit_addr(int(bit_addr))
+                    offset = int(byte_addr) * 8 + int(bit_addr)
                 else:
                     offset = int(addr_part[3:]) * 8
             elif addr_part.startswith("DBB"):
@@ -249,10 +256,8 @@ def parse_address(address_str: str) -> Tuple[S7Area, int, int]:
             if "." in address_str:
                 # Bit address: M10.5
                 byte_addr, bit_addr = address_str[1:].split(".")
-                bit_val = int(bit_addr)
-                if not 0 <= bit_val <= 7:
-                    raise ValueError(f"Bit address must be 0-7, got {bit_val}")
-                offset = int(byte_addr) * 8 + bit_val
+                _validate_bit_addr(int(bit_addr))
+                offset = int(byte_addr) * 8 + int(bit_addr)
             elif address_str.startswith("MW"):
                 # Word address: MW20
                 offset = int(address_str[2:])
@@ -270,10 +275,8 @@ def parse_address(address_str: str) -> Tuple[S7Area, int, int]:
             if "." in address_str:
                 # Bit address: I0.0
                 byte_addr, bit_addr = address_str[1:].split(".")
-                bit_val = int(bit_addr)
-                if not 0 <= bit_val <= 7:
-                    raise ValueError(f"Bit address must be 0-7, got {bit_val}")
-                offset = int(byte_addr) * 8 + bit_val
+                _validate_bit_addr(int(bit_addr))
+                offset = int(byte_addr) * 8 + int(bit_addr)
             elif address_str.startswith("IW"):
                 # Word address: IW10
                 offset = int(address_str[2:])
@@ -291,10 +294,8 @@ def parse_address(address_str: str) -> Tuple[S7Area, int, int]:
             if "." in address_str:
                 # Bit address: Q0.0
                 byte_addr, bit_addr = address_str[1:].split(".")
-                bit_val = int(bit_addr)
-                if not 0 <= bit_val <= 7:
-                    raise ValueError(f"Bit address must be 0-7, got {bit_val}")
-                offset = int(byte_addr) * 8 + bit_val
+                _validate_bit_addr(int(bit_addr))
+                offset = int(byte_addr) * 8 + int(bit_addr)
             elif address_str.startswith("QW"):
                 # Word address: QW10
                 offset = int(address_str[2:])