From 5fd64f217155efbc4034ed453451b043e56a8554 Mon Sep 17 00:00:00 2001 From: TD-er Date: Tue, 19 Oct 2021 16:02:40 +0200 Subject: [PATCH] [MQTT TLS] Add cert.py to extract certificate info --- platformio_core_defs.ini | 2 +- requirements.txt | 1 + tools/pio/cert.py | 127 +++++++++++++++++++++++++++++++++++++++ 3 files changed, 129 insertions(+), 1 deletion(-) create mode 100644 tools/pio/cert.py diff --git a/platformio_core_defs.ini b/platformio_core_defs.ini index bedee8a98..750355cb5 100644 --- a/platformio_core_defs.ini +++ b/platformio_core_defs.ini @@ -150,7 +150,7 @@ build_flags = -DESP32_STAGE [core_esp32_3_3_0] platform = espressif32 @ 3.3.0 platform_packages = framework-arduinoespressif32 -build_flags = -DESP32_STAGE +build_flags = [core_esp32_3_3_2] platform = espressif32 @ 3.3.2 diff --git a/requirements.txt b/requirements.txt index 4d9c71783..d33e57459 100644 --- a/requirements.txt +++ b/requirements.txt @@ -9,6 +9,7 @@ chardet==4.0.0 click==7.1.2 colorama==0.4.3 commonmark==0.9.1 +cryptography==35.0.0 docutils==0.16 idna==2.10 imagesize==1.2.0 diff --git a/tools/pio/cert.py b/tools/pio/cert.py new file mode 100644 index 000000000..3ade0f5cc --- /dev/null +++ b/tools/pio/cert.py @@ -0,0 +1,127 @@ +#!/usr/bin/env python3 + +# Script to download/update certificates and public keys +# and generate compilable source files for c++/Arduino. +# released to public domain + +# Original: https://github.com/esp8266/Arduino/blob/master/tools/cert.py + +import urllib.request +import re +import ssl +import sys +import socket +import argparse +import datetime + +from cryptography import x509 +from cryptography.hazmat.primitives import hashes +from cryptography.hazmat.primitives.serialization import pkcs7 +from cryptography.hazmat.primitives.serialization import Encoding +from cryptography.hazmat.primitives.serialization import PublicFormat + +def printData(data, showPub = True): + try: + xcert = x509.load_der_x509_certificate(data) + except: + try: + xcert = x509.load_pem_x509_certificate(data) + except: + try: + xcert = pkcs7.load_der_pkcs7_certificates(data) + except: + xcert = pkcs7.load_pem_pkcs7_certificates(data) + if len(xcert) > 1: + print('// Warning: TODO: pkcs7 has {} entries'.format(len(xcert))) + xcert = xcert[0] + + cn = '' + for dn in xcert.subject.rfc4514_string().split(','): + keyval = dn.split('=') + if keyval[0] == 'CN': + cn += keyval[1] + name = re.sub('[^a-zA-Z0-9_]', '_', cn) + print('// CN: {} => name: {}'.format(cn, name)) + + print('// not valid before:', xcert.not_valid_before) + print('// not valid after: ', xcert.not_valid_after) + + if showPub: + + fingerprint = xcert.fingerprint(hashes.SHA1()).hex(':') + print('const char fingerprint_{} [] PROGMEM = "{}";'.format(name, fingerprint)) + + pem = xcert.public_key().public_bytes(Encoding.PEM, PublicFormat.SubjectPublicKeyInfo).decode('utf-8') + print('const char pubkey_{} [] PROGMEM = R"PUBKEY('.format(name)) + print(pem + ')PUBKEY";') + + else: + + cert = xcert.public_bytes(Encoding.PEM).decode('utf-8') + print('const char cert_{} [] PROGMEM = R"CERT('.format(name)) + print(cert + ')CERT";') + + cas = [] + for ext in xcert.extensions: + if ext.oid == x509.ObjectIdentifier("1.3.6.1.5.5.7.1.1"): + for desc in ext.value: + if desc.access_method == x509.oid.AuthorityInformationAccessOID.CA_ISSUERS: + cas.append(desc.access_location.value) + for ca in cas: + with urllib.request.urlopen(ca) as crt: + print() + print('// ' + ca) + printData(crt.read(), False) + print() + +def get_certificate(hostname, port, name): + context = ssl.create_default_context() + context.check_hostname = False + context.verify_mode = ssl.CERT_NONE + with socket.create_connection((hostname, port)) as sock: + with context.wrap_socket(sock, server_hostname=hostname) as ssock: + print('////////////////////////////////////////////////////////////') + print('// certificate chain for {}:{}'.format(hostname, port)) + print() + if name: + print('const char* {}_host = "{}";'.format(name, hostname)); + print('const uint16_t {}_port = {};'.format(name, port)); + print() + printData(ssock.getpeercert(binary_form=True)) + print('// end of certificate chain for {}:{}'.format(hostname, port)) + print('////////////////////////////////////////////////////////////') + print() + return 0 + +def main(): + parser = argparse.ArgumentParser(description='download certificate chain and public keys under a C++/Arduino compilable form') + parser.add_argument('-s', '--server', action='store', required=True, help='TLS server dns name') + parser.add_argument('-p', '--port', action='store', required=False, help='TLS server port') + parser.add_argument('-n', '--name', action='store', required=False, help='variable name') + port = 443 + args = parser.parse_args() + server = args.server + port = 443 + try: + split = server.split(':') + server = split[0] + port = int(split[1]) + except: + pass + try: + port = int(args.port) + except: + pass + + print() + print('// this file is autogenerated - any modification will be overwritten') + print('// unused symbols will not be linked in the final binary') + print('// generated on {}'.format(datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"))) + print('// by {}'.format(sys.argv)) + print() + print('#pragma once') + print() + return get_certificate(server, port, args.name) + +if __name__ == '__main__': + sys.exit(main())