Feature/add key mgmt prov method

docker/docker-compose.yml

@@ -1,4 +1,3 @@
-version: '3.7'
 services:
     prov-service:
         container_name: wm-prov-srv

examples/provisioning_config.yml

@@ -19,7 +19,7 @@
 #     
 # node_name : (mandatory)(string) Ex: test_node
 #    method : (mandatory)(Unsecured:0, Secured:1, Extended UID:3) Ex: 0
-#    factory_key : (only for methods 1 and 3)(32 bytes string, [0:15 Auth key][16:31 Enc Key])
+#    factory_key : (for methods 1,3 and 7)(32 bytes string, [0:15 Auth key][16:31 Enc Key])
 #    node_uid : (optional)(16 bytes string) Ex: 0x7e 0x71 0xe5 0xd7 0x22 0xef 0x0f 0x4b 0xa8 0x7d 0x44 0xd4 0xe0 0xe5 0xb5 0x7d
 #    node_uid_type : (optional)(1 byte string) Ex: 0x01
 #    node_authenticator_uid_type : (optional)(1 byte string) Ex: 0x01
@@ -65,4 +65,16 @@ 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_key_extended_uid_key_mgmt:
+    authenticator_uid: 0xb3 0x43 0x33 0x00 0x93 0x81 0x08 0x4a 0x8d 0xb3 0xaa 0x9e 0x53 0xd2 0x2a 0x1e
+    authenticator_uid_type: 1
+    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
+    method: 7
+    network: network_prod
+    network_key_sequence: 1
+    node_id: 18
+    uid: 0x41 0xb1 0x85 0x7a 0x0f 0xb6 0xb1 0x48 0xa5 0xe4 0xb9 0xb6 0x03 0x53 0x1b 0x3b
+    management_encryption_key: 0x01 0x02 0x03 0x04 0x05 0x06 0x07 0x08 0x09 0x0a 0x0b 0x0c 0x0d 0x0e 0x0f 0x10
+    management_authentication_key: 0x10 0x0f 0x0e 0x0d 0x0c 0x0b 0x0a 0x09 0x08 0x07 0x06 0x05 0x04 0x03 0x02 0x01
+    management_key_sequence: 1
 version: 1

wirepas_provisioning_server/data.py

@@ -18,6 +18,21 @@
 from wirepas_provisioning_server.migrate_config import ConfigFileMigration
 
 
+class ProvisioningDataIds:
+    """CBOR IDs for provisioning data - must match values defined in SDK provisioning library"""
+
+    ENC_KEY = 0
+    AUTH_KEY = 1
+    NW_ADDR = 2
+    NW_CHANNEL = 3
+    NODE_ADDR = 4
+    NODE_ROLE = 5
+    NW_KEY_SEQ = 6
+    MGMT_ENC_KEY = 7
+    MGMT_AUTH_KEY = 8
+    MGMT_KEY_SEQ = 9
+
+
 @dataclasses.dataclass(frozen=True)
 class ProvisioningDataEndpoints:
     """Endpoints for provisioning data packets"""
@@ -140,7 +155,10 @@ def __init__(self, config: Optional[str] = None):
 
                 if "uid" in node_cfg.keys():
                     uid: str | int | bytes = node_cfg["uid"]
-                elif node_cfg["method"] == ProvisioningMethod.EXTENDED:
+                elif node_cfg["method"] in (
+                    ProvisioningMethod.EXTENDED,
+                    ProvisioningMethod.EXTENDED_UID_KEY_MGMT,
+                ):
                     try:
                         uid = _generate_extended_uid(
                             node_cfg["authenticator_uid_type"],
@@ -176,6 +194,26 @@ def __init__(self, config: Optional[str] = None):
                 else:
                     node_role = None
 
+                if "network_key_sequence" in node_cfg.keys():
+                    network_key_sequence = convert_to_int(node_cfg["network_key_sequence"])
+                else:
+                    network_key_sequence = None
+
+                if "management_encryption_key" in node_cfg.keys():
+                    management_encryption_key = convert_to_bytes(node_cfg["management_encryption_key"])
+                else:
+                    management_encryption_key = None
+
+                if "management_authentication_key" in node_cfg.keys():
+                    management_authentication_key = convert_to_bytes(node_cfg["management_authentication_key"])
+                else:
+                    management_authentication_key = None
+
+                if "management_key_sequence" in node_cfg.keys():
+                    management_key_sequence = convert_to_int(node_cfg["management_key_sequence"])
+                else:
+                    management_key_sequence = None
+
                 if "user_specific" in node_cfg.keys():
                     user_specific = dict()
                     for k in node_cfg["user_specific"]:
@@ -199,6 +237,10 @@ def __init__(self, config: Optional[str] = None):
                     network_channel,
                     node_id=node_id,
                     node_role=node_role,
+                    network_key_sequence=network_key_sequence,
+                    management_encryption_key=management_encryption_key,
+                    management_authentication_key=management_authentication_key,
+                    management_key_sequence=management_key_sequence,
                     user_specific=user_specific,
                     factory_key=factory_key,
                 )
