summaryrefslogtreecommitdiff
path: root/tools
diff options
context:
space:
mode:
Diffstat (limited to 'tools')
-rw-r--r--tools/bin2c.py74
-rw-r--r--tools/mcuboot/README1
-rw-r--r--tools/mcuboot/assemble.py131
-rw-r--r--tools/mcuboot/flash.sh18
-rw-r--r--tools/mcuboot/gdb-boot.sh27
-rw-r--r--tools/mcuboot/imgtool.nix30
-rwxr-xr-xtools/mcuboot/imgtool.py20
-rw-r--r--tools/mcuboot/imgtool/__init__.py15
-rw-r--r--tools/mcuboot/imgtool/boot_record.py47
-rw-r--r--tools/mcuboot/imgtool/image.py552
-rw-r--r--tools/mcuboot/imgtool/keys/__init__.py94
-rw-r--r--tools/mcuboot/imgtool/keys/ecdsa.py157
-rw-r--r--tools/mcuboot/imgtool/keys/ecdsa_test.py99
-rw-r--r--tools/mcuboot/imgtool/keys/ed25519.py105
-rw-r--r--tools/mcuboot/imgtool/keys/ed25519_test.py103
-rw-r--r--tools/mcuboot/imgtool/keys/general.py45
-rw-r--r--tools/mcuboot/imgtool/keys/rsa.py163
-rw-r--r--tools/mcuboot/imgtool/keys/rsa_test.py115
-rw-r--r--tools/mcuboot/imgtool/keys/x25519.py107
-rw-r--r--tools/mcuboot/imgtool/main.py352
-rw-r--r--tools/mcuboot/imgtool/version.py53
-rw-r--r--tools/mcuboot/jgdb.sh6
-rw-r--r--tools/mcuboot/jl.sh5
-rw-r--r--tools/mcuboot/mcubin.bt135
-rw-r--r--tools/mcuboot/requirements.txt4
-rw-r--r--tools/mcuboot/setup.py29
-rw-r--r--tools/rle_encode.py379
27 files changed, 2866 insertions, 0 deletions
diff --git a/tools/bin2c.py b/tools/bin2c.py
new file mode 100644
index 00000000..1d66656a
--- /dev/null
+++ b/tools/bin2c.py
@@ -0,0 +1,74 @@
+#!/usr/bin/env python
+#-*- coding: utf-8 -*-
+"""
+ bin2c
+ ~~~~~
+
+ Simple tool for creating C array from a binary file.
+
+ :copyright: (c) 2016 by Dmitry Alimov.
+ :license: The MIT License (MIT), see LICENSE for more details.
+"""
+
+import argparse
+import os
+import re
+import sys
+
+PY3 = sys.version_info[0] == 3
+
+
+def bin2c(filename, varname='data', linesize=80, indent=4):
+ """ Read binary data from file and return as a C array
+
+ :param filename: a filename of a file to read.
+ :param varname: a C array variable name.
+ :param linesize: a size of a line (min value is 40).
+ :param indent: an indent (number of spaces) that prepend each line.
+ """
+ if not os.path.isfile(filename):
+ print('File "%s" is not found!' % filename)
+ return ''
+ if not re.match('[a-zA-Z_][a-zA-Z0-9_]*', varname):
+ print('Invalid variable name "%s"' % varname)
+ return
+ with open(filename, 'rb') as in_file:
+ data = in_file.read()
+ # limit the line length
+ if linesize < 40:
+ linesize = 40
+ byte_len = 6 # '0x00, '
+ out = 'const char %s[%d] = {\n' % (varname, len(data))
+ line = ''
+ for byte in data:
+ line += '0x%02x, ' % (byte if PY3 else ord(byte))
+ if len(line) + indent + byte_len >= linesize:
+ out += ' ' * indent + line + '\n'
+ line = ''
+ # add the last line
+ if len(line) + indent + byte_len < linesize:
+ out += ' ' * indent + line + '\n'
+ # strip the last comma
+ out = out.rstrip(', \n') + '\n'
+ out += '};'
+ return out
+
+
+def main():
+ """ Main func """
+ parser = argparse.ArgumentParser()
+ parser.add_argument(
+ 'filename', help='filename to convert to C array')
+ parser.add_argument(
+ 'varname', nargs='?', help='variable name', default='data')
+ parser.add_argument(
+ 'linesize', nargs='?', help='line length', default=80, type=int)
+ parser.add_argument(
+ 'indent', nargs='?', help='indent size', default=4, type=int)
+ args = parser.parse_args()
+ # print out the data
+ print(bin2c(args.filename, args.varname, args.linesize, args.indent))
+
+
+if __name__ == '__main__':
+ main() \ No newline at end of file
diff --git a/tools/mcuboot/README b/tools/mcuboot/README
new file mode 100644
index 00000000..feb5d2f9
--- /dev/null
+++ b/tools/mcuboot/README
@@ -0,0 +1 @@
+This whole folder comes from MCUBoot source files (commit 9015a5d404c2c688166cab81067be53c860d98f4). \ No newline at end of file
diff --git a/tools/mcuboot/assemble.py b/tools/mcuboot/assemble.py
new file mode 100644
index 00000000..f2ce4a1b
--- /dev/null
+++ b/tools/mcuboot/assemble.py
@@ -0,0 +1,131 @@
+#! /usr/bin/env python3
+#
+# 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.
+
+"""
+Assemble multiple images into a single image that can be flashed on the device.
+"""
+
+import argparse
+import errno
+import io
+import re
+import os.path
+import sys
+
+ZEPHYR_BASE = os.getenv("ZEPHYR_BASE")
+if not ZEPHYR_BASE:
+ sys.exit("$ZEPHYR_BASE environment variable undefined")
+
+sys.path.insert(0, os.path.join(ZEPHYR_BASE, "scripts", "dts"))
+import edtlib
+
+def same_keys(a, b):
+ """Determine if the dicts a and b have the same keys in them"""
+ for ak in a.keys():
+ if ak not in b:
+ return False
+ for bk in b.keys():
+ if bk not in a:
+ return False
+ return True
+
+offset_re = re.compile(r"^#define DT_FLASH_AREA_([0-9A-Z_]+)_OFFSET(_0)?\s+(0x[0-9a-fA-F]+|[0-9]+)$")
+size_re = re.compile(r"^#define DT_FLASH_AREA_([0-9A-Z_]+)_SIZE(_0)?\s+(0x[0-9a-fA-F]+|[0-9]+)$")
+
+class Assembly():
+ def __init__(self, output, bootdir, edt):
+ self.find_slots(edt)
+ try:
+ os.unlink(output)
+ except OSError as e:
+ if e.errno != errno.ENOENT:
+ raise
+ self.output = output
+
+ def find_slots(self, edt):
+ offsets = {}
+ sizes = {}
+
+ part_nodes = edt.compat2nodes["fixed-partitions"]
+ for node in part_nodes:
+ for child in node.children.values():
+ if "label" in child.props:
+ label = child.props["label"].val
+ offsets[label] = child.regs[0].addr
+ sizes[label] = child.regs[0].size
+
+ if not same_keys(offsets, sizes):
+ raise Exception("Inconsistent data in devicetree.h")
+
+ # We care about the mcuboot, image-0, and image-1 partitions.
+ if 'mcuboot' not in offsets:
+ raise Exception("Board partition table does not have mcuboot partition")
+
+ if 'image-0' not in offsets:
+ raise Exception("Board partition table does not have image-0 partition")
+
+ if 'image-1' not in offsets:
+ raise Exception("Board partition table does not have image-1 partition")
+
+ self.offsets = offsets
+ self.sizes = sizes
+
+ def add_image(self, source, partition):
+ with open(self.output, 'ab') as ofd:
+ pos = ofd.tell()
+ print("partition {}, pos={}, offset={}".format(partition, pos, self.offsets[partition]))
+ if pos > self.offsets[partition]:
+ raise Exception("Partitions not in order, unsupported")
+ if pos < self.offsets[partition]:
+ buf = b'\xFF' * (self.offsets[partition] - pos)
+ ofd.write(buf)
+ with open(source, 'rb') as rfd:
+ ibuf = rfd.read()
+ if len(ibuf) > self.sizes[partition]:
+ raise Exception("Image {} is too large for partition".format(source))
+ ofd.write(ibuf)
+
+def main():
+ parser = argparse.ArgumentParser()
+
+ parser.add_argument('-b', '--bootdir', required=True,
+ help='Directory of built bootloader')
+ parser.add_argument('-p', '--primary', required=True,
+ help='Signed image file for primary image')
+ parser.add_argument('-s', '--secondary',
+ help='Signed image file for secondary image')
+ parser.add_argument('-o', '--output', required=True,
+ help='Filename to write full image to')
+
+ args = parser.parse_args()
+
+ # Extract board name from path
+ board = os.path.split(os.path.split(args.bootdir)[0])[1]
+
+ dts_path = os.path.join(args.bootdir, "zephyr", board + ".dts.pre.tmp")
+
+ edt = edtlib.EDT(dts_path, [os.path.join(ZEPHYR_BASE, "dts", "bindings")],
+ warn_reg_unit_address_mismatch=False)
+
+ output = Assembly(args.output, args.bootdir, edt)
+
+ output.add_image(os.path.join(args.bootdir, 'zephyr', 'zephyr.bin'), 'mcuboot')
+ output.add_image(args.primary, "image-0")
+ if args.secondary is not None:
+ output.add_image(args.secondary, "image-1")
+
+if __name__ == '__main__':
+ main()
diff --git a/tools/mcuboot/flash.sh b/tools/mcuboot/flash.sh
new file mode 100644
index 00000000..a2c58c75
--- /dev/null
+++ b/tools/mcuboot/flash.sh
@@ -0,0 +1,18 @@
+#! /bin/bash
+
+source $(dirname $0)/../target.sh
+
+lscript=/tmp/flash$$.jlink
+
+cat >$lscript <<EOF
+h
+r
+loadfile outdir/$BOARD/zephyr.bin $BASE_BOOT
+loadfile hello.signed.bin $BASE_PRIMARY_SLOT
+loadfile shell.signed.bin $BASE_SECONDARY_SLOT
+q
+EOF
+
+JLinkExe -device $SOC -si SWD -speed auto \
+ -CommanderScript $lscript
+rm $lscript
diff --git a/tools/mcuboot/gdb-boot.sh b/tools/mcuboot/gdb-boot.sh
new file mode 100644
index 00000000..dd8a34d2
--- /dev/null
+++ b/tools/mcuboot/gdb-boot.sh
@@ -0,0 +1,27 @@
+#! /bin/bash
+
+source $(dirname $0)/../target.sh
+
+gscript=/tmp/init$$.gdb
+
+cat > $gscript <<EOF
+target remote localhost:2331
+symbol-file outdir/$BOARD/zephyr.elf
+# symbol-file ../zephyr/samples/shell/outdir/$BOARD/zephyr.elf
+# dir apps/boot/src
+# dir libs/bootutil/src
+# dir hw/mcu/stm/stm32f4xx/src
+b main
+# b __reset
+# b bootutil_img_validate
+# b cmp_rsasig
+# b bootutil_verify_sig
+# b mbedtls_rsa_public
+# b boot_calloc
+mon reset 2
+layout src
+focus cmd
+EOF
+
+$gdbexe -x $gscript
+rm $gscript
diff --git a/tools/mcuboot/imgtool.nix b/tools/mcuboot/imgtool.nix
new file mode 100644
index 00000000..7d750df4
--- /dev/null
+++ b/tools/mcuboot/imgtool.nix
@@ -0,0 +1,30 @@
+#
+# Nix environment for imgtool
+#
+# To install the environment
+#
+# $ nix-env --file imgtool.nix --install env-imgtool
+#
+# To load the environment
+#
+# $ load-env-imgtool
+#
+with import <nixpkgs> {};
+let
+ # Nixpkgs has fairly recent versions of the dependencies, so we can
+ # rely on them without having to build our own derivations.
+ imgtoolPythonEnv = python37.withPackages (
+ _: [
+ python37.pkgs.click
+ python37.pkgs.cryptography
+ python37.pkgs.intelhex
+ python37.pkgs.setuptools
+ python37.pkgs.cbor
+ ]
+ );
+in
+myEnvFun {
+ name = "imgtool";
+
+ buildInputs = [ imgtoolPythonEnv ];
+}
diff --git a/tools/mcuboot/imgtool.py b/tools/mcuboot/imgtool.py
new file mode 100755
index 00000000..78614745
--- /dev/null
+++ b/tools/mcuboot/imgtool.py
@@ -0,0 +1,20 @@
+#! /usr/bin/env python3
+#
+# 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.
+
+from imgtool import main
+
+if __name__ == '__main__':
+ main.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 "<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 = 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:
+ "(<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, 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"))
diff --git a/tools/mcuboot/jgdb.sh b/tools/mcuboot/jgdb.sh
new file mode 100644
index 00000000..a79c87c6
--- /dev/null
+++ b/tools/mcuboot/jgdb.sh
@@ -0,0 +1,6 @@
+#! /bin/bash
+
+source $(dirname $0)/../target.sh
+
+# Start the jlink gdb server
+JLinkGDBServer -if swd -device $SOC -speed auto
diff --git a/tools/mcuboot/jl.sh b/tools/mcuboot/jl.sh
new file mode 100644
index 00000000..260206d5
--- /dev/null
+++ b/tools/mcuboot/jl.sh
@@ -0,0 +1,5 @@
+#!/bin/bash
+
+source $(dirname $0)/../target.sh
+
+JLinkExe -speed auto -si SWD -device $SOC
diff --git a/tools/mcuboot/mcubin.bt b/tools/mcuboot/mcubin.bt
new file mode 100644
index 00000000..e2ec3614
--- /dev/null
+++ b/tools/mcuboot/mcubin.bt
@@ -0,0 +1,135 @@
+// Copyright (C) 2019, Linaro Ltd
+//
+// 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.
+
+// This file is a Binary Template file for the 010 Editor
+// (http://www.sweetscape.com/010editor/) to allow it to show the
+// structure of an MCUboot image.
+
+LittleEndian();
+
+struct ENTRY {
+ uint32 id;
+ uint32 offset;
+ uint32 size;
+ uint32 pad;
+};
+
+// The simulator writes the partition table at the beginning of the
+// image, so that we can tell where the partitions are. If you are
+// trying to view an image captured from a device, you can either
+// construct a synthetic partition table in the file, or change code
+// described below to hardcode one.
+struct PTABLE {
+ uchar pheader[8];
+ if (ptable.pheader != "mcuboot\0") {
+ // NOTE: Put code here to hard code a partition table, and
+ // continue.
+ Warning("Invalid magic on ptable header");
+ return -1;
+ } else {
+ uint32 count;
+ struct ENTRY entries[count];
+ }
+};
+
+struct PTABLE ptable;
+
+struct IMAGE_VERSION {
+ uchar major;
+ uchar minor;
+ uint16 revision;
+ uint32 build_num;
+};
+
+struct IHDR {
+ uint32 magic <format=hex>;
+ uint32 load_addr <format=hex>;
+ uint16 hdr_size <format=hex>;
+ uint16 protect_size <format=hex>;
+ uint32 img_size <format=hex>;
+ uint32 flags;
+ struct IMAGE_VERSION ver;
+ uint32 _pad1;
+};
+
+struct TLV_HDR {
+ uint16 magic;
+ uint16 tlv_tot;
+};
+
+struct TLV {
+ uchar type <format=hex>;
+ uchar pad;
+ uint16 len;
+
+ switch (type) {
+ case 0x01: // keyhash
+ uchar keyhash[len];
+ break;
+ case 0x40: // dependency
+ if (len != 12) {
+ Warning("Invalid dependency size");
+ return -1;
+ }
+ uchar image_id;
+ uchar pad1;
+ uint16 pad2;
+ struct IMAGE_VERSION version;
+ break;
+ default:
+ // Other, just consume the data.
+ uchar data[len];
+ }
+};
+
+local int i;
+local int epos;
+
+for (i = 0; i < ptable.count; i++) {
+ FSeek(ptable.entries[i].offset);
+ switch (ptable.entries[i].id) {
+ case 1:
+ case 2:
+ case 4:
+ case 5:
+ struct IMAGE {
+ struct IHDR ihdr;
+
+ if (ihdr.magic == 0x96f3b83d) {
+ uchar payload[ihdr.img_size];
+
+ epos = FTell();
+ struct TLV_HDR tlv_hdr;
+
+ if (tlv_hdr.magic == 0x6907) {
+ epos += tlv_hdr.tlv_tot;
+ while (FTell() < epos) {
+ struct TLV tlv;
+ }
+ }
+ }
+ // uchar block[ptable.entries[i].size];
+ } image;
+ break;
+ case 3:
+ struct SCRATCH {
+ uchar data[ptable.entries[i].size];
+ } scratch;
+ break;
+ default:
+ break;
+ }
+}
diff --git a/tools/mcuboot/requirements.txt b/tools/mcuboot/requirements.txt
new file mode 100644
index 00000000..9481e2c1
--- /dev/null
+++ b/tools/mcuboot/requirements.txt
@@ -0,0 +1,4 @@
+cryptography>=2.6
+intelhex
+click
+cbor>=1.0.0
diff --git a/tools/mcuboot/setup.py b/tools/mcuboot/setup.py
new file mode 100644
index 00000000..058d0cb4
--- /dev/null
+++ b/tools/mcuboot/setup.py
@@ -0,0 +1,29 @@
+import setuptools
+from imgtool import imgtool_version
+
+setuptools.setup(
+ name="imgtool",
+ version=imgtool_version,
+ author="The MCUboot committers",
+ author_email="dev-mcuboot@lists.runtime.co",
+ description=("MCUboot's image signing and key management"),
+ license="Apache Software License",
+ url="http://github.com/JuulLabs-OSS/mcuboot",
+ packages=setuptools.find_packages(),
+ python_requires='>=3.6',
+ install_requires=[
+ 'cryptography>=2.4.2',
+ 'intelhex>=2.2.1',
+ 'click',
+ 'cbor>=1.0.0',
+ ],
+ entry_points={
+ "console_scripts": ["imgtool=imgtool.main:imgtool"]
+ },
+ classifiers=[
+ "Programming Language :: Python :: 3",
+ "Development Status :: 4 - Beta",
+ "Topic :: Software Development :: Build Tools",
+ "License :: OSI Approved :: Apache Software License",
+ ],
+)
diff --git a/tools/rle_encode.py b/tools/rle_encode.py
new file mode 100644
index 00000000..80a0926f
--- /dev/null
+++ b/tools/rle_encode.py
@@ -0,0 +1,379 @@
+#!/usr/bin/env python3
+
+# SPDX-License-Identifier: LGPL-3.0-or-later
+# Copyright (C) 2020 Daniel Thompson
+
+import argparse
+import sys
+import os.path
+from PIL import Image
+
+def clut8_rgb888(i):
+ """Reference CLUT for wasp-os.
+
+ Technically speaking this is not a CLUT because the we lookup the colours
+ algorithmically to avoid the cost of a genuine CLUT. The palette is
+ designed to be fairly easy to generate algorithmically.
+
+ The palette includes all 216 web-safe colours together 4 grays and
+ 36 additional colours that target "gaps" at the brighter end of the web
+ safe set. There are 11 greys (plus black and white) although two are
+ fairly close together.
+
+ :param int i: Index (from 0..255 inclusive) into the CLUT
+ :return: 24-bit colour in RGB888 format
+ """
+ if i < 216:
+ rgb888 = ( i % 6) * 0x33
+ rg = i // 6
+ rgb888 += (rg % 6) * 0x3300
+ rgb888 += (rg // 6) * 0x330000
+ elif i < 252:
+ i -= 216
+ rgb888 = 0x7f + (( i % 3) * 0x33)
+ rg = i // 3
+ rgb888 += 0x4c00 + ((rg % 4) * 0x3300)
+ rgb888 += 0x7f0000 + ((rg // 4) * 0x330000)
+ else:
+ i -= 252
+ rgb888 = 0x2c2c2c + (0x101010 * i)
+
+ return rgb888
+
+def clut8_rgb565(i):
+ """RBG565 CLUT for wasp-os.
+
+ This CLUT implements the same palette as :py:meth:`clut8_888` but
+ outputs RGB565 pixels.
+
+ .. note::
+
+ This function is unused within this file but needs to be
+ maintained alongside the reference clut so it is reproduced
+ here.
+
+ :param int i: Index (from 0..255 inclusive) into the CLUT
+ :return: 16-bit colour in RGB565 format
+ """
+ if i < 216:
+ rgb565 = (( i % 6) * 0x33) >> 3
+ rg = i // 6
+ rgb565 += ((rg % 6) * (0x33 << 3)) & 0x07e0
+ rgb565 += ((rg // 6) * (0x33 << 8)) & 0xf800
+ elif i < 252:
+ i -= 216
+ rgb565 = (0x7f + (( i % 3) * 0x33)) >> 3
+ rg = i // 3
+ rgb565 += ((0x4c << 3) + ((rg % 4) * (0x33 << 3))) & 0x07e0
+ rgb565 += ((0x7f << 8) + ((rg // 4) * (0x33 << 8))) & 0xf800
+ else:
+ i -= 252
+ gr6 = (0x2c + (0x10 * i)) >> 2
+ gr5 = gr6 >> 1
+ rgb565 = (gr5 << 11) + (gr6 << 5) + gr5
+
+ return rgb565
+
+class ReverseCLUT:
+ def __init__(self, clut):
+ l = []
+ for i in range(256):
+ l.append(clut(i))
+ self.clut = tuple(l)
+ self.lookup = {}
+
+ def __call__(self, rgb888):
+ """Compare rgb888 to every element of the CLUT and pick the
+ closest match.
+ """
+ if rgb888 in self.lookup:
+ return self.lookup[rgb888]
+
+ best = 200000
+ index = -1
+ clut = self.clut
+ r = rgb888 >> 16
+ g = (rgb888 >> 8) & 0xff
+ b = rgb888 & 0xff
+
+ for i in range(256):
+ candidate = clut[i]
+ rd = r - (candidate >> 16)
+ gd = g - ((candidate >> 8) & 0xff)
+ bd = b - (candidate & 0xff)
+ # This is the Euclidian distance (squared)
+ distance = rd * rd + gd * gd + bd * bd
+ if distance < best:
+ best = distance
+ index = i
+
+ self.lookup[rgb888] = index
+ #print(f'# #{rgb888:06x} -> #{clut8_rgb888(index):06x}')
+ return index
+
+def varname(p):
+ return os.path.basename(os.path.splitext(p)[0])
+
+def encode(im):
+ pixels = im.load()
+
+ rle = []
+ rl = 0
+ px = pixels[0, 0]
+
+ def encode_pixel(px, rl):
+ while rl > 255:
+ rle.append(255)
+ rle.append(0)
+ rl -= 255
+ rle.append(rl)
+
+ for y in range(im.height):
+ for x in range(im.width):
+ newpx = pixels[x, y]
+ if newpx == px:
+ rl += 1
+ assert(rl < (1 << 21))
+ continue
+
+ # Code the previous run
+ encode_pixel(px, rl)
+
+ # Start a new run
+ rl = 1
+ px = newpx
+
+ # Handle the final run
+ encode_pixel(px, rl)
+
+ return (im.width, im.height, bytes(rle))
+
+def encode_2bit(im):
+ """2-bit palette based RLE encoder.
+
+ This encoder has a reprogrammable 2-bit palette. This allows it to encode
+ arbitrary images with a full 8-bit depth but the 2-byte overhead each time
+ a new colour is introduced means it is not efficient unless the image is
+ carefully constructed to keep a good locality of reference for the three
+ non-background colours.
+
+ The encoding competes well with the 1-bit encoder for small monochrome
+ images but once run-lengths longer than 62 start to become frequent then
+ this encoding is about 30% larger than a 1-bit encoding.
+ """
+ pixels = im.load()
+ assert(im.width <= 255)
+ assert(im.height <= 255)
+
+ full_palette = ReverseCLUT(clut8_rgb888)
+
+ rle = []
+ rl = 0
+ px = pixels[0, 0]
+ # black, grey25, grey50, white
+ palette = [0, 254, 219, 215]
+ next_color = 1
+
+ def encode_pixel(px, rl):
+ nonlocal next_color
+ px = full_palette((px[0] << 16) + (px[1] << 8) + px[2])
+ if px not in palette:
+ rle.append(next_color << 6)
+ rle.append(px)
+ palette[next_color] = px
+ next_color += 1
+ if next_color >= len(palette):
+ next_color = 1
+ px = palette.index(px)
+ if rl >= 63:
+ rle.append((px << 6) + 63)
+ rl -= 63
+ while rl >= 255:
+ rle.append(255)
+ rl -= 255
+ rle.append(rl)
+ else:
+ rle.append((px << 6) + rl)
+
+ # Issue the descriptor
+ rle.append(2)
+ rle.append(im.width)
+ rle.append(im.height)
+
+ for y in range(im.height):
+ for x in range(im.width):
+ newpx = pixels[x, y]
+ if newpx == px:
+ rl += 1
+ assert(rl < (1 << 21))
+ continue
+
+ # Code the previous run
+ encode_pixel(px, rl)
+
+ # Start a new run
+ rl = 1
+ px = newpx
+
+ # Handle the final run
+ encode_pixel(px, rl)
+
+ return bytes(rle)
+
+def encode_8bit(im):
+ """Experimental 8-bit RLE encoder.
+
+ For monochrome images this is about 3x less efficient than the 1-bit
+ encoder. This encoder is not currently used anywhere in wasp-os and
+ currently there is no decoder either (so don't assume this code
+ actually works).
+ """
+ pixels = im.load()
+
+ rle = []
+ rl = 0
+ px = pixels[0, 0]
+
+ def encode_pixel(px, rl):
+ px = (px[0] & 0xe0) | ((px[1] & 0xe0) >> 3) | ((px[2] & 0xc0) >> 6)
+
+ rle.append(px)
+ if rl > 0:
+ rle.append(px)
+ rl -= 2
+ if rl > (1 << 14):
+ rle.append(0x80 | ((rl >> 14) & 0x7f))
+ if rl > (1 << 7):
+ rle.append(0x80 | ((rl >> 7) & 0x7f))
+ if rl >= 0:
+ rle.append( rl & 0x7f )
+
+ for y in range(im.height):
+ for x in range(im.width):
+ newpx = pixels[x, y]
+ if newpx == px:
+ rl += 1
+ assert(rl < (1 << 21))
+ continue
+
+ # Code the previous run
+ encode_pixel(px, rl)
+
+ # Start a new run
+ rl = 1
+ px = newpx
+
+ # Handle the final run
+ encode_pixel(px, rl)
+
+ return (im.width, im.height, bytes(rle))
+
+def render_c(image, fname, indent, depth):
+ extra_indent = ' ' * indent
+ if len(image) == 3:
+ print(f'{extra_indent}// {depth}-bit RLE, generated from {fname}, '
+ f'{len(image[2])} bytes')
+ (x, y, pixels) = image
+ else:
+ print(f'{extra_indent}// {depth}-bit RLE, generated from {fname}, '
+ f'{len(image)} bytes')
+ pixels = image
+
+ print(f'{extra_indent}static const uint8_t {varname(fname)}[] = {{')
+ print(f'{extra_indent} ', end='')
+ i = 0
+ for rl in pixels:
+ print(f' {hex(rl)},', end='')
+
+ i += 1
+ if i == 12:
+ print(f'\n{extra_indent} ', end='')
+ i = 0
+ print('\n};')
+
+def render_py(image, fname, indent, depth):
+ extra_indent = ' ' * indent
+ if len(image) == 3:
+ print(f'{extra_indent}# {depth}-bit RLE, generated from {fname}, '
+ f'{len(image[2])} bytes')
+ (x, y, pixels) = image
+ print(f'{extra_indent}{varname(fname)} = (')
+ print(f'{extra_indent} {x}, {y},')
+ else:
+ print(f'{extra_indent}# {depth}-bit RLE, generated from {fname}, '
+ f'{len(image)} bytes')
+ pixels = image[3:]
+ print(f'{extra_indent}{varname(fname)} = (')
+ print(f'{extra_indent} {image[0:1]}')
+ print(f'{extra_indent} {image[1:3]}')
+
+ # Split the bytestring to ensure each line is short enough to
+ # be absorbed on the target if needed.
+ for i in range(0, len(pixels), 16):
+ print(f'{extra_indent} {pixels[i:i+16]}')
+ print(f'{extra_indent})')
+
+
+def decode_to_ascii(image):
+ (sx, sy, rle) = image
+ data = bytearray(2*sx)
+ dp = 0
+ black = ord('#')
+ white = ord(' ')
+ color = black
+
+ for rl in rle:
+ while rl:
+ data[dp] = color
+ data[dp+1] = color
+ dp += 2
+ rl -= 1
+
+ if dp >= (2*sx):
+ print(data.decode('utf-8'))
+ dp = 0
+
+ if color == black:
+ color = white
+ else:
+ color = black
+
+ # Check the image is the correct length
+ assert(dp == 0)
+
+parser = argparse.ArgumentParser(description='RLE encoder tool.')
+parser.add_argument('files', nargs='+',
+ help='files to be encoded')
+parser.add_argument('--ascii', action='store_true',
+ help='Run the resulting image(s) through an ascii art decoder')
+parser.add_argument('--c', action='store_true',
+ help='Render the output as C instead of python')
+parser.add_argument('--indent', default=0, type=int,
+ help='Add extra indentation in the generated code')
+parser.add_argument('--2bit', action='store_true', dest='twobit',
+ help='Generate 2-bit image')
+parser.add_argument('--8bit', action='store_true', dest='eightbit',
+ help='Generate 8-bit image')
+
+args = parser.parse_args()
+if args.eightbit:
+ encoder = encode_8bit
+ depth = 8
+elif args.twobit:
+ encoder = encode_2bit
+ depth = 2
+else:
+ encoder = encode
+ depth =1
+
+for fname in args.files:
+ image = encoder(Image.open(fname))
+
+ if args.c:
+ render_c(image, fname, args.indent, depth)
+ else:
+ render_py(image, fname, args.indent, depth)
+
+ if args.ascii:
+ print()
+ decode_to_ascii(image) \ No newline at end of file