init:初始框架
This commit is contained in:
@@ -0,0 +1,17 @@
|
||||
# Copyright 2017-2020 Linaro Limited
|
||||
#
|
||||
# SPDX-License-Identifier: Apache-2.0
|
||||
#
|
||||
# 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.8.0"
|
||||
@@ -0,0 +1,49 @@
|
||||
# Copyright (c) 2019, Arm Limited.
|
||||
# Copyright (c) 2020, Linaro Limited
|
||||
#
|
||||
# SPDX-License-Identifier: Apache-2.0
|
||||
#
|
||||
# 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)
|
||||
@@ -0,0 +1,605 @@
|
||||
# Copyright 2018 Nordic Semiconductor ASA
|
||||
# Copyright 2017-2020 Linaro Limited
|
||||
# Copyright 2019-2021 Arm Limited
|
||||
#
|
||||
# SPDX-License-Identifier: Apache-2.0
|
||||
#
|
||||
# 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,
|
||||
'ENCRYPTED_AES128': 0x0000004,
|
||||
'ENCRYPTED_AES256': 0x0000008,
|
||||
'NON_BOOTABLE': 0x0000010,
|
||||
'RAM_LOAD': 0x0000020,
|
||||
'ROM_FIXED': 0x0000100,
|
||||
}
|
||||
|
||||
TLV_VALUES = {
|
||||
'KEYHASH': 0x01,
|
||||
'PUBKEY': 0x02,
|
||||
'SHA256': 0x10,
|
||||
'RSA2048': 0x20,
|
||||
'ECDSA224': 0x21,
|
||||
'ECDSA256': 0x22,
|
||||
'RSA3072': 0x23,
|
||||
'ED25519': 0x24,
|
||||
'ENCRSA2048': 0x30,
|
||||
'ENCKW': 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]
|
||||
if isinstance(kind, int):
|
||||
buf = struct.pack(e + 'BBH', kind, 0, len(payload))
|
||||
else:
|
||||
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,
|
||||
rom_fixed=None, erased_val=None, save_enctlv=False,
|
||||
security_counter=None):
|
||||
|
||||
if load_addr and rom_fixed:
|
||||
raise click.UsageError("Can not set rom_fixed and load_addr at the same time")
|
||||
|
||||
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.rom_fixed = rom_fixed
|
||||
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 "<Image version={}, header_size={}, security_counter={}, \
|
||||
base_addr={}, load_addr={}, align={}, slot_size={}, \
|
||||
max_sectors={}, overwrite_only={}, endian={} format={}, \
|
||||
payloadlen=0x{:x}>".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 = bytearray([self.erased_val] *
|
||||
(trailer_size - len(boot_magic)))
|
||||
if self.confirm and not self.overwrite_only:
|
||||
padding[-MAX_ALIGN] = 0x01 # image_ok = 0x01
|
||||
padding += boot_magic
|
||||
h.puts(trailer_addr, bytes(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, custom_tlvs=None, encrypt_keylen=128):
|
||||
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 custom_tlvs is not None:
|
||||
for value in custom_tlvs.values():
|
||||
protected_tlv_size += TLV_SIZE + len(value)
|
||||
|
||||
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 padding if image is not aligned to the 16 Bytes
|
||||
# in encrypted mode
|
||||
if self.enckey is not None:
|
||||
pad_len = len(self.payload) % 16
|
||||
if pad_len > 0:
|
||||
pad = bytes(16 - pad_len)
|
||||
if isinstance(self.payload, bytes):
|
||||
self.payload += pad
|
||||
else:
|
||||
self.payload.extend(pad)
|
||||
|
||||
# This adds the header to the payload as well
|
||||
if encrypt_keylen == 256:
|
||||
self.add_header(enckey, protected_tlv_size, 256)
|
||||
else:
|
||||
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)
|
||||
|
||||
if custom_tlvs is not None:
|
||||
for tag, value in custom_tlvs.items():
|
||||
prot_tlv.add(tag, value)
|
||||
|
||||
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:
|
||||
if encrypt_keylen == 256:
|
||||
plainkey = os.urandom(32)
|
||||
else:
|
||||
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, aes_length=128):
|
||||
"""Install the image header."""
|
||||
|
||||
flags = 0
|
||||
if enckey is not None:
|
||||
if aes_length == 128:
|
||||
flags |= IMAGE_F['ENCRYPTED_AES128']
|
||||
else:
|
||||
flags |= IMAGE_F['ENCRYPTED_AES256']
|
||||
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']
|
||||
if self.rom_fixed:
|
||||
flags |= IMAGE_F['ROM_FIXED']
|
||||
|
||||
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.rom_fixed or 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, None
|
||||
|
||||
tlv_off = header_size + img_size
|
||||
tlv_info = b[tlv_off:tlv_off+TLV_INFO_SIZE]
|
||||
magic, tlv_tot = struct.unpack('HH', tlv_info)
|
||||
if magic == TLV_PROT_INFO_MAGIC:
|
||||
tlv_off += tlv_tot
|
||||
tlv_info = b[tlv_off:tlv_off+TLV_INFO_SIZE]
|
||||
magic, tlv_tot = struct.unpack('HH', tlv_info)
|
||||
|
||||
if magic != TLV_INFO_MAGIC:
|
||||
return VerifyResult.INVALID_TLV_INFO_MAGIC, None, None
|
||||
|
||||
sha = hashlib.sha256()
|
||||
prot_tlv_size = tlv_off
|
||||
sha.update(b[:prot_tlv_size])
|
||||
digest = sha.digest()
|
||||
|
||||
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, digest
|
||||
else:
|
||||
return VerifyResult.INVALID_HASH, None, 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[:prot_tlv_size]
|
||||
try:
|
||||
if hasattr(key, 'verify'):
|
||||
key.verify(tlv_sig, payload)
|
||||
else:
|
||||
key.verify_digest(tlv_sig, digest)
|
||||
return VerifyResult.OK, version, digest
|
||||
except InvalidSignature:
|
||||
# continue to next TLV
|
||||
pass
|
||||
tlv_off += TLV_SIZE + tlv_len
|
||||
return VerifyResult.INVALID_SIGNATURE, None, None
|
||||
@@ -0,0 +1,96 @@
|
||||
# Copyright 2017 Linaro Limited
|
||||
#
|
||||
# SPDX-License-Identifier: Apache-2.0
|
||||
#
|
||||
# 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)))
|
||||
@@ -0,0 +1,159 @@
|
||||
"""
|
||||
ECDSA key management
|
||||
"""
|
||||
|
||||
# SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
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
|
||||
@@ -0,0 +1,101 @@
|
||||
"""
|
||||
Tests for ECDSA keys
|
||||
"""
|
||||
|
||||
# SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
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()
|
||||
@@ -0,0 +1,107 @@
|
||||
"""
|
||||
ED25519 key management
|
||||
"""
|
||||
|
||||
# SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
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)
|
||||
@@ -0,0 +1,105 @@
|
||||
"""
|
||||
Tests for ECDSA keys
|
||||
"""
|
||||
|
||||
# SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
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()
|
||||
@@ -0,0 +1,47 @@
|
||||
"""General key class."""
|
||||
|
||||
# SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
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: &[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)
|
||||
@@ -0,0 +1,165 @@
|
||||
"""
|
||||
RSA Key management
|
||||
"""
|
||||
|
||||
# SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
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())
|
||||
@@ -0,0 +1,117 @@
|
||||
"""
|
||||
Tests for RSA keys
|
||||
"""
|
||||
|
||||
# SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
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()
|
||||
@@ -0,0 +1,109 @@
|
||||
"""
|
||||
X25519 key management
|
||||
"""
|
||||
|
||||
# SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
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)
|
||||
@@ -0,0 +1,395 @@
|
||||
#! /usr/bin/env python3
|
||||
#
|
||||
# Copyright 2017-2020 Linaro Limited
|
||||
# Copyright 2019-2021 Arm Limited
|
||||
#
|
||||
# SPDX-License-Identifier: Apache-2.0
|
||||
#
|
||||
# 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, digest = image.Image.verify(imgfile, key)
|
||||
if ret == image.VerifyResult.OK:
|
||||
print("Image was correctly validated")
|
||||
print("Image version: {}.{}.{}+{}".format(*version))
|
||||
print("Image digest: {}".format(digest.hex()))
|
||||
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('--custom-tlv', required=False, nargs=2, default=[],
|
||||
multiple=True, metavar='[tag] [value]',
|
||||
help='Custom TLV that will be placed into protected area. '
|
||||
'Add "0x" prefix if the value should be interpreted as an '
|
||||
'integer, otherwise it will be interpreted as a string. '
|
||||
'Specify the option multiple times to add multiple TLVs.')
|
||||
@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('-F', '--rom-fixed', type=BasedIntParamType(), required=False,
|
||||
help='Set flash address the image is built for.')
|
||||
@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. '
|
||||
'(Not supported in direct-xip or ram-load mode.)')
|
||||
@click.option('--encrypt-keylen', default='128',
|
||||
type=click.Choice(['128','256']),
|
||||
help='When encrypting the image using AES, select a 128 bit or '
|
||||
'256 bit key len.')
|
||||
@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 (implies '
|
||||
'--pad)')
|
||||
@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. If the slots have different sizes, use '
|
||||
'the size of the secondary slot.')
|
||||
@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:
|
||||
"(<image_ID>,<image_version>), ... "''')
|
||||
@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_keylen, encrypt, infile, outfile, dependencies,
|
||||
load_addr, hex_addr, erased_val, save_enctlv, security_counter,
|
||||
boot_record, custom_tlv, rom_fixed):
|
||||
|
||||
if confirm:
|
||||
# Confirmed but non-padded images don't make much sense, because
|
||||
# otherwise there's no trailer area for writing the confirmed status.
|
||||
pad = True
|
||||
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, rom_fixed=rom_fixed,
|
||||
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
|
||||
|
||||
# Get list of custom protected TLVs from the command-line
|
||||
custom_tlvs = {}
|
||||
for tlv in custom_tlv:
|
||||
tag = int(tlv[0], 0)
|
||||
if tag in custom_tlvs:
|
||||
raise click.UsageError('Custom TLV %s already exists.' % hex(tag))
|
||||
if tag in image.TLV_VALUES.values():
|
||||
raise click.UsageError(
|
||||
'Custom TLV %s conflicts with predefined TLV.' % hex(tag))
|
||||
|
||||
value = tlv[1]
|
||||
if value.startswith('0x'):
|
||||
if len(value[2:]) % 2:
|
||||
raise click.UsageError('Custom TLV length is odd.')
|
||||
custom_tlvs[tag] = bytes.fromhex(value[2:])
|
||||
else:
|
||||
custom_tlvs[tag] = value.encode('utf-8')
|
||||
|
||||
img.create(key, public_key_format, enckey, dependencies, boot_record,
|
||||
custom_tlvs, int(encrypt_keylen))
|
||||
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()
|
||||
@@ -0,0 +1,55 @@
|
||||
# Copyright 2017 Linaro Limited
|
||||
#
|
||||
# SPDX-License-Identifier: Apache-2.0
|
||||
#
|
||||
# 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"))
|
||||
Reference in New Issue
Block a user