From f3059f410342936f689903f88bcb5a250f2c2a15 Mon Sep 17 00:00:00 2001 From: Gijs Molenaar Date: Tue, 18 Aug 2026 17:38:55 +0200 Subject: [PATCH 1/2] feat(s7commplus): add symbolic data subscriptions --- README.rst | 16 ++ s7commplus/__init__.py | 32 ++-- s7commplus/client.py | 74 +++++++-- s7commplus/connection.py | 47 ++++++ s7commplus/protocol.py | 6 + s7commplus/subscription.py | 285 ++++++++++++++++++++++++++++++++++ tests/test_s7_subscription.py | 166 ++++++++++++++++++++ 7 files changed, 598 insertions(+), 28 deletions(-) create mode 100644 s7commplus/subscription.py create mode 100644 tests/test_s7_subscription.py diff --git a/README.rst b/README.rst index ddc50f3d..3beae47a 100644 --- a/README.rst +++ b/README.rst @@ -95,6 +95,22 @@ PUT/GET enabled. * **S7 routing** -- connect to PLCs on remote subnets via a gateway PLC * **Symbolic addressing** -- read/write by tag name instead of raw addresses * **Live symbol browsing** -- resolve tag names directly from the PLC +* **Symbolic data subscriptions** -- monitor values using access sequences + returned by ``browse()``:: + + from s7commplus import Client + + client = Client() + client.connect("192.168.1.10", 0, 1, password="secret") + subscription_id = client.create_subscription(["8A0E0007.A"], cycle_ms=100) + notification = client.receive_subscription_notification() + value = notification.values[1] + client.delete_subscription(subscription_id) + client.disconnect() + + Reference IDs default to the one-based position of each access sequence. + Subscriptions use symbolic LIDs and therefore cannot be created from raw DB + byte offsets. * **TIA Portal XML import** -- import symbol tables from TIA Portal exports **Help us test!** If you have access to any Siemens S7 PLC, we would greatly diff --git a/s7commplus/__init__.py b/s7commplus/__init__.py index 7e5199d5..0289ef38 100644 --- a/s7commplus/__init__.py +++ b/s7commplus/__init__.py @@ -13,33 +13,39 @@ data = client.db_read(1, 0, 4) """ +from .async_client import S7CommPlusAsyncClient as AsyncClient from .blob_decompressor import decompress_blob, find_and_decompress from .client import S7CommPlusClient as Client -from .async_client import S7CommPlusAsyncClient as AsyncClient -from .server import S7CommPlusServer as Server, DataBlock, CPUState from .connection import S7CommPlusConnection +from .server import CPUState, DataBlock +from .server import S7CommPlusServer as Server +from .subscription import SubscriptionItem, SubscriptionNotification from .tag_browser import ( - Tag, - Member, DataBlock as ExploreDataBlock, - tags_from_explore, +) +from .tag_browser import ( + Member, + Tag, block_interface_from_explore, datablocks_from_explore, + tags_from_explore, ) __all__ = [ - "Client", "AsyncClient", - "Server", - "DataBlock", "CPUState", + "Client", + "DataBlock", + "ExploreDataBlock", + "Member", "S7CommPlusConnection", - "decompress_blob", - "find_and_decompress", + "Server", + "SubscriptionItem", + "SubscriptionNotification", "Tag", - "Member", - "ExploreDataBlock", - "tags_from_explore", "block_interface_from_explore", "datablocks_from_explore", + "decompress_blob", + "find_and_decompress", + "tags_from_explore", ] diff --git a/s7commplus/client.py b/s7commplus/client.py index 7a47c819..9212acc4 100644 --- a/s7commplus/client.py +++ b/s7commplus/client.py @@ -5,7 +5,8 @@ import logging import struct -from typing import Any, Callable, Optional, TypeVar +from collections.abc import Callable, Sequence +from typing import Any, Optional, TypeVar from snap7.error import S7ConnectionError @@ -16,9 +17,17 @@ encode_item_address, encode_object_qualifier, encode_pvalue_blob, + parse_create_object_session_id, ) from .connection import S7CommPlusConnection from .protocol import DataType, ElementID, FunctionCode, Ids, ObjectId, ProtocolVersion +from .subscription import ( + SubscriptionItem, + SubscriptionNotification, + build_delete_subscription_request, + build_subscription_request, + parse_subscription_notification, +) from .vlq import decode_uint32_vlq, decode_uint64_vlq, encode_uint32_vlq logger = logging.getLogger(__name__) @@ -37,6 +46,8 @@ def __init__(self) -> None: # Last-used connect() arguments, kept so operations can transparently # reconnect on firmware that RSTs the session after a symbolic read. self._connect_params: Optional[dict[str, Any]] = None + self._subscription_change_counter = 1 + self._subscription_relation_id = 0x7FFFC001 @property def connected(self) -> bool: @@ -614,31 +625,64 @@ def _explore_type_info_container(self) -> list["typeinfo.PObject"]: response = self._connection.send_request(FunctionCode.EXPLORE, payload, integrity_tail=5, reassemble=True) return typeinfo.extract_type_info_objects(response) - def create_subscription(self, items: list[tuple[int, int, int]], cycle_ms: int = 0) -> int: + def create_subscription( + self, + items: Sequence[SubscriptionItem | str], + cycle_ms: int = 100, + credit_limit: int = -1, + ) -> int: """Create a data change subscription. .. warning:: This method is **experimental** and may change. - The PLC will push data updates for the specified variables. Use - ``receive_notification()`` to receive the pushed data. + The PLC pushes an initial value and subsequent changes. Access-sequence + strings are returned by :meth:`browse`; explicit + :class:`SubscriptionItem` objects can supply a symbol CRC, sub-area, or + stable reference ID. Args: - items: List of (db_number, start_offset, size) tuples to monitor. - cycle_ms: Cycle time in milliseconds (0 = on change). + items: Symbolic access-sequence strings or subscription items. + cycle_ms: Sampling cycle in milliseconds. + credit_limit: Number of notification credits, or ``-1`` for + unlimited notifications. Returns: Subscription object ID assigned by the PLC. """ if self._connection is None: raise RuntimeError("Not connected") - - payload = _build_subscription_request(items, cycle_ms, self._connection.session_id) - response = self._connection.send_request(FunctionCode.CREATE_OBJECT, payload) - - # Parse the CreateObject response to get the subscription object ID - sub_id, consumed = decode_uint32_vlq(response, 0) - logger.info(f"Subscription created, id={sub_id:#x}") - return sub_id + if self._connection.subscription_container_id == 0: + raise RuntimeError("PLC did not provide a subscription container object") + + normalized = [SubscriptionItem.from_access_sequence(item) if isinstance(item, str) else item for item in items] + payload, integrity_tail = build_subscription_request( + self._connection.subscription_container_id, + normalized, + cycle_ms=cycle_ms, + credit_limit=credit_limit, + change_counter=self._subscription_change_counter, + relation_id=self._subscription_relation_id, + ) + response = self._connection.send_request( + FunctionCode.CREATE_OBJECT, + payload, + integrity_tail=integrity_tail, + ) + object_ids, _, return_value = parse_create_object_session_id(response) + if return_value != 0 or not object_ids: + raise RuntimeError(f"Subscription creation failed: PLC returned 0x{return_value:X}") + + self._subscription_change_counter = self._subscription_change_counter % 0xFF + 1 + self._subscription_relation_id = (self._subscription_relation_id + 1) & 0xFFFFFFFF + subscription_id = object_ids[0] + logger.info(f"Subscription created, id={subscription_id:#x}") + return subscription_id + + def receive_subscription_notification(self) -> SubscriptionNotification: + """Block until the PLC sends one data-subscription notification.""" + if self._connection is None: + raise RuntimeError("Not connected") + return parse_subscription_notification(self._connection.receive_notification()) def delete_subscription(self, subscription_id: int) -> None: """Delete a data change subscription. @@ -651,7 +695,7 @@ def delete_subscription(self, subscription_id: int) -> None: if self._connection is None: raise RuntimeError("Not connected") - payload = struct.pack(">I", subscription_id) + struct.pack(">I", 0) + payload = build_delete_subscription_request(subscription_id, self._connection.protocol_version) self._connection.send_request(FunctionCode.DELETE_OBJECT, payload) logger.info(f"Subscription {subscription_id:#x} deleted") diff --git a/s7commplus/connection.py b/s7commplus/connection.py index 756fd5e3..804ae78e 100644 --- a/s7commplus/connection.py +++ b/s7commplus/connection.py @@ -45,6 +45,7 @@ import ssl import struct import tempfile +from collections import deque from types import TracebackType from typing import Any, Optional, Type @@ -293,6 +294,7 @@ def __init__( self._incoming_bio: Optional[ssl.MemoryBIO] = None self._outgoing_bio: Optional[ssl.MemoryBIO] = None self._session_id: int = 0 + self._subscription_container_id: int = 0 self._sequence_number: int = 0 self._protocol_version: int = 0 # Detected from PLC response self._tls_active: bool = False @@ -328,6 +330,7 @@ def __init__( # Password for post-auth legitimation (V1-initial PLCs) self._connect_password: str = "" + self._notification_frames: deque[bytes] = deque() # Effective protection level, read once the session is up self._protection_level: Optional[int] = None @@ -346,6 +349,11 @@ def session_id(self) -> int: """Session ID assigned by the PLC.""" return self._session_id + @property + def subscription_container_id(self) -> int: + """Object ID assigned to the session's subscription container.""" + return self._subscription_container_id + @property def tls_active(self) -> bool: """Whether TLS encryption is active on this connection.""" @@ -671,6 +679,7 @@ def disconnect(self) -> None: self._outgoing_bio = None self._oms_secret = None self._session_id = 0 + self._subscription_container_id = 0 self._sequence_number = 0 self._protocol_version = 0 self._server_session_version = None @@ -678,6 +687,7 @@ def disconnect(self) -> None: self._integrity_id_read = 0 self._integrity_id_write = 0 self._protection_level = None + self._notification_frames.clear() self._iso_conn.disconnect() def send_request(self, function_code: int, payload: bytes = b"", integrity_tail: int = 4, reassemble: bool = False) -> bytes: @@ -785,6 +795,9 @@ def send_request(self, function_code: int, payload: bytes = b"", integrity_tail: # Receive response response_frame = self._recv_s7_data() + while self._is_notification_frame(response_frame): + self._notification_frames.append(response_frame) + response_frame = self._recv_s7_data() logger.debug(f"=== RECV RESPONSE === raw frame ({len(response_frame)} bytes): {response_frame.hex(' ')}") # Parse frame header, use data_length to exclude trailer @@ -844,6 +857,39 @@ def send_request(self, function_code: int, payload: bytes = b"", integrity_tail: return resp_payload + @staticmethod + def _is_notification_frame(frame: bytes) -> bool: + """Return whether a complete frame contains an unsolicited notification.""" + try: + version, data_length, consumed = decode_header(frame) + except (IndexError, ValueError): + return False + data = frame[consumed : consumed + data_length] + if version == ProtocolVersion.V3 and data: + hash_length = data[0] + if hash_length and len(data) > 1 + hash_length: + data = data[1 + hash_length :] + return bool(data) and data[0] == Opcode.NOTIFICATION + + def receive_notification(self) -> bytes: + """Receive one unsolicited S7CommPlus notification frame. + + Notifications observed while waiting for a request response are queued, + so callers do not lose updates when protocol traffic interleaves. This + method must not run concurrently with :meth:`send_request` because both + consume the same connection stream. + """ + if not self._connected: + from snap7.error import S7ConnectionError + + raise S7ConnectionError("Not connected") + frame = self._notification_frames.popleft() if self._notification_frames else self._recv_s7_data() + if not self._is_notification_frame(frame): + from snap7.error import S7ConnectionError + + raise S7ConnectionError("Expected an S7CommPlus notification") + return frame + # Sanity caps for fragment reassembly — generous vs. any real PLC EXPLORE response, # but bounded so a malformed/adversarial stream can't drive unbounded allocation. _MAX_REASSEMBLED_BYTES = 16 * 1024 * 1024 @@ -1084,6 +1130,7 @@ def _wstring_attr(attr_id: int, s: str) -> bytes: # First ObjectId is the new session id; second (if any) is for notifications. self._session_id = object_ids[0] + self._subscription_container_id = object_ids[1] if len(object_ids) > 1 else 0 self._protocol_version = version logger.debug( diff --git a/s7commplus/protocol.py b/s7commplus/protocol.py index 4d869630..da209c28 100644 --- a/s7commplus/protocol.py +++ b/s7commplus/protocol.py @@ -195,11 +195,17 @@ class Ids(IntEnum): # Subscription classes (for data change notifications) CLASS_SUBSCRIPTIONS = 255 CLASS_SUBSCRIPTION = 1001 + SUBSCRIPTION_MISSED_SENDINGS = 1002 + SUBSCRIPTION_SUBSYSTEM_ERROR = 1003 + SUBSCRIPTION_ROUTE_MODE = 1040 SUBSCRIPTION_CYCLE_TIME = 1049 SUBSCRIPTION_ACTIVE = 1041 SUBSCRIPTION_CREDIT_LIMIT = 1053 SUBSCRIPTION_REFERENCE_LIST = 1048 SUBSCRIPTION_FUNCTION_CLASS_ID = 1082 + SUBSCRIPTION_DISABLED = 1051 + SUBSCRIPTION_COUNT = 1052 + SUBSCRIPTION_TICKS = 1054 # Alarm subscription ALARM_SUBSCRIPTION_REF_CLASS_RID = 2662 diff --git a/s7commplus/subscription.py b/s7commplus/subscription.py new file mode 100644 index 00000000..e88f3f94 --- /dev/null +++ b/s7commplus/subscription.py @@ -0,0 +1,285 @@ +"""S7CommPlus data subscription encoding and notification parsing. + +The wire layout follows the subscription implementation in +``thomas-v2/S7CommPlusDriver`` and the TIA Portal captures attached to +GH-710. Subscriptions use symbolic access sequences; raw byte offsets do +not identify variables in optimized data blocks. +""" + +import struct +from collections.abc import Sequence +from dataclasses import dataclass + +from .codec import decode_header, decode_pvalue_to_bytes, encode_object_qualifier +from .protocol import DataType, ElementID, Ids, Opcode, ProtocolVersion +from .vlq import decode_uint32_vlq, encode_uint32_vlq + + +@dataclass(frozen=True) +class SubscriptionItem: + """One symbolic variable included in a data subscription.""" + + access_area: int + lids: tuple[int, ...] + access_sub_area: int | None = None + symbol_crc: int = 0 + reference_id: int = 0 + + def __post_init__(self) -> None: + if not 0 <= self.access_area <= 0xFFFFFFFF: + raise ValueError("access_area must fit in an unsigned 32-bit integer") + if not self.lids: + raise ValueError("a subscription item requires at least one LID") + if any(not 0 <= lid <= 0xFFFFFFFF for lid in self.lids): + raise ValueError("LIDs must fit in unsigned 32-bit integers") + if self.access_sub_area is not None and not 0 <= self.access_sub_area <= 0xFFFFFFFF: + raise ValueError("access_sub_area must fit in an unsigned 32-bit integer") + if not 0 <= self.symbol_crc <= 0xFFFFFFFF: + raise ValueError("symbol_crc must fit in an unsigned 32-bit integer") + if not 0 <= self.reference_id <= 0xFFFFFFFF: + raise ValueError("reference_id must fit in an unsigned 32-bit integer") + + @property + def resolved_sub_area(self) -> int: + """Return the explicit or inferred access sub-area.""" + if self.access_sub_area is not None: + return self.access_sub_area + if self.access_area & 0xFFFF0000 == Ids.DB_ACCESS_AREA_BASE: + return Ids.DB_VALUE_ACTUAL + return Ids.CONTROLLER_AREA_VALUE_ACTUAL + + @classmethod + def from_access_sequence( + cls, + access_sequence: str, + *, + symbol_crc: int = 0, + reference_id: int = 0, + access_sub_area: int | None = None, + ) -> "SubscriptionItem": + """Build an item from a ``browse()`` access sequence. + + For example, ``"8A0E0007.A.2"`` addresses LIDs ``0xA, 0x2`` in + DB7's actual-value area. + """ + parts = access_sequence.split(".") + if len(parts) < 2 or any(not part for part in parts): + raise ValueError("access_sequence must contain an access area and at least one LID") + try: + access_area = int(parts[0], 16) + lids = tuple(int(part, 16) for part in parts[1:]) + except ValueError as exc: + raise ValueError("access_sequence components must be hexadecimal") from exc + return cls(access_area, lids, access_sub_area, symbol_crc, reference_id) + + +@dataclass(frozen=True) +class SubscriptionNotification: + """Values and errors carried by one unsolicited subscription update.""" + + subscription_id: int + credit_tick: int + sequence_number: int + change_counter: int + values: dict[int, bytes] + errors: dict[int, int] + timestamp_microseconds: int | None = None + trailing_data: bytes = b"" + + +def _attribute(attribute_id: int, value: bytes) -> bytes: + return bytes([ElementID.ATTRIBUTE]) + encode_uint32_vlq(attribute_id) + value + + +def _scalar(data_type: DataType, data: bytes) -> bytes: + return bytes([0x00, data_type]) + data + + +def _reference_list(items: Sequence[SubscriptionItem], change_counter: int) -> bytes: + values = [0x80000000 | ((change_counter & 0xFF) << 16), 0, len(items)] + used_references: set[int] = set() + for index, item in enumerate(items, 1): + reference_id = item.reference_id or index + if reference_id in used_references: + raise ValueError(f"duplicate subscription reference_id {reference_id}") + used_references.add(reference_id) + values.extend( + ( + 0x80040000 | (1 + len(item.lids)), + reference_id, + 0, + item.access_area, + item.symbol_crc, + item.resolved_sub_area, + *item.lids, + ) + ) + + encoded = bytearray([0x20, DataType.UDINT]) + encoded += encode_uint32_vlq(len(values)) + for value in values: + encoded += encode_uint32_vlq(value) + return bytes(encoded) + + +def build_subscription_request( + subscription_container_id: int, + items: Sequence[SubscriptionItem], + *, + cycle_ms: int = 100, + credit_limit: int = -1, + change_counter: int = 1, + relation_id: int = 0x7FFFC001, + route_mode: int = 0x20, +) -> tuple[bytes, int]: + """Build a CreateObject payload and its IntegrityId insertion tail. + + The returned tail must be passed as ``integrity_tail`` to + :meth:`S7CommPlusConnection.send_request`; CreateObject carries its + IntegrityId between the request set and object tree, rather than near the + final padding used by most requests. + """ + if not items: + raise ValueError("a subscription requires at least one item") + if not 0 <= cycle_ms <= 0xFFFFFFFF: + raise ValueError("cycle_ms must fit in an unsigned 32-bit integer") + if not -1 <= credit_limit <= 0x7FFF: + raise ValueError("credit_limit must be -1 or a signed 16-bit positive value") + if not 1 <= change_counter <= 0xFF: + raise ValueError("change_counter must be between 1 and 255") + + payload = bytearray() + payload += struct.pack(">I", subscription_container_id) + payload += _scalar(DataType.UDINT, encode_uint32_vlq(0)) + payload += struct.pack(">I", 0) + request_set_size = len(payload) + + payload += bytes([ElementID.START_OF_OBJECT]) + payload += struct.pack(">I", relation_id) + payload += encode_uint32_vlq(Ids.CLASS_SUBSCRIPTION) + payload += encode_uint32_vlq(0) + payload += encode_uint32_vlq(0) + + name = f"python-snap7-subscription-{relation_id:08x}".encode() + payload += _attribute( + Ids.OBJECT_VARIABLE_TYPE_NAME, + _scalar(DataType.WSTRING, encode_uint32_vlq(len(name)) + name), + ) + payload += _attribute(Ids.SUBSCRIPTION_FUNCTION_CLASS_ID, _scalar(DataType.USINT, b"\x00")) + payload += _attribute(Ids.SUBSCRIPTION_MISSED_SENDINGS, _scalar(DataType.UINT, struct.pack(">H", 0))) + payload += _attribute(Ids.SUBSCRIPTION_SUBSYSTEM_ERROR, _scalar(DataType.LINT, struct.pack(">q", 0))) + payload += _attribute(Ids.SUBSCRIPTION_ROUTE_MODE, _scalar(DataType.USINT, bytes([route_mode & 0xFF]))) + payload += _attribute(Ids.SUBSCRIPTION_ACTIVE, _scalar(DataType.BOOL, b"\x01")) + payload += _attribute(Ids.SUBSCRIPTION_REFERENCE_LIST, _reference_list(items, change_counter)) + payload += _attribute(Ids.SUBSCRIPTION_CYCLE_TIME, _scalar(DataType.UDINT, encode_uint32_vlq(cycle_ms))) + payload += _attribute(Ids.SUBSCRIPTION_DISABLED, _scalar(DataType.USINT, b"\x00")) + payload += _attribute(Ids.SUBSCRIPTION_COUNT, _scalar(DataType.USINT, b"\x00")) + payload += _attribute(Ids.SUBSCRIPTION_CREDIT_LIMIT, _scalar(DataType.INT, struct.pack(">h", credit_limit))) + payload += _attribute(Ids.SUBSCRIPTION_TICKS, _scalar(DataType.UINT, struct.pack(">H", 0xFFFF))) + payload += _attribute(1055, _scalar(DataType.USINT, b"\x00")) + payload += bytes([ElementID.TERMINATING_OBJECT]) + payload += struct.pack(">I", 0) + + return bytes(payload), len(payload) - request_set_size + + +def build_delete_subscription_request(subscription_id: int, protocol_version: int) -> bytes: + """Build the payload for deleting one subscription object.""" + return ( + struct.pack(">I", subscription_id) + + b"\x00" + + encode_object_qualifier(protocol_version=protocol_version) + + struct.pack(">I", 0) + ) + + +def _decode_notification_value(data: bytes, offset: int) -> tuple[bytes, int]: + """Decode a notification PValue, including Siemens' padded BLOB form.""" + if len(data) >= offset + 4 and data[offset] & 0x10 == 0 and data[offset + 1] == DataType.BLOB and data[offset + 2] == 0: + length, width = decode_uint32_vlq(data, offset + 3) + value_offset = offset + 3 + width + value_end = value_offset + length + if value_end > len(data): + raise ValueError("subscription BLOB value is truncated") + return data[value_offset:value_end], value_end - offset + return decode_pvalue_to_bytes(data, offset) + + +def parse_subscription_notification(frame: bytes) -> SubscriptionNotification: + """Parse one complete unsolicited S7CommPlus notification frame.""" + version, data_length, consumed = decode_header(frame) + if len(frame) < consumed + data_length: + raise ValueError("truncated S7CommPlus notification frame") + data = frame[consumed : consumed + data_length] + if version == ProtocolVersion.V3 and data: + hash_length = data[0] + if hash_length and len(data) > 1 + hash_length: + data = data[1 + hash_length :] + if not data or data[0] != Opcode.NOTIFICATION: + raise ValueError("expected an S7CommPlus notification") + + offset = 1 + if len(data) < offset + 11: + raise ValueError("subscription notification header is truncated") + subscription_id = struct.unpack_from(">I", data, offset)[0] + offset += 4 + offset += 6 # three protocol-reserved UInt16 fields + credit_tick = data[offset] + offset += 1 + sequence_number, width = decode_uint32_vlq(data, offset) + offset += width + if offset >= len(data): + raise ValueError("subscription notification change counter is missing") + + timestamp: int | None = None + change_counter = data[offset] + offset += 1 + if change_counter == 0: + offset -= 1 + if len(data) < offset + 9: + raise ValueError("subscription notification timestamp is truncated") + timestamp = struct.unpack_from(">Q", data, offset)[0] + offset += 8 + change_counter = data[offset] + offset += 1 + + values: dict[int, bytes] = {} + errors: dict[int, int] = {} + while offset < len(data): + status = data[offset] + offset += 1 + if status == 0: + break + if status == 0x92: + if len(data) < offset + 4: + raise ValueError("subscription item reference is truncated") + reference_id = struct.unpack_from(">I", data, offset)[0] + offset += 4 + value, width = _decode_notification_value(data, offset) + offset += width + values[reference_id] = value + elif status == 0x9B: + reference_id, width = decode_uint32_vlq(data, offset) + offset += width + value, width = _decode_notification_value(data, offset) + offset += width + values[reference_id] = value + elif status in (0x03, 0x13): + if len(data) < offset + 4: + raise ValueError("subscription error reference is truncated") + reference_id = struct.unpack_from(">I", data, offset)[0] + offset += 4 + errors[reference_id] = status + else: + raise ValueError(f"unsupported subscription item status 0x{status:02X}") + + return SubscriptionNotification( + subscription_id=subscription_id, + credit_tick=credit_tick, + sequence_number=sequence_number, + change_counter=change_counter, + values=values, + errors=errors, + timestamp_microseconds=timestamp, + trailing_data=data[offset:], + ) diff --git a/tests/test_s7_subscription.py b/tests/test_s7_subscription.py new file mode 100644 index 00000000..44f1ad30 --- /dev/null +++ b/tests/test_s7_subscription.py @@ -0,0 +1,166 @@ +"""Tests for S7CommPlus symbolic data subscriptions.""" + +import struct +from unittest.mock import MagicMock + +import pytest + +from s7commplus.client import S7CommPlusClient +from s7commplus.codec import encode_header, encode_pvalue_blob +from s7commplus.connection import S7CommPlusConnection +from s7commplus.protocol import DataType, FunctionCode, Ids, Opcode, ProtocolVersion +from s7commplus.subscription import ( + SubscriptionItem, + build_delete_subscription_request, + build_subscription_request, + parse_subscription_notification, +) +from s7commplus.vlq import encode_uint32_vlq, encode_uint64_vlq + + +def _response_frame(function_code: int, sequence: int, payload: bytes) -> bytes: + response = struct.pack(">BHHHHB", Opcode.RESPONSE, 0, function_code, 0, sequence, 0x34) + payload + return encode_header(ProtocolVersion.V2, len(response)) + response + b"\x72\x02\x00\x00" + + +def _notification_frame(*, version: int = ProtocolVersion.V2, with_hmac: bool = False) -> bytes: + data = bytearray([Opcode.NOTIFICATION]) + data += struct.pack(">IHHH", 0x70400025, 4, 0, 0) + data += b"\x03" + encode_uint32_vlq(9) + b"\x01" + data += b"\x92" + struct.pack(">I", 7) + encode_pvalue_blob(b"\x12\x34") + data += b"\x9b" + encode_uint32_vlq(8) + bytes([0, DataType.USINT, 0x2A]) + data += b"\x13" + struct.pack(">I", 9) + data += b"\x00\xaa" + framed_data = (b"\x20" + bytes(32) if with_hmac else b"") + data + return encode_header(version, len(framed_data)) + framed_data + bytes([0x72, version, 0, 0]) + + +class TestSubscriptionItem: + def test_from_db_access_sequence(self) -> None: + item = SubscriptionItem.from_access_sequence("8A0E0007.A.2") + + assert item.access_area == Ids.DB_ACCESS_AREA_BASE + 7 + assert item.lids == (0xA, 0x2) + assert item.resolved_sub_area == Ids.DB_VALUE_ACTUAL + + def test_from_native_area_access_sequence(self) -> None: + item = SubscriptionItem.from_access_sequence("52.9") + assert item.resolved_sub_area == Ids.CONTROLLER_AREA_VALUE_ACTUAL + + @pytest.mark.parametrize("value", ["", "8A0E0007", "8A0E0007.not-hex", ".A"]) + def test_rejects_invalid_access_sequence(self, value: str) -> None: + with pytest.raises(ValueError): + SubscriptionItem.from_access_sequence(value) + + +class TestSubscriptionRequest: + def test_uses_subscription_container_and_symbolic_reference_list(self) -> None: + item = SubscriptionItem.from_access_sequence("8A0E0007.A.2", symbol_crc=0x1234, reference_id=7) + payload, integrity_tail = build_subscription_request(0x3C2, [item], cycle_ms=250) + + assert payload.startswith(struct.pack(">I", 0x3C2) + bytes([0, DataType.UDINT, 0]) + struct.pack(">I", 0)) + assert integrity_tail == len(payload) - 11 + expected_reference = b"".join( + encode_uint32_vlq(value) + for value in ( + 0x80010000, + 0, + 1, + 0x80040003, + 7, + 0, + Ids.DB_ACCESS_AREA_BASE + 7, + 0x1234, + Ids.DB_VALUE_ACTUAL, + 0xA, + 2, + ) + ) + assert bytes([0x20, DataType.UDINT]) + encode_uint32_vlq(11) + expected_reference in payload + + def test_requires_items(self) -> None: + with pytest.raises(ValueError, match="at least one"): + build_subscription_request(0x3C2, []) + + def test_delete_request_contains_object_qualifier(self) -> None: + payload = build_delete_subscription_request(0x70400025, ProtocolVersion.V2) + assert payload.startswith(struct.pack(">I", 0x70400025) + b"\x00") + assert payload.endswith(struct.pack(">I", 0)) + assert len(payload) > 9 + + +class TestSubscriptionNotification: + @pytest.mark.parametrize( + ("version", "with_hmac"), + [(ProtocolVersion.V2, False), (ProtocolVersion.V3, True)], + ) + def test_parses_values_errors_and_metadata(self, version: int, with_hmac: bool) -> None: + notification = parse_subscription_notification(_notification_frame(version=version, with_hmac=with_hmac)) + + assert notification.subscription_id == 0x70400025 + assert notification.credit_tick == 3 + assert notification.sequence_number == 9 + assert notification.change_counter == 1 + assert notification.values == {7: b"\x12\x34", 8: b"\x2a"} + assert notification.errors == {9: 0x13} + assert notification.trailing_data == b"\xaa" + + def test_rejects_response_frame(self) -> None: + with pytest.raises(ValueError, match="notification"): + parse_subscription_notification(_response_frame(FunctionCode.GET_VARIABLE, 1, b"\x00")) + + def test_parses_tia_portal_watch_notification_from_issue_710(self) -> None: + frame = bytes.fromhex( + "7203006520db9b8947109b14bc56e8bd25032cda2dd3a9488cbd1807017a69f90af8982371" + "337040002504000000000000050192000000070014001801000006278ab08c18c456364dfc0d" + "d0800000000f000000920000000800020092000000090001000000000000" + ) + + notification = parse_subscription_notification(frame) + + assert notification.subscription_id == 0x70400025 + assert notification.sequence_number == 5 + assert notification.values[7] == bytes.fromhex("01000006278ab08c18c456364dfc0dd0800000000f000000") + assert notification.values[8] == b"\x00" + assert notification.values[9] == b"\x00" + + +class TestSubscriptionClient: + def test_create_receive_and_delete(self) -> None: + connection = MagicMock() + connection.subscription_container_id = 0x3C2 + connection.protocol_version = ProtocolVersion.V1 + create_response = encode_uint64_vlq(0) + b"\x01" + encode_uint32_vlq(0x70400025) + connection.send_request.return_value = create_response + connection.receive_notification.return_value = _notification_frame() + + client = S7CommPlusClient() + client._connection = connection + subscription_id = client.create_subscription(["8A0E0007.A"], cycle_ms=100) + + assert subscription_id == 0x70400025 + create_call = connection.send_request.call_args_list[0] + assert create_call.args[0] == FunctionCode.CREATE_OBJECT + assert create_call.kwargs["integrity_tail"] > 4 + assert client.receive_subscription_notification().values[7] == b"\x12\x34" + + client.delete_subscription(subscription_id) + delete_call = connection.send_request.call_args_list[1] + assert delete_call.args[0] == FunctionCode.DELETE_OBJECT + assert delete_call.args[1].startswith(struct.pack(">I", subscription_id)) + + +class TestNotificationQueue: + def test_send_request_queues_interleaved_notification(self) -> None: + connection = S7CommPlusConnection("127.0.0.1") + connection._connected = True + connection._protocol_version = ProtocolVersion.V2 + connection._session_id = 1 + notification = _notification_frame() + response = _response_frame(FunctionCode.GET_VARIABLE, 0, b"\x00") + connection._send_s7_data = MagicMock() + connection._recv_s7_data = MagicMock(side_effect=[notification, response]) + + assert connection.send_request(FunctionCode.GET_VARIABLE, b"\x00\x00\x00\x00") == b"\x00" + assert connection.receive_notification() == notification + assert connection._recv_s7_data.call_count == 2 From 8b7fbf623c47eebdb7e4afc3f0a6666f57d8a146 Mon Sep 17 00:00:00 2001 From: Gijs Molenaar Date: Thu, 20 Aug 2026 07:08:23 +0200 Subject: [PATCH 2/2] fix: match real PLC subscription framing --- CHANGES.md | 1 + s7commplus/client.py | 12 ++++++++---- s7commplus/subscription.py | 10 +++++----- tests/test_s7_subscription.py | 32 ++++++++++++++++++++++++++++++-- 4 files changed, 44 insertions(+), 11 deletions(-) diff --git a/CHANGES.md b/CHANGES.md index 4474ffe9..13e5a2a5 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -12,6 +12,7 @@ Major release: new `s7commplus` package with S7CommPlus protocol support. * S7CommPlus PLC start/stop via INVOKE * S7CommPlus object browsing via EXPLORE * S7CommPlus live symbol browsing (`client.browse()`) and datablock listing (experimental) +* S7CommPlus symbolic data subscriptions and notification decoding (experimental) * TIA Portal XML import for SymbolTable (`SymbolTable.from_tia_xml()`) (experimental) * S7CommPlus CPU state reading and block transfer (upload/download) * **Symbolic (LID-based) access for optimized DBs** (experimental): diff --git a/s7commplus/client.py b/s7commplus/client.py index 9212acc4..5e2f611c 100644 --- a/s7commplus/client.py +++ b/s7commplus/client.py @@ -629,7 +629,7 @@ def create_subscription( self, items: Sequence[SubscriptionItem | str], cycle_ms: int = 100, - credit_limit: int = -1, + credit_limit: int = 10, ) -> int: """Create a data change subscription. @@ -643,8 +643,8 @@ def create_subscription( Args: items: Symbolic access-sequence strings or subscription items. cycle_ms: Sampling cycle in milliseconds. - credit_limit: Number of notification credits, or ``-1`` for - unlimited notifications. + credit_limit: Number of notification credits. The default of 10 + matches the value accepted by real S7-1500 PLCs. Returns: Subscription object ID assigned by the PLC. @@ -694,8 +694,12 @@ def delete_subscription(self, subscription_id: int) -> None: """ if self._connection is None: raise RuntimeError("Not connected") + if self._connection.subscription_container_id == 0: + raise RuntimeError("PLC did not provide a subscription container object") - payload = build_delete_subscription_request(subscription_id, self._connection.protocol_version) + # Subscription children are owned by the session's second CreateObject + # result. The reference driver deletes that container, not the child ID. + payload = build_delete_subscription_request(self._connection.subscription_container_id, self._connection.protocol_version) self._connection.send_request(FunctionCode.DELETE_OBJECT, payload) logger.info(f"Subscription {subscription_id:#x} deleted") diff --git a/s7commplus/subscription.py b/s7commplus/subscription.py index e88f3f94..5aecc70d 100644 --- a/s7commplus/subscription.py +++ b/s7commplus/subscription.py @@ -12,7 +12,7 @@ from .codec import decode_header, decode_pvalue_to_bytes, encode_object_qualifier from .protocol import DataType, ElementID, Ids, Opcode, ProtocolVersion -from .vlq import decode_uint32_vlq, encode_uint32_vlq +from .vlq import decode_uint32_vlq, encode_int64_vlq, encode_uint32_vlq @dataclass(frozen=True) @@ -127,10 +127,10 @@ def build_subscription_request( items: Sequence[SubscriptionItem], *, cycle_ms: int = 100, - credit_limit: int = -1, + credit_limit: int = 10, change_counter: int = 1, relation_id: int = 0x7FFFC001, - route_mode: int = 0x20, + route_mode: int = 0x14, ) -> tuple[bytes, int]: """Build a CreateObject payload and its IntegrityId insertion tail. @@ -160,14 +160,14 @@ def build_subscription_request( payload += encode_uint32_vlq(0) payload += encode_uint32_vlq(0) - name = f"python-snap7-subscription-{relation_id:08x}".encode() + name = f"Subscription_{relation_id}".encode() payload += _attribute( Ids.OBJECT_VARIABLE_TYPE_NAME, _scalar(DataType.WSTRING, encode_uint32_vlq(len(name)) + name), ) payload += _attribute(Ids.SUBSCRIPTION_FUNCTION_CLASS_ID, _scalar(DataType.USINT, b"\x00")) payload += _attribute(Ids.SUBSCRIPTION_MISSED_SENDINGS, _scalar(DataType.UINT, struct.pack(">H", 0))) - payload += _attribute(Ids.SUBSCRIPTION_SUBSYSTEM_ERROR, _scalar(DataType.LINT, struct.pack(">q", 0))) + payload += _attribute(Ids.SUBSCRIPTION_SUBSYSTEM_ERROR, _scalar(DataType.LINT, encode_int64_vlq(0))) payload += _attribute(Ids.SUBSCRIPTION_ROUTE_MODE, _scalar(DataType.USINT, bytes([route_mode & 0xFF]))) payload += _attribute(Ids.SUBSCRIPTION_ACTIVE, _scalar(DataType.BOOL, b"\x01")) payload += _attribute(Ids.SUBSCRIPTION_REFERENCE_LIST, _reference_list(items, change_counter)) diff --git a/tests/test_s7_subscription.py b/tests/test_s7_subscription.py index 44f1ad30..5ff62f16 100644 --- a/tests/test_s7_subscription.py +++ b/tests/test_s7_subscription.py @@ -54,6 +54,29 @@ def test_rejects_invalid_access_sequence(self, value: str) -> None: class TestSubscriptionRequest: + def test_matches_real_plc_reference_trace(self) -> None: + item = SubscriptionItem.from_access_sequence("8A0E0027.25.1A") + payload, integrity_tail = build_subscription_request( + 0x70000CB8, + [item], + cycle_ms=100, + relation_id=0x7FFFC001, + ) + + # Before TLS encryption, captured from the working C# reference driver. + # The request's IntegrityId 2 appears at offset 11 and is inserted later + # by send_request(), so remove it when comparing the builder output. + captured = bytes.fromhex( + "70000cb80004000000000002a17fffc00187690000a38169001517" + "537562736372697074696f6e5f32313437343637323635a3883a000200" + "a3876a00030000a3876b000900a38810000214a38811000101a388182004" + "0b888084800000018880908003010088d0b88027009376251aa38819000464" + "a3881b000200a3881c000200a3881d0007000aa3881e0003ffffa3881f000200" + "a200000000" + ) + assert payload == captured[:11] + captured[12:] + assert integrity_tail == len(payload) - 11 + def test_uses_subscription_container_and_symbolic_reference_list(self) -> None: item = SubscriptionItem.from_access_sequence("8A0E0007.A.2", symbol_crc=0x1234, reference_id=7) payload, integrity_tail = build_subscription_request(0x3C2, [item], cycle_ms=250) @@ -88,6 +111,11 @@ def test_delete_request_contains_object_qualifier(self) -> None: assert payload.endswith(struct.pack(">I", 0)) assert len(payload) > 9 + def test_delete_request_matches_real_plc_reference_trace(self) -> None: + payload = build_delete_subscription_request(0x70000CB8, ProtocolVersion.V2) + wire_payload = payload[:-4] + b"\x03" + payload[-4:] + assert wire_payload == bytes.fromhex("70000cb800000004e88969001200000000896a001300896b000400000300000000") + class TestSubscriptionNotification: @pytest.mark.parametrize( @@ -129,7 +157,7 @@ class TestSubscriptionClient: def test_create_receive_and_delete(self) -> None: connection = MagicMock() connection.subscription_container_id = 0x3C2 - connection.protocol_version = ProtocolVersion.V1 + connection.protocol_version = ProtocolVersion.V2 create_response = encode_uint64_vlq(0) + b"\x01" + encode_uint32_vlq(0x70400025) connection.send_request.return_value = create_response connection.receive_notification.return_value = _notification_frame() @@ -147,7 +175,7 @@ def test_create_receive_and_delete(self) -> None: client.delete_subscription(subscription_id) delete_call = connection.send_request.call_args_list[1] assert delete_call.args[0] == FunctionCode.DELETE_OBJECT - assert delete_call.args[1].startswith(struct.pack(">I", subscription_id)) + assert delete_call.args[1].startswith(struct.pack(">I", connection.subscription_container_id)) class TestNotificationQueue: