diff --git a/CHANGES.md b/CHANGES.md index f664baeb..8d46aad6 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -6,6 +6,14 @@ CHANGES Major release: new `s7commplus` package with S7CommPlus protocol support. +* Correlate S7CommPlus responses by opcode, function, and sequence; preserve + interleaved notifications and serialize synchronous wire requests. +* Verify authenticated V3 responses and cumulative fragment digests before + parsing, invalidating the connection with a dedicated integrity error on any + mismatch or truncated envelope. +* Complete symbolic subscription lifecycle handling with catalog-tag decoding, + bounded sync/async delivery, overflow and sequence-gap diagnostics, finite + credit replenishment, and stale-generation filtering. * Decode corroborating CPU execution attributes so S7CommPlus `get_cpu_state()` distinguishes RUN from STOP on S7-1500 and returns UNKNOWN for absent or inconsistent state attributes, including S7-1200 responses that omit them. diff --git a/README.rst b/README.rst index 6882e14d..559f5577 100644 --- a/README.rst +++ b/README.rst @@ -128,22 +128,32 @@ PUT/GET enabled. print(result.tag.name, result.value) else: print(result.tag.name, result.error) -* **Symbolic data subscriptions** -- monitor values using access sequences - returned by ``browse()``:: +* **Symbolic data subscriptions** -- monitor catalog tags or access sequences + through bounded sync/async notification streams:: 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.connect("192.168.1.10", password="secret") + tag = client.resolve_tag("DB1.Motor.Speed") + subscription_id = client.create_subscription([tag], cycle_ms=100) + notification = client.receive_subscription_notification(subscription_id) + value = notification.decoded_values[1] + raw_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. + ``iter_subscription_notifications()`` provides a bounded synchronous + iterator; the async client provides an async iterator and + ``subscription_queue()``. Finite notification credits are replenished + automatically. ``subscription_diagnostics()`` reports queue overflow and + sequence gaps. Reference IDs default to the one-based item position. Raw + values remain available when a datatype is unknown or structured. + + Deleting or disconnecting invalidates the local subscription state. Automatic + reconnect does not silently recreate subscriptions: create them again so the + caller can decide how to handle any update gap. Subscriptions use symbolic + LIDs and 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 29365f4c..8ceb6a7a 100644 --- a/s7commplus/__init__.py +++ b/s7commplus/__init__.py @@ -13,6 +13,7 @@ data = client.db_read(1, 0, 4) """ +from .async_client import AsyncSubscriptionQueue from .async_client import S7CommPlusAsyncClient as AsyncClient from .alarm import Alarm, AlarmNotification, AlarmText, LanguageId from .blob_decompressor import decompress_blob, find_and_decompress @@ -22,7 +23,7 @@ from .connection import S7CommPlusConnection from .server import CPUState, DataBlock from .server import S7CommPlusServer as Server -from .subscription import SubscriptionItem, SubscriptionNotification +from .subscription import SubscriptionDiagnostics, SubscriptionItem, SubscriptionNotification from .tag_browser import ( DataBlock as ExploreDataBlock, ) @@ -40,6 +41,7 @@ "AlarmText", "ArrayDimension", "AsyncClient", + "AsyncSubscriptionQueue", "CPUState", "Client", "DBWriteItem", @@ -51,6 +53,7 @@ "Server", "SubscriptionItem", "SubscriptionNotification", + "SubscriptionDiagnostics", "SymbolCatalog", "SymbolicReadItem", "SymbolicTag", diff --git a/s7commplus/async_client.py b/s7commplus/async_client.py index 21249443..225d402a 100644 --- a/s7commplus/async_client.py +++ b/s7commplus/async_client.py @@ -7,7 +7,8 @@ import logging import ssl import struct -from collections.abc import Mapping, Sequence +from collections import deque +from collections.abc import AsyncIterator, Mapping, Sequence from typing import Any, Awaitable, Callable, Optional, TypeVar from snap7.error import S7ConnectionError, S7ProtocolError @@ -26,7 +27,6 @@ _build_multi_symbolic_write_payload, _build_multi_symbolic_read_payload, _build_read_payload, - _build_subscription_request, _build_symbolic_read_payload, _build_symbolic_write_payload, _build_write_payload, @@ -47,16 +47,19 @@ parse_server_session_version, ) from .connection import ( + _MAX_QUEUED_NOTIFICATION_FRAMES, _MAX_SYSTEM_EVENTS_PER_RESPONSE, _S7_CIPHERS, _build_get_var_substreamed_payload, _build_set_variable_payload, _check_system_event, _check_set_variable_response, + _incoming_frame_opcode, _log_create_object_return_value, _parse_get_var_substreamed_response, _parse_protection_level_response, _set_s7_groups, + _validate_response_header, ) from .alarm import ( Alarm, @@ -90,6 +93,16 @@ Opcode, ProtocolVersion, ) +from .subscription import ( + SubscriptionDiagnostics, + SubscriptionItem, + SubscriptionNotification, + SubscriptionRegistry, + build_delete_subscription_request, + build_subscription_request, + notification_subscription_id, + parse_subscription_notification, +) from .vlq import decode_uint32_vlq, decode_uint64_vlq, encode_uint32_vlq logger = logging.getLogger(__name__) @@ -102,6 +115,26 @@ _COTP_DT = 0xF0 +class AsyncSubscriptionQueue: + """Bounded queue-like view over one async client's routed notifications.""" + + def __init__(self, client: "S7CommPlusAsyncClient", subscription_id: int) -> None: + self._client = client + self._subscription_id = subscription_id + + async def get(self, timeout: Optional[float] = None) -> SubscriptionNotification: + return await self._client.receive_subscription_notification(self._subscription_id, timeout=timeout) + + def get_nowait(self) -> SubscriptionNotification: + notification = self._client._subscriptions.pop(self._subscription_id) + if notification is None: + raise asyncio.QueueEmpty + return notification + + def qsize(self) -> int: + return self._client.subscription_diagnostics(self._subscription_id).queued_notifications + + class S7CommPlusAsyncClient: """Async S7CommPlus client for S7-1200/1500 PLCs. @@ -119,8 +152,15 @@ def __init__(self) -> None: self._session_ready = False self._connected = False self._lock = asyncio.Lock() + self._notification_frames: deque[bytes] = deque(maxlen=_MAX_QUEUED_NOTIFICATION_FRAMES) + self._notification_frame_overflows = 0 self._connect_params: Optional[dict[str, Any]] = None self._symbol_catalog: Optional[SymbolCatalog] = None + self._subscription_change_counter = 1 + self._subscription_relation_id = 0x7FFFC001 + self._subscriptions = SubscriptionRegistry() + self._alarm_subscription_ids: set[int] = set() + self._alarm_notification_frames: deque[bytes] = deque(maxlen=100) # V2+ IntegrityId tracking self._integrity_id_read: int = 0 @@ -488,6 +528,9 @@ async def _send_legitimation_legacy(self, response: bytes) -> None: async def disconnect(self) -> None: """Disconnect from PLC.""" + self._subscriptions.clear() + self._alarm_subscription_ids.clear() + self._alarm_notification_frames.clear() if self._session_ready and self._session_id: try: await self._delete_session() @@ -513,6 +556,8 @@ async def disconnect(self) -> None: self._server_session_version = None self._session_setup_ok = False self._protection_level = None + self._notification_frames.clear() + self._notification_frame_overflows = 0 if self._writer: try: @@ -678,24 +723,122 @@ async def download_block(self, block_type: int, block_number: int, data: bytes) await self._send_request(FunctionCode.SET_VAR_SUBSTREAMED, bytes(payload)) - async def create_subscription(self, items: list[tuple[int, int, int]], cycle_ms: int = 0) -> int: + async def create_subscription( + self, + items: Sequence[SubscriptionItem | SymbolicTag | str], + cycle_ms: int = 100, + credit_limit: int = 10, + credit_step: int = 5, + queue_size: int = 100, + ) -> int: """Create a data change subscription. .. warning:: This method is **experimental** and may change. Args: - items: List of (db_number, start_offset, size) tuples to monitor. - cycle_ms: Cycle time in milliseconds (0 = on change). + items: Symbolic access sequences, catalog tags, or explicit items. + cycle_ms: Sampling cycle in milliseconds. + credit_limit: Initial notification credit limit, or -1 for unlimited. + credit_step: Credits added before a finite limit expires. + queue_size: Maximum buffered notifications for this subscription. Returns: Subscription object ID assigned by the PLC. """ - payload = _build_subscription_request(items, cycle_ms, self._session_id) - response = await self._send_request(FunctionCode.CREATE_OBJECT, payload) - - sub_id, consumed = decode_uint32_vlq(response, 0) - logger.info(f"Subscription created, id={sub_id:#x}") - return sub_id + if self._subscription_container_id == 0: + raise RuntimeError("PLC did not provide a subscription container object") + if not 0 <= credit_step <= 255: + raise ValueError("credit_step must be between 0 and 255") + normalized = [ + SubscriptionItem.from_access_sequence(item) + if isinstance(item, str) + else SubscriptionItem.from_tag(item) + if isinstance(item, SymbolicTag) + else item + for item in items + ] + change_counter = self._subscription_change_counter + payload, integrity_tail = build_subscription_request( + self._subscription_container_id, + normalized, + cycle_ms=cycle_ms, + credit_limit=credit_limit, + change_counter=change_counter, + relation_id=self._subscription_relation_id, + ) + response = await self._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}") + subscription_id = object_ids[0] + self._subscriptions.register( + subscription_id, + normalized, + change_counter=change_counter, + credit_limit=credit_limit, + credit_step=credit_step, + queue_size=queue_size, + ) + self._subscription_change_counter = self._subscription_change_counter % 0xFF + 1 + self._subscription_relation_id = (self._subscription_relation_id + 1) & 0xFFFFFFFF + logger.info(f"Subscription created, id={subscription_id:#x}") + return subscription_id + + async def receive_subscription_notification( + self, subscription_id: int | None = None, timeout: Optional[float] = None + ) -> SubscriptionNotification: + """Wait for one routed data notification.""" + if subscription_id is not None: + queued = self._subscriptions.pop(subscription_id) + if queued is not None: + return queued + while True: + async with self._lock: + if not self._connected: + raise RuntimeError("Not connected") + if self._notification_frames: + frame = self._notification_frames.popleft() + else: + receive = self._recv_cotp_dt() + frame = await asyncio.wait_for(receive, timeout) if timeout is not None else await receive + frame_subscription_id = notification_subscription_id(frame) + if frame_subscription_id in self._alarm_subscription_ids: + self._alarm_notification_frames.append(frame) + continue + notification = parse_subscription_notification(frame) + matched, credit_update = self._subscriptions.route(notification) + if not matched: + continue + if matched and credit_update is not None: + await self._send_subscription_credit(notification.subscription_id, credit_update) + target_id = notification.subscription_id if subscription_id is None else subscription_id + queued = self._subscriptions.pop(target_id) + if queued is not None: + return queued + + async def iter_subscription_notifications( + self, subscription_id: int, limit: int | None = None + ) -> AsyncIterator[SubscriptionNotification]: + """Yield routed notifications, optionally stopping after ``limit``.""" + delivered = 0 + while limit is None or delivered < limit: + yield await self.receive_subscription_notification(subscription_id) + delivered += 1 + + def subscription_queue(self, subscription_id: int) -> AsyncSubscriptionQueue: + """Return a queue-like async view for one active subscription.""" + self._subscriptions.diagnostics(subscription_id) + return AsyncSubscriptionQueue(self, subscription_id) + + def subscription_diagnostics(self, subscription_id: int) -> SubscriptionDiagnostics: + diagnostics = self._subscriptions.diagnostics(subscription_id) + return SubscriptionDiagnostics( + diagnostics.subscription_id, + diagnostics.queued_notifications, + diagnostics.dropped_notifications, + diagnostics.missed_sequence_updates, + self._notification_frame_overflows, + ) async def delete_subscription(self, subscription_id: int) -> None: """Delete a data change subscription. @@ -705,8 +848,11 @@ async def delete_subscription(self, subscription_id: int) -> None: Args: subscription_id: ID returned by :meth:`create_subscription`. """ - payload = struct.pack(">I", subscription_id) + struct.pack(">I", 0) + if self._subscription_container_id == 0: + raise RuntimeError("PLC did not provide a subscription container object") + payload = build_delete_subscription_request(self._subscription_container_id, self._protocol_version) await self._send_request(FunctionCode.DELETE_OBJECT, payload) + self._subscriptions.unregister(subscription_id) logger.info(f"Subscription {subscription_id:#x} deleted") async def create_alarm_subscription( @@ -723,7 +869,9 @@ async def create_alarm_subscription( object_ids, _, return_value = parse_create_object_session_id(response) if return_value != 0 or not object_ids: raise RuntimeError(f"Alarm subscription failed: PLC returned {return_value:#x}") - return object_ids[0] + subscription_id = object_ids[0] + self._alarm_subscription_ids.add(subscription_id) + return subscription_id async def delete_alarm_subscription(self, subscription_id: int) -> None: """Delete an alarm subscription created by this client.""" @@ -731,6 +879,11 @@ async def delete_alarm_subscription(self, subscription_id: int) -> None: raise RuntimeError("PLC did not provide a subscription container object") payload = build_delete_alarm_subscription_request(self._subscription_container_id, self._protocol_version) await self._send_request(FunctionCode.DELETE_OBJECT, payload) + self._alarm_subscription_ids.discard(subscription_id) + self._alarm_notification_frames = deque( + (frame for frame in self._alarm_notification_frames if notification_subscription_id(frame) != subscription_id), + maxlen=self._alarm_notification_frames.maxlen, + ) logger.info(f"Alarm subscription {subscription_id:#x} deleted") async def receive_alarm_notification( @@ -738,15 +891,28 @@ async def receive_alarm_notification( ) -> AlarmNotification: """Wait for one alarm notification, optionally with a timeout in seconds. - Do not run this alongside a data-subscription receive loop on the same - connection: mixed notification dispatch is not supported yet. + Data notifications encountered first are routed to their bounded queues. """ - async with self._lock: - if not self._connected: - raise RuntimeError("Not connected") - receive = self._recv_cotp_dt() - frame = await asyncio.wait_for(receive, timeout) if timeout is not None else await receive - return parse_alarm_notification(frame, language_ids) + while True: + if self._alarm_notification_frames: + frame = self._alarm_notification_frames.popleft() + else: + async with self._lock: + if not self._connected: + raise RuntimeError("Not connected") + if self._notification_frames: + frame = self._notification_frames.popleft() + else: + receive = self._recv_cotp_dt() + frame = await asyncio.wait_for(receive, timeout) if timeout is not None else await receive + frame_subscription_id = notification_subscription_id(frame) + if self._subscriptions.contains(frame_subscription_id): + notification = parse_subscription_notification(frame) + matched, credit_update = self._subscriptions.route(notification) + if matched and credit_update is not None: + await self._send_subscription_credit(frame_subscription_id, credit_update) + continue + return parse_alarm_notification(frame, language_ids) async def read_alarms(self, language_ids: Optional[list[LanguageId | int]] = None) -> list[Alarm]: """Return a snapshot of the PLC's active alarms without consuming notifications.""" @@ -1000,6 +1166,36 @@ async def _explore_type_info_container(self) -> list["typeinfo.PObject"]: _MAX_REASSEMBLED_BYTES = 16 * 1024 * 1024 _MAX_REASSEMBLED_FRAGMENTS = 4096 + async def _send_subscription_credit(self, subscription_id: int, credit_limit: int) -> None: + """Send the reference driver's fire-and-forget credit update.""" + if not 1 <= credit_limit <= 255: + raise ValueError("credit_limit must be between 1 and 255") + value = bytes([0x00, DataType.INT]) + struct.pack(">h", credit_limit) + payload = _build_set_variable_payload(subscription_id, Ids.SUBSCRIPTION_CREDIT_LIMIT, value) + async with self._lock: + if not self._connected or self._writer is None or self._reader is None: + raise S7ConnectionError("Not connected") + sequence = self._next_sequence_number() + header = struct.pack( + ">BHHHHIB", + Opcode.REQUEST, + 0, + FunctionCode.SET_VARIABLE, + 0, + sequence, + self._session_id, + 0x74, + ) + integrity = b"" + if self._with_integrity_id and self._protocol_version >= ProtocolVersion.V2: + integrity = encode_uint32_vlq(self._integrity_id_write) + request = header + payload[:-4] + integrity + payload[-4:] + frame = encode_header(self._protocol_version, len(request)) + request + frame += struct.pack(">BBH", 0x72, self._protocol_version, 0) + await self._send_cotp_dt(frame) + if integrity: + self._integrity_id_write = (self._integrity_id_write + 1) & 0xFFFFFFFF + async def _send_request( self, function_code: int, @@ -1069,40 +1265,44 @@ async def _send_request( data = await self._recv_reassembled_payload(response_data) if len(data) < 10: raise S7ConnectionError("Response too short") - resp_func = struct.unpack_from(">H", data, 3)[0] - resp_seq = struct.unpack_from(">H", data, 7)[0] - if resp_seq != seq_num: - raise S7ProtocolError( - f"Response sequence mismatch: expected seq={seq_num}, got seq={resp_seq} for function=0x{resp_func:04X}" - ) + _validate_response_header(data, function_code, seq_num) return bytes(data[10:]) _, data_length, consumed = decode_header(response_data) response = response_data[consumed : consumed + data_length] - if len(response) < 10: - raise S7ConnectionError("Response too short") + _validate_response_header(response, function_code, seq_num) # RESPONSE header is 10 bytes (opcode+res+func+res+seqnr+transport) — responses # carry no SessionId field (requests do, hence their 14-byte header). For V2+ the # IntegrityId travels at the END of the payload and is ignored by the parsers. - resp_func = struct.unpack_from(">H", response, 3)[0] - resp_seq = struct.unpack_from(">H", response, 7)[0] - if resp_seq != seq_num: - raise S7ProtocolError( - f"Response sequence mismatch: expected seq={seq_num}, got seq={resp_seq} for function=0x{resp_func:04X}" - ) return response[10:] async def _recv_response_frame(self) -> bytes: - """Receive the next application response, consuming non-fatal SystemEvents.""" - for _ in range(_MAX_SYSTEM_EVENTS_PER_RESPONSE + 1): + """Receive the next response, queueing unsolicited application frames.""" + system_events = 0 + while True: response_data = await self._recv_cotp_dt() + if not response_data: + raise S7ConnectionError("Connection closed while waiting for an S7CommPlus response") version, data_length, consumed = decode_header(response_data) - if version != ProtocolVersion.SYSTEM_EVENT: + if version == ProtocolVersion.SYSTEM_EVENT: + _check_system_event(bytes(response_data[consumed : consumed + data_length])) + system_events += 1 + if system_events > _MAX_SYSTEM_EVENTS_PER_RESPONSE: + raise S7ProtocolError("Too many S7CommPlus SystemEvents while waiting for a response") + continue + if data_length < 10: return response_data - _check_system_event(bytes(response_data[consumed : consumed + data_length])) - raise S7ProtocolError("Too many S7CommPlus SystemEvents while waiting for a response") + opcode = _incoming_frame_opcode(response_data) + if opcode == Opcode.NOTIFICATION: + if len(self._notification_frames) == self._notification_frames.maxlen: + self._notification_frame_overflows += 1 + self._notification_frames.append(response_data) + continue + if opcode not in (Opcode.RESPONSE, Opcode.RESPONSE2): + raise S7ProtocolError(f"Unexpected S7CommPlus opcode 0x{opcode:02X} while waiting for a response") + return response_data async def _recv_reassembled_payload(self, initial_data: bytes = b"") -> bytes: """Receive a possibly-fragmented S7CommPlus response, returning its data section. diff --git a/s7commplus/catalog.py b/s7commplus/catalog.py index b879022d..4290f89c 100644 --- a/s7commplus/catalog.py +++ b/s7commplus/catalog.py @@ -4,6 +4,7 @@ from dataclasses import dataclass from collections.abc import Iterator +import struct from typing import Any, Optional from .protocol import DataType @@ -96,6 +97,44 @@ def from_browse(cls, item: dict[str, Any]) -> "SymbolicTag": nonopt_bitoffset=int(item.get("nonopt_bitoffset", 0)), ) + def decode_value(self, raw: bytes) -> Any: + """Decode a raw symbolic value when its scalar type is known. + + Unknown, structured, truncated, and array values remain bytes so + callers never lose firmware-specific data. + """ + if self.array_dimensions: + return raw + formats: dict[Softdatatype, str] = { + Softdatatype.BOOL: ">?", + Softdatatype.BBOOL: ">?", + Softdatatype.BYTE: ">B", + Softdatatype.WORD: ">H", + Softdatatype.INT: ">h", + Softdatatype.DWORD: ">I", + Softdatatype.DINT: ">i", + Softdatatype.REAL: ">f", + Softdatatype.LREAL: ">d", + Softdatatype.ULINT: ">Q", + Softdatatype.LINT: ">q", + Softdatatype.LWORD: ">Q", + Softdatatype.USINT: ">B", + Softdatatype.UINT: ">H", + Softdatatype.UDINT: ">I", + Softdatatype.SINT: ">b", + } + if self.softdatatype is Softdatatype.CHAR and len(raw) == 1: + return raw.decode("latin-1") + if self.softdatatype in (Softdatatype.STRING, Softdatatype.WSTRING): + try: + return raw.decode("utf-8") + except UnicodeDecodeError: + return raw + fmt = formats.get(self.softdatatype) + if fmt is None or len(raw) != struct.calcsize(fmt): + return raw + return struct.unpack(fmt, raw)[0] + @dataclass(frozen=True) class TagResult: diff --git a/s7commplus/client.py b/s7commplus/client.py index ef1cf86d..9302500a 100644 --- a/s7commplus/client.py +++ b/s7commplus/client.py @@ -5,7 +5,8 @@ import logging import struct -from collections.abc import Callable, Mapping, Sequence +from collections import deque +from collections.abc import Callable, Iterator, Mapping, Sequence from typing import Any, Optional, TypeAlias, TypeVar from snap7.error import S7ConnectionError, S7ProtocolError @@ -33,10 +34,13 @@ from .connection import S7CommPlusConnection from .protocol import DataType, ElementID, FunctionCode, Ids, ObjectId, ProtocolVersion from .subscription import ( + SubscriptionDiagnostics, SubscriptionItem, SubscriptionNotification, + SubscriptionRegistry, build_delete_subscription_request, build_subscription_request, + notification_subscription_id, parse_subscription_notification, ) from .vlq import decode_uint32_vlq, decode_uint64_vlq, encode_uint32_vlq @@ -77,6 +81,9 @@ def __init__(self) -> None: self._connect_params: Optional[dict[str, Any]] = None self._subscription_change_counter = 1 self._subscription_relation_id = 0x7FFFC001 + self._subscriptions = SubscriptionRegistry() + self._alarm_subscription_ids: set[int] = set() + self._alarm_notification_frames: deque[bytes] = deque(maxlen=100) self._symbol_catalog: Optional[SymbolCatalog] = None @property @@ -179,6 +186,9 @@ def _reconnect(self) -> None: symbolic ``GetMultiVariables`` read per connection, so multi-step flows such as :meth:`browse` need a fresh session to continue. """ + self._subscriptions.clear() + self._alarm_subscription_ids.clear() + self._alarm_notification_frames.clear() if self._connection is not None: try: self._connection.disconnect() @@ -201,6 +211,9 @@ def _with_reconnect(self, op: Callable[[], "_T"]) -> "_T": def disconnect(self) -> None: """Disconnect from PLC.""" + self._subscriptions.clear() + self._alarm_subscription_ids.clear() + self._alarm_notification_frames.clear() if self._connection: self._connection.disconnect() self._connection = None @@ -814,9 +827,11 @@ def _explore_type_info_container(self) -> list["typeinfo.PObject"]: def create_subscription( self, - items: Sequence[SubscriptionItem | str], + items: Sequence[SubscriptionItem | SymbolicTag | str], cycle_ms: int = 100, credit_limit: int = 10, + credit_step: int = 5, + queue_size: int = 100, ) -> int: """Create a data change subscription. @@ -832,6 +847,8 @@ def create_subscription( cycle_ms: Sampling cycle in milliseconds. credit_limit: Number of notification credits. The default of 10 matches the value accepted by real S7-1500 PLCs. + credit_step: Credits added one tick before a finite limit expires. + queue_size: Maximum buffered notifications for this subscription. Returns: Subscription object ID assigned by the PLC. @@ -841,13 +858,23 @@ def create_subscription( 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] + if not 0 <= credit_step <= 255: + raise ValueError("credit_step must be between 0 and 255") + normalized = [ + SubscriptionItem.from_access_sequence(item) + if isinstance(item, str) + else SubscriptionItem.from_tag(item) + if isinstance(item, SymbolicTag) + else item + for item in items + ] + change_counter = self._subscription_change_counter 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, + change_counter=change_counter, relation_id=self._subscription_relation_id, ) response = self._connection.send_request( @@ -862,14 +889,68 @@ def create_subscription( 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] + self._subscriptions.register( + subscription_id, + normalized, + change_counter=change_counter, + credit_limit=credit_limit, + credit_step=credit_step, + queue_size=queue_size, + ) 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.""" + def receive_subscription_notification(self, subscription_id: int | None = None) -> SubscriptionNotification: + """Block until one routed data notification is available.""" if self._connection is None: raise RuntimeError("Not connected") - return parse_subscription_notification(self._connection.receive_notification()) + if subscription_id is not None: + queued = self._subscriptions.pop(subscription_id) + if queued is not None: + return queued + while True: + frame = self._connection.receive_notification() + frame_subscription_id = notification_subscription_id(frame) + if frame_subscription_id in self._alarm_subscription_ids: + self._alarm_notification_frames.append(frame) + continue + notification = parse_subscription_notification(frame) + matched, credit_update = self._subscriptions.route(notification) + if not matched: + continue + if matched and credit_update is not None: + self._connection.send_subscription_credit(notification.subscription_id, credit_update) + target_id = notification.subscription_id if subscription_id is None else subscription_id + queued = self._subscriptions.pop(target_id) + if queued is not None: + return queued + + def iter_subscription_notifications( + self, subscription_id: int, limit: int | None = None + ) -> Iterator[SubscriptionNotification]: + """Yield routed notifications, optionally stopping after ``limit``.""" + delivered = 0 + while limit is None or delivered < limit: + yield self.receive_subscription_notification(subscription_id) + delivered += 1 + + def add_subscription_callback(self, subscription_id: int, callback: Callable[[SubscriptionNotification], None]) -> None: + """Invoke ``callback`` whenever this client dispatches an update.""" + self._subscriptions.add_callback(subscription_id, callback) + + def remove_subscription_callback(self, subscription_id: int, callback: Callable[[SubscriptionNotification], None]) -> None: + self._subscriptions.remove_callback(subscription_id, callback) + + def subscription_diagnostics(self, subscription_id: int) -> SubscriptionDiagnostics: + """Return bounded-queue overflow and sequence-gap counters.""" + diagnostics = self._subscriptions.diagnostics(subscription_id) + return SubscriptionDiagnostics( + diagnostics.subscription_id, + diagnostics.queued_notifications, + diagnostics.dropped_notifications, + diagnostics.missed_sequence_updates, + getattr(self._connection, "_notification_frame_overflows", 0), + ) def delete_subscription(self, subscription_id: int) -> None: """Delete a data change subscription. @@ -888,6 +969,7 @@ def delete_subscription(self, subscription_id: int) -> None: # 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) + self._subscriptions.unregister(subscription_id) logger.info(f"Subscription {subscription_id:#x} deleted") def create_alarm_subscription( @@ -923,7 +1005,9 @@ def create_alarm_subscription( object_ids, _, return_value = parse_create_object_session_id(response) if return_value != 0 or not object_ids: raise RuntimeError(f"Alarm subscription failed: PLC returned {return_value:#x}") - return object_ids[0] + subscription_id = object_ids[0] + self._alarm_subscription_ids.add(subscription_id) + return subscription_id def delete_alarm_subscription(self, subscription_id: int) -> None: """Delete an alarm subscription created by this client.""" @@ -935,17 +1019,34 @@ def delete_alarm_subscription(self, subscription_id: int) -> None: self._connection.subscription_container_id, self._connection.protocol_version ) self._connection.send_request(FunctionCode.DELETE_OBJECT, payload) + self._alarm_subscription_ids.discard(subscription_id) + self._alarm_notification_frames = deque( + (frame for frame in self._alarm_notification_frames if notification_subscription_id(frame) != subscription_id), + maxlen=self._alarm_notification_frames.maxlen, + ) logger.info(f"Alarm subscription {subscription_id:#x} deleted") def receive_alarm_notification(self, language_ids: Optional[list[LanguageId | int]] = None) -> AlarmNotification: """Block until the PLC sends one alarm notification. - Do not run this alongside a data-subscription receive loop on the same - connection: mixed notification dispatch is not supported yet. + Data notifications encountered first are routed to their bounded queues. """ if self._connection is None: raise RuntimeError("Not connected") - return parse_alarm_notification(self._connection.receive_notification(), language_ids) + while True: + frame = ( + self._alarm_notification_frames.popleft() + if self._alarm_notification_frames + else self._connection.receive_notification() + ) + frame_subscription_id = notification_subscription_id(frame) + if self._subscriptions.contains(frame_subscription_id): + notification = parse_subscription_notification(frame) + matched, credit_update = self._subscriptions.route(notification) + if matched and credit_update is not None: + self._connection.send_subscription_credit(frame_subscription_id, credit_update) + continue + return parse_alarm_notification(frame, language_ids) def read_alarms(self, language_ids: Optional[list[LanguageId | int]] = None) -> list[Alarm]: """Return the PLC's current active alarm state. diff --git a/s7commplus/connection.py b/s7commplus/connection.py index 39a73833..ef8be22d 100644 --- a/s7commplus/connection.py +++ b/s7commplus/connection.py @@ -45,6 +45,7 @@ import ssl import struct import tempfile +import threading from collections import deque from types import TracebackType from typing import Any, Optional, Type @@ -80,6 +81,47 @@ logger = logging.getLogger(__name__) +def _incoming_frame_opcode(frame: bytes) -> int: + """Return the application opcode from a complete non-SystemEvent frame.""" + from snap7.error import S7ConnectionError, S7ProtocolError + + if not frame: + raise S7ConnectionError("Connection closed while waiting for an S7CommPlus frame") + version, data_length, consumed = decode_header(frame) + data = bytes(frame[consumed : consumed + data_length]) + if version == ProtocolVersion.V3 and data: + hash_length = data[0] + if len(data) <= 1 + hash_length: + raise S7ProtocolError("Truncated S7CommPlus V3 integrity envelope") + data = data[1 + hash_length :] + if not data: + raise S7ProtocolError("S7CommPlus frame has no application opcode") + return data[0] + + +def _validate_response_header(response: bytes, expected_function: int, expected_sequence: int) -> None: + """Validate that application data is the response to one outstanding request.""" + from snap7.error import S7ConnectionError, S7ProtocolError + + if len(response) < 10: + raise S7ConnectionError("Response too short") + opcode = response[0] + function = struct.unpack_from(">H", response, 3)[0] + sequence = struct.unpack_from(">H", response, 7)[0] + if opcode not in (Opcode.RESPONSE, Opcode.RESPONSE2): + raise S7ProtocolError(f"Unexpected response opcode 0x{opcode:02X}") + # A PLC may answer any failed request with the protocol's generic ERROR + # function while retaining the request sequence number. + if function not in (expected_function, FunctionCode.ERROR): + raise S7ProtocolError( + f"Response function mismatch: expected function=0x{expected_function:04X}, got function=0x{function:04X}" + ) + if sequence != expected_sequence: + raise S7ProtocolError( + f"Response sequence mismatch: expected seq={expected_sequence}, got seq={sequence} for function=0x{function:04X}" + ) + + def _log_create_object_return_value(return_value: int, tls_active: bool) -> None: """Log a non-zero CreateObject status without guessing at TLS requirements.""" if return_value == 0: @@ -123,6 +165,7 @@ def _log_create_object_return_value(return_value: int, tls_active: bool) -> None _MAX_SYSTEM_EVENTS_PER_RESPONSE = 16 _SYSTEM_EVENT_RETURN_VALUE_ID = 40305 +_MAX_QUEUED_NOTIFICATION_FRAMES = 1000 def _system_event_return_value(payload: bytes) -> Optional[int]: @@ -217,20 +260,37 @@ def _set_s7_groups(ctx: ssl.SSLContext) -> None: ) -def _verify_v3_hmac(protected: bytes, session_key: bytes) -> bytes: - """Verify and remove the V3 HMAC prefix from application data.""" - from snap7.error import S7ConnectionError +def _verify_v3_hmac(protected: bytes, session_key: bytes, digest_state: hmac.HMAC | None = None) -> bytes: + """Verify and remove a V3 HMAC prefix. + + ``digest_state`` implements the legacy fragmented-response behavior: the + first digest covers the first fragment and each later digest covers all + application bytes accumulated so far. The state advances only after a + successful constant-time comparison. + """ + from snap7.error import S7IntegrityError if not protected: - raise S7ConnectionError("Empty V3 frame") + raise S7IntegrityError("Empty authenticated S7CommPlus V3 frame; reconnect before retrying") digest_length = protected[0] - if digest_length != hashlib.sha256().digest_size or len(protected) < 1 + digest_length: - raise S7ConnectionError(f"Invalid V3 HMAC length: {digest_length}") + expected_length = hashlib.sha256().digest_size + if digest_length != expected_length: + raise S7IntegrityError( + f"Invalid S7CommPlus V3 digest length {digest_length}, expected {expected_length}; reconnect before retrying" + ) + if len(protected) < 1 + digest_length: + raise S7IntegrityError( + f"Truncated S7CommPlus V3 digest: received {len(protected) - 1} of {digest_length} bytes; reconnect before retrying" + ) received_digest = protected[1 : 1 + digest_length] application_data = protected[1 + digest_length :] - expected_digest = hmac.new(session_key[:24], application_data, hashlib.sha256).digest() + verifier = digest_state.copy() if digest_state is not None else hmac.new(session_key[:24], digestmod=hashlib.sha256) + verifier.update(application_data) + expected_digest = verifier.digest() if not hmac.compare_digest(received_digest, expected_digest): - raise S7ConnectionError("Invalid V3 HMAC") + raise S7IntegrityError("S7CommPlus V3 response integrity check failed; reconnect before retrying") + if digest_state is not None: + digest_state.update(application_data) return bytes(application_data) @@ -477,7 +537,9 @@ def __init__( # Password for post-auth legitimation (V1-initial PLCs) self._connect_password: str = "" - self._notification_frames: deque[bytes] = deque() + self._notification_frames: deque[bytes] = deque(maxlen=_MAX_QUEUED_NOTIFICATION_FRAMES) + self._notification_frame_overflows = 0 + self._request_lock = threading.Lock() # Effective protection level, read once the session is up self._protection_level: Optional[int] = None @@ -777,19 +839,18 @@ def _send_legitimation_legacy(self, response: bytes) -> None: _check_set_variable_response(resp_payload) def collect_explore_frames(self, first_payload: bytes) -> bytes: - """Collect multi-fragment EXPLORE continuation frames for V3 PLCs. + """Collect unauthenticated multi-fragment EXPLORE continuation frames. On V3 PLCs (FW >= V4.5) a large EXPLORE response (e.g. RID 0x8A11FFFF) spans multiple TPKT frames. The first frame is the normal response (already stripped of its 10-byte header by send_request). Continuation - frames carry **no** response header — they are raw BLOB data protected - only by a V3 HMAC prefix. The caller must concatenate them before - parsing. + frames carry no response header. Authenticated callers must use + ``send_request(..., reassemble=True)`` because this legacy helper no + longer has the first frame bytes needed to verify cumulative digests. Termination: a ``frag_len == 0`` frame is the standard S7CommPlus - end-of-stream trailer. As a fallback, a frame whose body (after HMAC - strip) is measurably shorter than the first frame body is treated as the - last fragment (5-byte tolerance). + end-of-stream trailer. As a fallback, a measurably shorter frame body is + treated as the last fragment (5-byte tolerance). Collection is capped by ``_MAX_REASSEMBLED_FRAGMENTS`` and ``_MAX_REASSEMBLED_BYTES`` to prevent unbounded allocation on malformed @@ -802,6 +863,14 @@ def collect_explore_frames(self, first_payload: bytes) -> bytes: Returns: All fragment payloads concatenated (first_payload + continuations). """ + if self._session_key is not None: + from snap7.error import S7ProtocolError + + raise S7ProtocolError( + "Authenticated Explore continuations require send_request(..., reassemble=True) so cumulative digests " + "can be verified" + ) + # The first frame body (already header-stripped) was originally # len(first_payload) + 10 bytes on the wire (10-byte response header). # Continuation frames of the same "full" size will be that long after @@ -827,10 +896,6 @@ def collect_explore_frames(self, first_payload: bytes) -> bytes: if frag_len == 0: break # standard S7CommPlus end-of-stream trailer body = raw[4 : 4 + frag_len] - # V3 non-TLS: strip the HMAC prefix ([hash_len][hash_bytes]) - if self._protocol_version >= ProtocolVersion.V3 and len(body) > 33: - hash_len = body[0] - body = body[1 + hash_len :] if not body: break all_data += body @@ -873,9 +938,102 @@ def disconnect(self) -> None: self._integrity_id_write = 0 self._protection_level = None self._notification_frames.clear() + self._notification_frame_overflows = 0 self._iso_conn.disconnect() + def _invalidate_integrity_failure(self) -> None: + """Close an untrusted stream without sending protocol data on it.""" + self._session_ready = False + self._session_id = 0 + self.disconnect() + + def _verify_v3_hmac(self, protected: bytes, digest_state: hmac.HMAC | None = None) -> bytes: + """Verify authenticated data and make any failure terminal for this connection.""" + from snap7.error import S7IntegrityError + + if self._session_key is None: + self._invalidate_integrity_failure() + raise S7IntegrityError("Authenticated S7CommPlus V3 response arrived without a session key; reconnect") + try: + return _verify_v3_hmac(protected, self._session_key, digest_state) + except S7IntegrityError: + self._invalidate_integrity_failure() + raise + + def _verify_v3_frame(self, frame: bytes) -> None: + """Verify a complete V3 frame before its opcode is inspected or queued.""" + from snap7.error import S7IntegrityError + + version, data_length, consumed = decode_header(frame) + if version != ProtocolVersion.V3: + if self._session_key is not None: + self._invalidate_integrity_failure() + raise S7IntegrityError( + f"Authenticated S7CommPlus response used unauthenticated frame version {version}; reconnect" + ) + return + frame_end = consumed + data_length + if len(frame) < frame_end: + self._invalidate_integrity_failure() + raise S7IntegrityError( + f"Truncated authenticated S7CommPlus V3 frame: declared {data_length} data bytes, " + f"received {max(0, len(frame) - consumed)}; reconnect before retrying" + ) + self._verify_v3_hmac(bytes(frame[consumed:frame_end])) + def send_request(self, function_code: int, payload: bytes = b"", integrity_tail: int = 4, reassemble: bool = False) -> bytes: + """Serialize one request/response exchange on the connection.""" + with self._request_lock: + return self._send_request(function_code, payload, integrity_tail, reassemble) + + def send_subscription_credit(self, subscription_id: int, credit_limit: int) -> None: + """Replenish finite notification credits without waiting for a response.""" + if not 1 <= credit_limit <= 255: + raise ValueError("credit_limit must be between 1 and 255") + value = bytes([0x00, DataType.INT]) + struct.pack(">h", credit_limit) + payload = _build_set_variable_payload(subscription_id, Ids.SUBSCRIPTION_CREDIT_LIMIT, value) + with self._request_lock: + self._send_fire_and_forget(FunctionCode.SET_VARIABLE, payload, integrity_tail=4) + + def _send_fire_and_forget(self, function_code: int, payload: bytes, integrity_tail: int) -> None: + """Send a request with transport flags 0x74 and consume no response.""" + if not (self._connected or self._session_ready): + from snap7.error import S7ConnectionError + + raise S7ConnectionError("Not connected") + + seq_num = self._next_sequence_number() + request_header = struct.pack( + ">BHHHHIB", + Opcode.REQUEST, + 0, + function_code, + 0, + seq_num, + self._session_id, + 0x74, + ) + integrity_id_bytes = b"" + if self._with_integrity_id: + integrity_id_bytes = encode_uint32_vlq(self._integrity_id_write) + if integrity_id_bytes and len(payload) >= integrity_tail: + request = request_header + payload[:-integrity_tail] + integrity_id_bytes + payload[-integrity_tail:] + else: + request = request_header + integrity_id_bytes + payload + + if self._session_key is not None: + digest = hmac.new(self._session_key[:24], request, hashlib.sha256).digest() + frame_data = bytes([0x20]) + digest + request + frame = encode_header(ProtocolVersion.V3, len(frame_data)) + frame_data + frame += struct.pack(">BBH", 0x72, ProtocolVersion.V3, 0) + else: + frame = encode_header(self._protocol_version, len(request)) + request + frame += struct.pack(">BBH", 0x72, self._protocol_version, 0) + self._send_s7_data(frame) + if self._with_integrity_id: + self._integrity_id_write = (self._integrity_id_write + 1) & 0xFFFFFFFF + + def _send_request(self, function_code: int, payload: bytes, integrity_tail: int, reassemble: bool) -> bytes: """Send an S7CommPlus request and receive the response. For V2+ with IntegrityId tracking enabled, the IntegrityId is spliced into @@ -975,14 +1133,7 @@ def send_request(self, function_code: int, payload: bytes = b"", integrity_tail: from snap7.error import S7ConnectionError raise S7ConnectionError("Response too short") - resp_func = struct.unpack_from(">H", data, 3)[0] - resp_seq = struct.unpack_from(">H", data, 7)[0] - if resp_seq != seq_num: - from snap7.error import S7ProtocolError - - raise S7ProtocolError( - f"Response sequence mismatch: expected seq={seq_num}, got seq={resp_seq} for function=0x{resp_func:04X}" - ) + _validate_response_header(data, function_code, seq_num) logger.debug(f" Reassembled response ({len(data)} bytes), payload {len(data) - 10} bytes") resp_payload = bytes(data[10:]) if self._session_key is not None: @@ -1001,19 +1152,12 @@ def send_request(self, function_code: int, payload: bytes = b"", integrity_tail: # V3 responses have a hash-length byte + HMAC prefix before the payload. if version == ProtocolVersion.V3: - if self._session_key is None: - from snap7.error import S7ConnectionError - - raise S7ConnectionError("V3 response received without a session key") - response = _verify_v3_hmac(response, self._session_key) + response = self._verify_v3_hmac(response) logger.debug(" V3 HMAC verified") logger.debug(f" Response data ({len(response)} bytes): {response.hex(' ')}") - if len(response) < 10: - from snap7.error import S7ConnectionError - - raise S7ConnectionError("Response too short") + _validate_response_header(response, function_code, seq_num) # Parse the 10-byte response header for debug (responses carry no SessionId) resp_opcode = response[0] @@ -1024,13 +1168,6 @@ def send_request(self, function_code: int, payload: bytes = b"", integrity_tail: f" Response header: opcode=0x{resp_opcode:02X} function=0x{resp_func:04X} " f"seq={resp_seq} transport=0x{resp_transport:02X}" ) - if resp_seq != seq_num: - from snap7.error import S7ProtocolError - - raise S7ProtocolError( - f"Response sequence mismatch: expected seq={seq_num}, got seq={resp_seq} for function=0x{resp_func:04X}" - ) - # RESPONSE header is 10 bytes (opcode+res+func+res+seqnr+transport) — responses have # NO SessionId field (requests do, making their header 14 bytes). resp_offset = 10 @@ -1055,11 +1192,13 @@ def send_request(self, function_code: int, payload: bytes = b"", integrity_tail: def _recv_response_frame(self) -> bytes: """Receive the next response, queueing notifications and consuming non-fatal SystemEvents.""" - from snap7.error import S7ProtocolError + from snap7.error import S7ConnectionError, S7ProtocolError system_events = 0 while True: response_frame = self._recv_s7_data() + if not response_frame: + raise S7ConnectionError("Connection closed while waiting for an S7CommPlus response") version, data_length, consumed = decode_header(response_frame) if version == ProtocolVersion.SYSTEM_EVENT: _check_system_event(bytes(response_frame[consumed : consumed + data_length])) @@ -1067,43 +1206,49 @@ def _recv_response_frame(self) -> bytes: if system_events > _MAX_SYSTEM_EVENTS_PER_RESPONSE: raise S7ProtocolError("Too many S7CommPlus SystemEvents while waiting for a response") continue - if self._is_notification_frame(response_frame): + self._verify_v3_frame(response_frame) + if data_length < 10: + return response_frame + opcode = _incoming_frame_opcode(response_frame) + if opcode == Opcode.NOTIFICATION: + if len(self._notification_frames) == self._notification_frames.maxlen: + self._notification_frame_overflows += 1 self._notification_frames.append(response_frame) continue + if opcode not in (Opcode.RESPONSE, Opcode.RESPONSE2): + raise S7ProtocolError(f"Unexpected S7CommPlus opcode 0x{opcode:02X} while waiting for a response") return response_frame @staticmethod def _is_notification_frame(frame: bytes) -> bool: """Return whether a complete frame contains an unsolicited notification.""" + from snap7.error import S7ConnectionError, S7ProtocolError + try: - version, data_length, consumed = decode_header(frame) - except (IndexError, ValueError): + return _incoming_frame_opcode(frame) == Opcode.NOTIFICATION + except (IndexError, ValueError, S7ConnectionError, S7ProtocolError): 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. + It is serialized with :meth:`send_request` because both consume the + same connection stream. """ - if not self._connected: - from snap7.error import S7ConnectionError + with self._request_lock: + 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("Not connected") + frame = self._notification_frames.popleft() if self._notification_frames else self._recv_s7_data() + self._verify_v3_frame(frame) + if not self._is_notification_frame(frame): + from snap7.error import S7ConnectionError - raise S7ConnectionError("Expected an S7CommPlus notification") - return frame + 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. @@ -1132,11 +1277,27 @@ def ensure(n: int) -> None: data = bytearray() fragments = 0 + expected_version: int | None = None + digest_state = hmac.new(self._session_key[:24], digestmod=hashlib.sha256) if self._session_key is not None else None while True: ensure(4) if buf[0] != 0x72: raise S7ConnectionError("Expected S7CommPlus fragment header (0x72)") fragment_version = buf[1] + if expected_version is None: + expected_version = fragment_version + elif fragment_version != expected_version: + if self._session_key is not None: + from snap7.error import S7IntegrityError + + self._invalidate_integrity_failure() + raise S7IntegrityError( + f"Authenticated S7CommPlus response changed fragment version from {expected_version} " + f"to {fragment_version}; reconnect" + ) + raise S7ConnectionError( + f"S7CommPlus response changed fragment version from {expected_version} to {fragment_version}" + ) frag_len = (buf[2] << 8) | buf[3] del buf[:4] if frag_len == 0: @@ -1145,9 +1306,7 @@ def ensure(n: int) -> None: fragment_data = bytes(buf[:frag_len]) del buf[:frag_len] if fragment_version == ProtocolVersion.V3: - if self._session_key is None: - raise S7ConnectionError("V3 response received without a session key") - fragment_data = _verify_v3_hmac(fragment_data, self._session_key) + fragment_data = self._verify_v3_hmac(fragment_data, digest_state) data.extend(fragment_data) fragments += 1 if fragments > self._MAX_REASSEMBLED_FRAGMENTS or len(data) > self._MAX_REASSEMBLED_BYTES: @@ -1155,7 +1314,7 @@ def ensure(n: int) -> None: # The next 4 bytes are either the trailer (0x72 ver 0x0000) or the next # fragment's header (0x72 ver len>0). ensure(4) - if buf[0] == 0x72 and buf[2] == 0 and buf[3] == 0: + if buf[0] == 0x72 and buf[1] == expected_version and buf[2] == 0 and buf[3] == 0: del buf[:4] # consume trailer — last fragment break return bytes(data) diff --git a/s7commplus/subscription.py b/s7commplus/subscription.py index 5aecc70d..b47b097f 100644 --- a/s7commplus/subscription.py +++ b/s7commplus/subscription.py @@ -7,8 +7,13 @@ """ import struct -from collections.abc import Sequence -from dataclasses import dataclass +from collections import deque +from collections.abc import Callable, Sequence +from dataclasses import dataclass, field, replace +from typing import TYPE_CHECKING, Any + +if TYPE_CHECKING: + from .catalog import SymbolicTag from .codec import decode_header, decode_pvalue_to_bytes, encode_object_qualifier from .protocol import DataType, ElementID, Ids, Opcode, ProtocolVersion @@ -24,6 +29,7 @@ class SubscriptionItem: access_sub_area: int | None = None symbol_crc: int = 0 reference_id: int = 0 + tag: "SymbolicTag | None" = None def __post_init__(self) -> None: if not 0 <= self.access_area <= 0xFFFFFFFF: @@ -72,6 +78,11 @@ def from_access_sequence( raise ValueError("access_sequence components must be hexadecimal") from exc return cls(access_area, lids, access_sub_area, symbol_crc, reference_id) + @classmethod + def from_tag(cls, tag: "SymbolicTag", *, reference_id: int = 0) -> "SubscriptionItem": + """Build an item that retains a catalog tag for typed notifications.""" + return cls(tag.access_area, tag.lids, symbol_crc=tag.symbol_crc, reference_id=reference_id, tag=tag) + @dataclass(frozen=True) class SubscriptionNotification: @@ -85,6 +96,164 @@ class SubscriptionNotification: errors: dict[int, int] timestamp_microseconds: int | None = None trailing_data: bytes = b"" + decoded_values: dict[int, Any] = field(default_factory=dict) + tags: dict[int, "SymbolicTag"] = field(default_factory=dict) + + +@dataclass(frozen=True) +class SubscriptionDiagnostics: + """Queue and sequence-loss diagnostics for one active subscription.""" + + subscription_id: int + queued_notifications: int + dropped_notifications: int + missed_sequence_updates: int + transport_frame_overflows: int = 0 + + +@dataclass +class _SubscriptionState: + items: dict[int, SubscriptionItem] + change_counter: int + credit_limit: int + credit_step: int + queue_size: int + notifications: deque[SubscriptionNotification] = field(default_factory=deque) + callbacks: list[Callable[[SubscriptionNotification], None]] = field(default_factory=list) + dropped_notifications: int = 0 + missed_sequence_updates: int = 0 + last_sequence: int | None = None + next_credit_limit: int = 0 + + def __post_init__(self) -> None: + self.next_credit_limit = self.credit_limit + + +class SubscriptionRegistry: + """Bounded notification routing shared by the sync and async clients.""" + + def __init__(self, orphan_queue_size: int = 100) -> None: + self._states: dict[int, _SubscriptionState] = {} + self._orphans: deque[SubscriptionNotification] = deque(maxlen=orphan_queue_size) + self.unmatched_notifications = 0 + + def register( + self, + subscription_id: int, + items: Sequence[SubscriptionItem], + *, + change_counter: int, + credit_limit: int, + credit_step: int, + queue_size: int, + ) -> None: + if queue_size <= 0: + raise ValueError("queue_size must be positive") + references = {item.reference_id or index: item for index, item in enumerate(items, 1)} + state = _SubscriptionState(references, change_counter, credit_limit, credit_step, queue_size) + self._states[subscription_id] = state + retained: deque[SubscriptionNotification] = deque(maxlen=self._orphans.maxlen) + while self._orphans: + notification = self._orphans.popleft() + if notification.subscription_id == subscription_id and notification.change_counter == change_counter: + self._enqueue(state, self._decorate(notification, state)) + else: + retained.append(notification) + self._orphans = retained + + def unregister(self, subscription_id: int) -> None: + self._states.pop(subscription_id, None) + self._orphans = deque( + (notification for notification in self._orphans if notification.subscription_id != subscription_id), + maxlen=self._orphans.maxlen, + ) + + def clear(self) -> None: + self._states.clear() + self._orphans.clear() + + @property + def subscription_ids(self) -> tuple[int, ...]: + return tuple(self._states) + + def contains(self, subscription_id: int) -> bool: + return subscription_id in self._states + + def add_callback(self, subscription_id: int, callback: Callable[[SubscriptionNotification], None]) -> None: + self._state(subscription_id).callbacks.append(callback) + + def remove_callback(self, subscription_id: int, callback: Callable[[SubscriptionNotification], None]) -> None: + state = self._state(subscription_id) + if callback in state.callbacks: + state.callbacks.remove(callback) + + def route(self, notification: SubscriptionNotification) -> tuple[bool, int | None]: + state = self._states.get(notification.subscription_id) + if state is None: + if len(self._orphans) == self._orphans.maxlen: + self.unmatched_notifications += 1 + self._orphans.append(notification) + return False, None + if notification.change_counter != state.change_counter: + state.dropped_notifications += 1 + return False, None + + decorated = self._decorate(notification, state) + if state.last_sequence is not None and decorated.sequence_number > state.last_sequence + 1: + state.missed_sequence_updates += decorated.sequence_number - state.last_sequence - 1 + state.last_sequence = decorated.sequence_number + self._enqueue(state, decorated) + for callback in tuple(state.callbacks): + callback(decorated) + + credit_update = None + if state.credit_limit > 0 and state.credit_step > 0 and decorated.credit_tick >= state.next_credit_limit - 1: + state.next_credit_limit = (state.next_credit_limit + state.credit_step) % 255 or state.credit_step + credit_update = state.next_credit_limit + return True, credit_update + + def pop(self, subscription_id: int) -> SubscriptionNotification | None: + state = self._state(subscription_id) + return state.notifications.popleft() if state.notifications else None + + def diagnostics(self, subscription_id: int) -> SubscriptionDiagnostics: + state = self._state(subscription_id) + return SubscriptionDiagnostics( + subscription_id, + len(state.notifications), + state.dropped_notifications, + state.missed_sequence_updates, + ) + + def _state(self, subscription_id: int) -> _SubscriptionState: + try: + return self._states[subscription_id] + except KeyError as exc: + raise KeyError(f"Unknown subscription: {subscription_id:#x}") from exc + + @staticmethod + def _decorate(notification: SubscriptionNotification, state: _SubscriptionState) -> SubscriptionNotification: + tags: dict[int, SymbolicTag] = {} + decoded: dict[int, Any] = {} + for reference_id, raw in notification.values.items(): + item = state.items.get(reference_id) + if item is None or item.tag is None: + decoded[reference_id] = raw + continue + tags[reference_id] = item.tag + decoded[reference_id] = item.tag.decode_value(raw) + for reference_id in notification.errors: + item = state.items.get(reference_id) + if item is not None and item.tag is not None: + tags[reference_id] = item.tag + return replace(notification, tags=tags, decoded_values=decoded) + + @staticmethod + def _enqueue(state: _SubscriptionState, notification: SubscriptionNotification) -> None: + if len(state.notifications) >= state.queue_size: + state.notifications.popleft() + state.dropped_notifications += 1 + state.notifications.append(notification) def _attribute(attribute_id: int, value: bytes) -> bytes: @@ -205,6 +374,18 @@ def _decode_notification_value(data: bytes, offset: int) -> tuple[bytes, int]: return decode_pvalue_to_bytes(data, offset) +def notification_subscription_id(frame: bytes) -> int: + """Return the subscription object ID from any notification frame.""" + version, data_length, consumed = decode_header(frame) + data = frame[consumed : consumed + data_length] + if version == ProtocolVersion.V3 and data: + hash_length = data[0] + data = data[1 + hash_length :] + if len(data) < 5 or data[0] != Opcode.NOTIFICATION: + raise ValueError("expected an S7CommPlus notification with a subscription ID") + return struct.unpack_from(">I", data, 1)[0] + + def parse_subscription_notification(frame: bytes) -> SubscriptionNotification: """Parse one complete unsolicited S7CommPlus notification frame.""" version, data_length, consumed = decode_header(frame) diff --git a/snap7/error.py b/snap7/error.py index 16108475..aca1bd1d 100644 --- a/snap7/error.py +++ b/snap7/error.py @@ -28,6 +28,12 @@ class S7ProtocolError(S7Error): pass +class S7IntegrityError(S7ProtocolError): + """Raised when authenticated S7CommPlus traffic fails integrity checks.""" + + pass + + class S7TimeoutError(S7Error): """Raised when S7 operation times out.""" diff --git a/tests/test_error.py b/tests/test_error.py index 7e32f9e4..8792a149 100644 --- a/tests/test_error.py +++ b/tests/test_error.py @@ -6,6 +6,7 @@ S7Error, S7ConnectionError, S7ProtocolError, + S7IntegrityError, S7TimeoutError, S7AuthenticationError, S7StalePacketError, @@ -33,6 +34,7 @@ def test_s7error_without_code(self) -> None: def test_subclass_hierarchy(self) -> None: assert issubclass(S7ConnectionError, S7Error) assert issubclass(S7ProtocolError, S7Error) + assert issubclass(S7IntegrityError, S7ProtocolError) assert issubclass(S7TimeoutError, S7Error) assert issubclass(S7AuthenticationError, S7Error) assert issubclass(S7StalePacketError, S7ProtocolError) @@ -42,6 +44,7 @@ def test_all_subclasses_instantiate(self) -> None: for cls in ( S7ConnectionError, S7ProtocolError, + S7IntegrityError, S7TimeoutError, S7AuthenticationError, S7StalePacketError, diff --git a/tests/test_s7_alarm.py b/tests/test_s7_alarm.py index d5b2f157..29e1a48a 100644 --- a/tests/test_s7_alarm.py +++ b/tests/test_s7_alarm.py @@ -16,6 +16,7 @@ from s7commplus.async_client import S7CommPlusAsyncClient from s7commplus.client import S7CommPlusClient from s7commplus.protocol import DataType, ElementID, FunctionCode, Ids, Opcode, ProtocolVersion +from s7commplus.subscription import SubscriptionItem from s7commplus.vlq import encode_uint32_vlq, encode_uint64_vlq @@ -84,6 +85,13 @@ def _notification_frame() -> bytes: return struct.pack(">BBH", 0x72, ProtocolVersion.V2, len(body)) + body + struct.pack(">BBH", 0x72, ProtocolVersion.V2, 0) +def _data_notification_frame() -> bytes: + body = bytearray([Opcode.NOTIFICATION]) + body += struct.pack(">IHHH", 0x70400025, 0, 0, 0) + body += b"\x03" + encode_uint32_vlq(9) + b"\x01\x00" + return struct.pack(">BBH", 0x72, ProtocolVersion.V2, len(body)) + body + struct.pack(">BBH", 0x72, ProtocolVersion.V2, 0) + + def test_alarm_models_are_public() -> None: assert Alarm.__module__ == "s7commplus.alarm" assert AlarmNotification.__module__ == "s7commplus.alarm" @@ -203,9 +211,51 @@ async def test_async_alarm_client_apis() -> None: assert await client.create_alarm_subscription([1031]) == 0x55667788 assert (await client.read_alarms([1031]))[0].texts[1031].alarm_text == "Alarm 4 =F6+S2-G1" assert (await client.receive_alarm_notification(timeout=1)).credit_tick == 5 + client._notification_frames.append(_notification_frame()) + assert (await client.receive_alarm_notification(timeout=1)).sequence_number == 12 + client._recv_cotp_dt.assert_awaited_once() await client.delete_alarm_subscription(0x55667788) +def test_sync_alarm_receive_routes_interleaved_data_notification() -> None: + client = S7CommPlusClient() + connection = MagicMock() + connection.receive_notification.side_effect = [_data_notification_frame(), _notification_frame()] + client._connection = connection + client._alarm_subscription_ids.add(0x11223344) + client._subscriptions.register( + 0x70400025, + [SubscriptionItem.from_access_sequence("8A0E0007.A")], + change_counter=1, + credit_limit=-1, + credit_step=0, + queue_size=2, + ) + + assert client.receive_alarm_notification().subscription_id == 0x11223344 + assert client.receive_subscription_notification(0x70400025).sequence_number == 9 + assert connection.receive_notification.call_count == 2 + + +@pytest.mark.asyncio +async def test_async_alarm_receive_routes_interleaved_data_notification() -> None: + client = S7CommPlusAsyncClient() + client._connected = True + client._recv_cotp_dt = AsyncMock(side_effect=[_data_notification_frame(), _notification_frame()]) + client._alarm_subscription_ids.add(0x11223344) + client._subscriptions.register( + 0x70400025, + [SubscriptionItem.from_access_sequence("8A0E0007.A")], + change_counter=1, + credit_limit=-1, + credit_step=0, + queue_size=2, + ) + + assert (await client.receive_alarm_notification()).subscription_id == 0x11223344 + assert client.subscription_queue(0x70400025).get_nowait().sequence_number == 9 + + @pytest.mark.parametrize( "method,args", [ diff --git a/tests/test_s7_legacy_request_layout.py b/tests/test_s7_legacy_request_layout.py index 2769d61e..a2367da0 100644 --- a/tests/test_s7_legacy_request_layout.py +++ b/tests/test_s7_legacy_request_layout.py @@ -33,7 +33,9 @@ def test_v1_substreamed_request_matches_accepted_tia_packet() -> None: conn._integrity_id_read = 1 conn._send_s7_data = MagicMock() body = struct.pack(">BHHHHB", 0x32, 0, FunctionCode.GET_VAR_SUBSTREAMED, 0, 3, 0x34) + bytes(4) - conn._recv_s7_data = MagicMock(return_value=encode_header(ProtocolVersion.V2, len(body)) + body) + response_digest = hmac.new(conn._session_key, body, hashlib.sha256).digest() + protected_body = bytes([len(response_digest)]) + response_digest + body + conn._recv_s7_data = MagicMock(return_value=encode_header(ProtocolVersion.V3, len(protected_body)) + protected_body) conn.send_request(FunctionCode.GET_VAR_SUBSTREAMED, payload) frame = conn._send_s7_data.call_args.args[0] assert frame[37:-4] == expected diff --git a/tests/test_s7_server.py b/tests/test_s7_server.py index e6bfd701..89e2ad39 100644 --- a/tests/test_s7_server.py +++ b/tests/test_s7_server.py @@ -9,7 +9,7 @@ import pytest -from snap7.error import S7ConnectionError +from snap7.error import S7ConnectionError, S7IntegrityError from s7commplus.async_client import S7CommPlusAsyncClient from s7commplus.client import S7CommPlusClient from s7commplus.connection import _parse_get_var_substreamed_response, _verify_v3_hmac @@ -489,7 +489,7 @@ def test_v3_hmac_verification(self) -> None: tampered = protected[:-1] + bytes([protected[-1] ^ 0x01]) with pytest.raises(ConnectionError, match="Invalid V3 HMAC"): S7CommPlusServer._verify_v3_data(tampered, TEST_SESSION_KEY) - with pytest.raises(S7ConnectionError, match="Invalid V3 HMAC"): + with pytest.raises(S7IntegrityError, match="integrity check failed"): _verify_v3_hmac(tampered, TEST_SESSION_KEY) def test_create_object_response_contains_fingerprint_and_challenge(self) -> None: diff --git a/tests/test_s7_subscription.py b/tests/test_s7_subscription.py index 5ff62f16..0b436bdc 100644 --- a/tests/test_s7_subscription.py +++ b/tests/test_s7_subscription.py @@ -1,21 +1,29 @@ """Tests for S7CommPlus symbolic data subscriptions.""" +import asyncio +import hashlib +import hmac import struct -from unittest.mock import MagicMock +from unittest.mock import AsyncMock, MagicMock import pytest +from s7commplus.async_client import S7CommPlusAsyncClient +from s7commplus.catalog import SymbolicTag from s7commplus.client import S7CommPlusClient -from s7commplus.codec import encode_header, encode_pvalue_blob +from s7commplus.codec import decode_header, encode_header, encode_pvalue_blob from s7commplus.connection import S7CommPlusConnection from s7commplus.protocol import DataType, FunctionCode, Ids, Opcode, ProtocolVersion from s7commplus.subscription import ( SubscriptionItem, + SubscriptionRegistry, build_delete_subscription_request, build_subscription_request, parse_subscription_notification, ) +from s7commplus.typeinfo import Softdatatype from s7commplus.vlq import encode_uint32_vlq, encode_uint64_vlq +from snap7.error import S7IntegrityError def _response_frame(function_code: int, sequence: int, payload: bytes) -> bytes: @@ -23,10 +31,18 @@ def _response_frame(function_code: int, sequence: int, payload: bytes) -> bytes: 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: +def _notification_frame( + *, + version: int = ProtocolVersion.V2, + with_hmac: bool = False, + subscription_id: int = 0x70400025, + credit_tick: int = 3, + sequence_number: int = 9, + change_counter: int = 1, +) -> bytes: data = bytearray([Opcode.NOTIFICATION]) - data += struct.pack(">IHHH", 0x70400025, 4, 0, 0) - data += b"\x03" + encode_uint32_vlq(9) + b"\x01" + data += struct.pack(">IHHH", subscription_id, 4, 0, 0) + data += bytes([credit_tick]) + encode_uint32_vlq(sequence_number) + bytes([change_counter]) 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) @@ -52,6 +68,15 @@ def test_rejects_invalid_access_sequence(self, value: str) -> None: with pytest.raises(ValueError): SubscriptionItem.from_access_sequence(value) + def test_from_catalog_tag_retains_type_metadata(self) -> None: + tag = SymbolicTag("DB1.Count", 0x8A0E0001, (2,), Softdatatype.INT, DataType.INT, symbol_crc=7) + + item = SubscriptionItem.from_tag(tag, reference_id=4) + + assert item.tag is tag + assert item.reference_id == 4 + assert item.symbol_crc == 7 + class TestSubscriptionRequest: def test_matches_real_plc_reference_trace(self) -> None: @@ -177,6 +202,175 @@ def test_create_receive_and_delete(self) -> None: assert delete_call.args[0] == FunctionCode.DELETE_OBJECT assert delete_call.args[1].startswith(struct.pack(">I", connection.subscription_container_id)) + def test_catalog_tag_notification_is_decoded_and_raw_value_is_retained(self) -> None: + connection = MagicMock(subscription_container_id=0x3C2, protocol_version=ProtocolVersion.V2) + connection.send_request.return_value = encode_uint64_vlq(0) + b"\x01" + encode_uint32_vlq(0x70400025) + connection.receive_notification.return_value = _notification_frame() + tag = SymbolicTag("DB1.Count", 0x8A0E0001, (2,), Softdatatype.INT, DataType.INT) + client = S7CommPlusClient() + client._connection = connection + subscription_id = client.create_subscription([SubscriptionItem.from_tag(tag, reference_id=7)]) + + notification = client.receive_subscription_notification(subscription_id) + + assert notification.values[7] == b"\x12\x34" + assert notification.decoded_values[7] == 0x1234 + assert notification.tags[7] is tag + + def test_finite_credit_is_replenished_before_expiry(self) -> None: + connection = MagicMock(subscription_container_id=0x3C2, protocol_version=ProtocolVersion.V2) + connection.send_request.return_value = encode_uint64_vlq(0) + b"\x01" + encode_uint32_vlq(0x70400025) + connection.receive_notification.return_value = _notification_frame(credit_tick=9) + client = S7CommPlusClient() + client._connection = connection + subscription_id = client.create_subscription(["8A0E0007.A"], credit_limit=10, credit_step=5) + + client.receive_subscription_notification(subscription_id) + + connection.send_subscription_credit.assert_called_once_with(subscription_id, 15) + + def test_callback_and_bounded_iterator(self) -> None: + connection = MagicMock(subscription_container_id=0x3C2, protocol_version=ProtocolVersion.V2) + connection.send_request.return_value = encode_uint64_vlq(0) + b"\x01" + encode_uint32_vlq(0x70400025) + connection.receive_notification.side_effect = [ + _notification_frame(sequence_number=1), + _notification_frame(sequence_number=2), + ] + client = S7CommPlusClient() + client._connection = connection + subscription_id = client.create_subscription(["8A0E0007.A"]) + delivered = [] + client.add_subscription_callback(subscription_id, delivered.append) + + notifications = list(client.iter_subscription_notifications(subscription_id, limit=2)) + + assert [item.sequence_number for item in notifications] == [1, 2] + assert [item.sequence_number for item in delivered] == [1, 2] + + def test_notification_for_another_subscription_is_queued(self) -> None: + connection = MagicMock(subscription_container_id=0x3C2, protocol_version=ProtocolVersion.V2) + connection.send_request.side_effect = [ + encode_uint64_vlq(0) + b"\x01" + encode_uint32_vlq(0x70400025), + encode_uint64_vlq(0) + b"\x01" + encode_uint32_vlq(0x70400026), + ] + connection.receive_notification.side_effect = [ + _notification_frame(subscription_id=0x70400026, change_counter=2), + _notification_frame(subscription_id=0x70400025, change_counter=1), + ] + client = S7CommPlusClient() + client._connection = connection + first = client.create_subscription(["8A0E0007.A"]) + second = client.create_subscription(["8A0E0007.B"]) + + assert client.receive_subscription_notification(first).subscription_id == first + assert client.receive_subscription_notification(second).subscription_id == second + assert connection.receive_notification.call_count == 2 + + +class TestSubscriptionRegistry: + def test_buffers_pre_registration_notification(self) -> None: + registry = SubscriptionRegistry() + notification = parse_subscription_notification(_notification_frame()) + + assert registry.route(notification) == (False, None) + registry.register( + notification.subscription_id, + [SubscriptionItem.from_access_sequence("8A0E0007.A", reference_id=7)], + change_counter=1, + credit_limit=-1, + credit_step=0, + queue_size=2, + ) + + buffered = registry.pop(notification.subscription_id) + assert buffered is not None + assert buffered.sequence_number == notification.sequence_number + assert buffered.values == notification.values + + def test_queue_overflow_sequence_gap_and_stale_generation_are_diagnostic(self) -> None: + registry = SubscriptionRegistry() + subscription_id = 0x70400025 + registry.register( + subscription_id, + [SubscriptionItem.from_access_sequence("8A0E0007.A")], + change_counter=1, + credit_limit=-1, + credit_step=0, + queue_size=1, + ) + + registry.route(parse_subscription_notification(_notification_frame(sequence_number=3))) + registry.route(parse_subscription_notification(_notification_frame(sequence_number=5))) + matched, _ = registry.route(parse_subscription_notification(_notification_frame(change_counter=2))) + + diagnostics = registry.diagnostics(subscription_id) + assert not matched + assert diagnostics.queued_notifications == 1 + assert diagnostics.dropped_notifications == 2 + assert diagnostics.missed_sequence_updates == 1 + + def test_deleted_id_reuse_rejects_old_change_counter(self) -> None: + registry = SubscriptionRegistry() + subscription_id = 0x70400025 + item = SubscriptionItem.from_access_sequence("8A0E0007.A") + registry.register(subscription_id, [item], change_counter=1, credit_limit=-1, credit_step=0, queue_size=2) + registry.unregister(subscription_id) + registry.register(subscription_id, [item], change_counter=2, credit_limit=-1, credit_step=0, queue_size=2) + + assert registry.route(parse_subscription_notification(_notification_frame(change_counter=1))) == (False, None) + assert registry.pop(subscription_id) is None + + +@pytest.mark.asyncio +async def test_async_create_iterate_queue_and_delete_lifecycle() -> None: + client = S7CommPlusAsyncClient() + client._connected = True + client._reader = MagicMock() + client._writer = MagicMock() + client._subscription_container_id = 0x3C2 + client._protocol_version = ProtocolVersion.V2 + create_response = encode_uint64_vlq(0) + b"\x01" + encode_uint32_vlq(0x70400025) + client._send_request = AsyncMock(side_effect=[create_response, b""]) + client._recv_cotp_dt = AsyncMock(return_value=_notification_frame(credit_tick=9)) + client._send_subscription_credit = AsyncMock() + tag = SymbolicTag("DB1.Count", 0x8A0E0001, (2,), Softdatatype.INT, DataType.INT) + + subscription_id = await client.create_subscription( + [SubscriptionItem.from_tag(tag, reference_id=7)], credit_limit=10, credit_step=5, queue_size=2 + ) + queue = client.subscription_queue(subscription_id) + assert queue.qsize() == 0 + assert (await queue.get()).decoded_values[7] == 0x1234 + client._send_subscription_credit.assert_awaited_once_with(subscription_id, 15) + with pytest.raises(asyncio.QueueEmpty): + queue.get_nowait() + + client._notification_frames.append(_notification_frame(sequence_number=10)) + received = [item async for item in client.iter_subscription_notifications(subscription_id, limit=1)] + assert received[0].sequence_number == 10 + await client.delete_subscription(subscription_id) + with pytest.raises(KeyError, match="Unknown subscription"): + client.subscription_diagnostics(subscription_id) + + +def test_credit_update_uses_fire_and_forget_transport_flags() -> None: + connection = S7CommPlusConnection("127.0.0.1") + connection._connected = True + connection._protocol_version = ProtocolVersion.V2 + connection._session_id = 0x70000CB8 + connection._with_integrity_id = True + connection._integrity_id_write = 3 + connection._send_s7_data = MagicMock() + + connection.send_subscription_credit(0x70400025, 15) + + frame = connection._send_s7_data.call_args.args[0] + _, length, consumed = decode_header(frame) + request = frame[consumed : consumed + length] + assert request[13] == 0x74 + assert request[14:18] == struct.pack(">I", 0x70400025) + assert connection.integrity_id_write == 4 + class TestNotificationQueue: def test_send_request_queues_interleaved_notification(self) -> None: @@ -192,3 +386,26 @@ def test_send_request_queues_interleaved_notification(self) -> None: 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 + + def test_authenticated_notification_is_verified_before_queueing(self) -> None: + connection = S7CommPlusConnection("127.0.0.1") + connection._connected = True + connection._protocol_version = ProtocolVersion.V3 + connection._session_id = 1 + connection._session_key = bytes(range(24)) + connection._iso_conn.disconnect = MagicMock() + + notification = _notification_frame(version=ProtocolVersion.V3) + _, data_length, consumed = decode_header(notification) + data = notification[consumed : consumed + data_length] + digest = hmac.new(connection._session_key, data, hashlib.sha256).digest() + protected = bytearray(bytes([len(digest)]) + digest + data) + protected[1] ^= 1 + notification = encode_header(ProtocolVersion.V3, len(protected)) + protected + connection._send_s7_data = MagicMock() + connection._recv_s7_data = MagicMock(return_value=bytes(notification)) + + with pytest.raises(S7IntegrityError, match="integrity check failed"): + connection.send_request(FunctionCode.GET_VARIABLE, bytes(4)) + assert not connection.connected + assert not connection._notification_frames diff --git a/tests/test_s7_unit.py b/tests/test_s7_unit.py index e4673075..83a3ff71 100644 --- a/tests/test_s7_unit.py +++ b/tests/test_s7_unit.py @@ -3,7 +3,7 @@ import hashlib import hmac import struct -from unittest.mock import MagicMock, call +from unittest.mock import MagicMock, call, patch import pytest @@ -22,7 +22,7 @@ _build_symbolic_write_payload, _build_substreamed_write_payload, ) -from s7commplus.connection import S7CommPlusConnection, _strip_paom_string_in_session_version +from s7commplus.connection import S7CommPlusConnection, _strip_paom_string_in_session_version, _verify_v3_hmac from s7commplus.codec import encode_header, encode_object_qualifier, encode_pvalue_blob from s7commplus.codec import _pvalue_element_size as _element_size from s7commplus.codec import skip_typed_value, parse_server_session_version @@ -745,18 +745,53 @@ def test_multiple_fragments_split_across_reads(self) -> None: conn = self._conn_yielding([self._frag(b"abc"), self._frag(b"de"), self._TRAILER]) assert conn._recv_reassembled_payload() == b"abcde" - def test_v3_session_key_hmac_is_stripped_from_each_fragment(self) -> None: + def test_v3_session_key_hmac_uses_cumulative_fragment_digest(self) -> None: conn = self._conn_yielding([]) conn._session_key = bytes(24) + digest_state = hmac.new(conn._session_key, digestmod=hashlib.sha256) def v3_frag(data: bytes) -> bytes: - digest = hmac.new(conn._session_key, data, hashlib.sha256).digest() + digest_state.update(data) + digest = digest_state.digest() protected = bytes([len(digest)]) + digest + data return bytes([0x72, ProtocolVersion.V3, 0, len(protected)]) + protected initial = v3_frag(b"abc") + v3_frag(b"de") + bytes([0x72, ProtocolVersion.V3, 0, 0]) assert conn._recv_reassembled_payload(initial) == b"abcde" + def test_v3_cumulative_fragment_rejects_independent_second_digest(self) -> None: + from snap7.error import S7IntegrityError + + conn = self._conn_yielding([]) + conn._connected = True + conn._session_key = bytes(24) + + def independent_fragment(data: bytes) -> bytes: + digest = hmac.new(conn._session_key, data, hashlib.sha256).digest() + protected = bytes([len(digest)]) + digest + data + return bytes([0x72, ProtocolVersion.V3, 0, len(protected)]) + protected + + initial = independent_fragment(b"abc") + independent_fragment(b"de") + with pytest.raises(S7IntegrityError, match="integrity check failed"): + conn._recv_reassembled_payload(initial) + assert not conn.connected + + def test_authenticated_reassembly_rejects_fragment_version_downgrade(self) -> None: + from snap7.error import S7IntegrityError + + conn = self._conn_yielding([]) + conn._connected = True + conn._session_key = bytes(24) + first_data = b"abc" + digest = hmac.new(conn._session_key, first_data, hashlib.sha256).digest() + first_protected = b"\x20" + digest + first_data + first = encode_header(ProtocolVersion.V3, len(first_protected)) + first_protected + downgraded = bytes([0x72, ProtocolVersion.V2, 0, 2]) + b"de" + + with pytest.raises(S7IntegrityError, match="changed fragment version"): + conn._recv_reassembled_payload(first + downgraded) + assert not conn.connected + def test_bad_fragment_header_raises(self) -> None: from snap7.error import S7ConnectionError @@ -779,3 +814,81 @@ def test_fragment_count_cap(self) -> None: conn._MAX_REASSEMBLED_FRAGMENTS = 2 with pytest.raises(S7ConnectionError, match="exceeds limits"): conn._recv_reassembled_payload() + + +class TestV3ResponseIntegrity: + KEY = bytes(range(24)) + + @classmethod + def _protected(cls, data: bytes, key: bytes | None = None) -> bytes: + digest = hmac.new(key or cls.KEY, data, hashlib.sha256).digest() + return bytes([len(digest)]) + digest + data + + def test_valid_digest_uses_constant_time_comparison(self) -> None: + protected = self._protected(b"authenticated response") + with patch("s7commplus.connection.hmac.compare_digest", wraps=hmac.compare_digest) as compare: + assert _verify_v3_hmac(protected, self.KEY) == b"authenticated response" + compare.assert_called_once() + + @pytest.mark.parametrize("mutation", ["wrong-key", "payload", "digest"]) + def test_changed_digest_covered_data_is_rejected(self, mutation: str) -> None: + from snap7.error import S7IntegrityError + + signing_key = bytes(reversed(self.KEY)) if mutation == "wrong-key" else None + protected = self._protected(b"authenticated response", signing_key) + if mutation == "payload": + protected = protected[:-1] + bytes([protected[-1] ^ 1]) + elif mutation == "digest": + protected = protected[:1] + bytes([protected[1] ^ 1]) + protected[2:] + with pytest.raises(S7IntegrityError, match="integrity check failed"): + _verify_v3_hmac(protected, self.KEY) + + @pytest.mark.parametrize( + "protected, message", + [(b"", "Empty authenticated"), (b"\x1f" + bytes(31), "digest length"), (b"\x20" + bytes(12), "Truncated")], + ) + def test_invalid_or_truncated_digest_is_rejected(self, protected: bytes, message: str) -> None: + from snap7.error import S7IntegrityError + + with pytest.raises(S7IntegrityError, match=message): + _verify_v3_hmac(protected, self.KEY) + + def test_failure_invalidates_connection(self) -> None: + from snap7.error import S7IntegrityError + + conn = S7CommPlusConnection("127.0.0.1") + conn._connected = True + conn._session_ready = True + conn._session_id = 123 + conn._session_key = self.KEY + conn._iso_conn.disconnect = MagicMock() + response = struct.pack(">BHHHHB", Opcode.RESPONSE, 0, FunctionCode.GET_MULTI_VARIABLES, 0, 0, 0x34) + protected = bytearray(self._protected(response)) + protected[1] ^= 1 + frame = encode_header(ProtocolVersion.V3, len(protected)) + protected + conn._recv_s7_data = MagicMock(return_value=bytes(frame)) + conn._send_s7_data = MagicMock() + + with pytest.raises(S7IntegrityError, match="integrity check failed"): + conn.send_request(FunctionCode.GET_MULTI_VARIABLES) + assert not conn.connected + assert conn._session_key is None + conn._iso_conn.disconnect.assert_called_once_with() + + def test_authenticated_response_rejects_frame_version_downgrade(self) -> None: + from snap7.error import S7IntegrityError + + conn = S7CommPlusConnection("127.0.0.1") + conn._connected = True + conn._session_ready = True + conn._session_id = 123 + conn._session_key = self.KEY + conn._iso_conn.disconnect = MagicMock() + response = struct.pack(">BHHHHB", Opcode.RESPONSE, 0, FunctionCode.GET_MULTI_VARIABLES, 0, 0, 0x34) + frame = encode_header(ProtocolVersion.V2, len(response)) + response + conn._recv_s7_data = MagicMock(return_value=frame) + conn._send_s7_data = MagicMock() + + with pytest.raises(S7IntegrityError, match="unauthenticated frame version"): + conn.send_request(FunctionCode.GET_MULTI_VARIABLES) + assert not conn.connected diff --git a/tests/test_s7_v2.py b/tests/test_s7_v2.py index 9b70719e..b1d4ed18 100644 --- a/tests/test_s7_v2.py +++ b/tests/test_s7_v2.py @@ -9,6 +9,7 @@ import hmac import logging import struct +import threading from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -522,6 +523,193 @@ async def test_async_sequence_mismatch_raises_protocol_error(self, reassemble: b with pytest.raises(S7ProtocolError, match="Response sequence mismatch"): await client._send_request(FunctionCode.GET_MULTI_VARIABLES, b"", reassemble=reassemble) + @pytest.mark.parametrize("reassemble", [False, True]) + def test_sync_function_mismatch_raises_protocol_error(self, reassemble: bool) -> None: + conn = S7CommPlusConnection("127.0.0.1") + conn._connected = True + conn._protocol_version = ProtocolVersion.V2 + response = struct.pack(">BHHHHB", Opcode.RESPONSE, 0, FunctionCode.SET_MULTI_VARIABLES, 0, 0, 0x34) + response_frame = encode_header(ProtocolVersion.V2, len(response)) + response + response_frame += struct.pack(">BBH", 0x72, ProtocolVersion.V2, 0) + conn._send_s7_data = MagicMock() + conn._recv_s7_data = MagicMock(return_value=response_frame) + + with pytest.raises(S7ProtocolError, match="Response function mismatch"): + conn.send_request(FunctionCode.GET_MULTI_VARIABLES, b"", reassemble=reassemble) + + @pytest.mark.asyncio + @pytest.mark.parametrize("reassemble", [False, True]) + async def test_async_function_mismatch_raises_protocol_error(self, reassemble: bool) -> None: + client = S7CommPlusAsyncClient() + client._connected = True + client._reader = MagicMock() + client._writer = MagicMock() + client._protocol_version = ProtocolVersion.V2 + response = struct.pack(">BHHHHB", Opcode.RESPONSE, 0, FunctionCode.SET_MULTI_VARIABLES, 0, 0, 0x34) + response_frame = encode_header(ProtocolVersion.V2, len(response)) + response + response_frame += struct.pack(">BBH", 0x72, ProtocolVersion.V2, 0) + client._send_cotp_dt = AsyncMock() + client._recv_cotp_dt = AsyncMock(return_value=response_frame) + + with pytest.raises(S7ProtocolError, match="Response function mismatch"): + await client._send_request(FunctionCode.GET_MULTI_VARIABLES, b"", reassemble=reassemble) + + @pytest.mark.parametrize("opcode", [Opcode.REQUEST, 0x7F]) + def test_sync_unexpected_opcode_raises_protocol_error(self, opcode: int) -> None: + conn = S7CommPlusConnection("127.0.0.1") + conn._connected = True + conn._protocol_version = ProtocolVersion.V2 + response = struct.pack(">BHHHHB", opcode, 0, FunctionCode.GET_MULTI_VARIABLES, 0, 0, 0x34) + response_frame = encode_header(ProtocolVersion.V2, len(response)) + response + conn._send_s7_data = MagicMock() + conn._recv_s7_data = MagicMock(return_value=response_frame) + + with pytest.raises(S7ProtocolError, match="Unexpected S7CommPlus opcode"): + conn.send_request(FunctionCode.GET_MULTI_VARIABLES) + + @pytest.mark.asyncio + async def test_async_unexpected_opcode_raises_protocol_error(self) -> None: + client = S7CommPlusAsyncClient() + client._connected = True + client._reader = MagicMock() + client._writer = MagicMock() + client._protocol_version = ProtocolVersion.V2 + response = struct.pack(">BHHHHB", Opcode.REQUEST, 0, FunctionCode.GET_MULTI_VARIABLES, 0, 0, 0x34) + response_frame = encode_header(ProtocolVersion.V2, len(response)) + response + client._send_cotp_dt = AsyncMock() + client._recv_cotp_dt = AsyncMock(return_value=response_frame) + + with pytest.raises(S7ProtocolError, match="Unexpected S7CommPlus opcode"): + await client._send_request(FunctionCode.GET_MULTI_VARIABLES, b"") + + def test_sync_notification_before_response_is_queued(self) -> None: + conn = S7CommPlusConnection("127.0.0.1") + conn._connected = True + conn._protocol_version = ProtocolVersion.V2 + notification = struct.pack(">BHHHHB", Opcode.NOTIFICATION, 0, 0, 0, 12, 0x34) + notification_frame = encode_header(ProtocolVersion.V2, len(notification)) + notification + response = struct.pack(">BHHHHB", Opcode.RESPONSE, 0, FunctionCode.GET_MULTI_VARIABLES, 0, 0, 0x34) + response_frame = encode_header(ProtocolVersion.V2, len(response)) + response + response_frame += struct.pack(">BBH", 0x72, ProtocolVersion.V2, 0) + conn._send_s7_data = MagicMock() + conn._recv_s7_data = MagicMock(side_effect=[notification_frame, response_frame]) + + assert conn.send_request(FunctionCode.GET_MULTI_VARIABLES) == b"" + assert conn.receive_notification() == notification_frame + assert conn._recv_s7_data.call_count == 2 + + @pytest.mark.asyncio + async def test_async_notification_before_response_is_queued(self) -> None: + client = S7CommPlusAsyncClient() + client._connected = True + client._reader = MagicMock() + client._writer = MagicMock() + client._protocol_version = ProtocolVersion.V2 + notification = struct.pack(">BHHHHB", Opcode.NOTIFICATION, 0, 0, 0, 12, 0x34) + notification_frame = encode_header(ProtocolVersion.V2, len(notification)) + notification + response = struct.pack(">BHHHHB", Opcode.RESPONSE, 0, FunctionCode.GET_MULTI_VARIABLES, 0, 0, 0x34) + response_frame = encode_header(ProtocolVersion.V2, len(response)) + response + response_frame += struct.pack(">BBH", 0x72, ProtocolVersion.V2, 0) + client._send_cotp_dt = AsyncMock() + client._recv_cotp_dt = AsyncMock(side_effect=[notification_frame, response_frame]) + + assert await client._send_request(FunctionCode.GET_MULTI_VARIABLES, b"") == b"" + assert list(client._notification_frames) == [notification_frame] + + def test_duplicate_sync_response_cannot_satisfy_next_request(self) -> None: + conn = S7CommPlusConnection("127.0.0.1") + conn._connected = True + conn._protocol_version = ProtocolVersion.V2 + response = struct.pack(">BHHHHB", Opcode.RESPONSE, 0, FunctionCode.GET_MULTI_VARIABLES, 0, 0, 0x34) + response_frame = encode_header(ProtocolVersion.V2, len(response)) + response + response_frame += struct.pack(">BBH", 0x72, ProtocolVersion.V2, 0) + conn._send_s7_data = MagicMock() + conn._recv_s7_data = MagicMock(return_value=response_frame) + + assert conn.send_request(FunctionCode.GET_MULTI_VARIABLES) == b"" + with pytest.raises(S7ProtocolError, match="Response sequence mismatch"): + conn.send_request(FunctionCode.GET_MULTI_VARIABLES) + + @pytest.mark.asyncio + async def test_duplicate_async_response_cannot_satisfy_next_request(self) -> None: + client = S7CommPlusAsyncClient() + client._connected = True + client._reader = MagicMock() + client._writer = MagicMock() + client._protocol_version = ProtocolVersion.V2 + response = struct.pack(">BHHHHB", Opcode.RESPONSE, 0, FunctionCode.GET_MULTI_VARIABLES, 0, 0, 0x34) + response_frame = encode_header(ProtocolVersion.V2, len(response)) + response + client._send_cotp_dt = AsyncMock() + client._recv_cotp_dt = AsyncMock(return_value=response_frame) + + assert await client._send_request(FunctionCode.GET_MULTI_VARIABLES, b"") == b"" + with pytest.raises(S7ProtocolError, match="Response sequence mismatch"): + await client._send_request(FunctionCode.GET_MULTI_VARIABLES, b"") + + def test_sync_requests_are_serialized(self) -> None: + conn = S7CommPlusConnection("127.0.0.1") + first_entered = threading.Event() + release_first = threading.Event() + second_started = threading.Event() + second_entered = threading.Event() + call_count = 0 + count_lock = threading.Lock() + + def exchange(*_args: object) -> bytes: + nonlocal call_count + with count_lock: + call_count += 1 + current = call_count + if current == 1: + first_entered.set() + assert release_first.wait(1) + else: + second_entered.set() + return b"" + + conn._send_request = MagicMock(side_effect=exchange) + first = threading.Thread(target=conn.send_request, args=(FunctionCode.GET_MULTI_VARIABLES,)) + + def run_second() -> None: + second_started.set() + conn.send_request(FunctionCode.GET_MULTI_VARIABLES) + + second = threading.Thread(target=run_second) + first.start() + assert first_entered.wait(1) + second.start() + assert second_started.wait(1) + assert not second_entered.wait(0.05) + release_first.set() + first.join(1) + second.join(1) + assert not first.is_alive() + assert not second.is_alive() + assert second_entered.is_set() + + @pytest.mark.parametrize("client_kind", ["sync", "async"]) + @pytest.mark.asyncio + async def test_connection_close_while_waiting_is_connection_error(self, client_kind: str) -> None: + if client_kind == "sync": + conn = S7CommPlusConnection("127.0.0.1") + conn._connected = True + conn._protocol_version = ProtocolVersion.V2 + conn._send_s7_data = MagicMock() + conn._recv_s7_data = MagicMock(return_value=b"") + with pytest.raises(S7ConnectionError, match="Connection closed"): + conn.send_request(FunctionCode.GET_MULTI_VARIABLES) + return + + client = S7CommPlusAsyncClient() + client._connected = True + client._reader = MagicMock() + client._writer = MagicMock() + client._protocol_version = ProtocolVersion.V2 + client._send_cotp_dt = AsyncMock() + client._recv_cotp_dt = AsyncMock(return_value=b"") + with pytest.raises(S7ConnectionError, match="Connection closed"): + await client._send_request(FunctionCode.GET_MULTI_VARIABLES, b"") + class TestAsyncReassembledPayloadErrors: @pytest.mark.asyncio