WIP: Implement packet checksum validation

README.md

@@ -74,6 +74,19 @@ ness-cli events --logfile packets.log
 
 Include the generated log file with bug reports to assist with troubleshooting.
 
+### Checksum validation
+
+By default, the client accepts packets even if their checksums are invalid.
+Enable checksum verification to detect and reject malformed packets:
+
+- CLI users can pass `--validate-checksum` to the `events` and `server`
+  commands.
+- Library users can construct `Client` with
+  `validate_checksums=True`.
+
+When enabled, the library raises an error for packets with invalid checksums.
+The CLI logs such packets in red and ignores them.
+
 ## API Documentation
 You can find the full API documentation [here](https://nessclient.readthedocs.io/en/latest/api.html)
 

docs/cli.md

@@ -29,3 +29,16 @@ ness-cli events --logfile packets.log
 ```
 
 Include the generated file when raising an issue to help diagnose problems.
+
+### Checksum validation
+
+Packet checksums can be validated to drop malformed messages. When enabled,
+invalid packets are logged in red and ignored. Pass `--validate-checksum` to
+the `events` command to enable verification:
+
+```sh
+ness-cli events --host PANEL_HOST --port PORT --validate-checksum
+```
+
+The same flag is available on the `server` command to verify incoming
+packets from clients.

nessclient/cli/events.py

@@ -23,13 +23,19 @@
 @click.option("--update-interval", type=int, default=60)
 @click.option("--infer-arming-state/--no-infer-arming-state")
 @click.option("--logfile", type=click.Path(), help="Write raw TX/RX packets to file")
+@click.option(
+    "--validate-checksum",
+    is_flag=True,
+    help="Validate packet checksums",
+)
 def events(
     host: str,
     port: int,
     update_interval: int,
     infer_arming_state: bool,
     serial_tty: str | None,
     logfile: str | None,
+    validate_checksum: bool,
 ) -> None:
     asyncio.run(
         interactive_ui(
@@ -39,6 +45,7 @@ def events(
             infer_arming_state=infer_arming_state,
             serial_tty=serial_tty,
             packet_logfile=logfile,
+            validate_checksum=validate_checksum,
         )
     )
 
@@ -51,6 +58,7 @@ async def interactive_ui(
     infer_arming_state: bool,
     serial_tty: str | None,
     packet_logfile: str | None,
+    validate_checksum: bool,
 ) -> None:
     """Run an interactive TUI for the alarm."""
     log_fp: TextIO | None = open(packet_logfile, "a") if packet_logfile else None
@@ -70,6 +78,7 @@ async def interactive_ui(
         serial_tty=serial_tty if connection is None else None,
         infer_arming_state=infer_arming_state,
         update_interval=update_interval,
+        validate_checksums=validate_checksum,
     )
 
     panel_version: str | None = None
@@ -100,6 +109,10 @@ def on_zone_change(zone: int, triggered: bool) -> None:
     def on_state_change(state: ArmingState, arming_mode: ArmingMode | None) -> None:
         _add_log(logs, "RX", f"State: {state.value} Mode: {arming_mode}")
 
+    @client.on_decode_error
+    def on_decode_error(raw: str, exc: Exception) -> None:
+        _add_log(logs, "ERR", f"Failed to decode {raw!r}: {exc}")
+
     keepalive_task = asyncio.create_task(client.keepalive())
     # Initial update and panel info request
     _add_log(logs, "TX", "Update")
@@ -218,6 +231,8 @@ def on_state_change(state: ArmingState, arming_mode: ArmingMode | None) -> None:
                     attr |= curses.color_pair(2)
                 elif line.startswith("EVT"):
                     attr |= curses.A_DIM
+                elif line.startswith("ERR"):
+                    attr |= curses.color_pair(3)
                 pieces = textwrap.wrap(
                     line,
                     width=log_inner_w,

nessclient/cli/server/__init__.py

@@ -21,8 +21,18 @@
     default=PanelVersionUpdate.Model.D8X.name,
 )
 @click.option("--panel-version", default="0.0")
+@click.option(
+    "--validate-checksum",
+    is_flag=True,
+    help="Validate packet checksums",
+)
 def server(
-    host: str, port: int, zones: int, panel_model: str, panel_version: str
+    host: str,
+    port: int,
+    zones: int,
+    panel_model: str,
+    panel_version: str,
+    validate_checksum: bool,
 ) -> None:
     major_str, minor_str = panel_version.split(".")
     s = AlarmServer(
@@ -32,5 +42,6 @@ def server(
         panel_model=PanelVersionUpdate.Model[panel_model],
         panel_major_version=int(major_str),
         panel_minor_version=int(minor_str),
+        validate_checksums=validate_checksum,
     )
     s.start()

nessclient/cli/server/alarm_server.py

@@ -33,6 +33,7 @@ def __init__(
         panel_model: PanelVersionUpdate.Model,
         panel_major_version: int,
         panel_minor_version: int,
+        validate_checksums: bool,
     ):
         self._alarm = Alarm.create(
             num_zones=num_zones,
@@ -44,6 +45,7 @@ def __init__(
             handle_command=self._handle_command,
             log_callback=self._on_server_log,
             rx_callback=self._on_server_rx,
+            validate_checksums=validate_checksums,
         )
         self._host = host
         self._port = port
@@ -537,7 +539,7 @@ def _on_server_rx(self, line: str, pkt: Optional[Any]) -> None:
                 parsed = "<unknown>"
             self._add_log("RX", f"{line} -> {parsed}")
         else:
-            self._add_log("RX", line)
+            self._add_log("ERR", line)
 
 
 def mode_to_event(mode: Alarm.ArmingMode | None) -> SystemStatusEvent.EventType:

nessclient/cli/server/server.py

@@ -1,8 +1,7 @@
 import logging
 import socket
 import threading
-import traceback
-from typing import List, Callable, Any, Optional
+from typing import Any, Callable, List, Optional
 
 from .zone import Zone
 from ...event import BaseEvent, SystemStatusEvent
@@ -17,10 +16,13 @@ def __init__(
         handle_command: Callable[[str], None],
         log_callback: Optional[Callable[[str], None]] = None,
         rx_callback: Optional[Callable[[str, Optional[Packet]], None]] = None,
+        *,
+        validate_checksums: bool = False,
     ):
         self._handle_command = handle_command
         self._log_callback = log_callback
         self._rx_callback = rx_callback
+        self._validate_checksums = validate_checksums
         self._handle_event_lock: threading.Lock = threading.Lock()
         self._clients_lock: threading.Lock = threading.Lock()
         self._clients: List[socket.socket] = []
@@ -99,7 +101,7 @@ def _write_to_all_clients(self, data: str) -> None:
     def _handle_incoming_line(self, line: str) -> None:
         _LOGGER.debug("Received incoming line: %s", line)
         try:
-            pkt = Packet.decode(line)
+            pkt = Packet.decode(line, validate_checksum=self._validate_checksums)
             _LOGGER.debug("Packet is: %s", pkt)
             if self._rx_callback is not None:
                 try:
@@ -117,11 +119,7 @@ def _handle_incoming_line(self, line: str) -> None:
             else:
                 raise NotImplementedError()
         except Exception as e:
-            msg = (
-                f"Error decoding or handling line: {repr(line)}\n"
-                f"{e}\n"
-                f"{traceback.format_exc()}"
-            )
+            msg = f"Error decoding line {line!r}: {e}"
             if self._rx_callback is not None:
                 try:
                     self._rx_callback(line, None)

nessclient/client.py

@@ -22,6 +22,8 @@ class Client:
     :param infer_arming_state: Infer the `DISARMED` arming state only via
         system status events. This works around a bug with some panels
         (`<v5.8`) which emit `update.status = []` when they are armed.
+    :param validate_checksums: Verify packet checksums and drop malformed
+        packets when enabled.
     """
 
     def __init__(
@@ -34,6 +36,7 @@ def __init__(
         infer_arming_state: bool = False,
         alarm: Alarm | None = None,
         decode_options: DecodeOptions | None = None,
+        validate_checksums: bool = False,
     ):
         if connection is None:
             if host is not None and port is not None:
@@ -50,7 +53,9 @@ def __init__(
 
         self.alarm = alarm
         self._decode_options = decode_options
+        self._validate_checksums = validate_checksums
         self._on_event_received: Callable[[BaseEvent], None] | None = None
+        self._on_decode_error: Callable[[str, Exception], None] | None = None
         self._event_subscribers: List[asyncio.Queue[BaseEvent]] = []
         self._connection = connection
         self._closed = False
@@ -63,6 +68,10 @@ def __init__(
         self._pending_ui_requests: Dict[int, asyncio.Future[StatusUpdate]] = {}
         self._pending_ui_lock = asyncio.Lock()
 
+    def on_decode_error(self, callback: Callable[[str, Exception], None]) -> None:
+        """Register a callback for packet decode errors."""
+        self._on_decode_error = callback
+
     async def arm_away(self, code: str | None = None) -> None:
         command = "A{}E".format(code if code else "")
         return await self.send_command(command)
@@ -212,10 +221,19 @@ async def _recv_loop(self) -> None:
                 _LOGGER.debug("Decoding data: '%s'", decoded_data)
                 if len(decoded_data) > 0:
                     try:
-                        pkt = Packet.decode(decoded_data)
+                        pkt = Packet.decode(
+                            decoded_data, validate_checksum=self._validate_checksums
+                        )
                         event = BaseEvent.decode(pkt, self._decode_options)
-                    except Exception:
+                    except Exception as e:
                         _LOGGER.warning("Failed to decode packet", exc_info=True)
+                        if self._on_decode_error is not None:
+                            try:
+                                self._on_decode_error(decoded_data, e)
+                            except Exception:
+                                _LOGGER.debug(
+                                    "on_decode_error raised, ignoring", exc_info=True
+                                )
                         continue
 
                     self._dispatch_event(event, pkt)

nessclient/packet.py

@@ -45,9 +45,8 @@ def length(self) -> int:
 
     @property
     def checksum(self) -> int:
-        bytes = self.encode(with_checksum=False)
-        total = sum([ord(x) for x in bytes]) & 0xFF
-        return (256 - total) % 256
+        raw = self.encode(with_checksum=False)
+        return calculate_checksum(self.start, raw)
 
     def encode(self, with_checksum: bool = True) -> str:
         data = ""
@@ -71,7 +70,7 @@ def encode(self, with_checksum: bool = True) -> str:
         return data
 
     @classmethod
-    def decode(cls, _data: str) -> "Packet":
+    def decode(cls, _data: str, *, validate_checksum: bool = True) -> "Packet":
         """
         Packets are ASCII encoded data. Packet layout is as follows:
 
@@ -118,8 +117,13 @@ def decode(cls, _data: str) -> "Packet":
         if has_timestamp(start):
             timestamp = decode_timestamp(data.take_bytes(6))
 
-        # TODO(NW): Figure out checksum validation
-        checksum = data.take_hex()  # noqa
+        checksum = data.take_hex()
+        if validate_checksum:
+            expected = calculate_checksum(start, _data[:-2])
+            if checksum != expected:
+                raise ValueError(
+                    f"Checksum mismatch: expected {expected:02x} got {checksum:02x}"
+                )
 
         if not data.is_consumed():
             raise ValueError("Unable to consume all data")
@@ -182,6 +186,25 @@ def is_user_interface_resp(start: int) -> bool:
     return start == 0x82
 
 
+def calculate_checksum(start: int, raw: str) -> int:
+    """Calculate packet checksum.
+
+    For user interface requests (start byte 0x83) the checksum is the one's
+    complement of the sum of the ASCII values prior to conversion to hex. For
+    all other packets the checksum is calculated on the hexadecimal byte values
+    before ASCII encoding.
+    """
+
+    if is_user_interface_req(start):
+        total = sum(ord(c) for c in raw)
+    else:
+        total = 0
+        for i in range(0, len(raw), 2):
+            total += int(raw[i : i + 2], 16)
+
+    return (0x100 - (total & 0xFF)) & 0xFF
+
+
 def decode_timestamp(data: str) -> datetime.datetime:
     """
     Decode timestamp using bespoke decoder.