From 25f35c7d0e27af4308b8c263fde6661dbe29c2cc Mon Sep 17 00:00:00 2001 From: Jean-François Milants Date: Tue, 26 Jan 2021 20:31:45 +0100 Subject: Generate pinetime-recovery : a light version of InfiniTime design to be used as a recovery firmware : it only provides basic UI and BLE connectivity for OTA. This new FW is build on the same codebasse than the actual InfiniTime. Only the display task is different (this allows to remove lvgl from the recovery fw, which is very heavy). CMake builds and docker have been modified accordingly. Note than the fw is converted into an image and then into a DFU in the cmake build (previously, it was only done in the --- tools/mcuboot/imgtool/__init__.py | 15 + tools/mcuboot/imgtool/boot_record.py | 47 +++ tools/mcuboot/imgtool/image.py | 552 +++++++++++++++++++++++++++++ tools/mcuboot/imgtool/keys/__init__.py | 94 +++++ tools/mcuboot/imgtool/keys/ecdsa.py | 157 ++++++++ tools/mcuboot/imgtool/keys/ecdsa_test.py | 99 ++++++ tools/mcuboot/imgtool/keys/ed25519.py | 105 ++++++ tools/mcuboot/imgtool/keys/ed25519_test.py | 103 ++++++ tools/mcuboot/imgtool/keys/general.py | 45 +++ tools/mcuboot/imgtool/keys/rsa.py | 163 +++++++++ tools/mcuboot/imgtool/keys/rsa_test.py | 115 ++++++ tools/mcuboot/imgtool/keys/x25519.py | 107 ++++++ tools/mcuboot/imgtool/main.py | 352 ++++++++++++++++++ tools/mcuboot/imgtool/version.py | 53 +++ 14 files changed, 2007 insertions(+) create mode 100644 tools/mcuboot/imgtool/__init__.py create mode 100644 tools/mcuboot/imgtool/boot_record.py create mode 100644 tools/mcuboot/imgtool/image.py create mode 100644 tools/mcuboot/imgtool/keys/__init__.py create mode 100644 tools/mcuboot/imgtool/keys/ecdsa.py create mode 100644 tools/mcuboot/imgtool/keys/ecdsa_test.py create mode 100644 tools/mcuboot/imgtool/keys/ed25519.py create mode 100644 tools/mcuboot/imgtool/keys/ed25519_test.py create mode 100644 tools/mcuboot/imgtool/keys/general.py create mode 100644 tools/mcuboot/imgtool/keys/rsa.py create mode 100644 tools/mcuboot/imgtool/keys/rsa_test.py create mode 100644 tools/mcuboot/imgtool/keys/x25519.py create mode 100644 tools/mcuboot/imgtool/main.py create mode 100644 tools/mcuboot/imgtool/version.py (limited to 'tools/mcuboot/imgtool') diff --git a/tools/mcuboot/imgtool/__init__.py b/tools/mcuboot/imgtool/__init__.py new file mode 100644 index 00000000..c0c3ef21 --- /dev/null +++ b/tools/mcuboot/imgtool/__init__.py @@ -0,0 +1,15 @@ +# Copyright 2017 Linaro Limited +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +imgtool_version = "1.6.0rc2" diff --git a/tools/mcuboot/imgtool/boot_record.py b/tools/mcuboot/imgtool/boot_record.py new file mode 100644 index 00000000..4112b225 --- /dev/null +++ b/tools/mcuboot/imgtool/boot_record.py @@ -0,0 +1,47 @@ +# Copyright (c) 2019, Arm Limited. +# Copyright (c) 2020, Linaro Limited +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from enum import Enum +import cbor + + +class SwComponent(int, Enum): + """ + Software component property IDs specified by + Arm's PSA Attestation API 1.0 document. + """ + TYPE = 1 + MEASUREMENT_VALUE = 2 + VERSION = 4 + SIGNER_ID = 5 + MEASUREMENT_DESCRIPTION = 6 + + +def create_sw_component_data(sw_type, sw_version, sw_measurement_description, + sw_measurement_value, sw_signer_id): + + # List of software component properties (Key ID + value) + properties = { + SwComponent.TYPE: sw_type, + SwComponent.VERSION: sw_version, + SwComponent.SIGNER_ID: sw_signer_id, + SwComponent.MEASUREMENT_DESCRIPTION: sw_measurement_description, + } + + # Note: The measurement value must be the last item of the property + # list because later it will be modified by the bootloader. + properties[SwComponent.MEASUREMENT_VALUE] = sw_measurement_value + + return cbor.dumps(properties) diff --git a/tools/mcuboot/imgtool/image.py b/tools/mcuboot/imgtool/image.py new file mode 100644 index 00000000..291134d7 --- /dev/null +++ b/tools/mcuboot/imgtool/image.py @@ -0,0 +1,552 @@ +# Copyright 2018 Nordic Semiconductor ASA +# Copyright 2017 Linaro Limited +# Copyright 2019-2020 Arm Limited +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +""" +Image signing and management. +""" + +from . import version as versmod +from .boot_record import create_sw_component_data +import click +from enum import Enum +from intelhex import IntelHex +import hashlib +import struct +import os.path +from .keys import rsa, ecdsa, x25519 +from cryptography.hazmat.primitives.asymmetric import ec, padding +from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey +from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes +from cryptography.hazmat.primitives.kdf.hkdf import HKDF +from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat +from cryptography.hazmat.backends import default_backend +from cryptography.hazmat.primitives import hashes, hmac +from cryptography.exceptions import InvalidSignature + +IMAGE_MAGIC = 0x96f3b83d +IMAGE_HEADER_SIZE = 32 +BIN_EXT = "bin" +INTEL_HEX_EXT = "hex" +DEFAULT_MAX_SECTORS = 128 +MAX_ALIGN = 8 +DEP_IMAGES_KEY = "images" +DEP_VERSIONS_KEY = "versions" +MAX_SW_TYPE_LENGTH = 12 # Bytes + +# Image header flags. +IMAGE_F = { + 'PIC': 0x0000001, + 'NON_BOOTABLE': 0x0000010, + 'RAM_LOAD': 0x0000020, + 'ENCRYPTED': 0x0000004, +} + +TLV_VALUES = { + 'KEYHASH': 0x01, + 'PUBKEY': 0x02, + 'SHA256': 0x10, + 'RSA2048': 0x20, + 'ECDSA224': 0x21, + 'ECDSA256': 0x22, + 'RSA3072': 0x23, + 'ED25519': 0x24, + 'ENCRSA2048': 0x30, + 'ENCKW128': 0x31, + 'ENCEC256': 0x32, + 'ENCX25519': 0x33, + 'DEPENDENCY': 0x40, + 'SEC_CNT': 0x50, + 'BOOT_RECORD': 0x60, +} + +TLV_SIZE = 4 +TLV_INFO_SIZE = 4 +TLV_INFO_MAGIC = 0x6907 +TLV_PROT_INFO_MAGIC = 0x6908 + +boot_magic = bytes([ + 0x77, 0xc2, 0x95, 0xf3, + 0x60, 0xd2, 0xef, 0x7f, + 0x35, 0x52, 0x50, 0x0f, + 0x2c, 0xb6, 0x79, 0x80, ]) + +STRUCT_ENDIAN_DICT = { + 'little': '<', + 'big': '>' +} + +VerifyResult = Enum('VerifyResult', + """ + OK INVALID_MAGIC INVALID_TLV_INFO_MAGIC INVALID_HASH + INVALID_SIGNATURE + """) + + +class TLV(): + def __init__(self, endian, magic=TLV_INFO_MAGIC): + self.magic = magic + self.buf = bytearray() + self.endian = endian + + def __len__(self): + return TLV_INFO_SIZE + len(self.buf) + + def add(self, kind, payload): + """ + Add a TLV record. Kind should be a string found in TLV_VALUES above. + """ + e = STRUCT_ENDIAN_DICT[self.endian] + buf = struct.pack(e + 'BBH', TLV_VALUES[kind], 0, len(payload)) + self.buf += buf + self.buf += payload + + def get(self): + if len(self.buf) == 0: + return bytes() + e = STRUCT_ENDIAN_DICT[self.endian] + header = struct.pack(e + 'HH', self.magic, len(self)) + return header + bytes(self.buf) + + +class Image(): + + def __init__(self, version=None, header_size=IMAGE_HEADER_SIZE, + pad_header=False, pad=False, confirm=False, align=1, + slot_size=0, max_sectors=DEFAULT_MAX_SECTORS, + overwrite_only=False, endian="little", load_addr=0, + erased_val=None, save_enctlv=False, security_counter=None): + self.version = version or versmod.decode_version("0") + self.header_size = header_size + self.pad_header = pad_header + self.pad = pad + self.confirm = confirm + self.align = align + self.slot_size = slot_size + self.max_sectors = max_sectors + self.overwrite_only = overwrite_only + self.endian = endian + self.base_addr = None + self.load_addr = 0 if load_addr is None else load_addr + self.erased_val = 0xff if erased_val is None else int(erased_val, 0) + self.payload = [] + self.enckey = None + self.save_enctlv = save_enctlv + self.enctlv_len = 0 + + if security_counter == 'auto': + # Security counter has not been explicitly provided, + # generate it from the version number + self.security_counter = ((self.version.major << 24) + + (self.version.minor << 16) + + self.version.revision) + else: + self.security_counter = security_counter + + def __repr__(self): + return "".format( + self.version, + self.header_size, + self.security_counter, + self.base_addr if self.base_addr is not None else "N/A", + self.load_addr, + self.align, + self.slot_size, + self.max_sectors, + self.overwrite_only, + self.endian, + self.__class__.__name__, + len(self.payload)) + + def load(self, path): + """Load an image from a given file""" + ext = os.path.splitext(path)[1][1:].lower() + try: + if ext == INTEL_HEX_EXT: + ih = IntelHex(path) + self.payload = ih.tobinarray() + self.base_addr = ih.minaddr() + else: + with open(path, 'rb') as f: + self.payload = f.read() + except FileNotFoundError: + raise click.UsageError("Input file not found") + + # Add the image header if needed. + if self.pad_header and self.header_size > 0: + if self.base_addr: + # Adjust base_addr for new header + self.base_addr -= self.header_size + self.payload = bytes([self.erased_val] * self.header_size) + \ + self.payload + + self.check_header() + + def save(self, path, hex_addr=None): + """Save an image from a given file""" + ext = os.path.splitext(path)[1][1:].lower() + if ext == INTEL_HEX_EXT: + # input was in binary format, but HEX needs to know the base addr + if self.base_addr is None and hex_addr is None: + raise click.UsageError("No address exists in input file " + "neither was it provided by user") + h = IntelHex() + if hex_addr is not None: + self.base_addr = hex_addr + h.frombytes(bytes=self.payload, offset=self.base_addr) + if self.pad: + trailer_size = self._trailer_size(self.align, self.max_sectors, + self.overwrite_only, + self.enckey, + self.save_enctlv, + self.enctlv_len) + trailer_addr = (self.base_addr + self.slot_size) - trailer_size + padding = bytes([self.erased_val] * + (trailer_size - len(boot_magic))) + boot_magic + h.puts(trailer_addr, padding) + h.tofile(path, 'hex') + else: + if self.pad: + self.pad_to(self.slot_size) + with open(path, 'wb') as f: + f.write(self.payload) + + def check_header(self): + if self.header_size > 0 and not self.pad_header: + if any(v != 0 for v in self.payload[0:self.header_size]): + raise click.UsageError("Header padding was not requested and " + "image does not start with zeros") + + def check_trailer(self): + if self.slot_size > 0: + tsize = self._trailer_size(self.align, self.max_sectors, + self.overwrite_only, self.enckey, + self.save_enctlv, self.enctlv_len) + padding = self.slot_size - (len(self.payload) + tsize) + if padding < 0: + msg = "Image size (0x{:x}) + trailer (0x{:x}) exceeds " \ + "requested size 0x{:x}".format( + len(self.payload), tsize, self.slot_size) + raise click.UsageError(msg) + + def ecies_hkdf(self, enckey, plainkey): + if isinstance(enckey, ecdsa.ECDSA256P1Public): + newpk = ec.generate_private_key(ec.SECP256R1(), default_backend()) + shared = newpk.exchange(ec.ECDH(), enckey._get_public()) + else: + newpk = X25519PrivateKey.generate() + shared = newpk.exchange(enckey._get_public()) + derived_key = HKDF( + algorithm=hashes.SHA256(), length=48, salt=None, + info=b'MCUBoot_ECIES_v1', backend=default_backend()).derive(shared) + encryptor = Cipher(algorithms.AES(derived_key[:16]), + modes.CTR(bytes([0] * 16)), + backend=default_backend()).encryptor() + cipherkey = encryptor.update(plainkey) + encryptor.finalize() + mac = hmac.HMAC(derived_key[16:], hashes.SHA256(), + backend=default_backend()) + mac.update(cipherkey) + ciphermac = mac.finalize() + if isinstance(enckey, ecdsa.ECDSA256P1Public): + pubk = newpk.public_key().public_bytes( + encoding=Encoding.X962, + format=PublicFormat.UncompressedPoint) + else: + pubk = newpk.public_key().public_bytes( + encoding=Encoding.Raw, + format=PublicFormat.Raw) + return cipherkey, ciphermac, pubk + + def create(self, key, public_key_format, enckey, dependencies=None, + sw_type=None): + self.enckey = enckey + + # Calculate the hash of the public key + if key is not None: + pub = key.get_public_bytes() + sha = hashlib.sha256() + sha.update(pub) + pubbytes = sha.digest() + else: + pubbytes = bytes(hashlib.sha256().digest_size) + + protected_tlv_size = 0 + + if self.security_counter is not None: + # Size of the security counter TLV: header ('HH') + payload ('I') + # = 4 + 4 = 8 Bytes + protected_tlv_size += TLV_SIZE + 4 + + if sw_type is not None: + if len(sw_type) > MAX_SW_TYPE_LENGTH: + msg = "'{}' is too long ({} characters) for sw_type. Its " \ + "maximum allowed length is 12 characters.".format( + sw_type, len(sw_type)) + raise click.UsageError(msg) + + image_version = (str(self.version.major) + '.' + + str(self.version.minor) + '.' + + str(self.version.revision)) + + # The image hash is computed over the image header, the image + # itself and the protected TLV area. However, the boot record TLV + # (which is part of the protected area) should contain this hash + # before it is even calculated. For this reason the script fills + # this field with zeros and the bootloader will insert the right + # value later. + digest = bytes(hashlib.sha256().digest_size) + + # Create CBOR encoded boot record + boot_record = create_sw_component_data(sw_type, image_version, + "SHA256", digest, + pubbytes) + + protected_tlv_size += TLV_SIZE + len(boot_record) + + if dependencies is not None: + # Size of a Dependency TLV = Header ('HH') + Payload('IBBHI') + # = 4 + 12 = 16 Bytes + dependencies_num = len(dependencies[DEP_IMAGES_KEY]) + protected_tlv_size += (dependencies_num * 16) + + if protected_tlv_size != 0: + # Add the size of the TLV info header + protected_tlv_size += TLV_INFO_SIZE + + # At this point the image is already on the payload, this adds + # the header to the payload as well + self.add_header(enckey, protected_tlv_size) + + prot_tlv = TLV(self.endian, TLV_PROT_INFO_MAGIC) + + # Protected TLVs must be added first, because they are also included + # in the hash calculation + protected_tlv_off = None + if protected_tlv_size != 0: + + e = STRUCT_ENDIAN_DICT[self.endian] + + if self.security_counter is not None: + payload = struct.pack(e + 'I', self.security_counter) + prot_tlv.add('SEC_CNT', payload) + + if sw_type is not None: + prot_tlv.add('BOOT_RECORD', boot_record) + + if dependencies is not None: + for i in range(dependencies_num): + payload = struct.pack( + e + 'B3x'+'BBHI', + int(dependencies[DEP_IMAGES_KEY][i]), + dependencies[DEP_VERSIONS_KEY][i].major, + dependencies[DEP_VERSIONS_KEY][i].minor, + dependencies[DEP_VERSIONS_KEY][i].revision, + dependencies[DEP_VERSIONS_KEY][i].build + ) + prot_tlv.add('DEPENDENCY', payload) + + protected_tlv_off = len(self.payload) + self.payload += prot_tlv.get() + + tlv = TLV(self.endian) + + # Note that ecdsa wants to do the hashing itself, which means + # we get to hash it twice. + sha = hashlib.sha256() + sha.update(self.payload) + digest = sha.digest() + + tlv.add('SHA256', digest) + + if key is not None: + if public_key_format == 'hash': + tlv.add('KEYHASH', pubbytes) + else: + tlv.add('PUBKEY', pub) + + # `sign` expects the full image payload (sha256 done internally), + # while `sign_digest` expects only the digest of the payload + + if hasattr(key, 'sign'): + sig = key.sign(bytes(self.payload)) + else: + sig = key.sign_digest(digest) + tlv.add(key.sig_tlv(), sig) + + # At this point the image was hashed + signed, we can remove the + # protected TLVs from the payload (will be re-added later) + if protected_tlv_off is not None: + self.payload = self.payload[:protected_tlv_off] + + if enckey is not None: + plainkey = os.urandom(16) + + if isinstance(enckey, rsa.RSAPublic): + cipherkey = enckey._get_public().encrypt( + plainkey, padding.OAEP( + mgf=padding.MGF1(algorithm=hashes.SHA256()), + algorithm=hashes.SHA256(), + label=None)) + self.enctlv_len = len(cipherkey) + tlv.add('ENCRSA2048', cipherkey) + elif isinstance(enckey, (ecdsa.ECDSA256P1Public, + x25519.X25519Public)): + cipherkey, mac, pubk = self.ecies_hkdf(enckey, plainkey) + enctlv = pubk + mac + cipherkey + self.enctlv_len = len(enctlv) + if isinstance(enckey, ecdsa.ECDSA256P1Public): + tlv.add('ENCEC256', enctlv) + else: + tlv.add('ENCX25519', enctlv) + + nonce = bytes([0] * 16) + cipher = Cipher(algorithms.AES(plainkey), modes.CTR(nonce), + backend=default_backend()) + encryptor = cipher.encryptor() + img = bytes(self.payload[self.header_size:]) + self.payload[self.header_size:] = \ + encryptor.update(img) + encryptor.finalize() + + self.payload += prot_tlv.get() + self.payload += tlv.get() + + self.check_trailer() + + def add_header(self, enckey, protected_tlv_size): + """Install the image header.""" + + flags = 0 + if enckey is not None: + flags |= IMAGE_F['ENCRYPTED'] + if self.load_addr != 0: + # Indicates that this image should be loaded into RAM + # instead of run directly from flash. + flags |= IMAGE_F['RAM_LOAD'] + + e = STRUCT_ENDIAN_DICT[self.endian] + fmt = (e + + # type ImageHdr struct { + 'I' + # Magic uint32 + 'I' + # LoadAddr uint32 + 'H' + # HdrSz uint16 + 'H' + # PTLVSz uint16 + 'I' + # ImgSz uint32 + 'I' + # Flags uint32 + 'BBHI' + # Vers ImageVersion + 'I' # Pad1 uint32 + ) # } + assert struct.calcsize(fmt) == IMAGE_HEADER_SIZE + header = struct.pack(fmt, + IMAGE_MAGIC, + self.load_addr, + self.header_size, + protected_tlv_size, # TLV Info header + Protected TLVs + len(self.payload) - self.header_size, # ImageSz + flags, + self.version.major, + self.version.minor or 0, + self.version.revision or 0, + self.version.build or 0, + 0) # Pad1 + self.payload = bytearray(self.payload) + self.payload[:len(header)] = header + + def _trailer_size(self, write_size, max_sectors, overwrite_only, enckey, + save_enctlv, enctlv_len): + # NOTE: should already be checked by the argument parser + magic_size = 16 + if overwrite_only: + return MAX_ALIGN * 2 + magic_size + else: + if write_size not in set([1, 2, 4, 8]): + raise click.BadParameter("Invalid alignment: {}".format( + write_size)) + m = DEFAULT_MAX_SECTORS if max_sectors is None else max_sectors + trailer = m * 3 * write_size # status area + if enckey is not None: + if save_enctlv: + # TLV saved by the bootloader is aligned + keylen = (int((enctlv_len - 1) / MAX_ALIGN) + 1) * MAX_ALIGN + else: + keylen = 16 + trailer += keylen * 2 # encryption keys + trailer += MAX_ALIGN * 4 # image_ok/copy_done/swap_info/swap_size + trailer += magic_size + return trailer + + def pad_to(self, size): + """Pad the image to the given size, with the given flash alignment.""" + tsize = self._trailer_size(self.align, self.max_sectors, + self.overwrite_only, self.enckey, + self.save_enctlv, self.enctlv_len) + padding = size - (len(self.payload) + tsize) + pbytes = bytearray([self.erased_val] * padding) + pbytes += bytearray([self.erased_val] * (tsize - len(boot_magic))) + if self.confirm and not self.overwrite_only: + pbytes[-MAX_ALIGN] = 0x01 # image_ok = 0x01 + pbytes += boot_magic + self.payload += pbytes + + @staticmethod + def verify(imgfile, key): + with open(imgfile, "rb") as f: + b = f.read() + + magic, _, header_size, _, img_size = struct.unpack('IIHHI', b[:16]) + version = struct.unpack('BBHI', b[20:28]) + + if magic != IMAGE_MAGIC: + return VerifyResult.INVALID_MAGIC, None + + tlv_info = b[header_size+img_size:header_size+img_size+TLV_INFO_SIZE] + magic, tlv_tot = struct.unpack('HH', tlv_info) + if magic != TLV_INFO_MAGIC: + return VerifyResult.INVALID_TLV_INFO_MAGIC, None + + sha = hashlib.sha256() + sha.update(b[:header_size+img_size]) + digest = sha.digest() + + tlv_off = header_size + img_size + tlv_end = tlv_off + tlv_tot + tlv_off += TLV_INFO_SIZE # skip tlv info + while tlv_off < tlv_end: + tlv = b[tlv_off:tlv_off+TLV_SIZE] + tlv_type, _, tlv_len = struct.unpack('BBH', tlv) + if tlv_type == TLV_VALUES["SHA256"]: + off = tlv_off + TLV_SIZE + if digest == b[off:off+tlv_len]: + if key is None: + return VerifyResult.OK, version + else: + return VerifyResult.INVALID_HASH, None + elif key is not None and tlv_type == TLV_VALUES[key.sig_tlv()]: + off = tlv_off + TLV_SIZE + tlv_sig = b[off:off+tlv_len] + payload = b[:header_size+img_size] + try: + if hasattr(key, 'verify'): + key.verify(tlv_sig, payload) + else: + key.verify_digest(tlv_sig, digest) + return VerifyResult.OK, version + except InvalidSignature: + # continue to next TLV + pass + tlv_off += TLV_SIZE + tlv_len + return VerifyResult.INVALID_SIGNATURE, None diff --git a/tools/mcuboot/imgtool/keys/__init__.py b/tools/mcuboot/imgtool/keys/__init__.py new file mode 100644 index 00000000..f25e2aae --- /dev/null +++ b/tools/mcuboot/imgtool/keys/__init__.py @@ -0,0 +1,94 @@ +# Copyright 2017 Linaro Limited +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +""" +Cryptographic key management for imgtool. +""" + +from cryptography.hazmat.backends import default_backend +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric.rsa import ( + RSAPrivateKey, RSAPublicKey) +from cryptography.hazmat.primitives.asymmetric.ec import ( + EllipticCurvePrivateKey, EllipticCurvePublicKey) +from cryptography.hazmat.primitives.asymmetric.ed25519 import ( + Ed25519PrivateKey, Ed25519PublicKey) +from cryptography.hazmat.primitives.asymmetric.x25519 import ( + X25519PrivateKey, X25519PublicKey) + +from .rsa import RSA, RSAPublic, RSAUsageError, RSA_KEY_SIZES +from .ecdsa import ECDSA256P1, ECDSA256P1Public, ECDSAUsageError +from .ed25519 import Ed25519, Ed25519Public, Ed25519UsageError +from .x25519 import X25519, X25519Public, X25519UsageError + + +class PasswordRequired(Exception): + """Raised to indicate that the key is password protected, but a + password was not specified.""" + pass + + +def load(path, passwd=None): + """Try loading a key from the given path. Returns None if the password wasn't specified.""" + with open(path, 'rb') as f: + raw_pem = f.read() + try: + pk = serialization.load_pem_private_key( + raw_pem, + password=passwd, + backend=default_backend()) + # Unfortunately, the crypto library raises unhelpful exceptions, + # so we have to look at the text. + except TypeError as e: + msg = str(e) + if "private key is encrypted" in msg: + return None + raise e + except ValueError: + # This seems to happen if the key is a public key, let's try + # loading it as a public key. + pk = serialization.load_pem_public_key( + raw_pem, + backend=default_backend()) + + if isinstance(pk, RSAPrivateKey): + if pk.key_size not in RSA_KEY_SIZES: + raise Exception("Unsupported RSA key size: " + pk.key_size) + return RSA(pk) + elif isinstance(pk, RSAPublicKey): + if pk.key_size not in RSA_KEY_SIZES: + raise Exception("Unsupported RSA key size: " + pk.key_size) + return RSAPublic(pk) + elif isinstance(pk, EllipticCurvePrivateKey): + if pk.curve.name != 'secp256r1': + raise Exception("Unsupported EC curve: " + pk.curve.name) + if pk.key_size != 256: + raise Exception("Unsupported EC size: " + pk.key_size) + return ECDSA256P1(pk) + elif isinstance(pk, EllipticCurvePublicKey): + if pk.curve.name != 'secp256r1': + raise Exception("Unsupported EC curve: " + pk.curve.name) + if pk.key_size != 256: + raise Exception("Unsupported EC size: " + pk.key_size) + return ECDSA256P1Public(pk) + elif isinstance(pk, Ed25519PrivateKey): + return Ed25519(pk) + elif isinstance(pk, Ed25519PublicKey): + return Ed25519Public(pk) + elif isinstance(pk, X25519PrivateKey): + return X25519(pk) + elif isinstance(pk, X25519PublicKey): + return X25519Public(pk) + else: + raise Exception("Unknown key type: " + str(type(pk))) diff --git a/tools/mcuboot/imgtool/keys/ecdsa.py b/tools/mcuboot/imgtool/keys/ecdsa.py new file mode 100644 index 00000000..139d583d --- /dev/null +++ b/tools/mcuboot/imgtool/keys/ecdsa.py @@ -0,0 +1,157 @@ +""" +ECDSA key management +""" + +from cryptography.hazmat.backends import default_backend +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric import ec +from cryptography.hazmat.primitives.hashes import SHA256 + +from .general import KeyClass + +class ECDSAUsageError(Exception): + pass + +class ECDSA256P1Public(KeyClass): + def __init__(self, key): + self.key = key + + def shortname(self): + return "ecdsa" + + def _unsupported(self, name): + raise ECDSAUsageError("Operation {} requires private key".format(name)) + + def _get_public(self): + return self.key + + def get_public_bytes(self): + # The key is embedded into MBUboot in "SubjectPublicKeyInfo" format + return self._get_public().public_bytes( + encoding=serialization.Encoding.DER, + format=serialization.PublicFormat.SubjectPublicKeyInfo) + + def get_private_bytes(self, minimal): + self._unsupported('get_private_bytes') + + def export_private(self, path, passwd=None): + self._unsupported('export_private') + + def export_public(self, path): + """Write the public key to the given file.""" + pem = self._get_public().public_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PublicFormat.SubjectPublicKeyInfo) + with open(path, 'wb') as f: + f.write(pem) + + def sig_type(self): + return "ECDSA256_SHA256" + + def sig_tlv(self): + return "ECDSA256" + + def sig_len(self): + # Early versions of MCUboot (< v1.5.0) required ECDSA + # signatures to be padded to 72 bytes. Because the DER + # encoding is done with signed integers, the size of the + # signature will vary depending on whether the high bit is set + # in each value. This padding was done in a + # not-easily-reversible way (by just adding zeros). + # + # The signing code no longer requires this padding, and newer + # versions of MCUboot don't require it. But, continue to + # return the total length so that the padding can be done if + # requested. + return 72 + + def verify(self, signature, payload): + # strip possible paddings added during sign + signature = signature[:signature[1] + 2] + k = self.key + if isinstance(self.key, ec.EllipticCurvePrivateKey): + k = self.key.public_key() + return k.verify(signature=signature, data=payload, + signature_algorithm=ec.ECDSA(SHA256())) + + +class ECDSA256P1(ECDSA256P1Public): + """ + Wrapper around an ECDSA private key. + """ + + def __init__(self, key): + """key should be an instance of EllipticCurvePrivateKey""" + self.key = key + self.pad_sig = False + + @staticmethod + def generate(): + pk = ec.generate_private_key( + ec.SECP256R1(), + backend=default_backend()) + return ECDSA256P1(pk) + + def _get_public(self): + return self.key.public_key() + + def _build_minimal_ecdsa_privkey(self, der): + ''' + Builds a new DER that only includes the EC private key, removing the + public key that is added as an "optional" BITSTRING. + ''' + offset_PUB = 68 + EXCEPTION_TEXT = "Error parsing ecdsa key. Please submit an issue!" + if der[offset_PUB] != 0xa1: + raise ECDSAUsageError(EXCEPTION_TEXT) + len_PUB = der[offset_PUB + 1] + b = bytearray(der[:-offset_PUB]) + offset_SEQ = 29 + if b[offset_SEQ] != 0x30: + raise ECDSAUsageError(EXCEPTION_TEXT) + b[offset_SEQ + 1] -= len_PUB + offset_OCT_STR = 27 + if b[offset_OCT_STR] != 0x04: + raise ECDSAUsageError(EXCEPTION_TEXT) + b[offset_OCT_STR + 1] -= len_PUB + if b[0] != 0x30 or b[1] != 0x81: + raise ECDSAUsageError(EXCEPTION_TEXT) + b[2] -= len_PUB + return b + + def get_private_bytes(self, minimal): + priv = self.key.private_bytes( + encoding=serialization.Encoding.DER, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=serialization.NoEncryption()) + if minimal: + priv = self._build_minimal_ecdsa_privkey(priv) + return priv + + def export_private(self, path, passwd=None): + """Write the private key to the given file, protecting it with the optional password.""" + if passwd is None: + enc = serialization.NoEncryption() + else: + enc = serialization.BestAvailableEncryption(passwd) + pem = self.key.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=enc) + with open(path, 'wb') as f: + f.write(pem) + + def raw_sign(self, payload): + """Return the actual signature""" + return self.key.sign( + data=payload, + signature_algorithm=ec.ECDSA(SHA256())) + + def sign(self, payload): + sig = self.raw_sign(payload) + if self.pad_sig: + # To make fixed length, pad with one or two zeros. + sig += b'\000' * (self.sig_len() - len(sig)) + return sig + else: + return sig diff --git a/tools/mcuboot/imgtool/keys/ecdsa_test.py b/tools/mcuboot/imgtool/keys/ecdsa_test.py new file mode 100644 index 00000000..7982cad9 --- /dev/null +++ b/tools/mcuboot/imgtool/keys/ecdsa_test.py @@ -0,0 +1,99 @@ +""" +Tests for ECDSA keys +""" + +import io +import os.path +import sys +import tempfile +import unittest + +from cryptography.exceptions import InvalidSignature +from cryptography.hazmat.primitives.asymmetric import ec +from cryptography.hazmat.primitives.hashes import SHA256 + +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '../..'))) + +from imgtool.keys import load, ECDSA256P1, ECDSAUsageError + +class EcKeyGeneration(unittest.TestCase): + + def setUp(self): + self.test_dir = tempfile.TemporaryDirectory() + + def tname(self, base): + return os.path.join(self.test_dir.name, base) + + def tearDown(self): + self.test_dir.cleanup() + + def test_keygen(self): + name1 = self.tname("keygen.pem") + k = ECDSA256P1.generate() + k.export_private(name1, b'secret') + + self.assertIsNone(load(name1)) + + k2 = load(name1, b'secret') + + pubname = self.tname('keygen-pub.pem') + k2.export_public(pubname) + pk2 = load(pubname) + + # We should be able to export the public key from the loaded + # public key, but not the private key. + pk2.export_public(self.tname('keygen-pub2.pem')) + self.assertRaises(ECDSAUsageError, + pk2.export_private, self.tname('keygen-priv2.pem')) + + def test_emit(self): + """Basic sanity check on the code emitters.""" + k = ECDSA256P1.generate() + + ccode = io.StringIO() + k.emit_c_public(ccode) + self.assertIn("ecdsa_pub_key", ccode.getvalue()) + self.assertIn("ecdsa_pub_key_len", ccode.getvalue()) + + rustcode = io.StringIO() + k.emit_rust_public(rustcode) + self.assertIn("ECDSA_PUB_KEY", rustcode.getvalue()) + + def test_emit_pub(self): + """Basic sanity check on the code emitters.""" + pubname = self.tname("public.pem") + k = ECDSA256P1.generate() + k.export_public(pubname) + + k2 = load(pubname) + + ccode = io.StringIO() + k2.emit_c_public(ccode) + self.assertIn("ecdsa_pub_key", ccode.getvalue()) + self.assertIn("ecdsa_pub_key_len", ccode.getvalue()) + + rustcode = io.StringIO() + k2.emit_rust_public(rustcode) + self.assertIn("ECDSA_PUB_KEY", rustcode.getvalue()) + + def test_sig(self): + k = ECDSA256P1.generate() + buf = b'This is the message' + sig = k.raw_sign(buf) + + # The code doesn't have any verification, so verify this + # manually. + k.key.public_key().verify( + signature=sig, + data=buf, + signature_algorithm=ec.ECDSA(SHA256())) + + # Modify the message to make sure the signature fails. + self.assertRaises(InvalidSignature, + k.key.public_key().verify, + signature=sig, + data=b'This is thE message', + signature_algorithm=ec.ECDSA(SHA256())) + +if __name__ == '__main__': + unittest.main() diff --git a/tools/mcuboot/imgtool/keys/ed25519.py b/tools/mcuboot/imgtool/keys/ed25519.py new file mode 100644 index 00000000..fb000cd9 --- /dev/null +++ b/tools/mcuboot/imgtool/keys/ed25519.py @@ -0,0 +1,105 @@ +""" +ED25519 key management +""" + +from cryptography.hazmat.backends import default_backend +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric import ed25519 + +from .general import KeyClass + + +class Ed25519UsageError(Exception): + pass + + +class Ed25519Public(KeyClass): + def __init__(self, key): + self.key = key + + def shortname(self): + return "ed25519" + + def _unsupported(self, name): + raise Ed25519UsageError("Operation {} requires private key".format(name)) + + def _get_public(self): + return self.key + + def get_public_bytes(self): + # The key is embedded into MBUboot in "SubjectPublicKeyInfo" format + return self._get_public().public_bytes( + encoding=serialization.Encoding.DER, + format=serialization.PublicFormat.SubjectPublicKeyInfo) + + def get_private_bytes(self, minimal): + self._unsupported('get_private_bytes') + + def export_private(self, path, passwd=None): + self._unsupported('export_private') + + def export_public(self, path): + """Write the public key to the given file.""" + pem = self._get_public().public_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PublicFormat.SubjectPublicKeyInfo) + with open(path, 'wb') as f: + f.write(pem) + + def sig_type(self): + return "ED25519" + + def sig_tlv(self): + return "ED25519" + + def sig_len(self): + return 64 + + +class Ed25519(Ed25519Public): + """ + Wrapper around an ED25519 private key. + """ + + def __init__(self, key): + """key should be an instance of EllipticCurvePrivateKey""" + self.key = key + + @staticmethod + def generate(): + pk = ed25519.Ed25519PrivateKey.generate() + return Ed25519(pk) + + def _get_public(self): + return self.key.public_key() + + def get_private_bytes(self, minimal): + raise Ed25519UsageError("Operation not supported with {} keys".format( + self.shortname())) + + def export_private(self, path, passwd=None): + """ + Write the private key to the given file, protecting it with the + optional password. + """ + if passwd is None: + enc = serialization.NoEncryption() + else: + enc = serialization.BestAvailableEncryption(passwd) + pem = self.key.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=enc) + with open(path, 'wb') as f: + f.write(pem) + + def sign_digest(self, digest): + """Return the actual signature""" + return self.key.sign(data=digest) + + def verify_digest(self, signature, digest): + """Verify that signature is valid for given digest""" + k = self.key + if isinstance(self.key, ed25519.Ed25519PrivateKey): + k = self.key.public_key() + return k.verify(signature=signature, data=digest) diff --git a/tools/mcuboot/imgtool/keys/ed25519_test.py b/tools/mcuboot/imgtool/keys/ed25519_test.py new file mode 100644 index 00000000..31f43fe9 --- /dev/null +++ b/tools/mcuboot/imgtool/keys/ed25519_test.py @@ -0,0 +1,103 @@ +""" +Tests for ECDSA keys +""" + +import hashlib +import io +import os.path +import sys +import tempfile +import unittest + +from cryptography.exceptions import InvalidSignature +from cryptography.hazmat.primitives.asymmetric import ed25519 + +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '../..'))) + +from imgtool.keys import load, Ed25519, Ed25519UsageError + + +class Ed25519KeyGeneration(unittest.TestCase): + + def setUp(self): + self.test_dir = tempfile.TemporaryDirectory() + + def tname(self, base): + return os.path.join(self.test_dir.name, base) + + def tearDown(self): + self.test_dir.cleanup() + + def test_keygen(self): + name1 = self.tname("keygen.pem") + k = Ed25519.generate() + k.export_private(name1, b'secret') + + self.assertIsNone(load(name1)) + + k2 = load(name1, b'secret') + + pubname = self.tname('keygen-pub.pem') + k2.export_public(pubname) + pk2 = load(pubname) + + # We should be able to export the public key from the loaded + # public key, but not the private key. + pk2.export_public(self.tname('keygen-pub2.pem')) + self.assertRaises(Ed25519UsageError, + pk2.export_private, self.tname('keygen-priv2.pem')) + + def test_emit(self): + """Basic sanity check on the code emitters.""" + k = Ed25519.generate() + + ccode = io.StringIO() + k.emit_c_public(ccode) + self.assertIn("ed25519_pub_key", ccode.getvalue()) + self.assertIn("ed25519_pub_key_len", ccode.getvalue()) + + rustcode = io.StringIO() + k.emit_rust_public(rustcode) + self.assertIn("ED25519_PUB_KEY", rustcode.getvalue()) + + def test_emit_pub(self): + """Basic sanity check on the code emitters.""" + pubname = self.tname("public.pem") + k = Ed25519.generate() + k.export_public(pubname) + + k2 = load(pubname) + + ccode = io.StringIO() + k2.emit_c_public(ccode) + self.assertIn("ed25519_pub_key", ccode.getvalue()) + self.assertIn("ed25519_pub_key_len", ccode.getvalue()) + + rustcode = io.StringIO() + k2.emit_rust_public(rustcode) + self.assertIn("ED25519_PUB_KEY", rustcode.getvalue()) + + def test_sig(self): + k = Ed25519.generate() + buf = b'This is the message' + sha = hashlib.sha256() + sha.update(buf) + digest = sha.digest() + sig = k.sign_digest(digest) + + # The code doesn't have any verification, so verify this + # manually. + k.key.public_key().verify(signature=sig, data=digest) + + # Modify the message to make sure the signature fails. + sha = hashlib.sha256() + sha.update(b'This is thE message') + new_digest = sha.digest() + self.assertRaises(InvalidSignature, + k.key.public_key().verify, + signature=sig, + data=new_digest) + + +if __name__ == '__main__': + unittest.main() diff --git a/tools/mcuboot/imgtool/keys/general.py b/tools/mcuboot/imgtool/keys/general.py new file mode 100644 index 00000000..ce7a2d26 --- /dev/null +++ b/tools/mcuboot/imgtool/keys/general.py @@ -0,0 +1,45 @@ +"""General key class.""" + +import sys + +AUTOGEN_MESSAGE = "/* Autogenerated by imgtool.py, do not edit. */" + +class KeyClass(object): + def _emit(self, header, trailer, encoded_bytes, indent, file=sys.stdout, len_format=None): + print(AUTOGEN_MESSAGE, file=file) + print(header, end='', file=file) + for count, b in enumerate(encoded_bytes): + if count % 8 == 0: + print("\n" + indent, end='', file=file) + else: + print(" ", end='', file=file) + print("0x{:02x},".format(b), end='', file=file) + print("\n" + trailer, file=file) + if len_format is not None: + print(len_format.format(len(encoded_bytes)), file=file) + + def emit_c_public(self, file=sys.stdout): + self._emit( + header="const unsigned char {}_pub_key[] = {{".format(self.shortname()), + trailer="};", + encoded_bytes=self.get_public_bytes(), + indent=" ", + len_format="const unsigned int {}_pub_key_len = {{}};".format(self.shortname()), + file=file) + + def emit_rust_public(self, file=sys.stdout): + self._emit( + header="static {}_PUB_KEY: &'static [u8] = &[".format(self.shortname().upper()), + trailer="];", + encoded_bytes=self.get_public_bytes(), + indent=" ", + file=file) + + def emit_private(self, minimal, file=sys.stdout): + self._emit( + header="const unsigned char enc_priv_key[] = {", + trailer="};", + encoded_bytes=self.get_private_bytes(minimal), + indent=" ", + len_format="const unsigned int enc_priv_key_len = {};", + file=file) diff --git a/tools/mcuboot/imgtool/keys/rsa.py b/tools/mcuboot/imgtool/keys/rsa.py new file mode 100644 index 00000000..f8273bf5 --- /dev/null +++ b/tools/mcuboot/imgtool/keys/rsa.py @@ -0,0 +1,163 @@ +""" +RSA Key management +""" + +from cryptography.hazmat.backends import default_backend +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric import rsa +from cryptography.hazmat.primitives.asymmetric.padding import PSS, MGF1 +from cryptography.hazmat.primitives.hashes import SHA256 + +from .general import KeyClass + + +# Sizes that bootutil will recognize +RSA_KEY_SIZES = [2048, 3072] + + +class RSAUsageError(Exception): + pass + + +class RSAPublic(KeyClass): + """The public key can only do a few operations""" + def __init__(self, key): + self.key = key + + def key_size(self): + return self.key.key_size + + def shortname(self): + return "rsa" + + def _unsupported(self, name): + raise RSAUsageError("Operation {} requires private key".format(name)) + + def _get_public(self): + return self.key + + def get_public_bytes(self): + # The key embedded into MCUboot is in PKCS1 format. + return self._get_public().public_bytes( + encoding=serialization.Encoding.DER, + format=serialization.PublicFormat.PKCS1) + + def get_private_bytes(self, minimal): + self._unsupported('get_private_bytes') + + def export_private(self, path, passwd=None): + self._unsupported('export_private') + + def export_public(self, path): + """Write the public key to the given file.""" + pem = self._get_public().public_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PublicFormat.SubjectPublicKeyInfo) + with open(path, 'wb') as f: + f.write(pem) + + def sig_type(self): + return "PKCS1_PSS_RSA{}_SHA256".format(self.key_size()) + + def sig_tlv(self): + return"RSA{}".format(self.key_size()) + + def sig_len(self): + return self.key_size() / 8 + + def verify(self, signature, payload): + k = self.key + if isinstance(self.key, rsa.RSAPrivateKey): + k = self.key.public_key() + return k.verify(signature=signature, data=payload, + padding=PSS(mgf=MGF1(SHA256()), salt_length=32), + algorithm=SHA256()) + + +class RSA(RSAPublic): + """ + Wrapper around an RSA key, with imgtool support. + """ + + def __init__(self, key): + """The key should be a private key from cryptography""" + self.key = key + + @staticmethod + def generate(key_size=2048): + if key_size not in RSA_KEY_SIZES: + raise RSAUsageError("Key size {} is not supported by MCUboot" + .format(key_size)) + pk = rsa.generate_private_key( + public_exponent=65537, + key_size=key_size, + backend=default_backend()) + return RSA(pk) + + def _get_public(self): + return self.key.public_key() + + def _build_minimal_rsa_privkey(self, der): + ''' + Builds a new DER that only includes N/E/D/P/Q RSA parameters; + standard DER private bytes provided by OpenSSL also includes + CRT params (DP/DQ/QP) which can be removed. + ''' + OFFSET_N = 7 # N is always located at this offset + b = bytearray(der) + off = OFFSET_N + if b[off + 1] != 0x82: + raise RSAUsageError("Error parsing N while minimizing") + len_N = (b[off + 2] << 8) + b[off + 3] + 4 + off += len_N + if b[off + 1] != 0x03: + raise RSAUsageError("Error parsing E while minimizing") + len_E = b[off + 2] + 4 + off += len_E + if b[off + 1] != 0x82: + raise RSAUsageError("Error parsing D while minimizing") + len_D = (b[off + 2] << 8) + b[off + 3] + 4 + off += len_D + if b[off + 1] != 0x81: + raise RSAUsageError("Error parsing P while minimizing") + len_P = b[off + 2] + 3 + off += len_P + if b[off + 1] != 0x81: + raise RSAUsageError("Error parsing Q while minimizing") + len_Q = b[off + 2] + 3 + off += len_Q + # adjust DER size for removed elements + b[2] = (off - 4) >> 8 + b[3] = (off - 4) & 0xff + return b[:off] + + def get_private_bytes(self, minimal): + priv = self.key.private_bytes( + encoding=serialization.Encoding.DER, + format=serialization.PrivateFormat.TraditionalOpenSSL, + encryption_algorithm=serialization.NoEncryption()) + if minimal: + priv = self._build_minimal_rsa_privkey(priv) + return priv + + def export_private(self, path, passwd=None): + """Write the private key to the given file, protecting it with the + optional password.""" + if passwd is None: + enc = serialization.NoEncryption() + else: + enc = serialization.BestAvailableEncryption(passwd) + pem = self.key.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=enc) + with open(path, 'wb') as f: + f.write(pem) + + def sign(self, payload): + # The verification code only allows the salt length to be the + # same as the hash length, 32. + return self.key.sign( + data=payload, + padding=PSS(mgf=MGF1(SHA256()), salt_length=32), + algorithm=SHA256()) diff --git a/tools/mcuboot/imgtool/keys/rsa_test.py b/tools/mcuboot/imgtool/keys/rsa_test.py new file mode 100644 index 00000000..b0afa835 --- /dev/null +++ b/tools/mcuboot/imgtool/keys/rsa_test.py @@ -0,0 +1,115 @@ +""" +Tests for RSA keys +""" + +import io +import os +import sys +import tempfile +import unittest + +from cryptography.exceptions import InvalidSignature +from cryptography.hazmat.primitives.asymmetric.padding import PSS, MGF1 +from cryptography.hazmat.primitives.hashes import SHA256 + +# Setup sys path so 'imgtool' is in it. +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), + '../..'))) + +from imgtool.keys import load, RSA, RSAUsageError +from imgtool.keys.rsa import RSA_KEY_SIZES + + +class KeyGeneration(unittest.TestCase): + + def setUp(self): + self.test_dir = tempfile.TemporaryDirectory() + + def tname(self, base): + return os.path.join(self.test_dir.name, base) + + def tearDown(self): + self.test_dir.cleanup() + + def test_keygen(self): + # Try generating a RSA key with non-supported size + with self.assertRaises(RSAUsageError): + RSA.generate(key_size=1024) + + for key_size in RSA_KEY_SIZES: + name1 = self.tname("keygen.pem") + k = RSA.generate(key_size=key_size) + k.export_private(name1, b'secret') + + # Try loading the key without a password. + self.assertIsNone(load(name1)) + + k2 = load(name1, b'secret') + + pubname = self.tname('keygen-pub.pem') + k2.export_public(pubname) + pk2 = load(pubname) + + # We should be able to export the public key from the loaded + # public key, but not the private key. + pk2.export_public(self.tname('keygen-pub2.pem')) + self.assertRaises(RSAUsageError, pk2.export_private, + self.tname('keygen-priv2.pem')) + + def test_emit(self): + """Basic sanity check on the code emitters.""" + for key_size in RSA_KEY_SIZES: + k = RSA.generate(key_size=key_size) + + ccode = io.StringIO() + k.emit_c_public(ccode) + self.assertIn("rsa_pub_key", ccode.getvalue()) + self.assertIn("rsa_pub_key_len", ccode.getvalue()) + + rustcode = io.StringIO() + k.emit_rust_public(rustcode) + self.assertIn("RSA_PUB_KEY", rustcode.getvalue()) + + def test_emit_pub(self): + """Basic sanity check on the code emitters, from public key.""" + pubname = self.tname("public.pem") + for key_size in RSA_KEY_SIZES: + k = RSA.generate(key_size=key_size) + k.export_public(pubname) + + k2 = load(pubname) + + ccode = io.StringIO() + k2.emit_c_public(ccode) + self.assertIn("rsa_pub_key", ccode.getvalue()) + self.assertIn("rsa_pub_key_len", ccode.getvalue()) + + rustcode = io.StringIO() + k2.emit_rust_public(rustcode) + self.assertIn("RSA_PUB_KEY", rustcode.getvalue()) + + def test_sig(self): + for key_size in RSA_KEY_SIZES: + k = RSA.generate(key_size=key_size) + buf = b'This is the message' + sig = k.sign(buf) + + # The code doesn't have any verification, so verify this + # manually. + k.key.public_key().verify( + signature=sig, + data=buf, + padding=PSS(mgf=MGF1(SHA256()), salt_length=32), + algorithm=SHA256()) + + # Modify the message to make sure the signature fails. + self.assertRaises(InvalidSignature, + k.key.public_key().verify, + signature=sig, + data=b'This is thE message', + padding=PSS(mgf=MGF1(SHA256()), salt_length=32), + algorithm=SHA256()) + + +if __name__ == '__main__': + unittest.main() diff --git a/tools/mcuboot/imgtool/keys/x25519.py b/tools/mcuboot/imgtool/keys/x25519.py new file mode 100644 index 00000000..63c0b5a7 --- /dev/null +++ b/tools/mcuboot/imgtool/keys/x25519.py @@ -0,0 +1,107 @@ +""" +X25519 key management +""" + +from cryptography.hazmat.backends import default_backend +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric import x25519 + +from .general import KeyClass + + +class X25519UsageError(Exception): + pass + + +class X25519Public(KeyClass): + def __init__(self, key): + self.key = key + + def shortname(self): + return "x25519" + + def _unsupported(self, name): + raise X25519UsageError("Operation {} requires private key".format(name)) + + def _get_public(self): + return self.key + + def get_public_bytes(self): + # The key is embedded into MBUboot in "SubjectPublicKeyInfo" format + return self._get_public().public_bytes( + encoding=serialization.Encoding.DER, + format=serialization.PublicFormat.SubjectPublicKeyInfo) + + def get_private_bytes(self, minimal): + self._unsupported('get_private_bytes') + + def export_private(self, path, passwd=None): + self._unsupported('export_private') + + def export_public(self, path): + """Write the public key to the given file.""" + pem = self._get_public().public_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PublicFormat.SubjectPublicKeyInfo) + with open(path, 'wb') as f: + f.write(pem) + + def sig_type(self): + return "X25519" + + def sig_tlv(self): + return "X25519" + + def sig_len(self): + return 32 + + +class X25519(X25519Public): + """ + Wrapper around an X25519 private key. + """ + + def __init__(self, key): + """key should be an instance of EllipticCurvePrivateKey""" + self.key = key + + @staticmethod + def generate(): + pk = x25519.X25519PrivateKey.generate() + return X25519(pk) + + def _get_public(self): + return self.key.public_key() + + def get_private_bytes(self, minimal): + return self.key.private_bytes( + encoding=serialization.Encoding.DER, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=serialization.NoEncryption()) + + def export_private(self, path, passwd=None): + """ + Write the private key to the given file, protecting it with the + optional password. + """ + if passwd is None: + enc = serialization.NoEncryption() + else: + enc = serialization.BestAvailableEncryption(passwd) + pem = self.key.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=enc) + with open(path, 'wb') as f: + f.write(pem) + + def sign_digest(self, digest): + """Return the actual signature""" + return self.key.sign(data=digest) + + def verify_digest(self, signature, digest): + """Verify that signature is valid for given digest""" + k = self.key + if isinstance(self.key, x25519.X25519PrivateKey): + k = self.key.public_key() + return k.verify(signature=signature, data=digest) diff --git a/tools/mcuboot/imgtool/main.py b/tools/mcuboot/imgtool/main.py new file mode 100644 index 00000000..c93addc0 --- /dev/null +++ b/tools/mcuboot/imgtool/main.py @@ -0,0 +1,352 @@ +#! /usr/bin/env python3 +# +# Copyright 2017-2020 Linaro Limited +# Copyright 2019-2020 Arm Limited +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import re +import click +import getpass +import imgtool.keys as keys +import sys +from imgtool import image, imgtool_version +from imgtool.version import decode_version +from .keys import ( + RSAUsageError, ECDSAUsageError, Ed25519UsageError, X25519UsageError) + +MIN_PYTHON_VERSION = (3, 6) +if sys.version_info < MIN_PYTHON_VERSION: + sys.exit("Python %s.%s or newer is required by imgtool." + % MIN_PYTHON_VERSION) + + +def gen_rsa2048(keyfile, passwd): + keys.RSA.generate().export_private(path=keyfile, passwd=passwd) + + +def gen_rsa3072(keyfile, passwd): + keys.RSA.generate(key_size=3072).export_private(path=keyfile, + passwd=passwd) + + +def gen_ecdsa_p256(keyfile, passwd): + keys.ECDSA256P1.generate().export_private(keyfile, passwd=passwd) + + +def gen_ecdsa_p224(keyfile, passwd): + print("TODO: p-224 not yet implemented") + + +def gen_ed25519(keyfile, passwd): + keys.Ed25519.generate().export_private(path=keyfile, passwd=passwd) + + +def gen_x25519(keyfile, passwd): + keys.X25519.generate().export_private(path=keyfile, passwd=passwd) + + +valid_langs = ['c', 'rust'] +keygens = { + 'rsa-2048': gen_rsa2048, + 'rsa-3072': gen_rsa3072, + 'ecdsa-p256': gen_ecdsa_p256, + 'ecdsa-p224': gen_ecdsa_p224, + 'ed25519': gen_ed25519, + 'x25519': gen_x25519, +} + + +def load_key(keyfile): + # TODO: better handling of invalid pass-phrase + key = keys.load(keyfile) + if key is not None: + return key + passwd = getpass.getpass("Enter key passphrase: ").encode('utf-8') + return keys.load(keyfile, passwd) + + +def get_password(): + while True: + passwd = getpass.getpass("Enter key passphrase: ") + passwd2 = getpass.getpass("Reenter passphrase: ") + if passwd == passwd2: + break + print("Passwords do not match, try again") + + # Password must be bytes, always use UTF-8 for consistent + # encoding. + return passwd.encode('utf-8') + + +@click.option('-p', '--password', is_flag=True, + help='Prompt for password to protect key') +@click.option('-t', '--type', metavar='type', required=True, + type=click.Choice(keygens.keys()), prompt=True, + help='{}'.format('One of: {}'.format(', '.join(keygens.keys())))) +@click.option('-k', '--key', metavar='filename', required=True) +@click.command(help='Generate pub/private keypair') +def keygen(type, key, password): + password = get_password() if password else None + keygens[type](key, password) + + +@click.option('-l', '--lang', metavar='lang', default=valid_langs[0], + type=click.Choice(valid_langs)) +@click.option('-k', '--key', metavar='filename', required=True) +@click.command(help='Dump public key from keypair') +def getpub(key, lang): + key = load_key(key) + if key is None: + print("Invalid passphrase") + elif lang == 'c': + key.emit_c_public() + elif lang == 'rust': + key.emit_rust_public() + else: + raise ValueError("BUG: should never get here!") + + +@click.option('--minimal', default=False, is_flag=True, + help='Reduce the size of the dumped private key to include only ' + 'the minimum amount of data required to decrypt. This ' + 'might require changes to the build config. Check the docs!' + ) +@click.option('-k', '--key', metavar='filename', required=True) +@click.command(help='Dump private key from keypair') +def getpriv(key, minimal): + key = load_key(key) + if key is None: + print("Invalid passphrase") + try: + key.emit_private(minimal) + except (RSAUsageError, ECDSAUsageError, Ed25519UsageError, + X25519UsageError) as e: + raise click.UsageError(e) + + +@click.argument('imgfile') +@click.option('-k', '--key', metavar='filename') +@click.command(help="Check that signed image can be verified by given key") +def verify(key, imgfile): + key = load_key(key) if key else None + ret, version = image.Image.verify(imgfile, key) + if ret == image.VerifyResult.OK: + print("Image was correctly validated") + print("Image version: {}.{}.{}+{}".format(*version)) + return + elif ret == image.VerifyResult.INVALID_MAGIC: + print("Invalid image magic; is this an MCUboot image?") + elif ret == image.VerifyResult.INVALID_TLV_INFO_MAGIC: + print("Invalid TLV info magic; is this an MCUboot image?") + elif ret == image.VerifyResult.INVALID_HASH: + print("Image has an invalid sha256 digest") + elif ret == image.VerifyResult.INVALID_SIGNATURE: + print("No signature found for the given key") + else: + print("Unknown return code: {}".format(ret)) + sys.exit(1) + + +def validate_version(ctx, param, value): + try: + decode_version(value) + return value + except ValueError as e: + raise click.BadParameter("{}".format(e)) + + +def validate_security_counter(ctx, param, value): + if value is not None: + if value.lower() == 'auto': + return 'auto' + else: + try: + return int(value, 0) + except ValueError: + raise click.BadParameter( + "{} is not a valid integer. Please use code literals " + "prefixed with 0b/0B, 0o/0O, or 0x/0X as necessary." + .format(value)) + + +def validate_header_size(ctx, param, value): + min_hdr_size = image.IMAGE_HEADER_SIZE + if value < min_hdr_size: + raise click.BadParameter( + "Minimum value for -H/--header-size is {}".format(min_hdr_size)) + return value + + +def get_dependencies(ctx, param, value): + if value is not None: + versions = [] + images = re.findall(r"\((\d+)", value) + if len(images) == 0: + raise click.BadParameter( + "Image dependency format is invalid: {}".format(value)) + raw_versions = re.findall(r",\s*([0-9.+]+)\)", value) + if len(images) != len(raw_versions): + raise click.BadParameter( + '''There's a mismatch between the number of dependency images + and versions in: {}'''.format(value)) + for raw_version in raw_versions: + try: + versions.append(decode_version(raw_version)) + except ValueError as e: + raise click.BadParameter("{}".format(e)) + dependencies = dict() + dependencies[image.DEP_IMAGES_KEY] = images + dependencies[image.DEP_VERSIONS_KEY] = versions + return dependencies + + +class BasedIntParamType(click.ParamType): + name = 'integer' + + def convert(self, value, param, ctx): + try: + return int(value, 0) + except ValueError: + self.fail('%s is not a valid integer. Please use code literals ' + 'prefixed with 0b/0B, 0o/0O, or 0x/0X as necessary.' + % value, param, ctx) + + +@click.argument('outfile') +@click.argument('infile') +@click.option('-R', '--erased-val', type=click.Choice(['0', '0xff']), + required=False, + help='The value that is read back from erased flash.') +@click.option('-x', '--hex-addr', type=BasedIntParamType(), required=False, + help='Adjust address in hex output file.') +@click.option('-L', '--load-addr', type=BasedIntParamType(), required=False, + help='Load address for image when it should run from RAM.') +@click.option('--save-enctlv', default=False, is_flag=True, + help='When upgrading, save encrypted key TLVs instead of plain ' + 'keys. Enable when BOOT_SWAP_SAVE_ENCTLV config option ' + 'was set.') +@click.option('-E', '--encrypt', metavar='filename', + help='Encrypt image using the provided public key') +@click.option('-e', '--endian', type=click.Choice(['little', 'big']), + default='little', help="Select little or big endian") +@click.option('--overwrite-only', default=False, is_flag=True, + help='Use overwrite-only instead of swap upgrades') +@click.option('--boot-record', metavar='sw_type', help='Create CBOR encoded ' + 'boot record TLV. The sw_type represents the role of the ' + 'software component (e.g. CoFM for coprocessor firmware). ' + '[max. 12 characters]') +@click.option('-M', '--max-sectors', type=int, + help='When padding allow for this amount of sectors (defaults ' + 'to 128)') +@click.option('--confirm', default=False, is_flag=True, + help='When padding the image, mark it as confirmed') +@click.option('--pad', default=False, is_flag=True, + help='Pad image to --slot-size bytes, adding trailer magic') +@click.option('-S', '--slot-size', type=BasedIntParamType(), required=True, + help='Size of the slot where the image will be written') +@click.option('--pad-header', default=False, is_flag=True, + help='Add --header-size zeroed bytes at the beginning of the ' + 'image') +@click.option('-H', '--header-size', callback=validate_header_size, + type=BasedIntParamType(), required=True) +@click.option('--pad-sig', default=False, is_flag=True, + help='Add 0-2 bytes of padding to ECDSA signature ' + '(for mcuboot <1.5)') +@click.option('-d', '--dependencies', callback=get_dependencies, + required=False, help='''Add dependence on another image, format: + "(,), ... "''') +@click.option('-s', '--security-counter', callback=validate_security_counter, + help='Specify the value of security counter. Use the `auto` ' + 'keyword to automatically generate it from the image version.') +@click.option('-v', '--version', callback=validate_version, required=True) +@click.option('--align', type=click.Choice(['1', '2', '4', '8']), + required=True) +@click.option('--public-key-format', type=click.Choice(['hash', 'full']), + default='hash', help='In what format to add the public key to ' + 'the image manifest: full key or hash of the key.') +@click.option('-k', '--key', metavar='filename') +@click.command(help='''Create a signed or unsigned image\n + INFILE and OUTFILE are parsed as Intel HEX if the params have + .hex extension, otherwise binary format is used''') +def sign(key, public_key_format, align, version, pad_sig, header_size, + pad_header, slot_size, pad, confirm, max_sectors, overwrite_only, + endian, encrypt, infile, outfile, dependencies, load_addr, hex_addr, + erased_val, save_enctlv, security_counter, boot_record): + img = image.Image(version=decode_version(version), header_size=header_size, + pad_header=pad_header, pad=pad, confirm=confirm, + align=int(align), slot_size=slot_size, + max_sectors=max_sectors, overwrite_only=overwrite_only, + endian=endian, load_addr=load_addr, erased_val=erased_val, + save_enctlv=save_enctlv, + security_counter=security_counter) + img.load(infile) + key = load_key(key) if key else None + enckey = load_key(encrypt) if encrypt else None + if enckey and key: + if ((isinstance(key, keys.ECDSA256P1) and + not isinstance(enckey, keys.ECDSA256P1Public)) + or (isinstance(key, keys.RSA) and + not isinstance(enckey, keys.RSAPublic))): + # FIXME + raise click.UsageError("Signing and encryption must use the same " + "type of key") + + if pad_sig and hasattr(key, 'pad_sig'): + key.pad_sig = True + + img.create(key, public_key_format, enckey, dependencies, boot_record) + img.save(outfile, hex_addr) + + +class AliasesGroup(click.Group): + + _aliases = { + "create": "sign", + } + + def list_commands(self, ctx): + cmds = [k for k in self.commands] + aliases = [k for k in self._aliases] + return sorted(cmds + aliases) + + def get_command(self, ctx, cmd_name): + rv = click.Group.get_command(self, ctx, cmd_name) + if rv is not None: + return rv + if cmd_name in self._aliases: + return click.Group.get_command(self, ctx, self._aliases[cmd_name]) + return None + + +@click.command(help='Print imgtool version information') +def version(): + print(imgtool_version) + + +@click.command(cls=AliasesGroup, + context_settings=dict(help_option_names=['-h', '--help'])) +def imgtool(): + pass + + +imgtool.add_command(keygen) +imgtool.add_command(getpub) +imgtool.add_command(getpriv) +imgtool.add_command(verify) +imgtool.add_command(sign) +imgtool.add_command(version) + + +if __name__ == '__main__': + imgtool() diff --git a/tools/mcuboot/imgtool/version.py b/tools/mcuboot/imgtool/version.py new file mode 100644 index 00000000..030b012c --- /dev/null +++ b/tools/mcuboot/imgtool/version.py @@ -0,0 +1,53 @@ +# Copyright 2017 Linaro Limited +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +""" +Semi Semantic Versioning + +Implements a subset of semantic versioning that is supportable by the image +header. +""" + +from collections import namedtuple +import re + +SemiSemVersion = namedtuple('SemiSemVersion', ['major', 'minor', 'revision', + 'build']) + +version_re = re.compile( + r"""^([1-9]\d*|0)(\.([1-9]\d*|0)(\.([1-9]\d*|0)(\+([1-9]\d*|0))?)?)?$""") + + +def decode_version(text): + """Decode the version string, which should be of the form maj.min.rev+build + """ + m = version_re.match(text) + if m: + result = SemiSemVersion( + int(m.group(1)) if m.group(1) else 0, + int(m.group(3)) if m.group(3) else 0, + int(m.group(5)) if m.group(5) else 0, + int(m.group(7)) if m.group(7) else 0) + return result + else: + msg = "Invalid version number, should be maj.min.rev+build with later " + msg += "parts optional" + raise ValueError(msg) + + +if __name__ == '__main__': + print(decode_version("1.2")) + print(decode_version("1.0")) + print(decode_version("0.0.2+75")) + print(decode_version("0.0.0+00")) -- cgit v1.2.3