Serial encryption

paradox/event.py

@@ -1,10 +1,10 @@
+from collections import namedtuple
+from copy import copy
 import datetime
+from enum import Enum
 import logging
 import time
 import typing
-from collections import namedtuple
-from copy import copy
-from enum import Enum
 
 from construct import Container
 
@@ -70,7 +70,7 @@ def __init__(self, label_provider=None):
         if label_provider is not None:
             self.label_provider = label_provider
         else:
-            self.label_provider = lambda type, value: "[{}:{}]".format(type, value)
+            self.label_provider = lambda type, value: f"[{type}:{value}]"
 
     def __repr__(self):
         lvars = {}
@@ -81,11 +81,7 @@ def __repr__(self):
             str(self.__class__)
             + "\n"
             + "\n".join(
-                (
-                    "{} = {}".format(item, lvars[item])
-                    for item in lvars
-                    if not item.startswith("_")
-                )
+                f"{item} = {lvars[item]}" for item in lvars if not item.startswith("_")
             )
         )
 
@@ -126,21 +122,21 @@ def call_hook(self, *args, **kwargs):
             kwargs["event"] = self
             try:
                 self.hook_fn(*args, **kwargs)
-            except:
+            except Exception:
                 logger.exception("Failed to call event hook")
 
 
 class LiveEvent(Event):
     def __init__(self, event: Container, event_map: dict, label_provider=None):
         raw = event.fields.value
-        if raw.po.command != 0xE:
+        if raw.po.command != 0xE and hasattr(raw, "event"):
             raise AssertionError("Message is not an event")
 
         # parse event map
         if raw.event.major not in event_map:
-            raise AssertionError("Unknown event major: {}".format(raw))
+            raise AssertionError(f"Unknown event major: {raw}")
 
-        super(LiveEvent, self).__init__(label_provider=label_provider)
+        super().__init__(label_provider=label_provider)
 
         self.major = raw.event.major  # Event major code
         self.minor = raw.event.minor  # Event minor code
@@ -177,9 +173,7 @@ def __init__(self, event: Container, event_map: dict, label_provider=None):
                 for k in sub:
                     if k == "message":
                         event_map[k] = (
-                            "{}: {}".format(event_map[k], sub[k])
-                            if k in event_map
-                            else sub[k]
+                            f"{event_map[k]}: {sub[k]}" if k in event_map else sub[k]
                         )
                     elif isinstance(sub[k], typing.List):  # for tags or other lists
                         event_map[k] = event_map.get(k, []) + sub[k]

paradox/hardware/panel.py

@@ -1,19 +1,19 @@
-import binascii
-import inspect
-import logging
-import typing
 from abc import abstractmethod
+import binascii
 from collections import defaultdict, namedtuple
+import inspect
 from itertools import chain
+import logging
 from time import time
+import typing
 
 from construct import Construct, Container, EnumIntegerString
 
 from paradox.config import config as cfg, get_limits_for_type
 from paradox.lib.utils import construct_free, sanitize_key
 
-from ..lib import ps
 from . import parsers
+from ..lib import ps
 
 logger = logging.getLogger("PAI").getChild(__name__)
 
