Extend the pre-load entry type to support signing via hardware
security modules (HSM) using PKCS#11. The implementation integrates
with the pkcs11_tool btool to perform signing operations.

Signed-off-by: yan wang <[email protected]>
---
Changes in v2:
- Replace the hsm-support, hsm-library, hsm-token-label, hsm-token-serial,
  hsm-slot-index, hsm-key-label, hsm-key-id and hsm-pin properties with a
  single hsm-key PKCS#11 URI (RFC 7512), in line with FIT signing
- Do not store the PIN in the devicetree: use -a pre-load-hsm-pin or the
  PKCS11_SIGN_PIN environment variable
- Take the library path from -a pre-load-hsm-library, the hsm-library
  property or the module-path attribute of the URI
- Drop the _SignData() refactoring of v1; keep the file-based code
  unchanged and add a separate _SignHsmData() helper to reduce the diff
- Write the data to a file in the output directory, as sign() now takes a
  file
- Add ECDSA support
- Validate the HSM configuration early in ReadNode(), before the tool is
  looked up
- Check the length of the returned signature
- If pkcs11-tool is missing, record it as a missing bintool and emit a
  dummy header instead of raising an error

 tools/binman/etype/pre_load.py | 222 ++++++++++++++++++++++++++++++++-
 1 file changed, 221 insertions(+), 1 deletion(-)

diff --git a/tools/binman/etype/pre_load.py b/tools/binman/etype/pre_load.py
index 49a63ecaafa..14bfb23afb5 100644
--- a/tools/binman/etype/pre_load.py
+++ b/tools/binman/etype/pre_load.py
@@ -7,6 +7,7 @@
 
 import os
 import struct
+from urllib.parse import unquote
 from dtoc import fdt_util
 from u_boot_pylib import tools
 
@@ -21,6 +22,34 @@ from Cryptodome.Signature import pkcs1_15
 from Cryptodome.Signature import pss
 from Cryptodome.Signature import DSS
 
+def _parse_pkcs11_uri(uri):
+    """Parse a PKCS#11 URI (RFC 7512) into a dict of path and query attributes.
+
+    Path attributes (e.g. token=, object=, serial=, slot-id=) are separated by
+    semicolons. Query attributes (e.g. module-path=) follow '?' and are
+    separated by ampersands.
+
+    Args:
+        uri (str): PKCS#11 URI string starting with 'pkcs11:'
+
+    Returns:
+        dict: Mapping of attribute name to percent-decoded value
+
+    Raises:
+        ValueError: If the string does not start with 'pkcs11:'
+    """
+    if not uri.startswith('pkcs11:'):
+        raise ValueError(f'Not a PKCS#11 URI: {uri!r}')
+    rest = uri[7:]
+    path_part, _, query_part = rest.partition('?')
+    attrs = {}
+    for part in (path_part.split(';') + (query_part.split('&') if query_part 
else [])):
+        if '=' in part:
+            k, v = part.split('=', 1)
+            attrs[k.strip()] = unquote(v.strip())
+    return attrs
+
+
 PRE_LOAD_MAGIC = b'UBSH'
 
 RSAS = {
@@ -53,6 +82,37 @@ class Entry_pre_load(Entry_collection):
         - key-name: Filename of the private key to sign
         - header-size: Total size of the header
         - version: Version of the header
+        - pre-load-hsm-library: Path to the PKCS#11 module (.so). This is a
+          host-specific detail that varies by distribution and should not be
+          committed to the board DTS. Pass it at build time with
+          -a pre-load-hsm-library=... . It may also be set via the hsm-library
+          DT property when the build host is always the same, or embedded in
+          the hsm-key URI as the module-path= query attribute.
+        - pre-load-hsm-pin: PIN/passphrase to unlock the PKCS#11 token. Must
+          NOT be stored in the DT. May also be read from the PKCS11_SIGN_PIN
+          environment variable, mirroring how mkimage reads MKIMAGE_SIGN_PIN
+          for FIT signing.
+
+    HSM Signing Properties (optional):
+        When hsm-key is present, signing is done via a PKCS#11 token instead
+        of a key file, following the same convention as FIT signing.
+
+        - hsm-key: PKCS#11 URI (RFC 7512) identifying the token and private
+          key. Supported path attributes: token=, serial=, slot-id=, object=,
+          id=. The module-path= query attribute may embed the library path in
+          the URI as an alternative to the pre-load-hsm-library entry arg.
+          Examples::
+
+              hsm-key = "pkcs11:token=uboot-hsm;object=uboot-key";
+              hsm-key = "pkcs11:serial=abc123;object=uboot-key";
+              hsm-key = "pkcs11:slot-id=0;object=uboot-key";
+
+    This entry creates a pre-load header that contains a global image
+    signature. The signature can be created using either a traditional RSA
+    key file (key-name) or a Hardware Security Module (HSM) via PKCS#11
+    (hsm-key), enabling production keys to remain on the HSM and never touch
+    the build machine's filesystem.
+
 
     This entry creates a pre-load header that contains a global
     image signature.
@@ -77,6 +137,28 @@ class Entry_pre_load(Entry_collection):
                 };
             };
         };
