From d426eda5f8f74e7a9592121b416d9db789847900 Mon Sep 17 00:00:00 2001 From: Gijs Molenaar Date: Wed, 16 Sep 2026 08:24:49 +0200 Subject: [PATCH 1/2] fix(server): return read item errors for invalid ranges Closes #894. Co-authored-by: Codex --- CHANGES.md | 2 + snap7/server/__init__.py | 83 ++++++++++++++++++---------------------- tests/test_server.py | 19 ++++++--- 3 files changed, 54 insertions(+), 50 deletions(-) diff --git a/CHANGES.md b/CHANGES.md index 5c2b789f..786aaf24 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -6,6 +6,8 @@ CHANGES Major release: new `s7commplus` package with S7CommPlus protocol support. +* Return S7 item errors when pure-Python server reads target unregistered areas + or ranges outside registered memory instead of fabricating successful data. * Correlate S7CommPlus responses by opcode, function, and sequence; discard bounded stale replies from earlier requests, preserve interleaved notifications, and serialize synchronous wire requests. diff --git a/snap7/server/__init__.py b/snap7/server/__init__.py index 017966c0..52babdc6 100644 --- a/snap7/server/__init__.py +++ b/snap7/server/__init__.py @@ -875,9 +875,7 @@ def _handle_read_area(self, request: Dict[str, Any], client_address: Tuple[str, area, db_number, start, count = addr_info - read_data = self._read_from_memory_area(area, db_number, start, count) - if read_data is None: - return self._build_error_response(request, 0x8404) + return_code, read_data = self._read_from_memory_area(area, db_number, start, count) data_len = 4 + len(read_data) @@ -895,17 +893,19 @@ def _handle_read_area(self, request: Dict[str, Any], client_address: Tuple[str, parameters = struct.pack(">BB", S7Function.READ_AREA, 0x01) - data_section = struct.pack(">BBH", 0xFF, 0x04, len(read_data) * 8) + read_data - - self._emit_event( - EVC_DATA_READ, - param1=self._event_area(area), - param2=db_number, - param3=start, - param4=len(read_data), - sender=self._event_sender(client_address), - notify_read_callback=True, - ) + transport_size = 0x04 if return_code == 0xFF else 0x00 + data_section = struct.pack(">BBH", return_code, transport_size, len(read_data) * 8) + read_data + + if return_code == 0xFF: + self._emit_event( + EVC_DATA_READ, + param1=self._event_area(area), + param2=db_number, + param3=start, + param4=len(read_data), + sender=self._event_sender(client_address), + notify_read_callback=True, + ) return header + parameters + data_section @@ -942,26 +942,24 @@ def _handle_multi_read_area(self, request: Dict[str, Any], client_address: Tuple else: byte_count = count - read_data = self._read_from_memory_area(area, db_number, start, byte_count) - if read_data is None: - # Item error: not found - data_parts.extend(struct.pack(">BBH", 0x0A, 0x00, 0x0000)) - else: + return_code, read_data = self._read_from_memory_area(area, db_number, start, byte_count) + if return_code == 0xFF: data_parts.extend(struct.pack(">BBH", 0xFF, 0x04, len(read_data) * 8)) data_parts.extend(read_data) # Fill byte for even alignment (not after last item) if i < item_count - 1 and len(read_data) % 2 != 0: data_parts.append(0x00) - - self._emit_event( - EVC_DATA_READ, - param1=self._event_area(area), - param2=db_number, - param3=start, - param4=byte_count, - sender=self._event_sender(client_address), - notify_read_callback=True, - ) + self._emit_event( + EVC_DATA_READ, + param1=self._event_area(area), + param2=db_number, + param3=start, + param4=byte_count, + sender=self._event_sender(client_address), + notify_read_callback=True, + ) + else: + data_parts.extend(struct.pack(">BBH", return_code, 0x00, 0x0000)) data_len = len(data_parts) @@ -1025,7 +1023,7 @@ def _parse_read_address(self, request: Dict[str, Any]) -> Optional[Tuple[S7Area, logger.error(f"Error parsing read address: {e}") return None - def _read_from_memory_area(self, area: S7Area, db_number: int, start: int, count: int) -> Optional[bytearray]: + def _read_from_memory_area(self, area: S7Area, db_number: int, start: int, count: int) -> Tuple[int, bytearray]: """ Read data from registered memory area. @@ -1036,39 +1034,34 @@ def _read_from_memory_area(self, area: S7Area, db_number: int, start: int, count count: Number of bytes to read Returns: - Data read from memory area or None if area not found + Item return code and data. The return code is ``0xFF`` on success, + ``0x0A`` when the area is not registered, and ``0x05`` when the + requested range is outside the registered area. """ try: area_key = (area, db_number) if area_key not in self.memory_areas: logger.warning(f"Memory area {area}#{db_number} not registered") - # Return dummy data if area not found (for compatibility) - return bytearray([0x42, 0xFF, 0x12, 0x34])[:count] + return (0x0A, bytearray()) # Get area data with thread safety with self.area_locks[area_key]: area_data = self.memory_areas[area_key] # Check bounds - if start >= len(area_data): - logger.warning(f"Start address {start} beyond area size {len(area_data)}") - return bytearray([0x00] * count) - - # Read requested data, padding with zeros if needed - end = min(start + count, len(area_data)) - read_data = bytearray(area_data[start:end]) + if start < 0 or count < 0 or start + count > len(area_data): + logger.warning(f"Read range [{start}, {start + count}) exceeds area size {len(area_data)}") + return (0x05, bytearray()) - # Pad with zeros if we didn't read enough - if len(read_data) < count: - read_data.extend([0x00] * (count - len(read_data))) + read_data = bytearray(area_data[start : start + count]) logger.debug(f"Read {len(read_data)} bytes from {area}#{db_number} at offset {start}") - return read_data + return (0xFF, read_data) except Exception as e: logger.error(f"Error reading from memory area: {e}") - return bytearray([0x00] * count) + return (0x01, bytearray()) def _handle_write_area(self, request: Dict[str, Any], client_address: Tuple[str, int]) -> bytes: """Handle write area request.""" diff --git a/tests/test_server.py b/tests/test_server.py index 9fec91f7..ad5fed9b 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -11,7 +11,7 @@ from snap7.client import Client from snap7.datatypes import S7Area, S7WordLen -from snap7.error import S7ConnectionError, error_text, server_errors +from snap7.error import S7ConnectionError, S7ProtocolError, error_text, server_errors from snap7.server import EVC_DATA_READ, EVC_DATA_WRITE, EVC_SERVER_STARTED, EVC_SERVER_STOPPED, Server, ServerISOConnection from snap7.type import Block, Parameter, SrvArea, SrvEvent, mkEvent, mkLog @@ -991,10 +991,19 @@ def tearDown(self) -> None: self.client.destroy() def test_read_unregistered_db(self) -> None: - """Reading from an unregistered DB should still return data (server returns dummy data).""" - # The server returns dummy data for unregistered areas rather than an error - data = self.client.db_read(99, 0, 4) - self.assertEqual(len(data), 4) + """Reading from an unregistered DB returns item-not-available.""" + with self.assertRaisesRegex(S7ProtocolError, "0x0a"): + self.client.db_read(99, 0, 4) + + def test_read_start_beyond_area_bounds(self) -> None: + """Reading beyond a registered DB returns address-out-of-range.""" + with self.assertRaisesRegex(S7ProtocolError, "0x05"): + self.client.db_read(1, 100, 4) + + def test_read_crossing_area_bounds(self) -> None: + """A read crossing the end of a DB returns address-out-of-range.""" + with self.assertRaisesRegex(S7ProtocolError, "0x05"): + self.client.db_read(1, 8, 4) def test_write_beyond_area_bounds(self) -> None: """Writing beyond area bounds should raise an error.""" From 8071b8bd9493a64cc6e6fafdfe810c7a9529c03b Mon Sep 17 00:00:00 2001 From: Gijs Molenaar Date: Sun, 20 Sep 2026 10:54:06 +0200 Subject: [PATCH 2/2] fix: restore CI after S7CommPlus extraction --- .github/workflows/test.yml | 2 +- doc/API/client.rst | 116 ++----------------------------------- doc/API/server.rst | 26 ++------- tests/test_server.py | 10 ++++ 4 files changed, 21 insertions(+), 133 deletions(-) diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index b5ed16e3..9265c146 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -38,7 +38,7 @@ jobs: with: enable-cache: true - name: Install dependencies - run: uv sync --extra test --extra s7commplus --python python${{ matrix.python-version }} + run: uv sync --extra test --python python${{ matrix.python-version }} - name: Run pytest if: matrix.python-version != '3.14' || matrix.os != 'ubuntu-24.04' run: uv run pytest diff --git a/doc/API/client.rst b/doc/API/client.rst index 110a553a..9d5c2fcb 100644 --- a/doc/API/client.rst +++ b/doc/API/client.rst @@ -1,121 +1,17 @@ Client ====== -python-snap7 provides two client packages: +The ``s7`` package implements the classic S7 protocol for S7-300/400 PLCs and +PUT/GET access on S7-1200/1500. -- ``s7commplus``: S7CommPlus protocol for S7-1200/1500 PLCs -- ``s7``: Classic S7 protocol for S7-300/400 and PUT/GET access on S7-1200/1500 - -s7commplus.Client ------------------ - -.. code-block:: python - - from s7commplus import Client - - client = Client() - client.connect("192.168.1.10") - data = client.db_read(1, 0, 4) - client.disconnect() - -s7commplus.AsyncClient ----------------------- - -The asynchronous client currently supports the TLS connection and -legitimation path. Legacy V1 SessionKey authentication is available only on -the synchronous client. - -.. code-block:: python - - import asyncio - from s7commplus import AsyncClient - - async def main(): - client = AsyncClient() - await client.connect("192.168.1.10") - data = await client.db_read(1, 0, 4) - await client.disconnect() - - asyncio.run(main()) - -V2 connection with TLS ----------------------- - -S7-1500 PLCs with firmware 2.x use S7CommPlus V2, which requires TLS. Pass -``use_tls=True`` to the ``connect()`` method: - -.. code-block:: python - - from s7commplus import Client - - client = Client() - client.connect("192.168.1.10", use_tls=True) - data = client.db_read(1, 0, 4) - client.disconnect() - -For PLCs with custom certificates, provide the certificate paths: - -.. code-block:: python - - client.connect( - "192.168.1.10", - use_tls=True, - tls_cert="/path/to/client.pem", - tls_key="/path/to/client.key", - tls_ca="/path/to/ca.pem", - ) - -Password authentication ------------------------ - -The synchronous client accepts the ``password`` keyword for both TLS -legitimation and the post-SessionKey exchange used by older V1 PLCs: - -.. code-block:: python - - from s7commplus import Client - - client = Client() - client.connect("192.168.1.10", use_tls=True, password="my_plc_password") - data = client.db_read(1, 0, 4) - client.disconnect() - -For a V1 PLC, omit ``use_tls=True``. The asynchronous client has no -``password`` argument on ``connect``; on a TLS connection, authenticate -explicitly with ``await client.authenticate(password)``. - -Concurrent async reads ----------------------- - -An internal ``asyncio.Lock`` serialises each send/receive cycle so that -multiple coroutines can safely share a single connection: - -.. code-block:: python - - results = await asyncio.gather( - client.db_read(1, 0, 4), - client.db_read(1, 10, 4), - ) - ----- - -.. automodule:: s7commplus.client - :members: - -.. automodule:: s7commplus.async_client - :members: - -s7.Client (legacy) ---------------------- - -The ``s7.Client`` implements the classic S7 protocol for S7-300/400 PLCs -and PUT/GET access on S7-1200/1500. +s7.Client +--------- .. automodule:: snap7.client :members: -s7.AsyncClient (legacy) --------------------------- +s7.AsyncClient +-------------- .. automodule:: snap7.async_client :members: diff --git a/doc/API/server.rst b/doc/API/server.rst index f518a767..5ee1a709 100644 --- a/doc/API/server.rst +++ b/doc/API/server.rst @@ -1,20 +1,8 @@ Server ====== -python-snap7 provides two server implementations: - -- ``s7commplus.Server``: S7CommPlus server emulator -- ``s7.server.Server``: Legacy S7 server for testing - -.. code:: python - - from s7commplus import Server - - server = Server() - server.start(tcp_port=1102) - -For quick testing with the legacy server, you can also use the ``mainloop`` -helper: +python-snap7 provides a classic S7 server for testing. The ``mainloop`` helper +starts one quickly: .. code:: python @@ -24,14 +12,8 @@ helper: ---- -s7commplus.Server ------------------ - -.. automodule:: s7commplus.server - :members: - -snap7.Server (legacy) ---------------------- +s7.Server +--------- .. automodule:: snap7.server :members: diff --git a/tests/test_server.py b/tests/test_server.py index 520a2afa..e992e58d 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -1049,6 +1049,16 @@ def test_read_crossing_area_bounds(self) -> None: with self.assertRaisesRegex(S7ProtocolError, "0x05"): self.client.db_read(1, 8, 4) + def test_multi_read_preserves_item_error(self) -> None: + """A multi-read reports the failing item's return code.""" + items = [ + {"area": S7Area.DB, "db_number": 1, "start": 0, "size": 1}, + {"area": S7Area.DB, "db_number": 99, "start": 0, "size": 1}, + ] + + with self.assertRaisesRegex(S7ProtocolError, r"item 1 failed.*0x0a"): + self.client.read_multi_vars(items) + def test_write_beyond_area_bounds(self) -> None: """Writing beyond area bounds should raise an error.""" # DB1 is only 10 bytes, writing 20 bytes at offset 0 should fail