integration of comms and command_interface

Signed-off-by: Maximilian Deubel <[email protected]>

Author: maxd-nordic

src/nrfcredstore/at_client.py

@@ -1,10 +1,12 @@
 import argparse
 import sys
 import serial
+import logging
 
 from nrfcredstore.exceptions import ATCommandError, NoATClientException
-from nrfcredstore.at_client import ATClient
+from nrfcredstore.command_interface import ATCommandInterface
 from nrfcredstore.credstore import CredStore, CredType
+from nrfcredstore.comms import Comms
 
 FUN_MODE_OFFLINE = 4
 KEY_TYPES_OR_ANY = list(map(lambda type: type.name, CredType))
@@ -19,7 +21,7 @@
 
 def parse_args(in_args):
     parser = argparse.ArgumentParser(description='Manage certificates stored in a cellular modem.')
-    parser.add_argument('dev', help='Serial device used to communicate with the modem.')
+    parser.add_argument('dev', help='Device used to communicate with the modem.')
     parser.add_argument('--baudrate', type=int, default=115200, help='Serial baudrate')
     parser.add_argument('--timeout', type=int, default=3,
         help='Serial communication timeout in seconds')
@@ -55,6 +57,10 @@ def parse_args(in_args):
 
     deleteall_parser = subparsers.add_parser('deleteall', help='Delete all keys in a secure tag')
 
