Skip to content
Open
149 changes: 149 additions & 0 deletions pynitrokey/cli/lpcutils.py
Original file line number Diff line number Diff line change
@@ -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")
119 changes: 119 additions & 0 deletions pynitrokey/cli/nethsm_pvtkey.py
Original file line number Diff line number Diff line change
@@ -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()
Loading