feat: Add self-signed certificate generation for nRF91x1 devices

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=<sectag>,14,2` to generate a self-signed certificate and its attestation.
+6. Returns the modem to online mode (`AT+CFUN=1`).
+7. Prints `<deviceId>,<attestation>` 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-<IMEI>`), 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.

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"

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,

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()