Rough key-verification implementation

meshtastic/__main__.py

@@ -472,6 +472,21 @@ def onConnected(interface):
             waitForAckNak = True
             interface.getNode(args.dest, False, **getNode_kwargs).removeIgnored(args.remove_ignored_node)
 
+        if args.key_verification:
+            closeNow = True
+            waitForAckNak = True
+            interface.getNode(args.dest, False, **getNode_kwargs).keyVerification(args.key_verification)
+
+        if args.key_verification_number:
+            closeNow = True
+            waitForAckNak = True
+            interface.getNode(args.dest, False, **getNode_kwargs).keyVerificationNumber(args.key_verification_number, args.key_verification_nonce)
+
+        if args.key_verification_confirm:
+            closeNow = True
+            waitForAckNak = True
+            interface.getNode(args.dest, False, **getNode_kwargs).keyVerificationConfirm(args.key_verification_confirm)
+
         if args.reset_nodedb:
             closeNow = True
             waitForAckNak = True
@@ -1832,6 +1847,22 @@ def addRemoteAdminArgs(parser: argparse.ArgumentParser) -> argparse.ArgumentPars
         "Use the node ID with a '!' or '0x' prefix or the node number.",
         metavar="!xxxxxxxx"
     )
+    group.add_argument(
+        "--key-verification",
+        help="start key verification. "
+        "Use the node ID with a '!' or '0x' prefix or the node number.",
+        metavar="!xxxxxxxx"
+    )
+    group.add_argument(
+        "--key-verification-number",
+        help="start key verification. "
+        "Use the node ID with a '!' or '0x' prefix or the node number.",
+    )
+    group.add_argument(
+        "--key-verification-confirm",
+        help="start key verification. "
+        "Use the node ID with a '!' or '0x' prefix or the node number.",
+    )
     group.add_argument(
         "--reset-nodedb",
         help="Tell the destination node to clear its list of nodes",
@@ -1961,6 +1992,12 @@ def initParser():
         action="store_true",
     )
 
+    group.add_argument(
+        "--key-verification-nonce",
+        help="start key verification. "
+        "Use the node ID with a '!' or '0x' prefix or the node number.",
+    )
+
     power_group = parser.add_argument_group(
         "Power Testing", "Options for power testing/logging."
     )

meshtastic/node.py

@@ -770,6 +770,49 @@ def removeIgnored(self, nodeId: Union[int, str]):
             onResponse = self.onAckNak
         return self._sendAdmin(p, onResponse=onResponse)
 
+    def keyVerification(self, nodeId: Union[int, str]):
+        if isinstance(nodeId, str):
+            if nodeId.startswith("!"):
+                nodeId = int(nodeId[1:], 16)
+            else:
+                nodeId = int(nodeId)
+        p = admin_pb2.KeyVerificationAdmin()
+        p.message_type = p.MessageType.INITIATE_VERIFICATION
+        p.remote_nodenum = nodeId
+        p.nonce = 0
+        a = admin_pb2.AdminMessage()
+        a.key_verification_admin.CopyFrom(p)
+        if self == self.iface.localNode:
+            onResponse = None
+        else:
+            onResponse = self.onAckNak
+        return self._sendAdmin(a, onResponse=onResponse)
+    def keyVerificationNumber(self, number: int, nonce: int,):
+        print(int(number))
+        print(int(nonce))
+        p = admin_pb2.KeyVerificationAdmin()
+        p.message_type = p.MessageType.PROVIDE_SECURITY_NUMBER
+        p.nonce = int(nonce)
+        p.security_number = int(number)
+        a = admin_pb2.AdminMessage()
+        a.key_verification_admin.CopyFrom(p)
+        if self == self.iface.localNode:
+            onResponse = None
+        else:
+            onResponse = self.onAckNak
+        return self._sendAdmin(a, onResponse=onResponse)
+    def keyVerificationConfirm(self, nonce: int,):
+        print(int(nonce))
+        p = admin_pb2.KeyVerificationAdmin()
+        p.message_type = p.MessageType.DO_VERIFY
+        p.nonce = int(nonce)
+        a = admin_pb2.AdminMessage()
+        a.key_verification_admin.CopyFrom(p)
+        if self == self.iface.localNode:
+            onResponse = None
+        else:
+            onResponse = self.onAckNak
+        return self._sendAdmin(a, onResponse=onResponse)
     def resetNodeDb(self):
         """Tell the node to reset its list of nodes."""
         self.ensureSessionKey()