From 7eff9cd2cae18afe4f6c5c5da3586656d07c4459 Mon Sep 17 00:00:00 2001
From: Shubham Patil <shubham.patil@espressif.com>
Date: Mon, 20 May 2024 13:54:34 +0530
Subject: [PATCH 1/8] Some refactoring to dac revocation set generation script
---
credentials/generate-revocation-set.py | 140 ++++++++++++++++++-------
1 file changed, 104 insertions(+), 36 deletions(-)
diff --git a/credentials/generate-revocation-set.py b/credentials/generate-revocation-set.py
index 534edc15b0f203..6422bfe79cba57 100644
--- a/credentials/generate-revocation-set.py
+++ b/credentials/generate-revocation-set.py
@@ -1,7 +1,7 @@
-#!/usr/bin/python
+#!/usr/bin/env python3
#
-# Copyright (c) 2023 Project CHIP Authors
+# Copyright (c) 2023-2024 Project CHIP Authors
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
@@ -29,6 +29,7 @@
import click
import requests
from click_option_group import RequiredMutuallyExclusiveOptionGroup, optgroup
+from cryptography.hazmat.primitives.asymmetric import ec
from cryptography import x509
@@ -44,20 +45,61 @@ class RevocationType(Enum):
TEST_NODE_URL_REST = "https://on.test-net.dcl.csa-iot.org"
-def use_dcld(dcld, production, cmdlist):
- return [dcld] + cmdlist + (['--node', PRODUCTION_NODE_URL] if production else [])
-
-
def extract_single_integer_attribute(subject, oid):
attribute_list = subject.get_attributes_for_oid(oid)
if len(attribute_list) == 1:
- if attribute_list[0].value.isdigit():
- return int(attribute_list[0].value)
+ return int(attribute_list[0].value, 16)
return None
+class DCLDRequestsHelper:
+ def __init__(self, use_rest, dcld, production, rest_node_url):
+ self.use_rest = use_rest
+ self.dcld = dcld
+ self.production = production
+ self.rest_node_url = rest_node_url
+
+ def use_dcld(self, dcld, production, cmdlist):
+ return [dcld] + cmdlist + (['--node', PRODUCTION_NODE_URL] if production else [])
+
+ def get_dcld_cmd_output(self, cmdlist):
+ # Set the output as JSON
+ subprocess.Popen([self.dcld, 'config', 'output', 'json'])
+
+ cmdpipe = subprocess.Popen(self.use_dcld(self.dcld, self.production, cmdlist),
+ stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+ return json.loads(cmdpipe.stdout.read())
+
+ def get_revocation_points(self):
+ if self.use_rest:
+ response = requests.get(f"{self.rest_node_url}/dcl/pki/revocation-points").json()
+ else:
+ response = self.get_dcld_cmd_output(['query', 'pki', 'all-revocation-points'])
+
+ return response["PkiRevocationDistributionPoint"]
+
+ def get_paa_cert_for_crl_issuer(self, crl_signer_issuer_name, crl_signer_authority_key_id):
+ if self.use_rest:
+ response = requests.get(
+ f"{self.rest_node_url}/dcl/pki/certificates/{crl_signer_issuer_name}/{crl_signer_authority_key_id}").json()
+ else:
+ response = self.get_dcld_cmd_output(
+ ['query', 'pki', 'x509-cert', '-u', crl_signer_issuer_name, '-k', crl_signer_authority_key_id])
+
+ return response["approvedCertificates"]["certs"][0]["pemCert"]
+
+ def get_revocations_points_by_skid(self, issuer_subject_key_id):
+ if self.use_rest:
+ response = requests.get(f"{self.rest_node_url}/dcl/pki/revocation-points/{issuer_subject_key_id}").json()
+ else:
+ response = self.get_dcld_cmd_output(['query', 'pki', 'revocation-points',
+ '--issuer-subject-key-id', issuer_subject_key_id])
+
+ return response["pkiRevocationDistributionPointsByIssuerSubjectKeyID"]["points"]
+
+
@click.command()
@click.help_option('-h', '--help')
@optgroup.group('Input data sources', cls=RequiredMutuallyExclusiveOptionGroup)
@@ -68,7 +110,7 @@ def extract_single_integer_attribute(subject, oid):
@optgroup.group('Optional arguments')
@optgroup.option('--output', default='sample_revocation_set_list.json', type=str, metavar='FILEPATH', help="Output filename (default: sample_revocation_set_list.json)")
def main(use_main_net_dcld, use_test_net_dcld, use_main_net_http, use_test_net_http, output):
- """DCL PAA mirroring tools"""
+ """Tool to construct revocation set from DCL"""
production = False
dcld = use_test_net_dcld
@@ -83,28 +125,20 @@ def main(use_main_net_dcld, use_test_net_dcld, use_main_net_http, use_test_net_h
rest_node_url = PRODUCTION_NODE_URL_REST if production else TEST_NODE_URL_REST
- # TODO: Extract this to a helper function
- if use_rest:
- revocation_point_list = requests.get(f"{rest_node_url}/dcl/pki/revocation-points").json()["PkiRevocationDistributionPoint"]
- else:
- cmdlist = ['config', 'output', 'json']
- subprocess.Popen([dcld] + cmdlist)
-
- cmdlist = ['query', 'pki', 'all-revocation-points']
+ dcld_helper = DCLDRequestsHelper(use_rest, dcld, production, rest_node_url)
- cmdpipe = subprocess.Popen(use_dcld(dcld, production, cmdlist), stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-
- revocation_point_list = json.loads(cmdpipe.stdout.read())["PkiRevocationDistributionPoint"]
+ revocation_point_list = dcld_helper.get_revocation_points()
revocation_set = []
for revocation_point in revocation_point_list:
# 1. Validate Revocation Type
- if revocation_point["revocationType"] != RevocationType.CRL:
+ if revocation_point["revocationType"] != RevocationType.CRL.value:
+ print("Revocation Type is not CRL, continue...")
continue
# 2. Parse the certificate
- crl_signer_certificate = x509.load_pem_x509_certificate(revocation_point["crlSignerCertificate"])
+ crl_signer_certificate = x509.load_pem_x509_certificate(bytes(revocation_point["crlSignerCertificate"], 'utf-8'))
vid = revocation_point["vid"]
pid = revocation_point["pid"]
@@ -119,12 +153,15 @@ def main(use_main_net_dcld, use_test_net_dcld, use_main_net_http, use_test_net_h
if crl_vid is not None:
if vid != crl_vid:
# TODO: Need to log all situations where a continue is called
+ print("VID is not CRL VID, continue...")
continue
else:
if crl_vid is None or vid != crl_vid:
+ print("VID is not CRL VID, continue...")
continue
if crl_pid is not None:
if pid != crl_pid:
+ print("PID is not CRL PID, continue...")
continue
# 5. Validate the certification path containing CRLSignerCertificate.
@@ -133,30 +170,54 @@ def main(use_main_net_dcld, use_test_net_dcld, use_main_net_http, use_test_net_h
crl_signer_authority_key_id = crl_signer_certificate.extensions.get_extension_for_oid(
x509.OID_AUTHORITY_KEY_IDENTIFIER).value.key_identifier
- paa_certificate = None
+ # Convert CRL Signer AKID to colon separated hex
+ crl_signer_authority_key_id = crl_signer_authority_key_id.hex().upper()
+ crl_signer_authority_key_id = ':'.join([crl_signer_authority_key_id[i:i+2] for i in range(0, len(crl_signer_authority_key_id), 2)])
- # TODO: Extract this to a helper function
- if use_rest:
- response = requests.get(
- f"{rest_node_url}/dcl/pki/certificates/{crl_signer_issuer_name}/{crl_signer_authority_key_id}").json()["approvedCertificates"]["certs"][0]
- paa_certificate = response["pemCert"]
- else:
- cmdlist = ['query', 'pki', 'x509-cert', '-u', crl_signer_issuer_name, '-k', crl_signer_authority_key_id]
- cmdpipe = subprocess.Popen(use_dcld(dcld, production, cmdlist), stdout=subprocess.PIPE, stderr=subprocess.PIPE)
- paa_certificate = json.loads(cmdpipe.stdout.read())["approvedCertificates"]["certs"][0]["pemCert"]
+ paa_certificate = dcld_helper.get_paa_cert_for_crl_issuer(crl_signer_issuer_name, crl_signer_authority_key_id)
if paa_certificate is None:
+ print("PAA Certificate not found, continue...")
continue
- paa_certificate_object = x509.load_pem_x509_certificate(paa_certificate)
+ paa_certificate_object = x509.load_pem_x509_certificate(bytes(paa_certificate, 'utf-8'))
+ # This requires cryptography 40.0.0 and this haults on 40.0.0
+ # try:
+ # crl_signer_certificate.verify_directly_issued_by(paa_certificate_object)
+ # except Exception as e:
+ # print("CRL Signer Certificate is not issued by PAA Certificate, continue...")
+ # continue
+
+ # Verify issuer matches with subject
+ if crl_signer_certificate.issuer != paa_certificate_object.subject:
+ print("CRL Signer Certificate issuer does not match with PAA Certificate subject, continue...")
+ continue
+
+ # Check crl signers AKID matches with SKID of paa_certificate_object's AKID
+ paa_skid = paa_certificate_object.extensions.get_extension_for_oid(x509.OID_SUBJECT_KEY_IDENTIFIER).value.key_identifier
+ crl_akid = crl_signer_certificate.extensions.get_extension_for_oid(x509.OID_AUTHORITY_KEY_IDENTIFIER).value.key_identifier
+ if paa_skid != crl_akid:
+ print("CRL Signer's AKID does not match with PAA Certificate SKID, continue...")
+ continue
+
+ # verify if PAA singed the crl's public key
try:
- crl_signer_certificate.verify_directly_issued_by(paa_certificate_object)
- except Exception:
+ paa_certificate_object.public_key().verify(crl_signer_certificate.signature,
+ crl_signer_certificate.tbs_certificate_bytes,
+ ec.ECDSA(crl_signer_certificate.signature_hash_algorithm))
+ except Exception as e:
+ print("CRL Signer Certificate is not signed by PAA Certificate, continue...")
+ print("Error: ", e)
continue
# 6. Obtain the CRL
- r = requests.get(revocation_point["dataURL"])
+ try:
+ r = requests.get(revocation_point["dataURL"], timeout=5)
+ except requests.exceptions.Timeout:
+ print("Timeout fetching CRL from ", revocation_point["dataURL"])
+ continue
+
crl_file = x509.load_der_x509_crl(r.content)
# 7. Perform CRL File Validation
@@ -164,6 +225,7 @@ def main(use_main_net_dcld, use_test_net_dcld, use_main_net_http, use_test_net_h
crl_signer_subject_key_id = crl_signer_certificate.extensions.get_extension_for_oid(
x509.OID_SUBJECT_KEY_IDENTIFIER).value.key_identifier
if crl_authority_key_id != crl_signer_subject_key_id:
+ print("CRL Authority Key ID is not CRL Signer Subject Key ID, continue...")
continue
issuer_subject_key_id = ''.join('{:02X}'.format(x) for x in crl_authority_key_id)
@@ -192,13 +254,16 @@ def main(use_main_net_dcld, use_test_net_dcld, use_main_net_http, use_test_net_h
issuing_distribution_point = crl_file.extensions.get_extension_for_oid(
x509.OID_ISSUING_DISTRIBUTION_POINT).value
except Exception:
+ print("CRL Issuing Distribution Point not found, continue...")
continue
uri_list = issuing_distribution_point.full_name
if len(uri_list) == 1 and isinstance(uri_list[0], x509.UniformResourceIdentifier):
if uri_list[0].value != revocation_point["dataURL"]:
+ print("CRL Issuing Distribution Point URI is not CRL URL, continue...")
continue
else:
+ print("CRL Issuing Distribution Point URI is not CRL URL, continue...")
continue
# 9. Assign CRL File Issuer
@@ -213,6 +278,7 @@ def main(use_main_net_dcld, use_test_net_dcld, use_main_net_http, use_test_net_h
if revoked_cert_issuer is not None:
if revoked_cert_issuer != certificate_authority_name:
+ print("CRL Issuer is not CRL File Issuer, continue...")
continue
except Exception:
pass
@@ -223,8 +289,10 @@ def main(use_main_net_dcld, use_test_net_dcld, use_main_net_http, use_test_net_h
x509.OID_AUTHORITY_KEY_IDENTIFIER).value.key_identifier
if revoked_cert_authority_key_id is None or revoked_cert_authority_key_id != crl_signer_subject_key_id:
+ print("CRL Authority Key ID is not CRL Signer Subject Key ID, continue...")
continue
except Exception:
+ print("CRL Authority Key ID not found, continue...")
continue
# c. and d.
From c66885bc836e6bd80cdc241713fbf912368c12dc Mon Sep 17 00:00:00 2001
From: Shubham Patil <shubham.patil@espressif.com>
Date: Mon, 20 May 2024 14:41:32 +0530
Subject: [PATCH 2/8] restyle and linter fixes
---
credentials/generate-revocation-set.py | 55 +++++++++++---------------
1 file changed, 22 insertions(+), 33 deletions(-)
diff --git a/credentials/generate-revocation-set.py b/credentials/generate-revocation-set.py
index 6422bfe79cba57..4f7a94725b91c4 100644
--- a/credentials/generate-revocation-set.py
+++ b/credentials/generate-revocation-set.py
@@ -29,8 +29,8 @@
import click
import requests
from click_option_group import RequiredMutuallyExclusiveOptionGroup, optgroup
-from cryptography.hazmat.primitives.asymmetric import ec
from cryptography import x509
+from cryptography.hazmat.primitives.asymmetric import ec
class RevocationType(Enum):
@@ -172,7 +172,8 @@ def main(use_main_net_dcld, use_test_net_dcld, use_main_net_http, use_test_net_h
# Convert CRL Signer AKID to colon separated hex
crl_signer_authority_key_id = crl_signer_authority_key_id.hex().upper()
- crl_signer_authority_key_id = ':'.join([crl_signer_authority_key_id[i:i+2] for i in range(0, len(crl_signer_authority_key_id), 2)])
+ crl_signer_authority_key_id = ':'.join([crl_signer_authority_key_id[i:i+2]
+ for i in range(0, len(crl_signer_authority_key_id), 2)])
paa_certificate = dcld_helper.get_paa_cert_for_crl_issuer(crl_signer_issuer_name, crl_signer_authority_key_id)
@@ -204,8 +205,8 @@ def main(use_main_net_dcld, use_test_net_dcld, use_main_net_http, use_test_net_h
# verify if PAA singed the crl's public key
try:
paa_certificate_object.public_key().verify(crl_signer_certificate.signature,
- crl_signer_certificate.tbs_certificate_bytes,
- ec.ECDSA(crl_signer_certificate.signature_hash_algorithm))
+ crl_signer_certificate.tbs_certificate_bytes,
+ ec.ECDSA(crl_signer_certificate.signature_hash_algorithm))
except Exception as e:
print("CRL Signer Certificate is not signed by PAA Certificate, continue...")
print("Error: ", e)
@@ -230,26 +231,11 @@ def main(use_main_net_dcld, use_test_net_dcld, use_main_net_http, use_test_net_h
issuer_subject_key_id = ''.join('{:02X}'.format(x) for x in crl_authority_key_id)
- same_issuer_points = None
+ # b.
+ count_with_matching_vid_issuer_skid = sum(item.get('vid') == vid for item in same_issuer_points)
+ same_issuer_points = dcld_helper.get_revocations_points_by_skid(issuer_subject_key_id)
- # TODO: Extract this to a helper function
- if use_rest:
- response = requests.get(
- f"{rest_node_url}/dcl/pki/revocation-points/{issuer_subject_key_id}").json()["pkiRevocationDistributionPointsByIssuerSubjectKeyID"]
- same_issuer_points = response["points"]
- else:
- cmdlist = ['query', 'pki', 'revocation-points', '--issuer-subject-key-id', issuer_subject_key_id]
- cmdpipe = subprocess.Popen(use_dcld(dcld, production, cmdlist), stdout=subprocess.PIPE, stderr=subprocess.PIPE)
- same_issuer_points = json.loads(cmdpipe.stdout.read())[
- "pkiRevocationDistributionPointsByIssuerSubjectKeyID"]["points"]
-
- matching_entries = False
- for same_issuer_point in same_issuer_points:
- if same_issuer_point["vid"] == vid:
- matching_entries = True
- break
-
- if matching_entries:
+ if count_with_matching_vid_issuer_skid > 1:
try:
issuing_distribution_point = crl_file.extensions.get_extension_for_oid(
x509.OID_ISSUING_DISTRIBUTION_POINT).value
@@ -268,10 +254,13 @@ def main(use_main_net_dcld, use_test_net_dcld, use_main_net_http, use_test_net_h
# 9. Assign CRL File Issuer
certificate_authority_name = base64.b64encode(crl_file.issuer.public_bytes()).decode('utf-8')
+ print(f"CRL File Issuer: {certificate_authority_name}")
serialnumber_list = []
# 10. Iterate through the Revoked Certificates List
for revoked_cert in crl_file:
+ print(revoked_cert)
+ # a.
try:
revoked_cert_issuer = revoked_cert.extensions.get_extension_for_oid(
x509.CRLEntryExtensionOID.CERTIFICATE_ISSUER).value.get_values_for_type(x509.DirectoryName).value
@@ -284,16 +273,16 @@ def main(use_main_net_dcld, use_test_net_dcld, use_main_net_http, use_test_net_h
pass
# b.
- try:
- revoked_cert_authority_key_id = revoked_cert.extensions.get_extension_for_oid(
- x509.OID_AUTHORITY_KEY_IDENTIFIER).value.key_identifier
-
- if revoked_cert_authority_key_id is None or revoked_cert_authority_key_id != crl_signer_subject_key_id:
- print("CRL Authority Key ID is not CRL Signer Subject Key ID, continue...")
- continue
- except Exception:
- print("CRL Authority Key ID not found, continue...")
- continue
+ # try:
+ # revoked_cert_authority_key_id = revoked_cert.extensions.get_extension_for_oid(
+ # x509.OID_AUTHORITY_KEY_IDENTIFIER).value.key_identifier
+ #
+ # if revoked_cert_authority_key_id is None or revoked_cert_authority_key_id != crl_signer_subject_key_id:
+ # print("CRL Authority Key ID is not CRL Signer Subject Key ID, continue...")
+ # continue
+ # except Exception:
+ # print("CRL Authority Key ID not found, continue...")
+ # continue
# c. and d.
serialnumber_list.append(bytes(str('{:02X}'.format(revoked_cert.serial_number)), 'utf-8').decode('utf-8'))
From 3a3dc2e2cfbee177c24e1eb92e317064d621e8cc Mon Sep 17 00:00:00 2001
From: Shubham Patil <shubham.patil@espressif.com>
Date: Mon, 20 May 2024 14:52:42 +0530
Subject: [PATCH 3/8] one last lint fix
---
credentials/generate-revocation-set.py | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/credentials/generate-revocation-set.py b/credentials/generate-revocation-set.py
index 4f7a94725b91c4..984a10534ebd2e 100644
--- a/credentials/generate-revocation-set.py
+++ b/credentials/generate-revocation-set.py
@@ -232,8 +232,8 @@ def main(use_main_net_dcld, use_test_net_dcld, use_main_net_http, use_test_net_h
issuer_subject_key_id = ''.join('{:02X}'.format(x) for x in crl_authority_key_id)
# b.
- count_with_matching_vid_issuer_skid = sum(item.get('vid') == vid for item in same_issuer_points)
same_issuer_points = dcld_helper.get_revocations_points_by_skid(issuer_subject_key_id)
+ count_with_matching_vid_issuer_skid = sum(item.get('vid') == vid for item in same_issuer_points)
if count_with_matching_vid_issuer_skid > 1:
try:
From 2a3e71529163468cc2fa219527a7fbb09ca91787 Mon Sep 17 00:00:00 2001
From: Shubham Patil <shubham.patil@espressif.com>
Date: Wed, 22 May 2024 16:37:17 +0530
Subject: [PATCH 4/8] address comments
---
credentials/generate-revocation-set.py | 152 ++++++++++++++++++-------
1 file changed, 114 insertions(+), 38 deletions(-)
diff --git a/credentials/generate-revocation-set.py b/credentials/generate-revocation-set.py
index 984a10534ebd2e..a1915b877866ad 100644
--- a/credentials/generate-revocation-set.py
+++ b/credentials/generate-revocation-set.py
@@ -54,47 +54,130 @@ def extract_single_integer_attribute(subject, oid):
return None
-class DCLDRequestsHelper:
- def __init__(self, use_rest, dcld, production, rest_node_url):
+class DCLDClient:
+ '''
+ A client for interacting with DCLD using either the REST API or command line interface (CLI).
+
+ '''
+
+ def __init__(self, use_rest:bool, dcld_exe:str, production:bool, rest_node_url:str):
+ '''
+ Initialize the client
+
+ use_rest: bool
+ Use RESTful API with HTTPS against `rest_node_url`
+ dcld_exe: str
+ Path to `dcld` executable
+ production: bool
+ Use MainNet DCL URL with dcld executable
+ rest_node_url: str
+ RESTful API URL
+ '''
+
self.use_rest = use_rest
- self.dcld = dcld
+ self.dcld_exe = dcld_exe
self.production = production
self.rest_node_url = rest_node_url
- def use_dcld(self, dcld, production, cmdlist):
- return [dcld] + cmdlist + (['--node', PRODUCTION_NODE_URL] if production else [])
+ def build_dcld_command_line(self, cmdlist:list[str])->list[str]:
+ '''
+ Build command line for `dcld` executable.
+
+ Parameters
+ ----------
+ cmdlist: list[str]
+ List of command line arguments to append to some predefined arguments
+
+ Returns
+ -------
+ list[str]
+ The complete command list including the DCLD executable and node option if in production
+ '''
+
+ return [self.dcld_exe] + cmdlist + (['--node', PRODUCTION_NODE_URL] if self.production else [])
+
+ def get_dcld_cmd_output_json(self, cmdlist:list[str])->dict:
+ '''
+ Executes a DCLD CLI command and returns the JSON output.
+
+ Parameters
+ ----------
+ cmdlist: list[str]
+ List of command line arguments to append to some predefined arguments
+
+ Returns
+ -------
+ dict
+ The JSON output from the command
+ '''
- def get_dcld_cmd_output(self, cmdlist):
# Set the output as JSON
- subprocess.Popen([self.dcld, 'config', 'output', 'json'])
+ subprocess.Popen([self.dcld_exe, 'config', 'output', 'json'])
- cmdpipe = subprocess.Popen(self.use_dcld(self.dcld, self.production, cmdlist),
+ cmdpipe = subprocess.Popen(self.build_dcld_command_line(cmdlist),
stdout=subprocess.PIPE, stderr=subprocess.PIPE)
return json.loads(cmdpipe.stdout.read())
- def get_revocation_points(self):
+ def get_revocation_points(self)->list[dict]:
+ '''
+ Get revocation points from DCL
+
+ Returns
+ -------
+ list[dict]
+ List of revocation points
+ '''
+
if self.use_rest:
response = requests.get(f"{self.rest_node_url}/dcl/pki/revocation-points").json()
else:
- response = self.get_dcld_cmd_output(['query', 'pki', 'all-revocation-points'])
+ response = self.get_dcld_cmd_output_json(['query', 'pki', 'all-revocation-points'])
return response["PkiRevocationDistributionPoint"]
- def get_paa_cert_for_crl_issuer(self, crl_signer_issuer_name, crl_signer_authority_key_id):
+ def get_paa_cert_for_crl_issuer(self, crl_signer_issuer_name_b64, crl_signer_authority_key_id)->str:
+ '''
+ Get PAA certificate for CRL issuer
+
+ Parameters
+ ----------
+ crl_signer_issuer_name_b64: str
+ The issuer name of the CRL signer.
+ crl_signer_authority_key_id: str
+ The authority key ID of the CRL signer.
+
+ Returns
+ -------
+ str
+ PAA certificate in PEM format
+ '''
if self.use_rest:
response = requests.get(
- f"{self.rest_node_url}/dcl/pki/certificates/{crl_signer_issuer_name}/{crl_signer_authority_key_id}").json()
+ f"{self.rest_node_url}/dcl/pki/certificates/{crl_signer_issuer_name_b64}/{crl_signer_authority_key_id}").json()
else:
- response = self.get_dcld_cmd_output(
- ['query', 'pki', 'x509-cert', '-u', crl_signer_issuer_name, '-k', crl_signer_authority_key_id])
+ response = self.get_dcld_cmd_output_json(
+ ['query', 'pki', 'x509-cert', '-u', crl_signer_issuer_name_b64, '-k', crl_signer_authority_key_id])
return response["approvedCertificates"]["certs"][0]["pemCert"]
- def get_revocations_points_by_skid(self, issuer_subject_key_id):
+ def get_revocations_points_by_skid(self, issuer_subject_key_id)->list[dict]:
+ '''
+ Get revocation points by subject key ID
+
+ Parameters
+ ----------
+ issuer_subject_key_id: str
+ Subject key ID
+
+ Returns
+ -------
+ list[dict]
+ List of revocation points
+ '''
if self.use_rest:
response = requests.get(f"{self.rest_node_url}/dcl/pki/revocation-points/{issuer_subject_key_id}").json()
else:
- response = self.get_dcld_cmd_output(['query', 'pki', 'revocation-points',
+ response = self.get_dcld_cmd_output_json(['query', 'pki', 'revocation-points',
'--issuer-subject-key-id', issuer_subject_key_id])
return response["pkiRevocationDistributionPointsByIssuerSubjectKeyID"]["points"]
@@ -125,9 +208,9 @@ def main(use_main_net_dcld, use_test_net_dcld, use_main_net_http, use_test_net_h
rest_node_url = PRODUCTION_NODE_URL_REST if production else TEST_NODE_URL_REST
- dcld_helper = DCLDRequestsHelper(use_rest, dcld, production, rest_node_url)
+ dcld_client = DCLDClient(use_rest, dcld, production, rest_node_url)
- revocation_point_list = dcld_helper.get_revocation_points()
+ revocation_point_list = dcld_client.get_revocation_points()
revocation_set = []
@@ -175,7 +258,7 @@ def main(use_main_net_dcld, use_test_net_dcld, use_main_net_http, use_test_net_h
crl_signer_authority_key_id = ':'.join([crl_signer_authority_key_id[i:i+2]
for i in range(0, len(crl_signer_authority_key_id), 2)])
- paa_certificate = dcld_helper.get_paa_cert_for_crl_issuer(crl_signer_issuer_name, crl_signer_authority_key_id)
+ paa_certificate = dcld_client.get_paa_cert_for_crl_issuer(crl_signer_issuer_name, crl_signer_authority_key_id)
if paa_certificate is None:
print("PAA Certificate not found, continue...")
@@ -183,13 +266,7 @@ def main(use_main_net_dcld, use_test_net_dcld, use_main_net_http, use_test_net_h
paa_certificate_object = x509.load_pem_x509_certificate(bytes(paa_certificate, 'utf-8'))
- # This requires cryptography 40.0.0 and this haults on 40.0.0
- # try:
- # crl_signer_certificate.verify_directly_issued_by(paa_certificate_object)
- # except Exception as e:
- # print("CRL Signer Certificate is not issued by PAA Certificate, continue...")
- # continue
-
+ # TODO: use verify_directly_issued_by() method when we upgrade cryptography to v40.0.0
# Verify issuer matches with subject
if crl_signer_certificate.issuer != paa_certificate_object.subject:
print("CRL Signer Certificate issuer does not match with PAA Certificate subject, continue...")
@@ -232,7 +309,7 @@ def main(use_main_net_dcld, use_test_net_dcld, use_main_net_http, use_test_net_h
issuer_subject_key_id = ''.join('{:02X}'.format(x) for x in crl_authority_key_id)
# b.
- same_issuer_points = dcld_helper.get_revocations_points_by_skid(issuer_subject_key_id)
+ same_issuer_points = dcld_client.get_revocations_points_by_skid(issuer_subject_key_id)
count_with_matching_vid_issuer_skid = sum(item.get('vid') == vid for item in same_issuer_points)
if count_with_matching_vid_issuer_skid > 1:
@@ -259,7 +336,6 @@ def main(use_main_net_dcld, use_test_net_dcld, use_main_net_http, use_test_net_h
serialnumber_list = []
# 10. Iterate through the Revoked Certificates List
for revoked_cert in crl_file:
- print(revoked_cert)
# a.
try:
revoked_cert_issuer = revoked_cert.extensions.get_extension_for_oid(
@@ -273,16 +349,16 @@ def main(use_main_net_dcld, use_test_net_dcld, use_main_net_http, use_test_net_h
pass
# b.
- # try:
- # revoked_cert_authority_key_id = revoked_cert.extensions.get_extension_for_oid(
- # x509.OID_AUTHORITY_KEY_IDENTIFIER).value.key_identifier
- #
- # if revoked_cert_authority_key_id is None or revoked_cert_authority_key_id != crl_signer_subject_key_id:
- # print("CRL Authority Key ID is not CRL Signer Subject Key ID, continue...")
- # continue
- # except Exception:
- # print("CRL Authority Key ID not found, continue...")
- # continue
+ try:
+ revoked_cert_authority_key_id = revoked_cert.extensions.get_extension_for_oid(
+ x509.OID_AUTHORITY_KEY_IDENTIFIER).value.key_identifier
+
+ if revoked_cert_authority_key_id is None or revoked_cert_authority_key_id != crl_signer_subject_key_id:
+ print("CRL Authority Key ID is not CRL Signer Subject Key ID, continue...")
+ continue
+ except Exception:
+ print("CRL Authority Key ID not found, continue...")
+ continue
# c. and d.
serialnumber_list.append(bytes(str('{:02X}'.format(revoked_cert.serial_number)), 'utf-8').decode('utf-8'))
From e59b9eb86fbd53d65c49a187849ade8ea7b5dafb Mon Sep 17 00:00:00 2001
From: "Restyled.io" <commits@restyled.io>
Date: Wed, 22 May 2024 11:07:39 +0000
Subject: [PATCH 5/8] Restyled by autopep8
---
credentials/generate-revocation-set.py | 14 +++++++-------
1 file changed, 7 insertions(+), 7 deletions(-)
diff --git a/credentials/generate-revocation-set.py b/credentials/generate-revocation-set.py
index a1915b877866ad..21fc612fe3ec92 100644
--- a/credentials/generate-revocation-set.py
+++ b/credentials/generate-revocation-set.py
@@ -60,7 +60,7 @@ class DCLDClient:
'''
- def __init__(self, use_rest:bool, dcld_exe:str, production:bool, rest_node_url:str):
+ def __init__(self, use_rest: bool, dcld_exe: str, production: bool, rest_node_url: str):
'''
Initialize the client
@@ -79,7 +79,7 @@ def __init__(self, use_rest:bool, dcld_exe:str, production:bool, rest_node_url:s
self.production = production
self.rest_node_url = rest_node_url
- def build_dcld_command_line(self, cmdlist:list[str])->list[str]:
+ def build_dcld_command_line(self, cmdlist: list[str]) -> list[str]:
'''
Build command line for `dcld` executable.
@@ -96,7 +96,7 @@ def build_dcld_command_line(self, cmdlist:list[str])->list[str]:
return [self.dcld_exe] + cmdlist + (['--node', PRODUCTION_NODE_URL] if self.production else [])
- def get_dcld_cmd_output_json(self, cmdlist:list[str])->dict:
+ def get_dcld_cmd_output_json(self, cmdlist: list[str]) -> dict:
'''
Executes a DCLD CLI command and returns the JSON output.
@@ -118,7 +118,7 @@ def get_dcld_cmd_output_json(self, cmdlist:list[str])->dict:
stdout=subprocess.PIPE, stderr=subprocess.PIPE)
return json.loads(cmdpipe.stdout.read())
- def get_revocation_points(self)->list[dict]:
+ def get_revocation_points(self) -> list[dict]:
'''
Get revocation points from DCL
@@ -135,7 +135,7 @@ def get_revocation_points(self)->list[dict]:
return response["PkiRevocationDistributionPoint"]
- def get_paa_cert_for_crl_issuer(self, crl_signer_issuer_name_b64, crl_signer_authority_key_id)->str:
+ def get_paa_cert_for_crl_issuer(self, crl_signer_issuer_name_b64, crl_signer_authority_key_id) -> str:
'''
Get PAA certificate for CRL issuer
@@ -160,7 +160,7 @@ def get_paa_cert_for_crl_issuer(self, crl_signer_issuer_name_b64, crl_signer_aut
return response["approvedCertificates"]["certs"][0]["pemCert"]
- def get_revocations_points_by_skid(self, issuer_subject_key_id)->list[dict]:
+ def get_revocations_points_by_skid(self, issuer_subject_key_id) -> list[dict]:
'''
Get revocation points by subject key ID
@@ -178,7 +178,7 @@ def get_revocations_points_by_skid(self, issuer_subject_key_id)->list[dict]:
response = requests.get(f"{self.rest_node_url}/dcl/pki/revocation-points/{issuer_subject_key_id}").json()
else:
response = self.get_dcld_cmd_output_json(['query', 'pki', 'revocation-points',
- '--issuer-subject-key-id', issuer_subject_key_id])
+ '--issuer-subject-key-id', issuer_subject_key_id])
return response["pkiRevocationDistributionPointsByIssuerSubjectKeyID"]["points"]
From 532a0e8f1e19629e8d7128ca90d3468b24bf4d09 Mon Sep 17 00:00:00 2001
From: Shubham Patil <shubham.patil@espressif.com>
Date: Thu, 23 May 2024 08:07:42 +0530
Subject: [PATCH 6/8] add logger
---
credentials/generate-revocation-set.py | 60 +++++++++++++++++---------
1 file changed, 40 insertions(+), 20 deletions(-)
diff --git a/credentials/generate-revocation-set.py b/credentials/generate-revocation-set.py
index 21fc612fe3ec92..2fae7b2f5f43c3 100644
--- a/credentials/generate-revocation-set.py
+++ b/credentials/generate-revocation-set.py
@@ -22,6 +22,7 @@
import base64
import json
+import logging
import subprocess
import sys
from enum import Enum
@@ -32,6 +33,15 @@
from cryptography import x509
from cryptography.hazmat.primitives.asymmetric import ec
+# Supported log levels, mapping string values required for argument
+# parsing into logging constants
+__LOG_LEVELS__ = {
+ 'debug': logging.DEBUG,
+ 'info': logging.INFO,
+ 'warn': logging.WARN,
+ 'fatal': logging.FATAL,
+}
+
class RevocationType(Enum):
CRL = 1
@@ -191,10 +201,20 @@ def get_revocations_points_by_skid(self, issuer_subject_key_id) -> list[dict]:
@optgroup.option('--use-main-net-http', is_flag=True, type=str, help="Use RESTful API with HTTPS against public MainNet observer.")
@optgroup.option('--use-test-net-http', is_flag=True, type=str, help="Use RESTful API with HTTPS against public TestNet observer.")
@optgroup.group('Optional arguments')
-@optgroup.option('--output', default='sample_revocation_set_list.json', type=str, metavar='FILEPATH', help="Output filename (default: sample_revocation_set_list.json)")
-def main(use_main_net_dcld, use_test_net_dcld, use_main_net_http, use_test_net_http, output):
+@optgroup.option('--output', default='sample_revocation_set_list.json', type=str, metavar='FILEPATH',
+ help="Output filename (default: sample_revocation_set_list.json)")
+@optgroup.option('--log-level', default='INFO', show_default=True, type=click.Choice(__LOG_LEVELS__.keys(),
+ case_sensitive=False), callback=lambda c, p, v: __LOG_LEVELS__[v],
+ help='Determines the verbosity of script output')
+def main(use_main_net_dcld: str, use_test_net_dcld: str, use_main_net_http: bool, use_test_net_http: bool, output: str, log_level: str):
"""Tool to construct revocation set from DCL"""
+ logging.basicConfig(
+ level=log_level,
+ format='%(asctime)s %(name)s %(levelname)-7s %(message)s',
+ datefmt='%Y-%m-%d %H:%M:%S'
+ )
+
production = False
dcld = use_test_net_dcld
@@ -217,7 +237,7 @@ def main(use_main_net_dcld, use_test_net_dcld, use_main_net_http, use_test_net_h
for revocation_point in revocation_point_list:
# 1. Validate Revocation Type
if revocation_point["revocationType"] != RevocationType.CRL.value:
- print("Revocation Type is not CRL, continue...")
+ logging.warning("Revocation Type is not CRL, continue...")
continue
# 2. Parse the certificate
@@ -236,15 +256,15 @@ def main(use_main_net_dcld, use_test_net_dcld, use_main_net_http, use_test_net_h
if crl_vid is not None:
if vid != crl_vid:
# TODO: Need to log all situations where a continue is called
- print("VID is not CRL VID, continue...")
+ logging.warning("VID is not CRL VID, continue...")
continue
else:
if crl_vid is None or vid != crl_vid:
- print("VID is not CRL VID, continue...")
+ logging.warning("VID is not CRL VID, continue...")
continue
if crl_pid is not None:
if pid != crl_pid:
- print("PID is not CRL PID, continue...")
+ logging.warning("PID is not CRL PID, continue...")
continue
# 5. Validate the certification path containing CRLSignerCertificate.
@@ -261,7 +281,7 @@ def main(use_main_net_dcld, use_test_net_dcld, use_main_net_http, use_test_net_h
paa_certificate = dcld_client.get_paa_cert_for_crl_issuer(crl_signer_issuer_name, crl_signer_authority_key_id)
if paa_certificate is None:
- print("PAA Certificate not found, continue...")
+ logging.warning("PAA Certificate not found, continue...")
continue
paa_certificate_object = x509.load_pem_x509_certificate(bytes(paa_certificate, 'utf-8'))
@@ -269,14 +289,14 @@ def main(use_main_net_dcld, use_test_net_dcld, use_main_net_http, use_test_net_h
# TODO: use verify_directly_issued_by() method when we upgrade cryptography to v40.0.0
# Verify issuer matches with subject
if crl_signer_certificate.issuer != paa_certificate_object.subject:
- print("CRL Signer Certificate issuer does not match with PAA Certificate subject, continue...")
+ logging.warning("CRL Signer Certificate issuer does not match with PAA Certificate subject, continue...")
continue
# Check crl signers AKID matches with SKID of paa_certificate_object's AKID
paa_skid = paa_certificate_object.extensions.get_extension_for_oid(x509.OID_SUBJECT_KEY_IDENTIFIER).value.key_identifier
crl_akid = crl_signer_certificate.extensions.get_extension_for_oid(x509.OID_AUTHORITY_KEY_IDENTIFIER).value.key_identifier
if paa_skid != crl_akid:
- print("CRL Signer's AKID does not match with PAA Certificate SKID, continue...")
+ logging.warning("CRL Signer's AKID does not match with PAA Certificate SKID, continue...")
continue
# verify if PAA singed the crl's public key
@@ -285,15 +305,15 @@ def main(use_main_net_dcld, use_test_net_dcld, use_main_net_http, use_test_net_h
crl_signer_certificate.tbs_certificate_bytes,
ec.ECDSA(crl_signer_certificate.signature_hash_algorithm))
except Exception as e:
- print("CRL Signer Certificate is not signed by PAA Certificate, continue...")
- print("Error: ", e)
+ logging.warning("CRL Signer Certificate is not signed by PAA Certificate, continue...")
+ logging.error("Error: ", e)
continue
# 6. Obtain the CRL
try:
r = requests.get(revocation_point["dataURL"], timeout=5)
except requests.exceptions.Timeout:
- print("Timeout fetching CRL from ", revocation_point["dataURL"])
+ logging.warning("Timeout fetching CRL from ", revocation_point["dataURL"])
continue
crl_file = x509.load_der_x509_crl(r.content)
@@ -303,7 +323,7 @@ def main(use_main_net_dcld, use_test_net_dcld, use_main_net_http, use_test_net_h
crl_signer_subject_key_id = crl_signer_certificate.extensions.get_extension_for_oid(
x509.OID_SUBJECT_KEY_IDENTIFIER).value.key_identifier
if crl_authority_key_id != crl_signer_subject_key_id:
- print("CRL Authority Key ID is not CRL Signer Subject Key ID, continue...")
+ logging.warning("CRL Authority Key ID is not CRL Signer Subject Key ID, continue...")
continue
issuer_subject_key_id = ''.join('{:02X}'.format(x) for x in crl_authority_key_id)
@@ -317,21 +337,21 @@ def main(use_main_net_dcld, use_test_net_dcld, use_main_net_http, use_test_net_h
issuing_distribution_point = crl_file.extensions.get_extension_for_oid(
x509.OID_ISSUING_DISTRIBUTION_POINT).value
except Exception:
- print("CRL Issuing Distribution Point not found, continue...")
+ logging.warning("CRL Issuing Distribution Point not found, continue...")
continue
uri_list = issuing_distribution_point.full_name
if len(uri_list) == 1 and isinstance(uri_list[0], x509.UniformResourceIdentifier):
if uri_list[0].value != revocation_point["dataURL"]:
- print("CRL Issuing Distribution Point URI is not CRL URL, continue...")
+ logging.warning("CRL Issuing Distribution Point URI is not CRL URL, continue...")
continue
else:
- print("CRL Issuing Distribution Point URI is not CRL URL, continue...")
+ logging.warning("CRL Issuing Distribution Point URI is not CRL URL, continue...")
continue
# 9. Assign CRL File Issuer
certificate_authority_name = base64.b64encode(crl_file.issuer.public_bytes()).decode('utf-8')
- print(f"CRL File Issuer: {certificate_authority_name}")
+ logging.debug(f"CRL File Issuer: {certificate_authority_name}")
serialnumber_list = []
# 10. Iterate through the Revoked Certificates List
@@ -343,7 +363,7 @@ def main(use_main_net_dcld, use_test_net_dcld, use_main_net_http, use_test_net_h
if revoked_cert_issuer is not None:
if revoked_cert_issuer != certificate_authority_name:
- print("CRL Issuer is not CRL File Issuer, continue...")
+ logging.warning("CRL Issuer is not CRL File Issuer, continue...")
continue
except Exception:
pass
@@ -354,10 +374,10 @@ def main(use_main_net_dcld, use_test_net_dcld, use_main_net_http, use_test_net_h
x509.OID_AUTHORITY_KEY_IDENTIFIER).value.key_identifier
if revoked_cert_authority_key_id is None or revoked_cert_authority_key_id != crl_signer_subject_key_id:
- print("CRL Authority Key ID is not CRL Signer Subject Key ID, continue...")
+ logging.warning("CRL Authority Key ID is not CRL Signer Subject Key ID, continue...")
continue
except Exception:
- print("CRL Authority Key ID not found, continue...")
+ logging.warning("CRL Authority Key ID not found, continue...")
continue
# c. and d.
From e94208593fe96f76c3059a95c60e06b302a513e6 Mon Sep 17 00:00:00 2001
From: Shubham Patil <shubham.patil@espressif.com>
Date: Thu, 23 May 2024 12:26:08 +0530
Subject: [PATCH 7/8] Catch the exception when loading the CRL
---
credentials/generate-revocation-set.py | 13 +++++++++----
1 file changed, 9 insertions(+), 4 deletions(-)
diff --git a/credentials/generate-revocation-set.py b/credentials/generate-revocation-set.py
index 2fae7b2f5f43c3..889c245f60ec4b 100644
--- a/credentials/generate-revocation-set.py
+++ b/credentials/generate-revocation-set.py
@@ -299,7 +299,7 @@ def main(use_main_net_dcld: str, use_test_net_dcld: str, use_main_net_http: bool
logging.warning("CRL Signer's AKID does not match with PAA Certificate SKID, continue...")
continue
- # verify if PAA singed the crl's public key
+ # verify if PAA singed the crl signer certificate
try:
paa_certificate_object.public_key().verify(crl_signer_certificate.signature,
crl_signer_certificate.tbs_certificate_bytes,
@@ -310,13 +310,18 @@ def main(use_main_net_dcld: str, use_test_net_dcld: str, use_main_net_http: bool
continue
# 6. Obtain the CRL
+ logging.debug(f"Fetching CRL from {revocation_point['dataURL']}")
try:
r = requests.get(revocation_point["dataURL"], timeout=5)
- except requests.exceptions.Timeout:
- logging.warning("Timeout fetching CRL from ", revocation_point["dataURL"])
+ except Exception as e:
+ logging.error('Failed to fetch CRL')
continue
- crl_file = x509.load_der_x509_crl(r.content)
+ try:
+ crl_file = x509.load_der_x509_crl(r.content)
+ except Exception as e:
+ logging.error('Failed to load CRL')
+ continue
# 7. Perform CRL File Validation
crl_authority_key_id = crl_file.extensions.get_extension_for_oid(x509.OID_AUTHORITY_KEY_IDENTIFIER).value.key_identifier
From 5308d950a40522f17bf97be9a909ac3407a12a50 Mon Sep 17 00:00:00 2001
From: Shubham Patil <shubham.patil@espressif.com>
Date: Fri, 24 May 2024 10:07:48 +0530
Subject: [PATCH 8/8] CRL entry extension do not have a AKID extension
---
credentials/generate-revocation-set.py | 21 +++++++--------------
1 file changed, 7 insertions(+), 14 deletions(-)
diff --git a/credentials/generate-revocation-set.py b/credentials/generate-revocation-set.py
index 889c245f60ec4b..bfc5ce560c1f80 100644
--- a/credentials/generate-revocation-set.py
+++ b/credentials/generate-revocation-set.py
@@ -304,22 +304,21 @@ def main(use_main_net_dcld: str, use_test_net_dcld: str, use_main_net_http: bool
paa_certificate_object.public_key().verify(crl_signer_certificate.signature,
crl_signer_certificate.tbs_certificate_bytes,
ec.ECDSA(crl_signer_certificate.signature_hash_algorithm))
- except Exception as e:
+ except Exception:
logging.warning("CRL Signer Certificate is not signed by PAA Certificate, continue...")
- logging.error("Error: ", e)
continue
# 6. Obtain the CRL
logging.debug(f"Fetching CRL from {revocation_point['dataURL']}")
try:
r = requests.get(revocation_point["dataURL"], timeout=5)
- except Exception as e:
+ except Exception:
logging.error('Failed to fetch CRL')
continue
try:
crl_file = x509.load_der_x509_crl(r.content)
- except Exception as e:
+ except Exception:
logging.error('Failed to load CRL')
continue
@@ -374,16 +373,10 @@ def main(use_main_net_dcld: str, use_test_net_dcld: str, use_main_net_http: bool
pass
# b.
- try:
- revoked_cert_authority_key_id = revoked_cert.extensions.get_extension_for_oid(
- x509.OID_AUTHORITY_KEY_IDENTIFIER).value.key_identifier
-
- if revoked_cert_authority_key_id is None or revoked_cert_authority_key_id != crl_signer_subject_key_id:
- logging.warning("CRL Authority Key ID is not CRL Signer Subject Key ID, continue...")
- continue
- except Exception:
- logging.warning("CRL Authority Key ID not found, continue...")
- continue
+ # TODO: Verify that the certificate chain of the entry is linking to the same PAA
+ # that issued the CRLSignerCertificate for this entry, including path through
+ # CRLSignerDelegator if present. If the PAAs under which were issued the certificate
+ # and the CRLSignerCertificate are different, ignore the entry.
# c. and d.
serialnumber_list.append(bytes(str('{:02X}'.format(revoked_cert.serial_number)), 'utf-8').decode('utf-8'))