From 5d760b7a4a31545f6c477770ff82766bd163331d Mon Sep 17 00:00:00 2001 From: Pranav Sonawane Date: Wed, 5 Aug 2026 15:36:07 +0530 Subject: [PATCH] change(mbedtls/crt_bundle): Unify gen_crt_bundle output with esp-pylib gen_crt_bundle.py now uses the shared esp-pylib library instead of its own logging and argument parsing, so its messages look the same as the rest of the ESP-IDF Python tools. --- .../mbedtls/esp_crt_bundle/gen_crt_bundle.py | 161 ++++++++++-------- .../test_gen_crt_bundle.py | 121 +++++++++++-- 2 files changed, 192 insertions(+), 90 deletions(-) diff --git a/components/mbedtls/esp_crt_bundle/gen_crt_bundle.py b/components/mbedtls/esp_crt_bundle/gen_crt_bundle.py index c5e737ceaf2..27ca53ed3ff 100755 --- a/components/mbedtls/esp_crt_bundle/gen_crt_bundle.py +++ b/components/mbedtls/esp_crt_bundle/gen_crt_bundle.py @@ -8,16 +8,21 @@ # The bundle will have the format: number of certificates; crt 1 subject name length; crt 1 public key length; # crt 1 subject name; crt 1 public key; crt 2... # -# SPDX-FileCopyrightText: 2018-2025 Espressif Systems (Shanghai) CO LTD +# SPDX-FileCopyrightText: 2018-2026 Espressif Systems (Shanghai) CO LTD # SPDX-License-Identifier: Apache-2.0 -import argparse import csv import os import re import struct -import sys import warnings +import rich_click as click +from esp_pylib.cli_options import OptionEatAll +from esp_pylib.errors import FatalError +from esp_pylib.excepthook import install_exception_reporting +from esp_pylib.logger import log +from rich.markup import escape + DEFAULT_CERT_BUNDLE_MAX_CERTS = 200 # Ignore warning about non-positive serial numbers in certificates @@ -37,12 +42,11 @@ try: from cryptography.hazmat.backends import default_backend from cryptography.hazmat.primitives import serialization except ImportError: - print( - 'The cryptography package is not installed.' + log.die( + 'The cryptography package is not installed. ' 'Please refer to the Get Started section of the ESP-IDF Programming Guide for ' 'setting up the required packages.' ) - raise ca_bundle_bin_file = 'x509_crt_bundle' @@ -54,16 +58,9 @@ quiet = False def status(msg): - """Print status message to stderr""" + """Print a non-critical status message unless --quiet was given""" if not quiet: - critical(msg) - - -def critical(msg): - """Print critical message to stderr""" - sys.stderr.write('gen_crt_bundle.py: ') - sys.stderr.write(msg) - sys.stderr.write('\n') + log.print(msg) class CertificateBundle: @@ -77,7 +74,7 @@ class CertificateBundle: found |= self.add_from_file(os.path.join(crts_path, file_path), strict=False) if found is False: - raise InputError('No valid x509 certificates found in %s' % crts_path) + raise InputError(f'No valid x509 certificates found in {crts_path}') def add_from_file(self, file_path, strict=True): """Parse a certificate file @@ -88,14 +85,14 @@ class CertificateBundle: """ try: if file_path.endswith(PEM_FILE_EXTENSION): - status('Parsing certificates from %s' % file_path) - with open(file_path, 'r', encoding='utf-8') as f: + status(f'Parsing certificates from {escape(file_path)}') + with open(file_path, encoding='utf-8') as f: crt_str = f.read() self.add_from_pem(crt_str) return True elif file_path.endswith(DER_FILE_EXTENSION): - status('Parsing certificates from %s' % file_path) + status(f'Parsing certificates from {escape(file_path)}') with open(file_path, 'rb') as f: crt_str = f.read() self.add_from_der(crt_str) @@ -103,18 +100,16 @@ class CertificateBundle: elif os.path.isfile(file_path): msg = ( - 'Unsupported file extension in %s, certificates are only parsed from files with a ' - '%s (PEM encoded) or %s (DER encoded) extension. ' + f'Unsupported file extension in {file_path}, certificates are only parsed from files with a ' + f'{PEM_FILE_EXTENSION} (PEM encoded) or {DER_FILE_EXTENSION} (DER encoded) extension. ' 'Please rename the file to match its encoding.' - % (file_path, PEM_FILE_EXTENSION, DER_FILE_EXTENSION) ) if strict: raise InputError(msg) - critical('Warning: skipping file. %s' % msg) + log.warn(f'Skipping file. {escape(msg)}') except ValueError: - critical('Invalid certificate in %s' % file_path) - raise InputError('Invalid certificate') + raise InputError(f'Invalid certificate in {file_path}') return False @@ -140,7 +135,7 @@ class CertificateBundle: if count == 0: status('No certificate found') else: - status('Successfully added %d certificates' % count) + status(f'Successfully added {count} certificates') def add_from_der(self, crt_str): self.certificates.append(x509.load_der_x509_certificate(crt_str, default_backend())) @@ -148,12 +143,11 @@ class CertificateBundle: def create_bundle(self, max_certs=DEFAULT_CERT_BUNDLE_MAX_CERTS): if max_certs < len(self.certificates): - critical( - f'No. of certs in the certificate bundle = {len(self.certificates)} exceeds\n \ - Max allowed certificates in the certificate bundle = {max_certs} \ - Please update the menuconfig option with appropriate value' + raise InputError( + f'The certificate bundle has {len(self.certificates)} certificates, ' + f'exceeding the maximum of {max_certs}. ' + 'Please update the menuconfig option with an appropriate value.' ) - raise ValueError # Sort certificates in order to do binary search when looking up certificates self.certificates = sorted(self.certificates, key=lambda cert: cert.subject.public_bytes(default_backend())) @@ -186,13 +180,13 @@ class CertificateBundle: bundle += pub_key_der # Output all offsets before the first certificate - bundle = struct.pack('<{0:d}L'.format(len(offsets)), *offsets) + bundle + bundle = struct.pack(f'<{len(offsets):d}L', *offsets) + bundle return bundle def add_with_filter(self, crts_path, filter_path): filter_set = set() - with open(filter_path, 'r', encoding='utf-8') as f: + with open(filter_path, encoding='utf-8') as f: csv_reader = csv.reader(f, delimiter=',') # Skip header @@ -200,9 +194,9 @@ class CertificateBundle: for row in csv_reader: filter_set.add(row[1]) - status('Parsing certificates from %s' % crts_path) + status(f'Parsing certificates from {escape(crts_path)}') crt_str = [] - with open(crts_path, 'r', encoding='utf-8') as f: + with open(crts_path, encoding='utf-8') as f: crt_str = f.read() # Split all certs into a list of (name, certificate string) tuples @@ -216,66 +210,83 @@ class CertificateBundle: self.add_from_pem(filtered_crts) -class InputError(RuntimeError): - def __init__(self, e): - super(InputError, self).__init__(e) +class InputError(FatalError): + """Raised when the given certificate input cannot be turned into a bundle""" + + pass -def main(): +@click.command(context_settings={'help_option_names': ['-h', '--help']}) +@click.option( + '--quiet', + '-q', + 'quiet_flag', + is_flag=True, + help="Don't print non-critical status messages", +) +@click.option( + '--input', + '-i', + 'input_paths', + multiple=True, + required=True, + cls=OptionEatAll, + type=str, + help='Paths to the custom certificate folders or files to parse, parses all .pem or .der files', +) +@click.option( + '--filter', + '-f', + 'filter_path', + help='Path to CSV-file where the second column contains the name of the certificates ' + 'that should be included from cacrt_all.pem', +) +@click.option( + '--max-certs', + '-m', + type=int, + default=DEFAULT_CERT_BUNDLE_MAX_CERTS, + show_default=True, + help='Maximum number of certificates allowed in the certificate bundle', +) +def cli(quiet_flag, input_paths, filter_path, max_certs): + """ESP-IDF x509 certificate bundle utility""" global quiet - parser = argparse.ArgumentParser(description='ESP-IDF x509 certificate bundle utility') + # The build system captures this output through a pipe, where Rich would wrap at 80 + # columns and break the certificate paths in the messages across lines. + log.set_console_options(soft_wrap=True) - parser.add_argument('--quiet', '-q', help="Don't print non-critical status messages to stderr", action='store_true') - parser.add_argument( - '--input', - '-i', - nargs='+', - required=True, - help='Paths to the custom certificate folders or files to parse, parses all .pem or .der files', - ) - parser.add_argument( - '--filter', - '-f', - help='Path to CSV-file where the second columns contains the name of the certificates \ - that should be included from cacrt_all.pem', - ) - parser.add_argument( - '--max-certs', - '-m', - help='Maximum number of certificates allowed in the certificate bundle', - type=int, - default=DEFAULT_CERT_BUNDLE_MAX_CERTS, - ) - - args = parser.parse_args() - - quiet = args.quiet + quiet = quiet_flag bundle = CertificateBundle() - for path in args.input: + for path in input_paths: if os.path.isfile(path): - if os.path.basename(path) == 'cacrt_all.pem' and args.filter: - bundle.add_with_filter(path, args.filter) + if os.path.basename(path) == 'cacrt_all.pem' and filter_path: + bundle.add_with_filter(path, filter_path) else: bundle.add_from_file(path) elif os.path.isdir(path): bundle.add_from_path(path) else: - raise InputError('Invalid --input=%s, is neither file nor folder' % args.input) + raise InputError(f'Invalid --input={path}, is neither file nor folder') - status('Successfully added %d certificates in total' % len(bundle.certificates)) + status(f'Successfully added {len(bundle.certificates)} certificates in total') - crt_bundle = bundle.create_bundle(args.max_certs) + crt_bundle = bundle.create_bundle(max_certs) with open(ca_bundle_bin_file, 'wb') as f: f.write(crt_bundle) -if __name__ == '__main__': +def main(): + install_exception_reporting() try: - main() + cli() except InputError as e: - print(e) - sys.exit(2) + log.die(escape(str(e)), exit_code=2) + + +if __name__ == '__main__': + main() diff --git a/components/mbedtls/esp_crt_bundle/test_gen_crt_bundle/test_gen_crt_bundle.py b/components/mbedtls/esp_crt_bundle/test_gen_crt_bundle/test_gen_crt_bundle.py index 7cf9faa9402..433e5f03426 100755 --- a/components/mbedtls/esp_crt_bundle/test_gen_crt_bundle/test_gen_crt_bundle.py +++ b/components/mbedtls/esp_crt_bundle/test_gen_crt_bundle/test_gen_crt_bundle.py @@ -1,6 +1,7 @@ #!/usr/bin/env python import os import shutil +import subprocess import sys import tempfile import unittest @@ -12,26 +13,26 @@ except ImportError: import gen_crt_bundle -idf_path = os.environ['IDF_PATH'] -ca_crts_path = idf_path + '/components/mbedtls/esp_crt_bundle/' -test_crts_path = idf_path + '/components/mbedtls/esp_crt_bundle/test_gen_crt_bundle/' +idf_path = os.environ['IDF_PATH'] +ca_crts_path = idf_path + '/components/mbedtls/esp_crt_bundle/' +test_crts_path = idf_path + '/components/mbedtls/esp_crt_bundle/test_gen_crt_bundle/' -ca_bundle_bin_file = 'x509_crt_bundle' +ca_bundle_bin_file = 'x509_crt_bundle' -der_test_file = 'baltimore.der' -pem_test_file = 'entrust.pem' +der_test_file = 'baltimore.der' +pem_test_file = 'entrust.pem' verified_der_bundle = 'baltimore_crt_bundle' verified_pem_bundle = 'entrust_crt_bundle' -invalid_test_file = 'invalid_crt.pem' -non_ascii_file = 'non_ascii_crt.pem' -unsupported_file = 'unsupported_ext_crt.crt' -ca_crts_all_file = 'cacrt_all.pem' +invalid_test_file = 'invalid_crt.pem' +non_ascii_file = 'non_ascii_crt.pem' +unsupported_file = 'unsupported_ext_crt.crt' +ca_crts_all_file = 'cacrt_all.pem' +cmn_filter_file = 'cmn_crt_authorities.csv' class Py23TestCase(unittest.TestCase): - def __init__(self, *args, **kwargs): - super(Py23TestCase, self).__init__(*args, **kwargs) + super().__init__(*args, **kwargs) try: self.assertRaisesRegex except AttributeError: @@ -41,7 +42,6 @@ class Py23TestCase(unittest.TestCase): class GenCrtBundleTests(Py23TestCase): - # Verify generation from der vs known certificate def test_gen_from_der(self): bundle = gen_crt_bundle.CertificateBundle() @@ -70,7 +70,7 @@ class GenCrtBundleTests(Py23TestCase): bundle = gen_crt_bundle.CertificateBundle() with self.assertRaisesRegex(gen_crt_bundle.InputError, 'Invalid certificate'): - bundle.add_from_file(test_crts_path + invalid_test_file) + bundle.add_from_file(test_crts_path + invalid_test_file) bundle_prev_len = len(bundle.certificates) bundle.add_from_pem('') @@ -110,9 +110,100 @@ class GenCrtBundleTests(Py23TestCase): def test_non_ascii_crt_input(self): bundle = gen_crt_bundle.CertificateBundle() - bundle.add_from_file(test_crts_path + non_ascii_file) + bundle.add_from_file(test_crts_path + non_ascii_file) self.assertTrue(len(bundle.certificates)) +class GenCrtBundleCliTests(unittest.TestCase): + """Exercise the command line interface, including the flag shapes used by the build system""" + + def run_tool(self, args, cwd): + return subprocess.run( + [sys.executable, gen_crt_bundle.__file__] + args, + cwd=cwd, + capture_output=True, + text=True, + ) + + def test_help(self): + with tempfile.TemporaryDirectory() as workdir: + result = self.run_tool(['-h'], workdir) + + self.assertEqual(result.returncode, 0) + self.assertIn('certificate bundle utility', result.stdout) + + # The build system passes several paths to a single --input, then further options + def test_multiple_input_paths(self): + with tempfile.TemporaryDirectory() as workdir: + result = self.run_tool( + [ + '--input', + test_crts_path + der_test_file, + test_crts_path + pem_test_file, + '-q', + '--max-certs', + '200', + ], + workdir, + ) + + self.assertEqual(result.returncode, 0, result.stderr) + # --quiet keeps the status messages out of the build log + self.assertEqual(result.stdout, '') + self.assertTrue(os.path.isfile(os.path.join(workdir, ca_bundle_bin_file))) + + def test_status_messages_on_stdout(self): + with tempfile.TemporaryDirectory() as workdir: + result = self.run_tool(['--input', test_crts_path + pem_test_file], workdir) + + self.assertEqual(result.returncode, 0, result.stderr) + self.assertIn('Parsing certificates from', result.stdout) + # Long paths are not wrapped, so they stay usable in the build log + self.assertIn(test_crts_path + pem_test_file, result.stdout) + + def test_filter(self): + with tempfile.TemporaryDirectory() as workdir: + result = self.run_tool( + [ + '--input', + ca_crts_path + ca_crts_all_file, + '--filter', + ca_crts_path + cmn_filter_file, + '-q', + ], + workdir, + ) + + self.assertEqual(result.returncode, 0, result.stderr) + self.assertTrue(os.path.getsize(os.path.join(workdir, ca_bundle_bin_file)) > 0) + + def test_missing_input(self): + with tempfile.TemporaryDirectory() as workdir: + result = self.run_tool([], workdir) + + self.assertEqual(result.returncode, 2) + self.assertIn('--input', result.stderr) + + # Input errors are reported as a single error line, not as a traceback + def test_unsupported_extension(self): + with tempfile.TemporaryDirectory() as workdir: + result = self.run_tool(['--input', test_crts_path + unsupported_file], workdir) + + self.assertEqual(result.returncode, 2) + self.assertIn('Unsupported file extension', result.stderr) + self.assertNotIn('Traceback', result.stderr) + + def test_max_certs_exceeded(self): + with tempfile.TemporaryDirectory() as workdir: + result = self.run_tool( + ['--input', test_crts_path + pem_test_file, '--max-certs', '0'], + workdir, + ) + + self.assertEqual(result.returncode, 2) + self.assertIn('exceeding the maximum', result.stderr) + self.assertNotIn('Traceback', result.stderr) + + if __name__ == '__main__': unittest.main()