Refactor data model

examples/provisioning_config.yml

@@ -12,8 +12,8 @@
 # version: 1
 # networks:
 #   network_name: (mandatory)(string) Ex: test_network
-#     address: (optional)(uint) Ex: 0x1012EE
-#     channel: (optional)(uint) Ex: 13
+#     address: (mandatory)(uint) Ex: 0x1012EE
+#     channel: (mandatory)(uint) Ex: 13
 #     authentication_key : (mandatory)(16 bytes string) Ex: 0xFF 0xFF 0xFF 0xFF 0xFF 0xFF 0xFF 0xFF 0xFF 0xFF 0xFF 0xFF 0xFF 0xFF 0xFF 0xFF
 #     encryption_key : (mandatory)(16 bytes string) Ex: 0xFF 0xFF 0xFF 0xFF 0xFF 0xFF 0xFF 0xFF 0xFF 0xFF 0xFF 0xFF 0xFF 0xFF 0xFF 0xFF
 #     
@@ -65,4 +65,11 @@ nodes:
     network: network_prod
     node_id: 17
     uid: 0x58 0xc8 0x12 0xad 0x37 0xe8 0x36 0x4a 0xa1 0x1f 0x1c 0xbc 0x63 0x3e 0x8e 0x34
+  test_node_extended_uuid_only:
+    uid: 0x70 0xC8 0x33 0x00 0x00 0x00 0x00 0x00 0x00 0x87 0x00 0x00 0x00 0x00 0x00 0x01 0x01 0x7F 0xC8 0x33 0x00 0x00 0x00 0x00 0x00 0x00 0x87 0x00 0x00 0x00 0x00 0x00 0x01 0x01
+    method: 0x03
+    network: network_prod
+    node_id: 0x05
+    factory_key: 0x00 0x01 0x02 0x03 0x04 0x05 0x06 0x07 0x08 0X09 0X0A 0X0B 0X0C 0X0D 0X0E 0X0F 0x00 0x01 0x02 0x03 0x04 0x05 0x06 0x07 0x08 0X09 0X0A 0X0B 0X0C 0X0D 0X0E 0X0F
+
 version: 1

requirements-dev.txt

@@ -0,0 +1,3 @@
+# Copyright 2021 Wirepas Ltd 2021
+
+pytest==8.3.3

tests/test_models.py