+
+    Example with HSM signing::
+
+        binman {
+            pre-load {
+                content = <&image>;
+                algo-name = "sha256,rsa2048";
+                padding-name = "pss";
+                header-size = <4096>;
+                version = <1>;
+                hsm-key = "pkcs11:token=uboot-hsm;object=uboot-key";
+            };
+
+            image: blob-ext {
+                filename = "sandbox.itb";
+            };
+        };
+
+    Invoke binman with::
+
+        binman ... -a pre-load-hsm-library=/usr/lib/softhsm/libsofthsm2.so \\
+                   -a pre-load-hsm-pin=1234
     """
 
     def __init__(self, section, etype, node):
@@ -87,6 +169,18 @@ class Entry_pre_load(Entry_collection):
         self.header_size = fdt_util.GetInt(self._node, 'header-size')
         self.version = fdt_util.GetInt(self._node, 'version')
 
+        # HSM properties (optional); populated from hsm-key URI in ReadNode()
+        self.hsm_key_uri = fdt_util.GetString(self._node, 'hsm-key')
+        self.hsm_library = fdt_util.GetString(self._node, 'hsm-library')
+        self.hsm_token_label = None
+        self.hsm_token_serial = None
+        self.hsm_slot_index = None
+        self.hsm_key_label = None
+        self.hsm_key_id = None
+        self.hsm_pin = None
+
+        self.pkcs11_tool = None
+
     def ReadNode(self):
         super().ReadNode()
         self.key_path, = self.GetEntryArgsOrProps(
@@ -94,6 +188,50 @@ class Entry_pre_load(Entry_collection):
         if self.key_path is None:
             self.key_path = ''
 
+        # hsm-library may be supplied as an entry arg so the host-specific
+        # library path is not committed to the board DTS.
+        hsm_lib_arg, = self.GetEntryArgsOrProps(
+            [EntryArg('pre-load-hsm-library', str)])
+        if hsm_lib_arg:
+            self.hsm_library = hsm_lib_arg
+
+        # PIN from entry arg or environment variable, NOT from the DT.
+        pin_arg, = self.GetEntryArgsOrProps([EntryArg('pre-load-hsm-pin', 
str)])
+        self.hsm_pin = pin_arg or os.environ.get('PKCS11_SIGN_PIN')
+
+        # Validate HSM config early, before any tool is loaded, so a bad DT
+        # fails immediately regardless of whether pkcs11-tool is installed.
+        if self.hsm_key_uri:
+            try:
+                uri_attrs = _parse_pkcs11_uri(self.hsm_key_uri)
+            except ValueError as e:
+                self.Raise(str(e))
+            self.hsm_token_label = uri_attrs.get('token')
+            self.hsm_token_serial = uri_attrs.get('serial')
+            self.hsm_slot_index = uri_attrs.get('slot-id')
+            self.hsm_key_label = uri_attrs.get('object')
+            self.hsm_key_id = uri_attrs.get('id')
+            # module-path query attribute may carry the library path; the
+            # pre-load-hsm-library entry arg (already applied above) takes
+            # priority since it is more explicit.
+            if 'module-path' in uri_attrs and not self.hsm_library:
+                self.hsm_library = uri_attrs['module-path']
+
+            hash_name, sign_name = self.algo_name.split(',')
+            if hash_name not in SHAS:
+                self.Raise(f'{hash_name} is not supported')
+            if sign_name not in RSAS and sign_name not in ECDSAS:
+                self.Raise(f'{sign_name} is not supported')
+            padding_name = self.padding_name or 'pkcs-1.5'
+            if padding_name not in ('pss', 'pkcs-1.5'):
+                self.Raise(f'Unsupported padding scheme: {padding_name}')
+
+    def AddBintools(self, btools):
+        """Register pkcs11-tool if HSM signing is requested"""
+        super().AddBintools(btools)
+        if self.hsm_key_uri:
+            self.pkcs11_tool = self.AddBintool(btools, 'pkcs11_tool')
+
     def _CreateHeaderRsa(self, hash_name, sign_name, padding_name, key_name):
         # Check hash and signature name/type
         if hash_name not in SHAS:
@@ -197,8 +335,90 @@ class Entry_pre_load(Entry_collection):
 
         return data + pad
 
+    def _SignHsmData(self, data):
+        """Sign data with the configured HSM via PKCS#11."""
+        hash_name, sign_name = self.algo_name.split(',')
+        padding_name = self.padding_name or 'pkcs-1.5'
+
+        if hash_name not in SHAS:
+            self.Raise(f'{hash_name} is not supported')
+        if sign_name not in RSAS and sign_name not in ECDSAS:
+            self.Raise(f'{sign_name} is not supported')
+
+        padding = {'pss': 'pss', 'pkcs-1.5': 'pkcs1v15'}.get(padding_name)
+        if padding is None:
+            self.Raise(f'Unsupported padding scheme: {padding_name}')
+
+        if not self.pkcs11_tool.is_present():
+            self.record_missing_bintool(self.pkcs11_tool)
+            return None
+
+        input_fname = tools.get_output_filename(
+            f'{self.GetUniqueName()}_hsm_input.bin')
+        tools.write_file(input_fname, data)
+
+        try:
+            signature = self.pkcs11_tool.sign(
+                input_file=input_fname,
+                module=self.hsm_library,
+                token_label=self.hsm_token_label,
+                token_serial=self.hsm_token_serial,
+                slot_index=self.hsm_slot_index,
+                key_label=self.hsm_key_label,
+                key_id=self.hsm_key_id,
+                pin=self.hsm_pin,
+                hash_algo=hash_name,
+                sign_algo=sign_name,
+                padding=padding)
+        except (ValueError, RuntimeError) as e:
+            self.Raise(f'HSM signing failed: {e}')
+
+        expected = (RSAS[sign_name] if sign_name in RSAS
+                    else ECDSAS[sign_name])
+        if len(signature) != expected:
+            self.Raise(
+                f'HSM returned {len(signature)}-byte signature but '
+                f'{sign_name} requires {expected} bytes')
+        return signature
+
     def _CreateHeader(self):
