diff --git a/pynitrokey/cli/lpcutils.py b/pynitrokey/cli/lpcutils.py new file mode 100644 index 00000000..f98126fc --- /dev/null +++ b/pynitrokey/cli/lpcutils.py @@ -0,0 +1,149 @@ +import tempfile +from pathlib import Path +from typing import Any + +from cryptography.hazmat.primitives import hashes +from cryptography.hazmat.primitives.asymmetric import ec +from spsdk.crypto.signature_provider import SignatureProvider +from spsdk.image.mbi.mbi import MasterBootImage +from spsdk.sbfile.sb2.images import BootImageV21 +from spsdk.sbfile.sb2.sly_bd_parser import BDParser +from spsdk.utils.config import Config +from spsdk.utils.family import FamilyRevision +from spsdk.utils.misc import write_file + +# The following private key is dummy used to initialize the MBI class +dummy_priv_key = """ +-----BEGIN PRIVATE KEY----- +MIGHAgEAMBMGByqGSM49AgEGCCqGSM49AwEHBG0wawIBAQQgxpxN6kHTCoCRMGIR +4H4B58AHx5gjn3MnDgSa3qEF7kShRANCAATlvf3y1hlRnVdYcLE1UfBKSoEbwrWn +kjsIL8fJEMWfrcTY1Mlnz1eQ12F4xHIGG2sN014rXyUK+DlA8hLsJF34 +-----END PRIVATE KEY----- +""" + + +class EcSignatureProvider(SignatureProvider): + identifier = "ec" + + def __init__(self, key: ec.EllipticCurvePrivateKey) -> None: + self.key = key + + def sign(self, data: bytes) -> bytes: + return self.key.sign(data, ec.ECDSA(hashes.SHA256())) + + @property + def signature_length(self) -> int: + return (self.key.key_size + 7) // 8 * 2 + + +def _clean_tempfile(tfile: tempfile._TemporaryFileWrapper) -> None: # type: ignore + Path(tfile.name).unlink(missing_ok=True) + + +def mbi_export(cert_path: str, binary: str, signer: ec.EllipticCurvePrivateKey) -> bytes: + family = "lpc55s6x" + cert_block_yaml = f''' +family: {family} +imageBuildNumber: 0 + +rootCertificate0File: "{cert_path}/nk-firmware-root-cert.der" +rootCertificate1File: "{cert_path}/nk-firmware-ee2-cert.der" +rootCertificate2File: "{cert_path}/nk-firmware-ee3-cert.der" +rootCertificate3File: "{cert_path}/nk-firmware-ee4-cert.der" + +mainRootCertId: 0 + +chainCertificate0File0: "{cert_path}/nk-firmware-ee1-cert.der" +''' + + cert_block_file = tempfile.NamedTemporaryFile(suffix=".yaml", mode="w+t", delete=False) + cert_block_file.write(cert_block_yaml) + cert_block_file.close() + dummy_priv_file = tempfile.NamedTemporaryFile(suffix=".pem", mode="w+t", delete=False) + dummy_priv_file.write(dummy_priv_key) + dummy_priv_file.close() + + config_dict = { + "family": family, + "outputImageExecutionTarget": "Internal Flash (XIP)", + "outputImageAuthenticationType": "Signed", + "inputImageFile": binary, + "enableTrustZone": True, + "certBlock": cert_block_file.name, + "signer": dummy_priv_file.name, + } + + config = Config(config_dict) + familyrev = FamilyRevision.load_from_config(config) + mbi_cls = MasterBootImage.get_mbi_class(config)(family=familyrev) + for base in mbi_cls._get_mixins(): + base.mix_load_from_config(mbi_cls, config) # type: ignore + new_provider = EcSignatureProvider(signer) + mbi_cls.signature_provider = new_provider # type: ignore + mbi_data = mbi_cls.export_image() + _clean_tempfile(cert_block_file) + _clean_tempfile(dummy_priv_file) + + return mbi_data.export() + + +def _get_config_sb2(command_path: str, external_files: list[str]) -> Config: + family = "lpc55s6x" + with open(command_path, "r") as f: + content = f.read().replace("\t", " ") + + parser = BDParser() + parsed_conf = Config(parser.parse(content, extern=external_files)) + + assert "options" in parsed_conf + parsed_conf["options"]["family"] = family + options: dict[str, Any] = parsed_conf["options"] + parsed_conf["family"] = options.pop("family") + parsed_conf["revision"] = options.pop("revision", "latest") + return parsed_conf + + +def sb21_export( + parsed_config: Config, + key: str, + pkey: ec.EllipticCurvePrivateKey, + cert_path: str, + hash_of_hashes: str, +) -> bytes: + cert = [f"{cert_path}/nk-firmware-root-cert.der", f"{cert_path}/nk-firmware-ee1-cert.der"] + + root_key_cert = [ + f"{cert_path}/nk-firmware-root-cert.der", + f"{cert_path}/nk-firmware-ee2-cert.der", + f"{cert_path}/nk-firmware-ee3-cert.der", + f"{cert_path}/nk-firmware-ee4-cert.der", + ] + signature_provider = EcSignatureProvider(pkey) + sb2 = BootImageV21.load_from_config( + config=parsed_config, + key_file_path=key, + signature_provider=signature_provider, + signing_certificate_file_paths=cert, + root_key_certificate_paths=root_key_cert, + rkth_out_path=hash_of_hashes, + ) + return sb2.export() + + +def lpc55_sign_sb2( + cert_path: str, + binary_path: str, + commands: str, + key: str, + rkth: str, + out_file: str, + signer: ec.EllipticCurvePrivateKey, +) -> None: + mbi = mbi_export(cert_path, binary_path, signer) + signed_file = tempfile.NamedTemporaryFile(suffix=".bin", mode="w+b", delete=False) + signed_file.write(mbi) + signed_file.close() + config = _get_config_sb2(commands, [signed_file.name]) + sb21_file = sb21_export(config, key, signer, cert_path, rkth) + _clean_tempfile(signed_file) + write_file(sb21_file, out_file, mode="wb") diff --git a/pynitrokey/cli/nethsm_pvtkey.py b/pynitrokey/cli/nethsm_pvtkey.py new file mode 100644 index 00000000..81704f5f --- /dev/null +++ b/pynitrokey/cli/nethsm_pvtkey.py @@ -0,0 +1,119 @@ +import hashlib +import os +from typing import Any + +import nethsm as nethsm_sdk +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.asymmetric import ec +from nethsm import Base64, NetHSM + +from pynitrokey.cli.nethsm import Config + +try: + pass +except Exception: + print("Failed to import cryptography, cannot do signing") + + +class NetHSMKey(ec.EllipticCurvePrivateKey): + _sk: str + _public_key: ec.EllipticCurvePublicKey + _nethsm_config: Config + + def __init__(self, config: Config, sk: str) -> None: + if config.host is None: + v = "NETHSM_HOST" + if v not in os.environ: + raise AssertionError( + f"Missing NetHSM host: set the --host option or the {v} environment variable" + ) + config.host = os.environ.get(v) + self._nethsm_config = config + if sk: + self._load_key(sk) + + def _connect_nethsm(self) -> NetHSM: + config = self._nethsm_config + auth = None + if config.username and config.password: + auth = nethsm_sdk.Authentication(username=config.username, password=config.password) + assert config.host, "Host undefined" + nethsm = NetHSM( + config.host, auth=auth, verify_tls=config.verify_tls, ca_certs=config.ca_certs + ) + try: + return nethsm + except nethsm_sdk.NetHSMError as e: + raise AssertionError(f"NetHSM request failed: {e}") + except nethsm_sdk.NetHSMRequestError as e: + if e.type == nethsm_sdk.RequestErrorType.SSL_ERROR: + raise AssertionError( + f"Could not connect to the NetHSM: {e.reason}\nIf you use a self-signed certificate, please set the --no-verify-tls option." + ) + else: + raise AssertionError( + f"Cound not connect to the NetHSM: {e.reason}\nIs the NetHSM running and reachable?" + ) + + def _load_key(self, key_id: str) -> bool: + sk = key_id + client = self._connect_nethsm() + keys_list = client.list_keys(prefix=sk) + client.close() + if sk in keys_list: + self._sk = sk + pem_data = client.get_key_public_key(self._sk) + pub_temp = serialization.load_pem_public_key(pem_data.encode()) + assert isinstance(pub_temp, ec.EllipticCurvePublicKey) + self._public_key = pub_temp + + return False # Not using default key of nrfutil + + raise AssertionError(f"Key {sk} not found in the HSM") + + def sign(self, data: bytes, signature_algorithm: ec.EllipticCurveSignatureAlgorithm) -> bytes: + if self._sk is None: + raise AssertionError("Can't sign. No key created/loaded") + assert isinstance(signature_algorithm, ec.ECDSA) + assert isinstance(signature_algorithm.algorithm, hashes.SHA256) + + hash_data = hashlib.sha256(data).digest() + to_data = Base64.encode(hash_data) + + client = self._connect_nethsm() + der_signature = client.sign( + key_id=self._sk, data=to_data, mode=nethsm_sdk.SignMode.ECDSA + ).decode() + client.close() + return der_signature + + def exchange(self, algorithm: ec.ECDH, peer_public_key: ec.EllipticCurvePublicKey) -> bytes: + raise NotImplementedError() + + def public_key(self) -> ec.EllipticCurvePublicKey: + return self._public_key + + @property + def curve(self) -> ec.EllipticCurve: + return self._public_key.curve + + def private_numbers(self) -> ec.EllipticCurvePrivateNumbers: + raise NotImplementedError() + + @property + def key_size(self) -> int: + return self._public_key.key_size + + def private_bytes( + self, + encoding: serialization.Encoding, + format: serialization.PrivateFormat, + encryption_algorithm: serialization.KeySerializationEncryption, + ) -> bytes: + raise NotImplementedError() + + def __copy__(self) -> "NetHSMKey": + raise NotImplementedError() + + def __deepcopy__(self, memo: dict[Any, Any]) -> "NetHSMKey": + raise NotImplementedError() diff --git a/pynitrokey/cli/nrfutils.py b/pynitrokey/cli/nrfutils.py new file mode 100644 index 00000000..58f5ce31 --- /dev/null +++ b/pynitrokey/cli/nrfutils.py @@ -0,0 +1,238 @@ +import logging +import os +import re +from functools import wraps +from typing import Any, Callable, Optional + +import click +from cryptography.hazmat.primitives.asymmetric import ec +from nitrokey.trussed._bootloader.nrf52_upload.dfu.nrfutils import pkg_gen, pubview, usb_serial +from nitrokey.trussed._bootloader.nrf52_upload.dfu.signing import Signing + +from pynitrokey.cli.exceptions import CliException +from pynitrokey.cli.lpcutils import lpc55_sign_sb2 +from pynitrokey.cli.nethsm import Config +from pynitrokey.cli.nethsm_pvtkey import NetHSMKey +from pynitrokey.helpers import local_critical + +logger = logging.getLogger(__name__) + + +def get_pvt_key( + use_nethsm: bool, + nethsm_host: Optional[str] = None, + nethsm_username: Optional[str] = None, + nethsm_password: Optional[str] = None, + verify_tls: bool = True, + ca_certs: Optional[str] = None, + key_id: str = "", +) -> ec.EllipticCurvePrivateKey: + if not use_nethsm: + return Signing.get_key_from_file(key_id) + + config = Config( + host=nethsm_host, + username=nethsm_username, + password=nethsm_password, + verify_tls=verify_tls, + ca_certs=ca_certs, + debug=False, + ) + return NetHSMKey(config, key_id) + + +_AnyCallable = Callable[..., Any] + + +def _extract_key_id(text: str) -> str: # Extract from NetHSM output + # Pattern explanation: + # ^Key\s+ Starts with "Key" followed by spaces + # ([a-fA-F0-9]+) Captures the hex Key ID (Group 1) + # \s+generated on NetHSM\s+ Matches middle label + # \S+$ Matches the host URL at the end + pattern = r"^Key\s+([a-fA-F0-9]+)\s+generated on NetHSM\s+\S+$" + match = re.search(pattern, text.strip()) + if match: + return match.group(1) + return text + + +def with_signer(f: _AnyCallable) -> _AnyCallable: + """Decorator that adds signer options and injects the signer object.""" + + @click.option( + "--key", required=True, help="Key file for file from disk, key id for using NetHSM" + ) + @click.option("--use-nethsm", is_flag=True, default=False, help="Use NetHSM for signing.") + @click.option( + "--use-key-file", + is_flag=True, + default=False, + help="Use if --key parameter has a file pointing to NetHSM key id.", + ) + @click.option( + "--nethsm-host", + default=None, + help="NetHSM host address. Leave empty if environment variable is set", + ) + @click.option("--nethsm-username", default=None, help="NetHSM username.") + @click.option("--nethsm-password", default=None, help="NetHSM password.") + @click.option( + "--verify-tls", is_flag=True, default=False, help="Enable TLS certificate verification." + ) + @click.option("--ca-certs", default=None, help="CA Certificates.") + @wraps(f) + def wrapper( + key: str, + use_nethsm: bool, + use_key_file: bool, + nethsm_host: Optional[str] = None, + nethsm_username: Optional[str] = None, + nethsm_password: Optional[str] = None, + verify_tls: bool = True, + ca_certs: Optional[str] = None, + *args: Any, + **kwargs: Any, + ) -> Any: + if use_key_file: + with open(key, "r") as fl: + key = _extract_key_id(fl.read()) + signer = get_pvt_key( + use_nethsm=use_nethsm, + nethsm_host=nethsm_host, + nethsm_username=nethsm_username, + nethsm_password=nethsm_password, + verify_tls=verify_tls, + ca_certs=ca_certs, + key_id=key, + ) + return f(*args, signer=signer, **kwargs) + + return wrapper + + +@click.group() +def nrf() -> None: + """Nordic nRF52 DFU/signing utilities.""" + + +@nrf.group() +def keys() -> None: + """Key generation and inspection.""" + + +@keys.command("display") +@click.option("--format", "fmt", required=True, type=click.Choice(["code", "pem"])) +@click.option("--out-file", "out_file", required=True) +@with_signer +def keys_display(fmt: str, out_file: str, signer: ec.EllipticCurvePrivateKey) -> None: + """Display/export the public key derived from KEY_FILE in the given format.""" + pubview(fmt, signer, out_file) + + +@nrf.group() +def dfu() -> None: + """Perform a DFU transfer.""" + + +@dfu.command("usb-serial") +@click.option("--package", "-pkg", required=True) +@click.option("--port", "-p", required=True) +def dfu_usb_serial(package: str, port: str) -> None: + """Send a DFU package over a USB serial port.""" + usb_serial(package, port) + + +@nrf.group() +def pkg() -> None: + """Generate DFU packages.""" + + +@pkg.command("generate") +@click.option("--hw-version", required=True, type=int) +@click.option("--sd-req", required=True) +@click.option("--application-version", "app_version", type=int, default=None) +@click.option("--bootloader-version", type=int, default=None) +@click.option("--application", "application", default=None) +@click.option("--bootloader", "bootloader", default=None) +@click.option( + "--app-boot-validation", + "ecdsa_validation", + is_flag=True, + default=False, + help="Set to enable VALIDATE_ECDSA_P256_SHA256.", +) +@click.argument("out_path") +@with_signer +def pkg_generate( + hw_version: int, + sd_req: str, + app_version: Optional[int], + bootloader_version: Optional[int], + application: Optional[str], + bootloader: Optional[str], + ecdsa_validation: bool, + out_path: str, + signer: ec.EllipticCurvePrivateKey, +) -> None: + """Generate an application and/or bootloader DFU package.""" + pkg_gen( + hw_version=hw_version, + sd_req=sd_req, + key_file=signer, + out_path=out_path, + app_version=app_version, + bootloader_version=bootloader_version, + application=application, + bootloader=bootloader, + ecdsa_validation=ecdsa_validation, + ) + + +@click.group() +def lpc55() -> None: + """Generate LPC55 Package.""" + + +@lpc55.command("generate") +@click.option("--cert-path", required=True, type=str) +@click.option("--binary-path", required=True, type=str) +@click.option("--commands", required=True, type=str, help="Path to BD file.") +@click.option("--sbkek", required=True, type=str, help="Path to sbkek file.") +@click.option("--rkth", required=True, type=str, help="Path to rkth output file.") +@click.argument("out_path") +@with_signer +def lpc55_generate( + cert_path: str, + binary_path: str, + commands: str, + sbkek: str, + rkth: str, + out_path: str, + signer: ec.EllipticCurvePrivateKey, +) -> None: + """Generate a signed SB2 file from unsigned BIN file for LPC55""" + lpc55_sign_sb2(cert_path, binary_path, commands, sbkek, rkth, out_path, signer) + + +def cmd_main(func: Callable[[], None]) -> None: + development = os.environ.get("NKDEV") + try: + func() + except CliException as e: + if development: + raise + e.show() + except Exception as e: + if development: + raise + logger.warning("An unhandled exception occurred", exc_info=True) + local_critical("An unhandled exception occurred", e) + + +def nrf_main() -> None: + cmd_main(nrf) + + +def lpc_main() -> None: + cmd_main(lpc55) diff --git a/pyproject.toml b/pyproject.toml index 6ce38265..2ce773e4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -42,12 +42,15 @@ dependencies = [ [project.optional-dependencies] pcsc = ["pyscard >=2, <3"] +spsdk = ["spsdk >=3, <4"] [project.urls] repository = "https://github.com/Nitrokey/pynitrokey" [project.scripts] nitropy = "pynitrokey.cli:main" +nitropy-nrf = "pynitrokey.cli.nrfutils:nrf_main" +nitropy-lpc = "pynitrokey.cli.nrfutils:lpc_main" [tool.mypy] mypy_path = "stubs"