Add a read_only flag to Server.register_area

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 (#992).
+
 3.5.0
 -----
 

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
 --------------------------
 

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,14 +217,21 @@ 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.
 
         Args:
             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]
 

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()