-        """Create a pre load header"""
+        """Create a pre-load header.
+
+        When HSM signing is active the pkcs11-tool btool is called twice per
+        _CreateHeader() invocation (once for the image signature and once for
+        the header signature). Binman may call ProcessContents() several times
+        as entries settle, so consider caching the slot and mechanism results
+        in a real deployment against a rate-limited HSM.
+        """
+        if self.hsm_key_uri:
+            # HSM path: use the HSM signing helper
+            img_sig = self._SignHsmData(self.image)
+            if img_sig is None:
+                # pkcs11-tool is missing; return a dummy header
+                return tools.get_bytes(0, self.header_size)
+
+            hash_sig = SHA256.new(img_sig)
+
+            version = self.version
+            header_size = self.header_size
+            image_size = len(self.image)
+            ofs_img_sig = 64 + len(img_sig)
+            flags = 0
+            reserved0 = 0
+            reserved1 = 0
+
+            first_header = struct.pack('>4sIIIIIII32s', PRE_LOAD_MAGIC,
+                                       version, header_size, image_size,
+                                       ofs_img_sig, flags, reserved0,
+                                       reserved1, hash_sig.digest())
+
+            hdr_sig = self._SignHsmData(first_header)
+            if hdr_sig is None:
+                return tools.get_bytes(0, self.header_size)
+
+            data = first_header + hdr_sig + img_sig
+            pad = bytearray(self.header_size - len(data))
+            return data + pad
 
         # File-based path: dispatch on signing algorithm
         if not self.key_name:
-- 
2.43.0

Reply via email to