From d3389ed55dc7c6bc5cb3c4def7dba31b238c6b58 Mon Sep 17 00:00:00 2001 From: Gijs Molenaar Date: Thu, 12 Nov 2026 10:21:47 +0200 Subject: [PATCH 1/2] Add a read_only flag to Server.register_area A real S7-1214C V4.2 answers a Write Var to a write-protected DB with item return code 0x03, "accessing the object not allowed", and no header error (capture in #976). The server can now emulate that. - register_area() takes a keyword-only read_only argument. Registering the area again, or unregistering it, drops the flag. - A write to a read-only area is refused with item code 0x03 in both the single-item and the multi-item path. Nothing is written and no data-write event is emitted. Read-only is checked before the range; there is no capture of an out-of-range write to a protected DB. - Reads, a block download and changes the server side makes to the buffer are not affected. --- CHANGES.md | 9 ++++ doc/server.rst | 17 ++++++++ snap7/server/__init__.py | 51 +++++++++++++++++++---- tests/test_server.py | 89 ++++++++++++++++++++++++++++++++++++++++ 4 files changed, 157 insertions(+), 9 deletions(-) diff --git a/CHANGES.md b/CHANGES.md index d1545317..6a4360c6 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -1,6 +1,15 @@ CHANGES ======== +Unreleased +---------- + +### New features + +* `Server.register_area()` takes a `read_only` keyword. The server refuses a client write to a read-only area with + item return code `0x03`, "accessing the object not allowed", and no header error, as an S7-1214C V4.2 does for a + write-protected DB. Reads still work, and a block download is not refused (#976). + 3.5.0 ----- diff --git a/doc/server.rst b/doc/server.rst index 460b05d5..31b1b614 100644 --- a/doc/server.rst +++ b/doc/server.rst @@ -99,6 +99,23 @@ Registering Multiple Areas range. +Read-only Areas +--------------- + +Pass ``read_only=True`` to make the server refuse client writes to an area. +Reads keep working, and the server side can still change the data. + +.. code-block:: python + + server.register_area(SrvArea.DB, 1, bytearray(100), read_only=True) + +A write (Write Var) to such an area is answered with item return code ``0x03``, +"accessing the object not allowed", and no error in the header. The client +raises ``S7ProtocolError`` for it. This is what an S7-1214C (firmware V4.2) +answers for a write to a write-protected DB; S7-300/400 has not been captured. +A block download to a read-only DB is not refused. + + Using the Mainloop Helper -------------------------- diff --git a/snap7/server/__init__.py b/snap7/server/__init__.py index 47e637bb..bb45bbda 100644 --- a/snap7/server/__init__.py +++ b/snap7/server/__init__.py @@ -14,7 +14,7 @@ from ctypes import Array, c_char from enum import IntEnum from types import TracebackType -from typing import Any, Callable, Dict, List, Optional, Tuple, Type, Union +from typing import Any, Callable, Dict, List, Optional, Set, Tuple, Type, Union from ..datatypes import S7Area, S7WordLen from ..error import S7ConnectionError, S7ProtocolError @@ -37,6 +37,11 @@ _EVENT_QUEUE_CAPACITY = 1024 +# Item return code "accessing the object not allowed". An S7-1214C V4.2 answers a +# Write Var to a write-protected DB with it, at item level and without a header +# error (capture in python-snap7 issue #976). +_ITEM_ACCESS_NOT_ALLOWED = 0x03 + class ServerState(IntEnum): """S7 server states.""" @@ -94,6 +99,7 @@ def __init__(self, log: bool = True, max_clients: int = 64, **kwargs: object) -> # Memory areas self.memory_areas: Dict[Tuple[S7Area, int], bytearray] = {} self.area_locks: Dict[Tuple[S7Area, int], threading.Lock] = {} + self._read_only_areas: Set[Tuple[S7Area, int]] = set() self._upload_contexts: Dict[Tuple[str, int], Dict[str, Any]] = {} self._download_contexts: Dict[Tuple[str, int], Dict[str, Any]] = {} @@ -211,7 +217,9 @@ def stop(self) -> int: logger.info("S7 Server stopped") return 0 - def register_area(self, area: SrvArea, index: int, userdata: Union[bytearray, "Array[c_char]"]) -> int: + def register_area( + self, area: SrvArea, index: int, userdata: Union[bytearray, "Array[c_char]"], *, read_only: bool = False + ) -> int: """ Register a memory area with the server. @@ -219,6 +227,11 @@ def register_area(self, area: SrvArea, index: int, userdata: Union[bytearray, "A area: Memory area type index: Area index/number userdata: Initial data for the area (bytearray or ctypes array) + read_only: Refuse client writes (Write Var) to this area with item + return code ``0x03``, "accessing the object not allowed", as an + S7-1200 does for a write-protected DB. Reads still work, and so + do a block download and changes the server side makes to + ``userdata``. Registering the area again replaces the flag. Returns: 0 on success @@ -246,8 +259,12 @@ def register_area(self, area: SrvArea, index: int, userdata: Union[bytearray, "A area_key = (s7_area, index) self.memory_areas[area_key] = data self.area_locks[area_key] = threading.Lock() + if read_only: + self._read_only_areas.add(area_key) + else: + self._read_only_areas.discard(area_key) - logger.info(f"Registered area {area.name} index {index}, size {len(data)}") + logger.info(f"Registered area {area.name} index {index}, size {len(data)}{', read-only' if read_only else ''}") return 0 def unregister_area(self, area: SrvArea, index: int) -> int: @@ -278,6 +295,7 @@ def unregister_area(self, area: SrvArea, index: int) -> int: if area_key in self.memory_areas: del self.memory_areas[area_key] del self.area_locks[area_key] + self._read_only_areas.discard(area_key) logger.info(f"Unregistered area {area.name} index {index}") return 0 @@ -1083,10 +1101,16 @@ def _handle_write_area(self, request: Dict[str, Any], client_address: Tuple[str, area, db_number, start, count, write_data = addr_info - # Write data to registered memory area - success = self._write_to_memory_area(area, db_number, start, write_data) - if not success: - return self._build_error_response(request, 0x8404) # Area not found or write error + # A read-only area refuses the write at item level, with no header + # error, as a real S7-1200 does for a write-protected DB. + read_only = (area, db_number) in self._read_only_areas + if read_only: + logger.warning(f"Write to read-only area {area}#{db_number} refused") + else: + # Write data to registered memory area + success = self._write_to_memory_area(area, db_number, start, write_data) + if not success: + return self._build_error_response(request, 0x8404) # Area not found or write error # Build successful response with error bytes header = struct.pack( @@ -1109,6 +1133,9 @@ def _handle_write_area(self, request: Dict[str, Any], client_address: Tuple[str, ) # Data section (write response) + if read_only: + return header + parameters + bytes((_ITEM_ACCESS_NOT_ALLOWED,)) + data_section = b"\xff" # Success return code self._emit_event( @@ -1310,8 +1337,10 @@ def _write_item_to_memory_area(self, area: S7Area, db_number: int, start: int, w Returns: Item return code: ``0xFF`` on success, ``0x0A`` when the area is not - registered, and ``0x05`` when the range is outside the registered area. - Nothing is written unless the whole item fits. + registered, ``0x03`` when it is read-only, and ``0x05`` when the range + is outside the registered area. Nothing is written unless the whole + item fits. That read-only is checked before the range is a choice: + there is no capture of an out-of-range write to a protected DB. """ try: area_key = (area, db_number) @@ -1320,6 +1349,10 @@ def _write_item_to_memory_area(self, area: S7Area, db_number: int, start: int, w logger.warning(f"Memory area {area}#{db_number} not registered for write") return 0x0A + if area_key in self._read_only_areas: + logger.warning(f"Write to read-only area {area}#{db_number} refused") + return _ITEM_ACCESS_NOT_ALLOWED + with self.area_locks[area_key]: area_data = self.memory_areas[area_key] diff --git a/tests/test_server.py b/tests/test_server.py index 6d00a6a6..5961cb6b 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -1471,5 +1471,94 @@ def test_multi_item_write_reports_per_item_errors() -> None: assert buffer == bytearray([5, 6, 0, 0, 0, 0, 0, 0, 0, 0]) +@pytest.mark.server +def test_read_only_area_refuses_single_item_write_at_item_level() -> None: + """A write to a read-only area gets item code 0x03 and no header error. + + This is what an S7-1214C V4.2 answers for a write-protected DB (#976). + """ + server = Server() + buffer = bytearray(4) + server.register_area(SrvArea.DB, 1, buffer, read_only=True) + server.clear_events() + + request = server.protocol.build_write_request(S7Area.DB, 1, 0, S7WordLen.BYTE, b"\x01\x02") + response = server._process_request(request, ("127.0.0.1", 12345)) + + assert response is not None + assert response[1] == S7PDUType.ACK_DATA + assert response[10:12] == b"\x00\x00" # error class and code: no header error + assert response[12:] == bytes([S7Function.WRITE_AREA, 1, 0x03]) + assert buffer == bytearray(4) + assert not server.pick_event() # a refused write is not a data-write event + + +@pytest.mark.server +def test_read_only_area_refuses_multi_item_write_per_item() -> None: + """In a multi-item write only the items for read-only areas are refused.""" + server = Server() + protected = bytearray(4) + writable = bytearray(4) + server.register_area(SrvArea.DB, 1, protected, read_only=True) + server.register_area(SrvArea.DB, 2, writable) + + items = [ + (2, 0, bytes([1, 2])), + (1, 0, bytes([3, 4])), + (1, 3, bytes([5, 6])), # also out of range: read-only is reported first + ] + response = server._process_request(_build_multi_write_request(items), ("127.0.0.1", 12345)) + + assert response is not None + assert response[10:12] == b"\x00\x00" + assert response[14:] == b"\xff\x03\x03" + assert protected == bytearray(4) + assert writable == bytearray([1, 2, 0, 0]) + + +@pytest.mark.server +def test_read_only_flag_is_cleared_by_register_and_unregister() -> None: + """Registering an area again, or unregistering it, drops the read-only flag.""" + server = Server() + request = server.protocol.build_write_request(S7Area.DB, 1, 0, S7WordLen.BYTE, b"\x01") + + server.register_area(SrvArea.DB, 1, bytearray(4), read_only=True) + buffer = bytearray(4) + server.register_area(SrvArea.DB, 1, buffer) + response = server._process_request(request, ("127.0.0.1", 12345)) + assert response is not None + assert response[12:] == bytes([S7Function.WRITE_AREA, 1, 0xFF]) + assert buffer[0] == 1 + + server.register_area(SrvArea.DB, 1, bytearray(4), read_only=True) + server.unregister_area(SrvArea.DB, 1) + buffer = bytearray(4) + server.register_area(SrvArea.DB, 1, buffer) + response = server._process_request(request, ("127.0.0.1", 12345)) + assert response is not None + assert response[12:] == bytes([S7Function.WRITE_AREA, 1, 0xFF]) + + +@pytest.mark.server +def test_client_write_to_read_only_area() -> None: + """The client raises for a write to a read-only DB, and can still read it.""" + server = Server() + buffer = bytearray([7, 8, 9, 10]) + server.register_area(SrvArea.DB, 1, buffer, read_only=True) + server.start(tcp_port=SERVER_PORT + 4) + client = Client() + try: + client.connect(ip, 0, 1, SERVER_PORT + 4) + with pytest.raises(S7ProtocolError, match="0x03"): + client.db_write(1, 0, bytearray([1])) + assert client.db_read(1, 0, 4) == bytearray([7, 8, 9, 10]) + buffer[0] = 42 # the server side can still change the data + assert client.db_read(1, 0, 1) == bytearray([42]) + finally: + client.disconnect() + server.stop() + server.destroy() + + if __name__ == "__main__": unittest.main() From addf2fd1ff03718e39803f268580ed74fc2ba6a4 Mon Sep 17 00:00:00 2001 From: Gijs Molenaar Date: Thu, 12 Nov 2026 18:57:47 +0200 Subject: [PATCH 2/2] Cite the pull request in the changelog entry --- CHANGES.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/CHANGES.md b/CHANGES.md index 6a4360c6..ea8eef02 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -8,7 +8,7 @@ Unreleased * `Server.register_area()` takes a `read_only` keyword. The server refuses a client write to a read-only area with item return code `0x03`, "accessing the object not allowed", and no header error, as an S7-1214C V4.2 does for a - write-protected DB. Reads still work, and a block download is not refused (#976). + write-protected DB. Reads still work, and a block download is not refused (#992). 3.5.0 -----