@@ -36,6 +36,9 @@ def parse_message(self, message, direction="topanel") -> typing.Optional[Contain
         if message is None or len(message) == 0:
             return None
 
+        if message[0] >> 4 == 0xE and message[1] == 0xFE:
+            return parsers.Encrypted.parse(message)
+
         if direction == "topanel":
             if message[0] == 0x72 and message[1] == 0:
                 return parsers.InitiateCommunication.parse(message)
@@ -46,15 +49,15 @@ def parse_message(self, message, direction="topanel") -> typing.Optional[Contain
                 return parsers.InitiateCommunicationResponse.parse(message)
             elif message[0] == 0x00 and message[4] > 0:
                 return parsers.StartCommunicationResponse.parse(message)
-            else:
-                return None
+
+        return None
 
     def get_message(self, name) -> Construct:
         clsmembers = dict(inspect.getmembers(parsers))
         if name in clsmembers:
             return clsmembers[name]
         else:
-            raise ResourceWarning("{} parser not found".format(name))
+            raise ResourceWarning(f"{name} parser not found")
 
     @staticmethod
     def get_error_message(error_code) -> str:
@@ -170,7 +173,9 @@ async def load_definitions(self):
                         if definition != "disabled":
                             enabled_indexes.add(index)
 
-                cfg.LIMITS[elem_type] = get_limits_for_type(elem_type, list(enabled_indexes))
+                cfg.LIMITS[elem_type] = get_limits_for_type(
+                    elem_type, list(enabled_indexes)
+                )
                 cfg.LIMITS[elem_type] = list(
                     set(cfg.LIMITS[elem_type]).intersection(enabled_indexes)
                 )
@@ -339,10 +344,8 @@ def handle_status(message: Container, parser_map):
             if cfg.LOGGING_DUMP_MESSAGES:
                 logger.debug(f"Status parsed({mvars.address}): {res}")
             return res
-        except:
-            logger.exception(
-                "Unable to parse RAM Status Block ({})".format(mvars.address)
-            )
+        except Exception:
+            logger.exception(f"Unable to parse RAM Status Block ({mvars.address})")
             return
 
     @abstractmethod

paradox/hardware/parsers.py

@@ -1,8 +1,30 @@
-from construct import (BitsInteger, BitStruct, Bytes, Const, Default, Enum,
-                       Flag, Int8ub, Int16ub, Nibble, Padding, RawCopy, Struct)
+from construct import (
+    BitsInteger,
+    BitStruct,
+    Bytes,
+    Checksum,
+    Const,
+    Default,
+    Enum,
+    Flag,
+    Int8ub,
+    Int16ub,
+    Nibble,
+    Padding,
+    RawCopy,
+    Struct,
+    this,
+)
 
-from .common import (CommunicationSourceIDEnum, HexInt, PacketChecksum,
-                     PacketLength, ProductIdEnum, FamilyIdEnum)
+from .common import (
+    CommunicationSourceIDEnum,
+    FamilyIdEnum,
+    HexInt,
+    PacketChecksum,
+    PacketLength,
+    ProductIdEnum,
+    calculate_checksum,
+)
 
 InitiateCommunication = Struct(
     "fields"
@@ -120,3 +142,30 @@
     ),
     "checksum" / PacketChecksum(Bytes(1)),
 )
+
+Encrypted = Struct(
+    "fields"
+    / RawCopy(
+        Struct(
+            "po"
+            / BitStruct(
+                "command" / Const(0xE, Nibble),
+                "status"
+                / Struct(
+                    "reserved" / Flag,
+                    "alarm_reporting_pending" / Flag,
+                    "Winload_connected" / Flag,
+                    "NeWare_connected" / Flag,
+                ),
+            ),
+            "source" / Const(0xFE, Int8ub),
+            "length" / PacketLength(Int8ub),
+            "_not_used0" / Bytes(1),
+            "request_nr" / Int8ub,
+            "data" / Bytes(lambda this: this.length - 7),
+        )
+    ),
+    "checksum"
+    / Checksum(Bytes(1), lambda data: calculate_checksum(data), this.fields.data),
+    "end" / Int8ub,
+)

paradox/lib/async_message_manager.py

@@ -4,8 +4,7 @@
 
 from construct import Container
 
-from paradox.lib.handlers import (FutureHandler, HandlerRegistry,
-                                  PersistentHandler)
+from paradox.lib.handlers import FutureHandler, HandlerRegistry, PersistentHandler
 
 logger = logging.getLogger("PAI").getChild(__name__)
 
@@ -14,7 +13,7 @@ class EventMessageHandler(PersistentHandler):
     def can_handle(self, data: Container) -> bool:
         assert isinstance(data, Container)
         values = data.fields.value
-        return values.po.command == 0xE and (not hasattr(values, "requested_event_nr"))
+        return values.po.command == 0xE and hasattr(values, "event")
 
 
 class ErrorMessageHandler(PersistentHandler):
@@ -27,7 +26,7 @@ def can_handle(self, data: Container) -> bool:
 
 class AsyncMessageManager:
     def __init__(self, loop=None):
-        super(AsyncMessageManager, self).__init__()
+        super().__init__()
 
         if not loop:
             loop = asyncio.get_event_loop()

tests/hardware/test_encryption.py

@@ -0,0 +1,85 @@
+import pytest
+
+from paradox.hardware.parsers import Encrypted
+
+
+@pytest.mark.parametrize(
+    "payload_hex",
+    [
+        # from EVO192 7.50.000+ firmware
+        # tx
+        (
+            "E0 FE 2E 00 12 C5 CA 4A B7 DC B3 C5 92 06 F6 E9 EB 47 76 1E C9 28 BF 27 54 EE 41 DD D3 AB B4 D0 88 BB B3 EE 36 9B E2 17 50 FD 52 CC 91 19"
+        ),
+        # rx
+        (
+            "E0 FE 2E 00 12 C5 3F 0A B7 DC 83 97 D4 06 F6 E9 EB 47 56 5C 89 38 BF 35 F0 EA A5 DC C3 2B 95 D2 80 E9 B3 EE 36 9B E2 17 50 FD B4 CC 7C 19"
+        ),
+        # tx
+        (
+            "E0 FE 2E 00 12 01 79 71 35 21 C7 F1 C5 3F 0A B7 DC B3 E5 D0 46 E6 E9 F9 E3 72 FA C8 38 3F 06 56 E6 13 DD D3 AB B4 D0 88 BB B3 77 B4 11 19"
+        ),
+        # rx
+        ("E0 FE 0F 00 12 01 3D 77 35 21 F7 03 B4 B8 04"),
+        (
+            "E0 FE 2E 00 13 80 4F F6 7E FD 6A 3B 91 85 52 E2 45 A5 52 DB 28 77 5D DC 9C 64 42 DB BA BE 47 79 71 35 21 F7 A3 83 3F 0A B7 DC E0 69 9B 16"
+        ),
+        (
+            "E0 FE 2E 00 14 61 CB D8 54 3E 81 E5 F1 2B BC E0 EE BB B2 EE 36 9B E2 17 50 FD 2D 15 F3 05 D7 2F B5 59 19 60 FC 6B F3 CC 76 B8 28 D3 4A 18"
+        ),
+        # tx
+        ("E0 FE 11 00 14 F5 78 3D 21 F7 59 83 BF 54 BC 70 07"),
+        # rx
+        (
+            "E0 FE 50 00 14 F5 7A 90 21 F7 59 83 0F 0A B7 DC B3 C5 92 06 F6 E9 EB 47 76 1E C9 28 BF 27 54 EE 41 DD D3 AB B4 D0 88 BB B3 EE 36 9B E2 17 50 FD 2D 15 F3 05 D7 2F B5 59 19 60 FC 6B F3 CC 76 B8 8E 81 F7 D4 49 FC 06 BE 6E E3 4E 29 99 BC E5 2A"
+        ),
+        # tx
+        ("E0 FE 11 00 14 E3 42 95 95 56 5A DB 25 19 8C A7 06"),
+        # rx
+        (
+            "E0 FE 50 00 14 E3 40 38 95 56 5A DB A5 46 DB 28 77 5D DC 9C 64 42 DB BA BE 47 79 71 35 21 F7 A3 83 3F 0A B7 DC B3 C5 92 06 F6 E9 EB 47 76 1E C9 28 BF 27 54 EE 41 DD D3 AB B4 D0 88 B3 A3 FE B6 8B E2 17 50 ED 07 15 F3 05 D7 2F B5 F1 8C FD 29"
+        ),
+        # tx
+        ("E0 FE 11 00 14 A8 76 23 8A F9 6A EF 5A E3 24 81 07"),
+        # rx
+        (
+            "E0 FE 50 00 14 A8 74 8E 8A F9 6A EF DA 3D B2 83 09 7E BC 6F 4B 9D 95 56 A0 DF A5 46 DB 28 77 5D DC 9C 64 42 DB BA BE 47 79 71 B5 21 F7 A3 83 3F 0A B7 DC B3 C5 92 06 F6 E9 EB 47 76 1E C9 28 BF 27 54 EE 41 DD D3 AB B4 D0 88 BB B3 16 24 69 2A"
+        ),
+        # tx
+        ("E0 FE 11 00 14 B7 E4 97 24 7F E4 CE FB 4F B8 8C 08"),
+        # rx
+        (
+            "E0 FE 50 00 14 B7 F4 0A 24 7F E4 CE F9 90 CF DA 3D B2 83 09 7E BC 6F 4B 9D 95 56 A0 DF A5 46 DB 28 77 5D DC 9C 64 42 DB BA BE 47 79 71 35 21 F7 A3 83 3F 0A B7 DC B3 C5 92 06 F6 E9 EB 47 76 1E C9 28 BF 27 54 EE 41 DD D3 AB B4 D0 25 B8 72 2A"
+        ),
+        # tx
+        ("E0 FE 11 00 14 F7 BC 57 EC F4 65 C7 A4 1F 84 60 08"),
+        # rx
+        (
+            "E0 FE 50 00 14 F7 BE FA EC F4 65 C7 24 7F 2B 8A F9 90 CF DA 3D B2 83 09 7E BC 6F 4B 9D 95 56 A0 DF A5 46 DB 28 77 5D DC 9C 64 42 DB BA BE 47 79 71 35 21 F7 A3 83 3F 0A B7 DC B3 C5 92 06 F6 E9 EB 47 76 1E C9 28 BF 27 54 EE 41 DD A3 84 C3 2A"
+        ),
+        # tx
+        ("E0 FE 11 00 14 F9 3F 57 79 71 8F C8 26 F8 10 01 07"),
+        # rx
+        (
+            "E0 FE 4C 00 14 F9 2F 4A 79 71 8F C8 F7 A3 83 3F 0A B7 DC B3 C5 92 06 F6 E9 EB 47 76 1E C9 28 BF 27 54 EE 41 DD D3 AB B4 D0 88 BB B3 EE 36 9B E2 17 50 FD 2D 15 F3 05 D7 2F B5 59 19 60 FC 6B F3 CC 76 B8 8E 81 F7 D4 49 5D 10 7E 28"
+        ),
+        # from SP6000+
+        (
+            "e0 fe 2e 00 00 78 04 c1 92 06 f6 e9 eb 47 76 1e c9 28 bf 27 54 ee 41 dd d3 ab b4 d0 88 bb b3 ee 36 9b e2 17 50 fd 2d 15 f3 05 20 6b 7f 16"
+        ),
+    ],
+)
+def test_parse(payload_hex: str):
+    payload = bytes.fromhex(payload_hex)
+    data = Encrypted.parse(payload)
+    print(
+        f"Expected length: {len(payload[5:-2])}, actual length: {len(data.fields.value.data)}"
+    )
+    assert data.fields.value.data == payload[5:-2]
+    print(data)
+
+
+# def test_payload_decryption():
+#     payload = bytes.fromhex("12 01 3D 77 35 21 F7 03 B4")
+#     l = len(payload)
+#     print(f"Length: {l}")

tests/lib/test_async_message_manager.py

@@ -1,10 +1,9 @@
 import asyncio
 
-import pytest
 from construct import Container
+import pytest
 
-from paradox.lib.async_message_manager import (AsyncMessageManager,
-                                               EventMessageHandler)
+from paradox.lib.async_message_manager import AsyncMessageManager, EventMessageHandler
 from paradox.lib.handlers import PersistentHandler
 
 
@@ -89,7 +88,11 @@ async def test_wait_for_message(mocker):
 @pytest.mark.asyncio
 async def test_handler_exception(mocker):
     msg = Container(
-        fields=Container(value=Container(po=Container(command=0xE), event_source=0xFF))
+        fields=Container(
+            value=Container(
+                po=Container(command=0xE), event_source=0xFF, event=Container()
+            )
+        )
     )
 
     mm = AsyncMessageManager()