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