Implement the sign() method in the pkcs11_tool btool to enable HSM-based cryptographic signing using the PKCS#11 standard. This provides the core PKCS#11 interface needed for secure signing with hardware security modules in binman workflows.
Signed-off-by: Yan WANG <[email protected]> --- tools/binman/btool/pkcs11_tool.py | 275 ++++++++++++++++++++++++++++++ 1 file changed, 275 insertions(+) diff --git a/tools/binman/btool/pkcs11_tool.py b/tools/binman/btool/pkcs11_tool.py index ca80943f98c..e380f585fc7 100644 --- a/tools/binman/btool/pkcs11_tool.py +++ b/tools/binman/btool/pkcs11_tool.py @@ -3,7 +3,12 @@ # """Bintool implementation for pkcs11-tool""" +import os +import re +import tempfile + from binman import bintool +from u_boot_pylib import tools class Bintoolpkcs11_tool(bintool.Bintool): @@ -22,6 +27,276 @@ class Bintoolpkcs11_tool(bintool.Bintool): 'PKCS #11 tokens managing tool', version_args='--show-info') + def _build_mechanism(self, hash_algo, sign_algo, padding): + """Build a PKCS#11 mechanism name from algorithm parameters. + + Derives the mechanism name dynamically from the hash algorithm, + signing algorithm, and padding, supporting any PKCS#11-compatible + combination without a fixed lookup table. + + The signing algorithm may include a key-size suffix (e.g. 'rsa2048', + 'ec256') which is stripped; only the algorithm family is used. + + Args: + hash_algo (str): Hash algorithm (e.g. 'sha256', 'sha384', 'sha512') + sign_algo (str): Signing algorithm (e.g. 'rsa2048', 'rsa4096', + 'ec256', 'ecdsa384') + padding (str): Padding scheme ('pkcs1v15', 'pkcs-1.5', or 'pss') + + Returns: + str: PKCS#11 mechanism name (e.g. 'SHA256-RSA-PKCS') + + Raises: + ValueError: If the algorithm family cannot be mapped to a mechanism + """ + hash_upper = hash_algo.upper() + + # Strip trailing digits to extract the algorithm family: + # 'rsa2048' -> 'RSA', 'ec256' -> 'EC', 'ecdsa384' -> 'ECDSA' + m = re.match(r'([a-zA-Z]+)', sign_algo) + if not m: + raise ValueError(f"Cannot parse signing algorithm: {sign_algo!r}") + algo_family = m.group(1).upper() + + # Normalize padding aliases + pad = padding.lower() + if pad == 'pkcs-1.5': + pad = 'pkcs1v15' + + if algo_family == 'RSA': + if pad == 'pss': + return f'{hash_upper}-RSA-PKCS-PSS' + else: + return f'{hash_upper}-RSA-PKCS' + elif algo_family in ('EC', 'ECDSA'): + return f'ECDSA-{hash_upper}' + else: + raise ValueError( + f"Unsupported signing algorithm: {sign_algo!r}" + ) + + def _verify_mechanism(self, mechanism, module, slot_index=None, + token_label=None): + """Verify a mechanism against the token's allowed mechanisms list. + + Calls pkcs11-tool --list-mechanisms to retrieve the set of mechanisms + supported by the token and raises an error if the requested mechanism + is absent. Verification is silently skipped when the list cannot be + retrieved (tool unavailable, non-zero exit, or unparseable output). + + Args: + mechanism (str): PKCS#11 mechanism name to verify + (e.g. 'SHA256-RSA-PKCS') + module (str): Path to PKCS#11 library (.so file) + slot_index (str): Slot index to target (optional) + token_label (str): Token label (optional, used when no slot_index) + + Raises: + ValueError: If mechanism is not in the token's allowed mechanisms + """ + args = ['--list-mechanisms', '--module', module] + if slot_index is not None: + args += ['--slot-index', slot_index] + elif token_label: + args += ['--token-label', token_label] + + result = self.run_cmd_result(*args, raise_on_error=False) + if result is None or result.return_code != 0: + return # Cannot verify; proceed and let pkcs11-tool report errors + + stdout = (result.stdout.decode('utf-8', errors='replace') + if isinstance(result.stdout, bytes) else result.stdout) + + # Parse mechanism names from lines like: + # SHA256-RSA-PKCS, keySize={512,4096}, hw, sign, verify + # Only accept names that match the PKCS#11 identifier format + # (uppercase alphanumeric with dashes, e.g. SHA256-RSA-PKCS). + allowed = [] + for line in stdout.splitlines(): + line = line.strip() + if not line or line.startswith(('Using ', 'Supported ', 'Mechanism')): + continue + mech_name = line.split(',')[0].strip() + if re.match(r'^[A-Z][A-Z0-9-]+$', mech_name): + allowed.append(mech_name) + + if allowed and mechanism not in allowed: + raise ValueError( + f"Mechanism '{mechanism}' is not in the token's allowed " + f"mechanisms.\nAllowed: {', '.join(sorted(allowed))}" + ) + + def find_slot_by_serial(self, module, serial): + """Find slot index by token serial number. + + Args: + module: Path to PKCS#11 library + serial: Token serial number to find + + Returns: + Slot index as string, or None if not found + """ + args = ['--module', module, '--list-slots'] + result = self.run_cmd(*args) + + # Parse output line by line to find matching serial + lines = result.stdout.split('\n') + current_slot = None + + for line in lines: + # Track current slot number + slot_match = re.search(r'Slot (\d+)', line) + if slot_match: + current_slot = slot_match.group(1) + + # Check for serial number match + if 'token serial num' in line.lower(): + serial_match = re.search(r'token serial num\s*:\s*(\S+)', line, + re.IGNORECASE) + if serial_match and serial_match.group(1) == serial: + return current_slot + + return None + + def sign(self, data, module, token_label=None, token_serial=None, + slot_index=None, key_label=None, key_id=None, pin=None, + hash_algo='sha256', sign_algo='rsa2048', padding='pkcs1v15'): + """Sign data using PKCS#11 HSM + + This method signs data using a private key stored in a PKCS#11 token + (HSM, smart card, or software token like SoftHSM). The data is hashed + and signed by the HSM using the specified mechanism. + + The mechanism name is derived dynamically from the algorithm parameters + and validated against the token's allowed mechanisms list before + signing. + + Args: + data (bytes): Data to sign (will be hashed and signed by the HSM) + module (str): Path to PKCS#11 library (.so file) + token_label (str): Token label (alternative to token_serial) + token_serial (str): Token serial number (alternative to token_label) + key_label (str): Private key label (alternative to key_id) + key_id (str): Private key ID (alternative to key_label) + pin (str): PIN/password to unlock the token + hash_algo (str): Hash algorithm ('sha256', 'sha384', 'sha512', …) + Default: 'sha256' + sign_algo (str): Signature algorithm with optional key size + (e.g. 'rsa2048', 'rsa4096', 'ec256') + Default: 'rsa2048' + padding (str): Padding scheme ('pkcs1v15', 'pkcs-1.5', or 'pss') + Default: 'pkcs1v15' + + Returns: + bytes: Signature bytes + + Raises: + ValueError: If mechanism is unsupported, parameters are invalid, + or the mechanism is not in the token's allowed list + RuntimeError: If signing operation fails + """ + # Build mechanism name dynamically from algorithm parameters + mechanism = self._build_mechanism(hash_algo, sign_algo, padding) + + # Validate token identification + if not token_label and not token_serial and not slot_index: + raise ValueError( + "Must specify either token_label or token_serial or slot_index" + ) + + # Validate key identification + if not key_label and not key_id: + raise ValueError( + "Must specify either key_label or key_id" + ) + + # Validate module path + if not module: + raise ValueError("PKCS#11 module path is required") + + if not os.path.exists(module): + raise ValueError(f"PKCS#11 module not found: {module}") + + # Resolve slot from serial number if needed (done early so that + # mechanism verification targets the correct token) + resolved_slot = slot_index + if not slot_index and token_serial: + slot_idx = self.find_slot_by_serial(module, token_serial) + if slot_idx: + resolved_slot = slot_idx + else: + raise ValueError( + f"Token with serial {token_serial} not found" + ) + + # Verify mechanism is in the token's allowed mechanisms list + self._verify_mechanism( + mechanism, module, + slot_index=resolved_slot, + token_label=token_label if not resolved_slot else None, + ) + + # Write data to temporary file (pkcs11-tool requires file input) + input_fd, input_file = tempfile.mkstemp(suffix='.bin', + prefix='binman_sign_') + try: + os.write(input_fd, data) + os.close(input_fd) + + # Build pkcs11-tool command + args = [ + '--sign', + '--mechanism', mechanism, + '--module', module, + '--input-file', input_file, + ] + + # Add token identification + if resolved_slot: + args += ['--slot-index', resolved_slot] + elif token_label: + args += ['--token-label', token_label] + + # Add key identification + if key_label: + args += ['--label', key_label] + elif key_id: + args += ['--id', key_id] + + # Add PIN and login + if pin: + args += ['--pin', pin] + args += ['--login'] + + # Execute signing (signature goes to stdout in binary format) + result = self.run_cmd_result(*args, binary=True, + raise_on_error=False) + + if result is None: + raise RuntimeError( + "pkcs11-tool not available. Install the 'opensc' package" + ) + + if result.return_code != 0: + stderr = (result.stderr.decode('utf-8', errors='replace') + if result.stderr else '') + raise RuntimeError( + f"pkcs11-tool signing failed " + f"(exit code {result.return_code}):\n{stderr}" + ) + + if not result.stdout: + raise RuntimeError("pkcs11-tool produced no signature output") + + return result.stdout + + finally: + # Clean up temporary file + try: + os.unlink(input_file) + except OSError: + pass + def fetch(self, method): """Install opensc via APT """ if method != bintool.FETCH_BIN: -- 2.43.0
