Mirrors/esp-idf
1
0
Fork 0
mirror of https://github.com/espressif/esp-idf.git synced 2026-09-22 13:01:16 +03:00
Code Issues Packages Projects Releases Wiki Activity

change(mbedtls/crt_bundle): Unify gen_crt_bundle output with esp-pylib

Browse Source 
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.
This commit is contained in:
Pranav Sonawane
2026-08-05 15:36:07 +05:30
committed by Mahavir Jain
parent 30e851857c
commit 5d760b7a4a
2 changed files with 192 additions and 90 deletions
Download Patch File Download Diff File Expand all files Collapse all files

161
components/mbedtls/esp_crt_bundle/gen_crt_bundle.py
View File

@@ -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()

121
components/mbedtls/esp_crt_bundle/test_gen_crt_bundle/test_gen_crt_bundle.py
View File

@@ -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()