@@ -0,0 +1,145 @@
+import json
+import pydantic
+import pytest
+from typing import Any
+from wirepas_provisioning_server.message import ProvisioningMethod
+from wirepas_provisioning_server.models import NodeV1, NetworkV1
+
+
+@pytest.fixture
+def network_data() -> dict[str, Any]:
+    return dict(
+        name="test_network",
+        address=0x1012EE,
+        channel=13,
+        authentication_key="0xFF 0xFF 0xFF 0xFF 0xFF 0xFF 0xFF 0xFF 0xFF 0xFF 0xFF 0xFF 0xFF 0xFF 0xFF 0xFF",
+        encryption_key="0xAA 0xBB 0xCC 0xDD 0xEE 0xFF 0x00 0x11 0x22 0x33 0x44 0x55 0x66 0x77 0x88 0x99",
+    )
+
+
+@pytest.fixture
+def node_data_secured(network_data) -> dict[str, Any]:
+    return dict(
+        factory_key="0xAABBCCDDEEFF00112233445566778899AABBCCDDEEFF00112233445566778899",
+        method=1,
+        name="test_node",
+        network=NetworkV1(**network_data),
+        node_id=0x654321,
+        role=0x01,
+        uid="0x00 0x11 0x12 0x13",
+        user_specific={
+            128: 0xAA,
+            255: 0xBB,
+        },
+    )
+
+
+def test_node_v1_basics(
+    network_data: dict[str, Any], node_data_secured: dict[str, Any]
+):
+
+    node = NodeV1(**node_data_secured)
+
+    assert node.factory_key == bytes.fromhex(
+        "AABBCCDDEEFF00112233445566778899AABBCCDDEEFF00112233445566778899"
+    )
+    assert node.method == ProvisioningMethod.SECURED
+    assert node.name == "test_node"
+    assert node.network == NetworkV1(**network_data)
+    assert node.node_id == 0x654321
+    assert node.role == bytes.fromhex("01")
+    assert node.uid == bytes.fromhex("00111213")
+    assert node.user_specific == {128: 0xAA, 255: 0xBB}
+
+
+def test_node_v1_validation_key(node_data_secured: dict[str, Any]):
+
+    with pytest.raises(ValueError) as e:
+        node_data_secured["factory_key"] = b"\x00\x11\x12\x13"
+        NodeV1(**node_data_secured)
+    assert 'Factory key must be 32 bytes, got "4"' in str(e)
+
+
+def test_node_v1_method(node_data_secured: dict[str, Any]):
+
+    for i in [-1, 2, 4]:
+        with pytest.raises(ValueError) as e:
+            node_data_secured["method"] = i
+            NodeV1(**node_data_secured)
+        assert "Input should be 0, 1 or 3 [type=enum" in str(e)
+
+    for i in [0, 1]:
+        node_data_secured["method"] = i
+        assert NodeV1(**node_data_secured).method == i
+
+
+
+def test_node_v1_node_id(node_data_secured: dict[str, Any]):
+
+    for i in [0x00000000, 0x80000000, 0x80FFFFFF, 0xFFFFFFFF]:
+        with pytest.raises(pydantic.ValidationError) as e:
+            node_data_secured["node_id"] = i
+            NodeV1(**node_data_secured)
+        print(str(e))
+        assert "Node ID must be None, between [0x1; 0x7FFFFFFF] or [0x81000000" in str(e)
+
+    for i in [1, 0x7FFFFFFF, 0x81000000, 0xFFFFFFFD]:
+        node_data_secured["node_id"] = i
+        assert NodeV1(**node_data_secured).node_id == i
+
+def test_network_v1_basics(network_data: dict[str, Any]):
+
+    network = NetworkV1(**network_data)
+
+    assert network.name == "test_network"
+    assert network.address == 0x1012EE
+    assert network.channel == 13
+    assert network.authentication_key == bytes.fromhex(
+        "FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF"
+    )
+    assert network.encryption_key == bytes.fromhex("AABBCCDDEEFF00112233445566778899")
+
+    with pytest.raises(pydantic.ValidationError):
+        network.name = "new_name"
+
+    json_data = NetworkV1(**network_data).model_dump_json()
+    assert json.loads(json_data) == {
+        "name": "test_network",
+        "address": 0x1012EE,
+        "channel": 13,
+        "authentication_key": "0xffffffffffffffffffffffffffffffff",
+        "encryption_key": "0xaabbccddeeff00112233445566778899",
+    }
+
+
+def test_network_v1_optional_fields(network_data: dict[str, Any]):
+    del network_data["address"]
+    del network_data["channel"]
+
+    network = NetworkV1(**network_data)
+    assert network.name == "test_network"
+    assert network.address is None
+    assert network.channel is None
+    assert network.authentication_key == bytes.fromhex(
+        "FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF"
+    )
+    assert network.encryption_key == bytes.fromhex("AABBCCDDEEFF00112233445566778899")
+
+
+def test_network_v1_keys(network_data: dict[str, Any]):
+    network_data["authentication_key"] = "0x01020304050607080910111213141516"
+    network_data["encryption_key"] = 0x01020304050607080910111213141516
+
+    network = NetworkV1(**network_data)
+    assert network.authentication_key == bytes.fromhex(
+        "01020304050607080910111213141516"
+    )
+    assert network.encryption_key == bytes.fromhex("01020304050607080910111213141516")
+
+    with pytest.raises(ValueError) as e:
+        NetworkV1(
+            name="invalid_key_length",
+            authentication_key=b"\x01\x02",
+            encryption_key=b"\x01\x02",
+        )
+    assert "Keys must be 16 bytes" in str(e)

wirepas_provisioning_server/data.py

@@ -11,10 +11,11 @@
 import yaml
 import logging
 