+    imei_parser = subparsers.add_parser('imei', help='Get IMEI of the modem')
+
+    attoken_parser = subparsers.add_parser('attoken', help='Get attestation token of the modem')
+
     # add generate command and args
     generate_parser = subparsers.add_parser('generate', help='Generate private key')
     generate_parser.add_argument('tag', type=int,
@@ -68,7 +74,8 @@ def parse_args(in_args):
 
 def exec_cmd(args, credstore):
     if args.subcommand:
-        credstore.func_mode(FUN_MODE_OFFLINE)
+        if not credstore.func_mode(FUN_MODE_OFFLINE):
+            raise RuntimeError("Failed to set modem to offline mode.")
 
     if args.subcommand == 'list':
         ct = CredType[args.type]
@@ -104,30 +111,43 @@ def exec_cmd(args, credstore):
         credstore.keygen(args.tag, args.file, args.attributes)
         print(f'New private key generated in secure tag {args.tag}')
         print(f'Wrote CSR in DER format to {args.file.name}')
+    elif args.subcommand=='imei':
+        imei = credstore.command_interface.get_imei()
+        if imei is None:
+            raise RuntimeError("Failed to get IMEI.")
+        print(f'IMEI: {imei}')
+    elif args.subcommand=='attoken':
+        attoken = credstore.command_interface.get_attestation_token()
+        if attoken is None:
+            raise RuntimeError("Failed to get attestation token.")
+        print(f'Attestation token: {attoken}')
 
 def exit_with_msg(exitcode, msg):
     print(msg)
     exit(exitcode)
 
-def main(in_args, credstore):
-    at_client = credstore.at_client
-    try:
-        args = parse_args(in_args)
-        if args.dev:
-            at_client.connect(args.dev, args.baudrate, args.timeout)
-            at_client.verify()
-            at_client.enable_error_codes()
-        exec_cmd(args, credstore)
-    except NoATClientException:
-        exit_with_msg(ERR_NO_AT_CLIENT, 'The device does not respond to AT commands. Please flash at_client sample.')
-    except ATCommandError as err:
-        exit_with_msg(ERR_AT_COMMAND, err)
-    except TimeoutError as err:
-        exit_with_msg(ERR_TIMEOUT, 'The device did not respond in time. Please try again.')
-    except serial.SerialException as err:
-        exit_with_msg(ERR_SERIAL, f'Serial error: {err}')
-    except Exception as err:
-        exit_with_msg(ERR_UNKNOWN, f'Unhandled Error: {err}')
-
-def run():
-    main(sys.argv[1:], CredStore(ATClient(serial.Serial())))
+def main(args, credstore):
+    credstore.command_interface.detect_shell_mode()
+    credstore.command_interface.enable_error_codes()
+    exec_cmd(args, credstore)
+
+def run(argv=sys.argv):
+    logging.basicConfig(level='DEBUG')
+    args = parse_args(argv[1:])
+    comms = None
+
+    # use inquirer to find the device
+    if args.dev == 'auto':
+        comms = Comms(list_all=True, baudrate=args.baudrate, timeout=args.timeout)
+    elif args.dev == 'rtt':
+        comms = Comms(rtt=True, baudrate=args.baudrate, timeout=args.timeout)
+    # if dev is just numbers, assume it's an rtt device
+    elif args.dev.isdigit():
+        comms = Comms(rtt = True, serial_number = int(args.dev), timeout=args.timeout)
+    # otherwise, assume it's a serial device
+    else:
+        comms = Comms(port=args.dev, baudrate=args.baudrate, timeout=args.timeout)
+
+    cred_if = ATCommandInterface(comms)
+
+    main(args, CredStore(cred_if))

src/nrfcredstore/cli.py

@@ -27,16 +27,16 @@ def __init__(self, tag: int, type: int, sha: str):
         self.sha = sha
 
 class CredStore:
-    def __init__(self, at_client):
-        self.at_client = at_client
+    def __init__(self, command_interface):
+        self.command_interface = command_interface
 
     def func_mode(self, mode):
         """Set modem functioning mode
 
         See AT Command Reference Guide for valid modes.
         """
 
-        return self.at_client.at_command(f'AT+CFUN={mode}') == []
+        return self.command_interface.at_command(f'AT+CFUN={mode}', wait_for_result=True)
 
     def list(self, tag = None, type: CredType = CredType.ANY) -> List[Credential]:
         """List stored credentials
@@ -57,8 +57,12 @@ def list(self, tag = None, type: CredType = CredType.ANY) -> List[Credential]:
             if type != CredType.ANY:
                 cmd = f'{cmd},{CredType(type).value}'
 
-        response_lines = self.at_client.at_command(cmd)
-        response_lines = [line for line in response_lines if line.strip()]
+        self.command_interface.at_command(cmd, wait_for_result=False)
+        result, response = self.command_interface.comms.expect_response("OK", "ERROR", "%CMNG: ")
+        if not result:
+            raise RuntimeError("Failed to list credentials")
+        response_lines = response.splitlines()
+        response_lines = [line.strip() for line in response_lines if line.strip()]
 
         columns = map(lambda line: line.replace('%CMNG: ', '').replace('"', '').split(','), response_lines)
         cred_map = map(lambda columns:
@@ -77,7 +81,8 @@ def write(self, tag: int, type: CredType, file: io.TextIOBase):
         if type == CredType.ANY:
             raise ValueError
         cert = file.read().rstrip()
-        return self.at_client.at_command(f'AT%CMNG=0,{tag},{type.value},"{cert}"')
+        if not self.command_interface.at_command(f'AT%CMNG=0,{tag},{type.value},"{cert}"', wait_for_result=True):
+            raise RuntimeError("Failed to write credential")
 
     def delete(self, tag: int, type: CredType):
         """Delete a credential from the modem
@@ -87,22 +92,19 @@ def delete(self, tag: int, type: CredType):
 
         if type == CredType.ANY:
             raise ValueError
-        return self.at_client.at_command(f'AT%CMNG=3,{tag},{type.value}')
+        if not self.command_interface.at_command(f'AT%CMNG=3,{tag},{type.value}', wait_for_result=True):
+            raise RuntimeError("Failed to delete credential")
 
     def keygen(self, tag: int, file: io.BufferedIOBase, attributes: str = ''):
         """Generate a new private key and return a certificate signing request in DER format"""
-        cmd = f'AT%KEYGEN={tag},2,0'
 
-        if attributes != '':
-            cmd = f'{cmd},"{attributes}"'
+        keygen_output = self.command_interface.get_csr(sectag=tag, attributes=attributes)
 
-        response_lines = self.at_client.at_command(cmd)
-        for l in response_lines:
-            if not l.startswith('%KEYGEN'):
-                continue
-            keygen_output = l.replace('%KEYGEN: "', '')
-            csr_der_b64 = keygen_output.split('.')[0]
-            csr_der_bytes = base64.urlsafe_b64decode(csr_der_b64 + '===')
-            file.write(csr_der_bytes)
+        if not keygen_output:
+            raise RuntimeError("Failed to generate key")
 
+        csr_der_b64 = keygen_output.split('.')[0]
+        csr_der_bytes = base64.urlsafe_b64decode(csr_der_b64 + '===')
+
+        file.write(csr_der_bytes)
         file.close()