diff --git a/ADVANCED.md b/ADVANCED.md index 56e6c82..0ee65fa 100644 --- a/ADVANCED.md +++ b/ADVANCED.md @@ -9,6 +9,7 @@ These Python scripts are designed to assist users in provisioning devices with t - [Device Credentials Installer](#device-credentials-installer) - [nRF Cloud Device Onboarding](#nrf-cloud-device-onboarding) - [nRF93M1 Device Onboarding](#nrf93m1-device-onboarding) +- [nRF91x1 Self-Signed Certificate Onboarding](#nrf91x1-self-signed-certificate-onboarding) - [Modem Credentials Parser](#modem-credentials-parser) - [Create Device Credentials](#create-device-credentials) - [Claim and Provision Device](#claim-and-provision-device) @@ -101,6 +102,49 @@ Your nRF Cloud REST API key is required and can be found on your [User Account p nrf93_onboard --port /dev/ttyACM0 --api-key $API_KEY ``` +## nRF91x1 Self-Signed Certificate Onboarding + +The `nrf91_gather_self_signed_certs` script generates a self-signed device certificate directly on an nRF91x1 device and produces an onboarding CSV row that nRF Cloud can use to register the device. Compared to the [Device Credentials Installer](#device-credentials-installer) flow, this approach does not require a local CA certificate or private key, and the device private key never leaves the modem. + +The script: + +1. Connects to the device over serial (or RTT) and verifies that the modem firmware is supported. +2. Reads the device UUID via `AT%DEVICEUUID`. +3. Switches the modem to offline mode (`AT+CFUN=4`). +4. Optionally clears the target security tag. +5. Runs `AT%KEYGEN=,14,2` to generate a self-signed certificate and its attestation. +6. Returns the modem to online mode (`AT+CFUN=1`). +7. Prints `,` to stdout and, if `--csv` is provided, appends the same pair to an onboarding CSV with headers `deviceId,selfSignedCertificateAttestation`. + +The resulting CSV is intended for upload to the Memfault side of nRF Cloud through the web frontend. **Note**: the frontend upload flow for self-signed certificate attestations is not yet released, so the CSV cannot be onboarded today. It is **not** compatible with the [`nrf_cloud_onboard`](#nrf-cloud-device-onboarding) script. + +### Limitations + +- Only supported on **nRF91x1** devices (nRF9151 / nRF9161). nRF9160 is not supported. +- Requires modem firmware **>= 2.0.2**. +- The device must be configured to use its **internal UUID** as the nRF Cloud client ID (`CONFIG_NRF_CLOUD_CLIENT_ID_SRC_INTERNAL_UUID=y`). The script emits the UUID read from `AT%DEVICEUUID` as the `deviceId`; if the device connects to nRF Cloud under a different ID (for example, `nrf-`), the onboarded entry will not match the device and the connection will be refused. +- Requires AT command support (AT Host or AT Shell). The TLS Credentials Shell mode (`--cmd-type tls_cred_shell`) is not supported, since the flow issues raw AT commands. +- If the security tag is already populated, generation fails — re-run with `-c`/`--clear-sectag` to delete the existing credentials first. + +### Examples + +#### Gather a single device, print to stdout +``` +nrf91_gather_self_signed_certs --port /dev/ttyACM0 +``` + +#### Append the result to an onboarding CSV +``` +nrf91_gather_self_signed_certs --port /dev/ttyACM0 --csv onboard.csv +``` + +Run the command again for each device to accumulate rows. Use `-o`/`--overwrite` to start a new file instead of appending, or `--keep` to preserve existing rows when a device ID is already present. + +#### Use a non-default security tag and clear it first +``` +nrf91_gather_self_signed_certs --port /dev/ttyACM0 --sectag 12345 -c +``` + ## Modem Credentials Parser The script above, `device_credentials_installer` makes use of this script, `modem_credentials_parser`, so if you use the former, you do not need to also follow the directions below. If `device_credentials_installer` does not meet your needs, you can use `modem_credentials_parser` directly to take advantage of additional options. diff --git a/pyproject.toml b/pyproject.toml index ffcab25..b896721 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -37,6 +37,7 @@ create_proxy_jwt = "nrfcloud_utils.create_proxy_jwt:run" device_credentials_installer = "nrfcloud_utils.device_credentials_installer:run" gather_attestation_tokens = "nrfcloud_utils.gather_attestation_tokens:run" modem_credentials_parser = "nrfcloud_utils.modem_credentials_parser:run" +nrf91_gather_self_signed_certs = "nrfcloud_utils.nrf91_gather_self_signed_certs:run" nrf_cloud_device_mgmt = "nrfcloud_utils.nrf_cloud_device_mgmt:run" nrf_cloud_onboard = "nrfcloud_utils.nrf_cloud_onboard:run" nrf93_onboard = "nrfcloud_utils.nrf93_onboard:run" diff --git a/src/nrfcloud_utils/cli.py b/src/nrfcloud_utils/cli.py index 6c00cf0..89cecb0 100644 --- a/src/nrfcloud_utils/cli.py +++ b/src/nrfcloud_utils/cli.py @@ -15,6 +15,7 @@ device_credentials_installer, gather_attestation_tokens, modem_credentials_parser, + nrf91_gather_self_signed_certs, nrf_cloud_device_mgmt, nrf_cloud_onboard, nrf93_onboard, @@ -29,6 +30,7 @@ "device_credentials_installer": device_credentials_installer, "gather_attestation_tokens": gather_attestation_tokens, "modem_credentials_parser": modem_credentials_parser, + "nrf91_gather_self_signed_certs": nrf91_gather_self_signed_certs, "nrf_cloud_device_mgmt": nrf_cloud_device_mgmt, "nrf_cloud_onboard": nrf_cloud_onboard, "nrf93_onboard": nrf93_onboard, diff --git a/src/nrfcloud_utils/nrf91_gather_self_signed_certs.py b/src/nrfcloud_utils/nrf91_gather_self_signed_certs.py new file mode 100644 index 0000000..7080aac --- /dev/null +++ b/src/nrfcloud_utils/nrf91_gather_self_signed_certs.py @@ -0,0 +1,253 @@ +#!/usr/bin/env python3 +# +# Copyright (c) 2026 Nordic Semiconductor ASA +# +# SPDX-License-Identifier: BSD-3-Clause + +import argparse +import csv +import logging +import os +import sys +import semver + +from nrfcloud_utils.cli_helpers import ( + setup_logging, + parser_add_comms_args, + user_request_open_mode, + CMD_TERM_DICT, CMD_TYPE_AUTO, CMD_TYPE_AT, CMD_TYPE_AT_SHELL, +) +from nrfcloud_utils.device_credentials_installer import parse_mfw_ver +from nrfcredstore.command_interface import ATCommandInterface +from nrfcredstore.comms import Comms + +logger = logging.getLogger(__name__) + +MIN_REQD_MFW_VER = "2.0.2" +DEFAULT_SECTAG = 16842753 +CSV_HEADERS = ["deviceId", "selfSignedCertificateAttestation"] +KEYGEN_TIMEOUT_S = 30 + + +def get_parser(): + parser = argparse.ArgumentParser( + description="Generate a self-signed certificate on an nRF91x1 device " + "and emit (deviceId, attestation) for nRF Cloud onboarding.", + formatter_class=argparse.ArgumentDefaultsHelpFormatter, + add_help=False, + ) + parser_add_comms_args(parser) + parser.add_argument("--csv", type=str, default="", + help="Filepath to onboarding CSV file. " + "If empty (default), only print to stdout.") + parser.add_argument("-o", "--overwrite", action="store_true", default=False, + help="When saving CSV, overwrite the file instead of appending") + parser.add_argument("--keep", action="store_true", default=False, + help="When appending: if device already exists in CSV, " + "keep old data instead of replacing") + parser.add_argument("--sectag", type=int, default=DEFAULT_SECTAG, + help="Security tag to use for the self-signed certificate") + parser.add_argument("-c", "--clear-sectag", action="store_true", default=False, + help="Clear the existing certificate and key in the " + "sectag before generating a new one. Required if " + "the slot is already populated.") + parser.add_argument("-P", "--plain", action="store_true", default=False, + help="Plain output (no colors)") + parser.add_argument("--log-level", default="info", + choices=["debug", "info", "warning", "error", "critical"], + help="Set the logging level") + return parser + + +def parse_args(in_args): + _p = get_parser() + parser = argparse.ArgumentParser(parents=[_p], description=_p.description, + formatter_class=_p.formatter_class) + args = parser.parse_args(in_args) + setup_logging(level=args.log_level, use_color=not args.plain) + return args + + +def error_exit(msg, code=1): + logger.error(msg) + sys.exit(code) + + +def check_mfw_version(cred_if): + ver = cred_if.get_mfw_version() + if not ver: + error_exit("Failed to obtain modem firmware version") + logger.info(f"Modem FW version: {ver}") + + parsed = parse_mfw_ver(ver) + if parsed is None: + error_exit(f"Could not parse modem FW version from '{ver}'") + if semver.Version.parse(parsed).compare(MIN_REQD_MFW_VER) < 0: + error_exit(f"Modem FW version must be >= {MIN_REQD_MFW_VER}, got {parsed}") + return ver + + +def get_device_uuid(cred_if): + if not cred_if.at_command("AT%DEVICEUUID", wait_for_result=False): + return None + ok, output = cred_if.comms.expect_response("OK", "ERROR", "%DEVICEUUID:") + if not ok: + return None + for line in output.split("\n"): + line = line.strip() + if line.startswith("%DEVICEUUID:"): + uuid_str = line.split(":", 1)[1].strip() + if uuid_str: + return uuid_str + return None + + +def gen_self_signed_cert(cred_if, sectag): + cmd = f"AT%KEYGEN={sectag},14,2" + if not cred_if.at_command(cmd, wait_for_result=False): + return None + ok, output = cred_if.comms.expect_response( + "OK", "ERROR", "%KEYGEN:", timeout=KEYGEN_TIMEOUT_S + ) + if not ok: + return None + for line in output.split("\n"): + line = line.strip() + if line.startswith("%KEYGEN:"): + value = line.split(":", 1)[1].strip() + return value.strip('"') + return None + + +def check_if_device_exists_in_csv(csv_filename, dev_id, delete_duplicates): + row_count = 0 + duplicate_rows = [] + keep_rows = [] if delete_duplicates else None + try: + with open(csv_filename) as f: + for row in csv.reader(f): + if not row: + continue + if row[0] == CSV_HEADERS[0]: + if delete_duplicates: + keep_rows.append(row) + continue + row_count += 1 + if row[0] == dev_id: + duplicate_rows.append(row) + elif delete_duplicates: + keep_rows.append(row) + except OSError: + logger.error(f"Error opening (read) file {csv_filename}") + return duplicate_rows, row_count + + if delete_duplicates and duplicate_rows: + try: + with open(csv_filename, "w", newline="\n") as f: + w = csv.writer(f, delimiter=",", lineterminator="\n", + quoting=csv.QUOTE_MINIMAL) + w.writerows(keep_rows) + except OSError: + logger.error(f"Error opening (write) file {csv_filename}") + + return duplicate_rows, row_count + + +def save_csv(csv_filename, append, replace, dev_id, attestation): + mode = user_request_open_mode(csv_filename, append) + if mode is None: + return + + write_header = mode == "w" or not os.path.isfile(csv_filename) + + if mode == "a" and not write_header: + duplicate_rows, _ = check_if_device_exists_in_csv(csv_filename, dev_id, replace) + if duplicate_rows: + if replace: + logger.warning(f"Removed existing row(s):\n\t{duplicate_rows}") + else: + logger.error( + f"Device {dev_id} already exists in {csv_filename}; row NOT added" + ) + return + + try: + with open(csv_filename, mode, newline="\n") as f: + w = csv.writer(f, delimiter=",", lineterminator="\n", + quoting=csv.QUOTE_MINIMAL) + if write_header: + w.writerow(CSV_HEADERS) + w.writerow([dev_id, attestation]) + logger.info(f"CSV file {csv_filename} saved") + except OSError: + logger.error(f"Error opening file {csv_filename}") + + +def main(in_args): + args = parse_args(in_args) + + if args.cmd_type not in (CMD_TYPE_AT, CMD_TYPE_AT_SHELL, CMD_TYPE_AUTO): + error_exit("Self-signed certificate generation requires AT command support") + + serial_interface = Comms( + port=args.port, + serial=args.serial_number, + baudrate=args.baud, + xonxoff=args.xonxoff, + rtscts=not args.rtscts_off, + dsrdtr=args.dsrdtr, + line_ending=CMD_TERM_DICT[args.term], + list_all=args.all, + rtt=args.rtt, + ) + + cred_if = ATCommandInterface(serial_interface) + if args.cmd_type == CMD_TYPE_AUTO: + cred_if.detect_shell_mode() + elif args.cmd_type == CMD_TYPE_AT_SHELL: + cred_if.set_shell_mode(True) + elif args.rtt: + cred_if.write_raw("at at_cmd_mode start") + + check_mfw_version(cred_if) + + logger.info("Reading device UUID...") + dev_id = get_device_uuid(cred_if) + if not dev_id: + error_exit("Failed to read device UUID") + logger.info(f"Device UUID: {dev_id}") + + logger.info("Switching modem to offline mode...") + if not cred_if.go_offline(): + error_exit("Failed to switch modem to offline mode") + + try: + if args.clear_sectag: + logger.info(f"Clearing existing credentials in sectag {args.sectag}...") + cred_if.delete_credential(args.sectag, 1) + cred_if.delete_credential(args.sectag, 2) + + logger.info(f"Generating self-signed certificate (sectag {args.sectag})...") + attestation = gen_self_signed_cert(cred_if, args.sectag) + if not attestation: + error_exit("Failed to generate self-signed certificate, use --clear-sectag if the slot is already occupied") + finally: + # Always try to return the modem to online mode, even if keygen failed, + # so the user isn't left with a modem stuck in CFUN=4. + logger.info("Returning modem to online mode...") + if not cred_if.at_command("AT+CFUN=1", wait_for_result=True): + logger.warning("Failed to return modem to online mode") + + print(f"{dev_id},{attestation}") + + if args.csv: + save_csv(args.csv, append=not args.overwrite, replace=not args.keep, + dev_id=dev_id, attestation=attestation) + + +def run(): + main(sys.argv[1:]) + + +if __name__ == "__main__": + run() diff --git a/tests/test_nrf91_gather_self_signed_certs.py b/tests/test_nrf91_gather_self_signed_certs.py new file mode 100644 index 0000000..8236dcd --- /dev/null +++ b/tests/test_nrf91_gather_self_signed_certs.py @@ -0,0 +1,427 @@ +""" +Tests for nrf91_gather_self_signed_certs.py +""" + +import os +import csv +from collections import namedtuple +from tempfile import TemporaryDirectory +from unittest.mock import patch, Mock, MagicMock + +import pytest + +from nrfcloud_utils import nrf91_gather_self_signed_certs as dut + + +TEST_UUID = "50363154-3931-44f0-8022-121b6401627d" +TEST_KEYGEN_BLOB = ( + "MIIBCzCBrwIBADAvMS0wKwYDVQQDDCQ1MDM2MzE1NC0zOTMxLTQ0ZjAtODAyMi0xMjFiNjQwMTYyN2Q" + ".0oRDoQEmoQRBIfZYQGuXwJliinHc6xDPruiyjsaXyXZbZVpUuOhHG9YS8L05VuglCcJhMN4EUhWVGpaHgNnHHno6ahi-d5tOeZmAcNY" +) +TEST_MFW_VERSION_OK = "mfw_nrf91x1_2.0.4" +TEST_MFW_VERSION_TOO_OLD = "mfw_nrf91x1_2.0.1" + + +def make_cred_if(at_command_retval=True, expect_retval=True, expect_output=""): + cred_if = Mock() + cred_if.at_command.return_value = at_command_retval + cred_if.comms.expect_response.return_value = (expect_retval, expect_output) + return cred_if + + +# --------------------------------------------------------------------------- +# get_device_uuid +# --------------------------------------------------------------------------- + +class TestGetDeviceUuid: + def test_success(self): + output = f"%DEVICEUUID: {TEST_UUID}\nOK\n" + cred_if = make_cred_if(expect_output=output) + assert dut.get_device_uuid(cred_if) == TEST_UUID + cred_if.at_command.assert_called_once_with("AT%DEVICEUUID", wait_for_result=False) + + def test_at_command_fails(self): + cred_if = make_cred_if(at_command_retval=False) + assert dut.get_device_uuid(cred_if) is None + + def test_expect_response_fails(self): + cred_if = make_cred_if(expect_retval=False) + assert dut.get_device_uuid(cred_if) is None + + def test_uuid_not_in_response(self): + cred_if = make_cred_if(expect_output="OK\n") + assert dut.get_device_uuid(cred_if) is None + + def test_empty_uuid_value_rejected(self): + cred_if = make_cred_if(expect_output="%DEVICEUUID: \nOK\n") + assert dut.get_device_uuid(cred_if) is None + + +# --------------------------------------------------------------------------- +# gen_self_signed_cert +# --------------------------------------------------------------------------- + +class TestGenSelfSignedCert: + def test_success_strips_quotes(self): + output = f'%KEYGEN: "{TEST_KEYGEN_BLOB}"\nOK\n' + cred_if = make_cred_if(expect_output=output) + result = dut.gen_self_signed_cert(cred_if, dut.DEFAULT_SECTAG) + assert result == TEST_KEYGEN_BLOB + + def test_command_uses_14_2_params(self): + # Lock in the AT%KEYGEN parameters (sectag,14,2 for self-signed cert, + # NOT sectag,2,0 which is for CSR). + cred_if = make_cred_if(expect_output=f'%KEYGEN: "{TEST_KEYGEN_BLOB}"\nOK\n') + dut.gen_self_signed_cert(cred_if, 16842753) + cred_if.at_command.assert_called_once_with( + "AT%KEYGEN=16842753,14,2", wait_for_result=False + ) + + def test_custom_sectag_in_command(self): + cred_if = make_cred_if(expect_output=f'%KEYGEN: "{TEST_KEYGEN_BLOB}"\nOK\n') + dut.gen_self_signed_cert(cred_if, 12345) + assert cred_if.at_command.call_args[0][0] == "AT%KEYGEN=12345,14,2" + + def test_at_command_fails(self): + cred_if = make_cred_if(at_command_retval=False) + assert dut.gen_self_signed_cert(cred_if, dut.DEFAULT_SECTAG) is None + + def test_expect_response_fails(self): + cred_if = make_cred_if(expect_retval=False) + assert dut.gen_self_signed_cert(cred_if, dut.DEFAULT_SECTAG) is None + + def test_keygen_not_in_response(self): + cred_if = make_cred_if(expect_output="OK\n") + assert dut.gen_self_signed_cert(cred_if, dut.DEFAULT_SECTAG) is None + + def test_uses_long_timeout(self): + cred_if = make_cred_if(expect_output=f'%KEYGEN: "{TEST_KEYGEN_BLOB}"\nOK\n') + dut.gen_self_signed_cert(cred_if, dut.DEFAULT_SECTAG) + kwargs = cred_if.comms.expect_response.call_args.kwargs + assert kwargs.get("timeout", 0) >= dut.KEYGEN_TIMEOUT_S + + +# --------------------------------------------------------------------------- +# check_mfw_version +# --------------------------------------------------------------------------- + +class TestCheckMfwVersion: + def test_accepts_minimum_version(self): + cred_if = Mock() + cred_if.get_mfw_version.return_value = "mfw_nrf91x1_2.0.2" + assert dut.check_mfw_version(cred_if) == "mfw_nrf91x1_2.0.2" + + def test_accepts_newer_version(self): + cred_if = Mock() + cred_if.get_mfw_version.return_value = TEST_MFW_VERSION_OK + assert dut.check_mfw_version(cred_if) == TEST_MFW_VERSION_OK + + def test_rejects_older_version(self): + cred_if = Mock() + cred_if.get_mfw_version.return_value = TEST_MFW_VERSION_TOO_OLD + with pytest.raises(SystemExit): + dut.check_mfw_version(cred_if) + + def test_rejects_legacy_1_3(self): + cred_if = Mock() + cred_if.get_mfw_version.return_value = "mfw_nrf9160_1.3.5" + with pytest.raises(SystemExit): + dut.check_mfw_version(cred_if) + + def test_no_version_returned_exits(self): + cred_if = Mock() + cred_if.get_mfw_version.return_value = None + with pytest.raises(SystemExit): + dut.check_mfw_version(cred_if) + + def test_unparseable_version_exits(self): + cred_if = Mock() + cred_if.get_mfw_version.return_value = "garbage" + with pytest.raises(SystemExit): + dut.check_mfw_version(cred_if) + + +# --------------------------------------------------------------------------- +# CSV helpers +# --------------------------------------------------------------------------- + +class TestSaveCsv: + def test_writes_header_and_row_when_creating(self): + with TemporaryDirectory() as tmp: + path = os.path.join(tmp, "out.csv") + dut.save_csv(path, append=True, replace=False, + dev_id=TEST_UUID, attestation=TEST_KEYGEN_BLOB) + with open(path) as f: + rows = list(csv.reader(f)) + assert rows[0] == dut.CSV_HEADERS + assert rows[1] == [TEST_UUID, TEST_KEYGEN_BLOB] + + def test_overwrite_replaces_file(self): + with TemporaryDirectory() as tmp: + path = os.path.join(tmp, "out.csv") + with open(path, "w") as f: + f.write("garbage\n") + with patch("builtins.input", return_value="y"): + dut.save_csv(path, append=False, replace=False, + dev_id=TEST_UUID, attestation=TEST_KEYGEN_BLOB) + with open(path) as f: + rows = list(csv.reader(f)) + assert rows[0] == dut.CSV_HEADERS + assert rows[1] == [TEST_UUID, TEST_KEYGEN_BLOB] + assert len(rows) == 2 + + def test_overwrite_quit_leaves_file_untouched(self): + with TemporaryDirectory() as tmp: + path = os.path.join(tmp, "out.csv") + with open(path, "w") as f: + f.write("garbage\n") + with patch("builtins.input", return_value="n"): + dut.save_csv(path, append=False, replace=False, + dev_id=TEST_UUID, attestation=TEST_KEYGEN_BLOB) + with open(path) as f: + assert f.read() == "garbage\n" + + def test_append_adds_row_without_duplicate_header(self): + with TemporaryDirectory() as tmp: + path = os.path.join(tmp, "out.csv") + dut.save_csv(path, append=True, replace=False, + dev_id="aaa", attestation="blob-aaa") + dut.save_csv(path, append=True, replace=False, + dev_id="bbb", attestation="blob-bbb") + with open(path) as f: + rows = list(csv.reader(f)) + assert rows == [ + dut.CSV_HEADERS, + ["aaa", "blob-aaa"], + ["bbb", "blob-bbb"], + ] + + def test_append_duplicate_device_with_replace_overwrites_row(self): + with TemporaryDirectory() as tmp: + path = os.path.join(tmp, "out.csv") + dut.save_csv(path, append=True, replace=False, + dev_id="aaa", attestation="blob-old") + dut.save_csv(path, append=True, replace=True, + dev_id="aaa", attestation="blob-new") + with open(path) as f: + rows = list(csv.reader(f)) + # Header + the freshly added row only + assert rows[0] == dut.CSV_HEADERS + assert ["aaa", "blob-new"] in rows + assert ["aaa", "blob-old"] not in rows + + def test_append_duplicate_device_without_replace_skips(self): + with TemporaryDirectory() as tmp: + path = os.path.join(tmp, "out.csv") + dut.save_csv(path, append=True, replace=False, + dev_id="aaa", attestation="blob-old") + dut.save_csv(path, append=True, replace=False, + dev_id="aaa", attestation="blob-new") + with open(path) as f: + rows = list(csv.reader(f)) + assert ["aaa", "blob-old"] in rows + assert ["aaa", "blob-new"] not in rows + + +# --------------------------------------------------------------------------- +# get_parser argument defaults +# --------------------------------------------------------------------------- + +class TestParser: + def test_default_csv_is_empty(self): + args = dut.get_parser().parse_args(["--port", "/dev/null"]) + assert args.csv == "" + + def test_default_sectag(self): + args = dut.get_parser().parse_args(["--port", "/dev/null"]) + assert args.sectag == dut.DEFAULT_SECTAG + + def test_custom_sectag(self): + args = dut.get_parser().parse_args(["--port", "/dev/null", "--sectag", "42"]) + assert args.sectag == 42 + + +# --------------------------------------------------------------------------- +# main() — high-level flow with mocks +# --------------------------------------------------------------------------- + +MODULE = "nrfcloud_utils.nrf91_gather_self_signed_certs" + + +def _base_main_patches(): + return { + "Comms": MagicMock(), + "ATCommandInterface": MagicMock(), + "check_mfw_version": Mock(return_value=TEST_MFW_VERSION_OK), + "get_device_uuid": Mock(return_value=TEST_UUID), + "gen_self_signed_cert": Mock(return_value=TEST_KEYGEN_BLOB), + } + + +class TestMain: + def _run(self, patches, extra_args=""): + args = f"--port /dev/ttyACM0 --cmd-type at {extra_args}".strip().split() + with patch.multiple(MODULE, **patches): + patches["ATCommandInterface"].return_value.at_command.return_value = True + patches["ATCommandInterface"].return_value.go_offline.return_value = True + dut.main(args) + + def test_default_run_prints_csv_row(self, capsys): + patches = _base_main_patches() + self._run(patches) + captured = capsys.readouterr() + assert f"{TEST_UUID},{TEST_KEYGEN_BLOB}" in captured.out + + def test_uuid_failure_exits(self): + patches = _base_main_patches() + patches["get_device_uuid"] = Mock(return_value=None) + with pytest.raises(SystemExit): + self._run(patches) + + def test_offline_failure_exits(self): + patches = _base_main_patches() + with patch.multiple(MODULE, **patches): + patches["ATCommandInterface"].return_value.at_command.return_value = True + patches["ATCommandInterface"].return_value.go_offline.return_value = False + with pytest.raises(SystemExit): + dut.main("--port /dev/ttyACM0 --cmd-type at".split()) + + def test_keygen_failure_exits(self): + patches = _base_main_patches() + patches["gen_self_signed_cert"] = Mock(return_value=None) + with pytest.raises(SystemExit): + self._run(patches) + + def test_csv_flag_writes_file(self): + with TemporaryDirectory() as tmp: + path = os.path.join(tmp, "out.csv") + self._run(_base_main_patches(), extra_args=f"--csv {path}") + assert os.path.exists(path) + with open(path) as f: + rows = list(csv.reader(f)) + assert rows[0] == dut.CSV_HEADERS + assert rows[1] == [TEST_UUID, TEST_KEYGEN_BLOB] + + def test_no_csv_means_no_file(self): + with TemporaryDirectory() as tmp: + self._run(_base_main_patches()) + # No file created; the directory stays empty + assert os.listdir(tmp) == [] + + def test_brings_modem_back_online(self): + patches = _base_main_patches() + with patch.multiple(MODULE, **patches): + ati = patches["ATCommandInterface"].return_value + ati.at_command.return_value = True + ati.go_offline.return_value = True + dut.main("--port /dev/ttyACM0 --cmd-type at".split()) + calls = [c.args[0] for c in ati.at_command.call_args_list] + assert "AT+CFUN=1" in calls + + def test_modem_returned_online_even_when_keygen_fails(self): + patches = _base_main_patches() + patches["gen_self_signed_cert"] = Mock(return_value=None) + with patch.multiple(MODULE, **patches): + ati = patches["ATCommandInterface"].return_value + ati.at_command.return_value = True + ati.go_offline.return_value = True + with pytest.raises(SystemExit): + dut.main("--port /dev/ttyACM0 --cmd-type at".split()) + # Even though keygen failed and the script exited, AT+CFUN=1 + # must still have been issued so the modem isn't stuck offline. + calls = [c.args[0] for c in ati.at_command.call_args_list] + assert "AT+CFUN=1" in calls + + def test_default_does_not_clear_sectag(self): + patches = _base_main_patches() + with patch.multiple(MODULE, **patches): + ati = patches["ATCommandInterface"].return_value + ati.at_command.return_value = True + ati.go_offline.return_value = True + dut.main("--port /dev/ttyACM0 --cmd-type at".split()) + ati.delete_credential.assert_not_called() + + def test_clear_sectag_flag_clears_cert_and_key(self): + patches = _base_main_patches() + with patch.multiple(MODULE, **patches): + ati = patches["ATCommandInterface"].return_value + ati.at_command.return_value = True + ati.go_offline.return_value = True + dut.main("--port /dev/ttyACM0 --cmd-type at --sectag 99 --clear-sectag".split()) + ati.delete_credential.assert_any_call(99, 1) + ati.delete_credential.assert_any_call(99, 2) + + def test_clear_sectag_short_flag(self): + patches = _base_main_patches() + with patch.multiple(MODULE, **patches): + ati = patches["ATCommandInterface"].return_value + ati.at_command.return_value = True + ati.go_offline.return_value = True + dut.main("--port /dev/ttyACM0 --cmd-type at -c".split()) + ati.delete_credential.assert_any_call(dut.DEFAULT_SECTAG, 1) + ati.delete_credential.assert_any_call(dut.DEFAULT_SECTAG, 2) + + +# --------------------------------------------------------------------------- +# Integration test with FakeSerial (mirrors test_gather_attestation_tokens.py) +# --------------------------------------------------------------------------- + +class FakeSerial(Mock): + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self.response = [] + + def write(self, data): + cmd = data.decode("utf-8").strip() + if cmd == "AT+CGMR": + self.response = [b"OK\r\n", b"mfw_nrf91x1_2.0.4\r\n"] + elif cmd == "AT%DEVICEUUID": + self.response = [b"OK\r\n", f"%DEVICEUUID: {TEST_UUID}\r\n".encode()] + elif cmd.startswith("AT%KEYGEN="): + self.response = [ + b"OK\r\n", + f'%KEYGEN: "{TEST_KEYGEN_BLOB}"\r\n'.encode(), + ] + elif cmd.startswith("AT%CMNG="): + self.response = [b"OK\r\n"] + elif cmd in ("AT+CFUN=4", "AT+CFUN=1"): + self.response = [b"OK\r\n"] + elif cmd == "": + self.response = [b"OK\r\n"] + else: + self.response = [b"ERROR\r\n"] + + def readline(self): + if not self.response: + return b"" + return self.response.pop() + + +FakeSerialPort = namedtuple("FakeSerialPort", ["device"]) + + +class TestIntegration: + @patch("nrfcredstore.comms.select_device", + return_value=(FakeSerialPort("/not/a/real/device"), "TEST_DEVICE")) + @patch("nrfcredstore.comms.serial.Serial", return_value=FakeSerial()) + def test_end_to_end_with_csv(self, ser, select_device, capsys): + with TemporaryDirectory() as tmp: + csv_file = os.path.join(tmp, "certs.csv") + args = f"--port /not/a/real/device --cmd-type at --csv {csv_file}".split() + dut.main(args) + assert os.path.exists(csv_file) + with open(csv_file) as f: + rows = list(csv.reader(f)) + assert rows[0] == dut.CSV_HEADERS + assert rows[1] == [TEST_UUID, TEST_KEYGEN_BLOB] + captured = capsys.readouterr() + assert f"{TEST_UUID},{TEST_KEYGEN_BLOB}" in captured.out + + @patch("nrfcredstore.comms.select_device", + return_value=(FakeSerialPort("/not/a/real/device"), "TEST_DEVICE")) + @patch("nrfcredstore.comms.serial.Serial", return_value=FakeSerial()) + def test_end_to_end_stdout_only(self, ser, select_device, capsys): + args = "--port /not/a/real/device --cmd-type at".split() + dut.main(args) + captured = capsys.readouterr() + assert f"{TEST_UUID},{TEST_KEYGEN_BLOB}" in captured.out