-from typing import Final, Optional
+from typing import Optional
 
 from wirepas_provisioning_server.helpers import convert_to_bytes, convert_to_int, ProvisioningDataException
 from wirepas_provisioning_server.message import ProvisioningMethod
+from wirepas_provisioning_server.models import NetworkV1, NodeV1
 from wirepas_provisioning_server.migrate_config import ConfigFileMigration
 
 
@@ -65,169 +66,34 @@ def __init__(self, config: Optional[str] = None):
             if cfg.get("version") != 1:
                 raise ProvisioningDataException("Invalid data config file. Version must be 1")
 
-            # Validate network parameters
+            networks: dict[str, NetworkV1] = {}
             for name, network in cfg["networks"].items():
-                try:
-                    for parameter in [
-                        "authentication_key",
-                        "encryption_key",
-                    ]:
-                        network[parameter]
-                except KeyError as e:
-                    raise ProvisioningDataException(f"Invalid data config file. Network {name} must include {str(e)}.")
-
-            for node_name, node_cfg in cfg["nodes"].items():
-                if "network" not in node_cfg.keys():
-                    raise ProvisioningDataException(f"Invalid data config file. Node {node_name} must include network.")
-                network_name = node_cfg["network"]
-
-                if "method" not in node_cfg.keys():
-                    raise ProvisioningDataException(f"Invalid data config file. Node {node_name} must include method.")
-
-                provision_methods = [e.value for e in ProvisioningMethod]
-                if node_cfg["method"] not in provision_methods:
-                    raise ProvisioningDataException(f"Node method must be one of {provision_methods}")
-
-                if "uid" in node_cfg.keys():
-                    uid: str | int | bytes = node_cfg["uid"]
-                elif node_cfg["method"] == ProvisioningMethod.EXTENDED:
-                    try:
-                        uid = _generate_extended_uid(
-                            node_cfg["authenticator_uid_type"],
-                            node_cfg["authenticator_uid"],
-                            node_cfg["node_uid_type"],
-                            node_cfg["node_uid"],
-                        )
-
-                    except KeyError:
-                        raise ProvisioningDataException(
-                            f"Invalid data config file. Node {node_name} must include UID information."
-                        )
-                else:
-                    raise ProvisioningDataException(f"Invalid data config file. Node {node_name} must include UID information")
-
-                if "address" in cfg["networks"][network_name].keys():
-                    network_address = convert_to_int(cfg["networks"][network_name]["address"])
-                else:
-                    network_address = None
-
-                if "channel" in cfg["networks"][network_name].keys():
-                    network_channel = convert_to_int(cfg["networks"][network_name]["channel"])
-                else:
-                    network_channel = None
-
-                if "node_id" in node_cfg.keys():
-                    node_id = convert_to_int(node_cfg["node_id"])
-                else:
-                    node_id = None
-
-                if "node_role" in node_cfg.keys():
-                    node_role = convert_to_bytes(node_cfg["node_role"])
-                else:
-                    node_role = None
-
-                if "user_specific" in node_cfg.keys():
-                    user_specific = dict()
-                    for k in node_cfg["user_specific"]:
-                        if k < 128 or k > 255:
-                            raise KeyError
-                        user_specific[k] = node_cfg["user_specific"][k]
-                else:
-                    user_specific = None
+                networks[name] = NetworkV1(
+                    **network,
+                    name=name,
+                )
 