@@ -213,6 +255,10 @@ def append(
         network_channel: Optional[int],
         node_id: Optional[int] = None,
         node_role: Optional[bytes] = None,
+        network_key_sequence: Optional[int] = None,
+        management_encryption_key: Optional[bytes] = None,
+        management_authentication_key: Optional[bytes] = None,
+        management_key_sequence: Optional[int] = None,
         user_specific: Optional[dict[int, bytes | str]] = None,
         factory_key: Optional[bytes] = None,
     ) -> None:
@@ -235,6 +281,18 @@ def append(
         if node_role is not None:
             self[uid]["node_role"] = node_role
 
+        if network_key_sequence is not None:
+            self[uid]["network_key_sequence"] = network_key_sequence
+
+        if management_encryption_key is not None:
+            self[uid]["management_encryption_key"] = management_encryption_key
+
+        if management_authentication_key is not None:
+            self[uid]["management_authentication_key"] = management_authentication_key
+
+        if management_key_sequence is not None:
+            self[uid]["management_key_sequence"] = management_key_sequence
+
         if user_specific is not None:
             self[uid]["user_specific"] = dict()
             for k in user_specific:
@@ -254,27 +312,43 @@ def append(
         logging.debug(" -  network_channel: %s", network_channel)
         logging.debug(" -  node_id: %s", node_id)
         logging.debug(" -  node_role: %s", node_role)
+        logging.debug(" -  network_key_sequence: %s", network_key_sequence)
+        logging.debug(" -  management_encryption_key: %s", management_encryption_key)
+        logging.debug(" -  management_authentication_key: %s", management_authentication_key)
+        logging.debug(" -  management_key_sequence: %s", management_key_sequence)
         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"]
+        self_dic[ProvisioningDataIds.ENC_KEY] = self[uid]["encryption_key"]
+        self_dic[ProvisioningDataIds.AUTH_KEY] = self[uid]["authentication_key"]
 
         if "network_address" in self[uid].keys():
-            self_dic[2] = self[uid]["network_address"]
+            self_dic[ProvisioningDataIds.NW_ADDR] = self[uid]["network_address"]
 
         if "network_channel" in self[uid].keys():
-            self_dic[3] = self[uid]["network_channel"]
+            self_dic[ProvisioningDataIds.NW_CHANNEL] = self[uid]["network_channel"]
 
         if "node_id" in self[uid].keys():
-            self_dic[4] = self[uid]["node_id"]
+            self_dic[ProvisioningDataIds.NODE_ADDR] = self[uid]["node_id"]
 
         if "node_role" in self[uid].keys():
-            self_dic[5] = self[uid]["node_role"]
+            self_dic[ProvisioningDataIds.NODE_ROLE] = self[uid]["node_role"]
+
+        if "network_key_sequence" in self[uid].keys():
+            self_dic[ProvisioningDataIds.NW_KEY_SEQ] = self[uid]["network_key_sequence"]
+
+        if "management_encryption_key" in self[uid].keys():
+            self_dic[ProvisioningDataIds.MGMT_ENC_KEY] = self[uid]["management_encryption_key"]
+
+        if "management_authentication_key" in self[uid].keys():
+            self_dic[ProvisioningDataIds.MGMT_AUTH_KEY] = self[uid]["management_authentication_key"]
+
+        if "management_key_sequence" in self[uid].keys():
+            self_dic[ProvisioningDataIds.MGMT_KEY_SEQ] = self[uid]["management_key_sequence"]
 
         if "user_specific" in self[uid].keys():
             for key in self[uid]["user_specific"]:

wirepas_provisioning_server/message.py

@@ -34,6 +34,7 @@ class ProvisioningMethod(IntEnum):
     UNSECURED = 0
     SECURED = 1
     EXTENDED = 3
+    EXTENDED_UID_KEY_MGMT = 7
 
 
 class ProvisioningNackReason(IntEnum):