diff --git a/src/nrfcloud_utils/claim_and_provision_device.py b/src/nrfcloud_utils/claim_and_provision_device.py index b3a66e1..500621a 100644 --- a/src/nrfcloud_utils/claim_and_provision_device.py +++ b/src/nrfcloud_utils/claim_and_provision_device.py @@ -7,7 +7,7 @@ import time import json import argparse -import coloredlogs, logging +import logging from nrfcloud_utils import ( ca_certs, nrf_cloud_onboard, @@ -15,8 +15,8 @@ nrf_cloud_diap, create_device_credentials ) -from nrfcloud_utils.cli_helpers import is_linux, is_windows, is_macos -from nrfcloud_utils.cli_helpers import CMD_TERM_DICT, CMD_TYPE_AUTO, CMD_TYPE_AT, CMD_TYPE_AT_SHELL, CMD_TYPE_TLS_SHELL, parser_add_comms_args +from nrfcloud_utils.cli_helpers import setup_logging +from nrfcloud_utils.cli_helpers import CMD_TERM_DICT, CMD_TYPE_AUTO, CMD_TYPE_AT, CMD_TYPE_AT_SHELL, parser_add_comms_args from cryptography import x509 from cryptography.hazmat.primitives import serialization from nrfcredstore.comms import Comms @@ -82,10 +82,7 @@ def parse_args(in_args): help='Set the logging level' ) args = parser.parse_args(in_args) - if args.plain: - logging.basicConfig(level=args.log_level.upper()) - else: - coloredlogs.install(level=args.log_level.upper(), fmt='%(levelname)-8s %(message)s') + setup_logging(level=args.log_level, use_color=not args.plain) return args def error_exit(err_msg): @@ -132,7 +129,7 @@ def main(in_args): error_exit(f'Device ID must not exceed {DEV_ID_MAX_LEN} characters') if 'CN=' in args.csr_attr: - error_exit(f'Do not include CN in --csr_attr. The device ID will be used as the CN') + error_exit('Do not include CN in --csr_attr. The device ID will be used as the CN') # load local CA cert and key if needed; assume not needed if using provisioning tags if args.provisioning_tags is None: @@ -195,7 +192,7 @@ def main(in_args): logger.info(f'Unclaiming device {dev_uuid}...') api_res = nrf_cloud_diap.unclaim_device(args.api_key, dev_uuid) if api_res.status_code == 204: - logger.info(f'...success') + logger.info('...success') else: nrf_cloud_diap.print_api_result("Unclaim device response", api_res) logger.info("Device may not have been claimed before, continuing...") diff --git a/src/nrfcloud_utils/claim_devices.py b/src/nrfcloud_utils/claim_devices.py index 2381bfa..45408d8 100644 --- a/src/nrfcloud_utils/claim_devices.py +++ b/src/nrfcloud_utils/claim_devices.py @@ -7,9 +7,9 @@ import sys import csv import argparse -import coloredlogs, logging +import logging from nrfcloud_utils import nrf_cloud_diap -from nrfcloud_utils.cli_helpers import is_linux, is_windows, is_macos +from nrfcloud_utils.cli_helpers import setup_logging logger = logging.getLogger(__name__) @@ -40,10 +40,7 @@ def parse_args(in_args): help='Set the logging level' ) args = parser.parse_args(in_args) - if args.plain: - logging.basicConfig(level=args.log_level.upper()) - else: - coloredlogs.install(level=args.log_level.upper(), fmt='%(levelname)-8s %(message)s') + setup_logging(level=args.log_level, use_color=not args.plain) return args def bulk_claim(api_key, array_of_claims): diff --git a/src/nrfcloud_utils/cli_helpers.py b/src/nrfcloud_utils/cli_helpers.py index 020628d..9ea5dbe 100644 --- a/src/nrfcloud_utils/cli_helpers.py +++ b/src/nrfcloud_utils/cli_helpers.py @@ -10,6 +10,7 @@ import os import csv import logging +import coloredlogs MAX_CSV_ROWS = 1000 @@ -20,6 +21,39 @@ logger = logging.getLogger(__name__) +def setup_logging(level='info', fmt='%(levelname)-8s %(message)s', use_color=True): + """ + Configure colored logging with dark-terminal-friendly colors. + + Args: + level: logging level as string (debug, info, warning, error, critical) + fmt: log format string + use_color: whether to use colored output (False for --plain mode) + """ + log_level = getattr(logging, level.upper(), logging.INFO) + + if use_color: + # Define colors for dark terminals with better contrast + level_styles = { + 'debug': {'color': 'cyan'}, + 'info': {'color': 'green'}, + 'warning': {'color': 'yellow'}, + 'error': {'color': 'red'}, + 'critical': {'color': 'red', 'bold': True}, + } + field_styles = { + 'asctime': {'color': 'white'}, + 'levelname': {'color': 'white', 'bold': True}, + } + coloredlogs.install( + level=log_level, + fmt=fmt, + level_styles=level_styles, + field_styles=field_styles + ) + else: + logging.basicConfig(level=log_level, format=fmt) + def write_file(pathname, filename, bytes): """ save bytes to file @@ -28,7 +62,7 @@ def write_file(pathname, filename, bytes): if not path.isdir(pathname): try: makedirs(pathname, exist_ok=True) - except OSError as e: + except OSError: raise RuntimeError(f"Error creating file path [{pathname}]") full_path = path.join(pathname, filename) diff --git a/src/nrfcloud_utils/create_ca_cert.py b/src/nrfcloud_utils/create_ca_cert.py index adb24a4..16f3787 100644 --- a/src/nrfcloud_utils/create_ca_cert.py +++ b/src/nrfcloud_utils/create_ca_cert.py @@ -22,8 +22,8 @@ SubjectKeyIdentifier, ) -from nrfcloud_utils.cli_helpers import write_file -import coloredlogs, logging +from nrfcloud_utils.cli_helpers import write_file, setup_logging +import logging logger = logging.getLogger(__name__) @@ -51,9 +51,7 @@ def parse_args(in_args): help='Set the logging level' ) args = parser.parse_args(in_args) - level = getattr(logging, args.log_level.upper(), logging.INFO) - fmt = '%(levelname)-8s %(message)s' - coloredlogs.install(level=level, fmt=fmt) + setup_logging(level=args.log_level) return args diff --git a/src/nrfcloud_utils/create_device_credentials.py b/src/nrfcloud_utils/create_device_credentials.py index a1664fa..0c3d7d9 100644 --- a/src/nrfcloud_utils/create_device_credentials.py +++ b/src/nrfcloud_utils/create_device_credentials.py @@ -15,17 +15,12 @@ from cryptography import x509 import uuid from cryptography.x509 import ( - Name, - NameAttribute, - BasicConstraints, KeyUsage, - AuthorityKeyIdentifier, - SubjectKeyIdentifier, ) -from nrfcloud_utils.cli_helpers import write_file, save_onboarding_csv +from nrfcloud_utils.cli_helpers import write_file, save_onboarding_csv, setup_logging from nrfcloud_utils import ca_certs -import coloredlogs, logging +import logging logger = logging.getLogger(__name__) @@ -61,9 +56,7 @@ def parse_args(in_args): help='Set the logging level' ) args = parser.parse_args(in_args) - level = getattr(logging, args.log_level.upper(), logging.INFO) - fmt = '%(levelname)-8s %(message)s' - coloredlogs.install(level=level, fmt=fmt) + setup_logging(level=args.log_level) if len(args.csr) == 0 and len(args.cn) == 0: args.cn = str(uuid.uuid4()) return args diff --git a/src/nrfcloud_utils/create_proxy_jwt.py b/src/nrfcloud_utils/create_proxy_jwt.py index 683ba21..ed8a7db 100644 --- a/src/nrfcloud_utils/create_proxy_jwt.py +++ b/src/nrfcloud_utils/create_proxy_jwt.py @@ -9,7 +9,8 @@ import jwt from datetime import datetime, timezone, timedelta from os import path -import coloredlogs, logging +import logging +from nrfcloud_utils.cli_helpers import setup_logging logger = logging.getLogger(__name__) @@ -43,9 +44,7 @@ def parse_args(in_args): help='Set the logging level' ) args = parser.parse_args(in_args) - level = getattr(logging, args.log_level.upper(), logging.INFO) - fmt = '%(levelname)-8s %(message)s' - coloredlogs.install(level=level, fmt=fmt) + setup_logging(level=args.log_level) return args def create_nrf_cloud_jwt(prv_key_bytes, team_id, dev_id, days_valid): diff --git a/src/nrfcloud_utils/device_credentials_installer.py b/src/nrfcloud_utils/device_credentials_installer.py index 547962c..17ee4a2 100644 --- a/src/nrfcloud_utils/device_credentials_installer.py +++ b/src/nrfcloud_utils/device_credentials_installer.py @@ -10,9 +10,9 @@ import sys import uuid import semver -import coloredlogs, logging +import logging from nrfcloud_utils import create_device_credentials, ca_certs, modem_credentials_parser -from nrfcloud_utils.cli_helpers import write_file, save_devinfo_csv, save_onboarding_csv, is_linux, is_windows, is_macos, full_encoding +from nrfcloud_utils.cli_helpers import write_file, save_devinfo_csv, save_onboarding_csv, full_encoding, setup_logging from nrfcloud_utils.cli_helpers import CMD_TERM_DICT, CMD_TYPE_AUTO, CMD_TYPE_AT, CMD_TYPE_AT_SHELL, CMD_TYPE_TLS_SHELL, parser_add_comms_args from nrfcredstore.command_interface import ATCommandInterface, TLSCredShellInterface from nrfcredstore.comms import Comms @@ -106,10 +106,7 @@ def parse_args(in_args): help='Set the logging level' ) args = parser.parse_args(in_args) - if args.plain: - logging.basicConfig(level=args.log_level.upper()) - else: - coloredlogs.install(level=args.log_level.upper(), fmt='%(levelname)-8s %(message)s') + setup_logging(level=args.log_level, use_color=not args.plain) return args def parse_mfw_ver(ver_str): @@ -380,17 +377,32 @@ def main(in_args): # write CA cert(s) to device nrf_ca_cert_text = format_cred(ca_certs.get_ca_certs(args.coap, stage=args.stage)) - logger.info(f'Writing CA cert(s) to device...') - cred_if.write_credential(args.sectag, 0, nrf_ca_cert_text) + logger.info('Writing CA cert(s) to device...') + try: + cred_if.write_credential(args.sectag, 0, nrf_ca_cert_text) + except Exception as e: + logger.error('Failed to write CA certificate to device') + logger.debug(f'Error details: {e}') + raise # write dev cert to device - logger.info(f'Writing dev cert to device...') - cred_if.write_credential(args.sectag, 1, dev_text) + logger.info('Writing dev cert to device...') + try: + cred_if.write_credential(args.sectag, 1, dev_text) + except Exception as e: + logger.error('Failed to write device certificate to device') + logger.debug(f'Error details: {e}') + raise # If the private key was locally generated, write it to the device if prv_text is not None: - logger.info(f'Writing private key to device...') - cred_if.write_credential(args.sectag, 2, prv_text) + logger.info('Writing private key to device...') + try: + cred_if.write_credential(args.sectag, 2, prv_text) + except Exception as e: + logger.error('Failed to write private key to device') + logger.debug(f'Error details: {e}') + raise if args.verify: logger.info('Verifying credentials...') @@ -451,7 +463,14 @@ def verify_credential(cred_if, sec_tag, cred_type, cred = None, get_hash = False logger.error('Invalid credential string') return False - present, hash = cred_if.check_credential_exists(sec_tag, cred_type, get_hash = get_hash) + try: + present, hash = cred_if.check_credential_exists(sec_tag, cred_type, get_hash = get_hash) + except IndexError as e: + logger.error(f'...Failed to verify {cred_type_name}: Modem returned malformed response') + logger.debug(f'Error details: {e}') + logger.error('This may indicate the credential was not properly written to the device.') + logger.error('Try re-running the provisioning process.') + return False if not present: logger.error(f'...{cred_type_name} not found') @@ -462,7 +481,13 @@ def verify_credential(cred_if, sec_tag, cred_type, cred = None, get_hash = False return False if verify_hash: - expected_hash = cred_if.calculate_expected_hash(cred) + try: + expected_hash = cred_if.calculate_expected_hash(cred) + except (IndexError, Exception) as e: + logger.error(f'...Failed to calculate hash for {cred_type_name}') + logger.debug(f'Error details: {e}') + return False + if hash != expected_hash: logger.error(f'{cred_type_name} - SHA mismatch:') logger.error(f'\tDevice : {hash}') diff --git a/src/nrfcloud_utils/gather_attestation_tokens.py b/src/nrfcloud_utils/gather_attestation_tokens.py index 85c94de..f40e52c 100644 --- a/src/nrfcloud_utils/gather_attestation_tokens.py +++ b/src/nrfcloud_utils/gather_attestation_tokens.py @@ -8,12 +8,12 @@ import csv import argparse from nrfcloud_utils import modem_credentials_parser -from nrfcloud_utils.cli_helpers import is_linux, is_windows, is_macos -from nrfcloud_utils.cli_helpers import CMD_TERM_DICT, CMD_TYPE_AUTO, CMD_TYPE_AT, CMD_TYPE_AT_SHELL, CMD_TYPE_TLS_SHELL, parser_add_comms_args +from nrfcloud_utils.cli_helpers import setup_logging +from nrfcloud_utils.cli_helpers import CMD_TERM_DICT, CMD_TYPE_AUTO, CMD_TYPE_AT, CMD_TYPE_AT_SHELL, parser_add_comms_args from nrfcredstore.comms import Comms from nrfcredstore.command_interface import ATCommandInterface from datetime import datetime, timezone -import coloredlogs, logging +import logging logger = logging.getLogger(__name__) @@ -41,10 +41,7 @@ def parse_args(in_args): help='Set the logging level' ) args = parser.parse_args(in_args) - if args.plain: - logging.basicConfig(level=args.log_level.upper()) - else: - coloredlogs.install(level=args.log_level.upper(), fmt='%(levelname)-8s %(message)s') + setup_logging(level=args.log_level, use_color=not args.plain) return args def check_if_device_exists_in_csv(csv_filename, uuid, delete_duplicates): diff --git a/src/nrfcloud_utils/modem_credentials_parser.py b/src/nrfcloud_utils/modem_credentials_parser.py index 7988c47..7ee7870 100644 --- a/src/nrfcloud_utils/modem_credentials_parser.py +++ b/src/nrfcloud_utils/modem_credentials_parser.py @@ -6,15 +6,13 @@ import argparse import sys -from os import path -from os import makedirs from cbor2 import loads import base64 import hashlib -import coloredlogs, logging +import logging from cryptography.hazmat.primitives import serialization from cryptography import x509 -from nrfcloud_utils.cli_helpers import write_file +from nrfcloud_utils.cli_helpers import write_file, setup_logging logger = logging.getLogger(__name__) @@ -59,10 +57,7 @@ def parse_args(in_args): help="bool: Plain output (no colors)", action='store_true', default=False) args = parser.parse_args(in_args) - if args.plain: - logging.basicConfig(level=args.log_level.upper()) - else: - coloredlogs.install(level=args.log_level.upper(), fmt='%(levelname)-8s %(message)s') + setup_logging(level=args.log_level, use_color=not args.plain) return args def base64_decode(string): diff --git a/src/nrfcloud_utils/nrf_cloud_diap.py b/src/nrfcloud_utils/nrf_cloud_diap.py index 98333af..fd9a77f 100644 --- a/src/nrfcloud_utils/nrf_cloud_diap.py +++ b/src/nrfcloud_utils/nrf_cloud_diap.py @@ -5,7 +5,7 @@ # SPDX-License-Identifier: BSD-3-Clause import requests -import coloredlogs, logging +import logging from nrfcloud_utils import ca_certs DEV_STAGE_DICT = {'dev': '.dev.', diff --git a/src/nrfcloud_utils/nrf_cloud_onboard.py b/src/nrfcloud_utils/nrf_cloud_onboard.py index 247e234..5572685 100644 --- a/src/nrfcloud_utils/nrf_cloud_onboard.py +++ b/src/nrfcloud_utils/nrf_cloud_onboard.py @@ -10,15 +10,12 @@ import csv import time import json -import requests import os import io -import coloredlogs, logging -from os import path +import logging from os import makedirs -from ast import literal_eval from enum import Enum -from nrfcloud_utils.cli_helpers import write_file +from nrfcloud_utils.cli_helpers import write_file, setup_logging logger = logging.getLogger(__name__) @@ -86,10 +83,7 @@ def parse_args(in_args): help="bool: Plain output (no colors)", action='store_true', default=False) args = parser.parse_args(in_args) - if args.plain: - logging.basicConfig(level=args.log_level.upper()) - else: - coloredlogs.install(level=args.log_level.upper(), fmt='%(levelname)-8s %(message)s') + setup_logging(level=args.log_level, use_color=not args.plain) return args def set_dev_stage(stage = ''): @@ -353,7 +347,7 @@ def check_file_path(file): if not os.path.exists(path): try: makedirs(path, exist_ok=True) - except OSError as e: + except OSError: logger.error("Error creating file path: " + path) return '' diff --git a/tests/test_claim_and_provision_device.py b/tests/test_claim_and_provision_device.py index b945ce2..32a77cc 100644 --- a/tests/test_claim_and_provision_device.py +++ b/tests/test_claim_and_provision_device.py @@ -3,10 +3,7 @@ """ from unittest.mock import patch, Mock -from serial import Serial -import pytest from nrfcloud_utils import claim_and_provision_device -from tempfile import TemporaryDirectory from requests import Response from collections import namedtuple @@ -46,7 +43,7 @@ class TestClaimAndProvisionDevice: def test_provisioning_tags(self, ser, select_device, diap): diap.claim_device = Mock(return_value=TEST_RESPONSE) diap.can_device_be_claimed = Mock(return_value=(True, "")) - args = f"--port /not/a/real/device --cmd-type at --api-key NOTAKEY --provisioning-tags nrf-cloud-onboarding".split() + args = "--port /not/a/real/device --cmd-type at --api-key NOTAKEY --provisioning-tags nrf-cloud-onboarding".split() # call DUT claim_and_provision_device.main(args) diap.claim_device.assert_called_once() diff --git a/tests/test_create_ca_cert.py b/tests/test_create_ca_cert.py index 335a127..15ece41 100644 --- a/tests/test_create_ca_cert.py +++ b/tests/test_create_ca_cert.py @@ -2,9 +2,7 @@ Test for create_ca_cert.py """ -import pytest -from unittest.mock import Mock from nrfcloud_utils import create_ca_cert from tempfile import TemporaryDirectory from cryptography.hazmat.primitives import serialization diff --git a/tests/test_create_device_credentials.py b/tests/test_create_device_credentials.py index bbe05b9..6479c2d 100644 --- a/tests/test_create_device_credentials.py +++ b/tests/test_create_device_credentials.py @@ -2,9 +2,7 @@ Test for create_device_credentials.py """ -import pytest -from unittest.mock import Mock from nrfcloud_utils import create_device_credentials from tempfile import TemporaryDirectory from cryptography.hazmat.primitives import serialization diff --git a/tests/test_device_credentials_installer.py b/tests/test_device_credentials_installer.py index 3e44b2f..91a4c59 100644 --- a/tests/test_device_credentials_installer.py +++ b/tests/test_device_credentials_installer.py @@ -3,8 +3,6 @@ """ from unittest.mock import patch, Mock -from serial import Serial -import pytest from nrfcloud_utils import device_credentials_installer from tempfile import TemporaryDirectory import os diff --git a/tests/test_gather_attestation_tokens.py b/tests/test_gather_attestation_tokens.py index dc85517..6e8b9f6 100644 --- a/tests/test_gather_attestation_tokens.py +++ b/tests/test_gather_attestation_tokens.py @@ -3,8 +3,6 @@ """ from unittest.mock import patch, Mock -from serial import Serial -import pytest from nrfcloud_utils import gather_attestation_tokens from tempfile import TemporaryDirectory import os diff --git a/tests/test_modem_credentials_parser.py b/tests/test_modem_credentials_parser.py index 7c23fee..956bf8e 100644 --- a/tests/test_modem_credentials_parser.py +++ b/tests/test_modem_credentials_parser.py @@ -3,9 +3,7 @@ """ -import pytest -from unittest.mock import Mock from nrfcloud_utils import modem_credentials_parser KEYGEN_CSR = "MIIBCjCBrwIBADAvMS0wKwYDVQQDDCQ1MDM2MzE1NC0zOTMxLTQ0ZjAtODAyMi0xMjFiNjQwMTYyN2QwWTATBgcqhkjOPQIBBggqhkjOPQMBBwNCAAQqD6pNfa29o_EXnw62bnQWr8-JqsNh_HZxS3k3bMD4KZ8-qxnvgeoiqQ5zAycEP_Wcmzqypvwyf3qWMrZ2VB5aoB4wHAYJKoZIhvcNAQkOMQ8wDTALBgNVHQ8EBAMCA-gwDAYIKoZIzj0EAwIFAANIADBFAiEAv7OLZ_dXbszfhhjcLMUT72wTmw-z6GlgWxVhyWgR27ACIAvY_lPu3yfYZY5AL6uYTkUFp4GQkbSOUC_lsHyCxOuG.0oRDoQEmoQRBIVhL2dn3hQlQUDYxVDkxRPCAIhIbZAFifUERWCBwKj1W8FsvclMdZQgl4gBB4unZMYw0toU6uQZuXHLoDFAbhyLuHetYFWbiyxNZsnzSWEDUiTl7wwFt0hEsCiEQsxj-hCtpBk8Za8UXfdAycpx2faCOPJIrkfmiSS8-Y6_2tTAoAMN1BiWiTOimY1wZE3Ud"