-                if "factory_key" in node_cfg.keys():
-                    factory_key = convert_to_bytes(node_cfg["factory_key"])
-                else:
-                    factory_key = None
-
-                self.append(
-                    convert_to_bytes(uid),
-                    node_cfg["method"],
-                    convert_to_bytes(cfg["networks"][network_name]["encryption_key"]),
-                    convert_to_bytes(cfg["networks"][network_name]["authentication_key"]),
-                    network_address,
-                    network_channel,
-                    node_id=node_id,
-                    node_role=node_role,
-                    user_specific=user_specific,
-                    factory_key=factory_key,
+            for name, raw_node in cfg["nodes"].items():
+                raw_node["network"] = networks[raw_node["network"]]
+                node = NodeV1(
+                    **raw_node,
+                    name=name,
                 )
+                self[node.uid] = node
 
-    def append(
-        self,
-        uid: bytes,
-        method: int,
-        encryption_key: bytes,
-        authentication_key: bytes,
-        network_address: Optional[int],
-        network_channel: Optional[int],
-        node_id: Optional[int] = None,
-        node_role: Optional[bytes] = None,
-        user_specific: Optional[dict[int, bytes | str]] = None,
-        factory_key: Optional[bytes] = None,
-    ) -> None:
-
-        # TODO : parameter checks
-        self[uid] = dict(
-            method=method,
-            encryption_key=encryption_key,
-            authentication_key=authentication_key,
-        )
-        if network_address is not None:
-            self[uid]["network_address"] = network_address
-
-        if network_channel is not None:
-            self[uid]["network_channel"] = network_channel
-
-        if node_id is not None:
-            self[uid]["node_id"] = node_id
-
-        if node_role is not None:
-            self[uid]["node_role"] = node_role
-
-        if user_specific is not None:
-            self[uid]["user_specific"] = dict()
-            for k in user_specific:
-                # k should be an integer [128:255]
-                # authorized type for value are string, byte string, integers
-                self[uid]["user_specific"][k] = user_specific[k]
-
-        if factory_key is not None:
-            self[uid]["factory_key"] = factory_key
-
-        logging.info("Append new UID: %s", uid.hex())
-        logging.debug(" -  method: %s", method)
-        logging.debug(" -  factory_key: %s", factory_key)
-        logging.debug(" -  encryption_key: %s", encryption_key)
-        logging.debug(" -  authentication_key: %s", authentication_key)
-        logging.debug(" -  network_address: %s", network_address)
-        logging.debug(" -  network_channel: %s", network_channel)
-        logging.debug(" -  node_id: %s", node_id)
-        logging.debug(" -  node_role: %s", node_role)
-        if "user_specific" in self[uid].keys():
-            for k in self[uid]["user_specific"]:
-                logging.debug(" - %d : %s", k, self[uid]["user_specific"][k])
-
-    def getCbor(self, uid: bytes) -> bytes:
-        self_dic = dict()
-
-        self_dic[0] = self[uid]["encryption_key"]
-        self_dic[1] = self[uid]["authentication_key"]
-
-        if "network_address" in self[uid].keys():
-            self_dic[2] = self[uid]["network_address"]
-
-        if "network_channel" in self[uid].keys():
-            self_dic[3] = self[uid]["network_channel"]
-
-        if "node_id" in self[uid].keys():
-            self_dic[4] = self[uid]["node_id"]
-
-        if "node_role" in self[uid].keys():
-            self_dic[5] = self[uid]["node_role"]
-
-        if "user_specific" in self[uid].keys():
-            for key in self[uid]["user_specific"]:
-                self_dic[key] = self[uid]["user_specific"][key]
-
-        return cbor2.dumps(self_dic)
+                logging.info("Append new UID: 0x%s", node.uid.hex().upper())
+                logging.debug(" -  method: %s", node.method)
+                if node.factory_key is not None:
+                    logging.debug(" -  factory_key: 0x%s", node.factory_key.hex().upper())
+                else:
+                    logging.debug(" -  factory_key: None")
+                logging.debug(" -  encryption_key: 0x%s", node.network.encryption_key.hex().upper())
+                logging.debug(" -  authentication_key: 0x%s", node.network.authentication_key.hex().upper())
+                logging.debug(" -  network_address: %s", node.network.address)
+                logging.debug(" -  network_channel: %s", node.network.channel)
+                logging.debug(" -  node_id: %s", node.node_id)
+                logging.debug(" -  node_role: %s", node.role)
+                if node.user_specific is not None:
+                    logging.debug(" -  User specific data:: %s", node.role)
+                    for index, value in node.user_specific.items():
+                        logging.debug(" -    %d: %s", index, value)