Add S7 routing support for multi-subnet PLC access

doc/connecting.rst

@@ -122,6 +122,32 @@ Using a Non-Standard Port
    client = Client()
    client.connect("192.168.1.10", 0, 1, tcp_port=1102)
 
+Routing (Multi-Subnet Access)
+------------------------------
+
+.. warning::
+
+   Routing support is experimental and may change in future versions.
+
+When the target PLC sits on a different subnet behind a gateway PLC, use
+``connect_routed`` to let the gateway forward the connection:
+
+.. code-block:: python
+
+   import snap7
+
+   client = snap7.Client()
+   client.connect_routed(
+       host="192.168.1.1",       # gateway PLC address
+       router_rack=0,            # gateway rack
+       router_slot=2,            # gateway slot
+       subnet=0x0001,            # target subnet ID
+       dest_rack=0,              # target PLC rack
+       dest_slot=3,              # target PLC slot
+   )
+   data = client.db_read(1, 0, 4)
+   client.disconnect()
+
 Legacy ``snap7`` Package
 -------------------------
 

pyproject.toml

@@ -59,6 +59,7 @@ markers =[
     "logo",
     "mainloop",
     "partner",
+    "routing",
     "server",
     "util",
     "conformance: protocol conformance tests"

snap7/client.py

@@ -404,6 +404,78 @@ def connect(self, address: str, rack: int, slot: int, tcp_port: int = 102) -> "C
 
         return self
 
+    def connect_routed(
+        self,
+        host: str,
+        router_rack: int,
+        router_slot: int,
+        subnet: int,
+        dest_rack: int,
+        dest_slot: int,
+        port: int = 102,
+        timeout: float = 5.0,
+    ) -> "Client":
+        """Connect to an S7 PLC via a routing gateway on another subnet.
+
+        The gateway PLC (identified by *host*, *router_rack*, *router_slot*)
+        forwards the connection to the target PLC (identified by *subnet*,
+        *dest_rack*, *dest_slot*) through S7 routing parameters embedded in
+        the COTP Connection Request.
+
+        .. warning:: This method is experimental and may change in future versions.
+
+        Args:
+            host: IP address of the routing gateway PLC
+            router_rack: Rack number of the gateway PLC
+            router_slot: Slot number of the gateway PLC
+            subnet: Subnet ID of the target network (0x0000-0xFFFF)
+            dest_rack: Rack number of the destination PLC
+            dest_slot: Slot number of the destination PLC
+            port: TCP port (default 102)
+            timeout: Connection timeout in seconds
+
+        Returns:
+            Self for method chaining
+        """
+        self.host = host
+        self.port = port
+        self.rack = router_rack
+        self.slot = router_slot
+        self._params[Parameter.RemotePort] = port
+
+        # Remote TSAP targets the gateway rack/slot
+        self.remote_tsap = 0x0100 | (router_rack << 5) | router_slot
+
+        try:
+            start_time = time.time()
+
+            self.connection = ISOTCPConnection(
+                host=host,
+                port=port,
+                local_tsap=self.local_tsap,
+                remote_tsap=self.remote_tsap,
+            )
+            self.connection.set_routing(subnet, dest_rack, dest_slot)
+            self.connection.connect(timeout=timeout)
+
+            # Setup communication and negotiate PDU length
+            self._setup_communication()
+
+            self.connected = True
+            self._exec_time = int((time.time() - start_time) * 1000)
+            logger.info(
+                f"Connected (routed) to {host}:{port} via rack {router_rack} slot {router_slot}, "
+                f"subnet {subnet:#06x} -> rack {dest_rack} slot {dest_slot}"
+            )
+        except Exception as e:
+            self.disconnect()
+            if isinstance(e, S7Error):
+                raise
+            else:
+                raise S7ConnectionError(f"Routed connection failed: {e}")
+
+        return self
+
     def disconnect(self) -> int:
         """Disconnect from S7 PLC.
 

snap7/connection.py

@@ -61,6 +61,10 @@ class ISOTCPConnection:
     COTP_PARAM_CALLING_TSAP = 0xC1
     COTP_PARAM_CALLED_TSAP = 0xC2
 
+    # S7 routing parameter codes
+    COTP_PARAM_SUBNET_ID = 0xC6
+    COTP_PARAM_ROUTING_TSAP = 0xC7
+
     def __init__(
         self,
         host: str,
@@ -94,6 +98,31 @@ def __init__(
         self.src_ref = 0x0001  # Source reference
         self.dst_ref = 0x0000  # Destination reference (assigned by peer)
 
+        # Routing parameters (set via connect_routed)
+        self._routing: bool = False
+        self._subnet_id: int = 0
+        self._routing_tsap: int = 0
+
+    def set_routing(self, subnet_id: int, dest_rack: int, dest_slot: int) -> None:
+        """Configure S7 routing parameters for multi-subnet access.
+
+        When routing is enabled, the COTP Connection Request includes
+        additional parameters that instruct the gateway PLC to forward
+        the connection to a target PLC on another subnet.
+
+        .. warning:: This method is experimental and may change in future versions.
+
+        Args:
+            subnet_id: Subnet ID of the target network (2 bytes)
+            dest_rack: Rack number of the destination PLC
+            dest_slot: Slot number of the destination PLC
+        """
+        self._routing = True
+        self._subnet_id = subnet_id & 0xFFFF
+        # Routing TSAP encodes the final target rack/slot the same way
+        # as a normal remote TSAP.
+        self._routing_tsap = 0x0100 | (dest_rack << 5) | dest_slot
+
     def connect(self, timeout: float = 5.0) -> None:
         """
         Establish ISO on TCP connection.
@@ -279,6 +308,13 @@ def _build_cotp_cr(self) -> bytes:
 
         parameters = calling_tsap + called_tsap + pdu_size_param
 
+        # Append routing parameters when routing is enabled
+        if self._routing:
+            subnet_param = struct.pack(">BBH", self.COTP_PARAM_SUBNET_ID, 2, self._subnet_id)
+            routing_tsap_param = struct.pack(">BBH", self.COTP_PARAM_ROUTING_TSAP, 2, self._routing_tsap)
+            parameters += subnet_param + routing_tsap_param
+            logger.debug(f"COTP CR with routing: subnet={self._subnet_id:#06x}, routing_tsap={self._routing_tsap:#06x}")
+
         # Update PDU length to include parameters
         total_length = 6 + len(parameters)
         pdu = struct.pack(">B", total_length) + base_pdu[1:] + parameters