| #!/usr/bin/env python3 |
| # |
| # 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. |
| # You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, software |
| # distributed under the License is distributed on an "AS IS" BASIS, |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| # See the License for the specific language governing permissions and |
| # limitations under the License. |
| # |
| # Note: py cryptography supports indirect CRLs trom version 44.0.0 and above. |
| # You may need to update your cryptography version. |
| # |
| # Generates a basic RevocationSet from TestNet |
| # Usage: |
| # python ./credentials/generate-revocation-set.py --help |
| |
| import base64 |
| import dataclasses |
| import json |
| import logging |
| import os |
| import subprocess |
| import sys |
| import unittest |
| from enum import Enum |
| |
| import click |
| import requests |
| from click_option_group import AllOptionGroup, RequiredMutuallyExclusiveOptionGroup, optgroup |
| from cryptography import x509 |
| from cryptography.hazmat.primitives import serialization |
| from cryptography.hazmat.primitives.asymmetric import ec |
| from cryptography.x509.extensions import ExtensionNotFound |
| from cryptography.x509.oid import NameOID |
| |
| log = logging.getLogger(__name__) |
| |
| # Supported log levels, mapping string values required for argument |
| # parsing into logging constants |
| __LOG_LEVELS__ = { |
| 'debug': logging.DEBUG, |
| 'info': logging.INFO, |
| 'warn': logging.WARNING, |
| 'fatal': logging.FATAL, |
| } |
| |
| |
| class RevocationType(Enum): |
| CRL = 1 |
| |
| |
| class CertVerificationResult(Enum): |
| SUCCESS = 1 |
| SKID_NOT_FOUND = 2 |
| AKID_NOT_FOUND = 3 |
| SIGNATURE_VERIFICATION_FAILED = 4 |
| ISSUER_MISMATCH = 5 |
| AKID_MISMATCH = 6 |
| |
| |
| @dataclasses.dataclass |
| class RevocationPoint: |
| vid: int |
| label: str |
| issuerSubjectKeyID: str |
| pid: int |
| isPAA: bool |
| crlSignerCertificate: str |
| dataURL: str |
| dataFileSize: str |
| dataDigest: str |
| dataDigestType: int |
| revocationType: int |
| schemaVersion: int |
| crlSignerDelegator: str |
| |
| |
| @dataclasses.dataclass |
| class RevocationSet: |
| type: str |
| issuer_subject_key_id: str |
| issuer_name: str |
| revoked_serial_numbers: [str] |
| crl_signer_cert: str |
| crl_signer_delegator: str = None |
| |
| def asDict(self): |
| return dataclasses.asdict(self) |
| |
| |
| OID_VENDOR_ID = x509.ObjectIdentifier("1.3.6.1.4.1.37244.2.1") |
| OID_PRODUCT_ID = x509.ObjectIdentifier("1.3.6.1.4.1.37244.2.2") |
| |
| PRODUCTION_NODE_URL = "https://on.dcl.csa-iot.org:26657" |
| PRODUCTION_NODE_URL_REST = "https://on.dcl.csa-iot.org" |
| TEST_NODE_URL_REST = "https://on.test-net.dcl.csa-iot.org" |
| |
| |
| def extract_single_integer_attribute(subject, oid): |
| attribute_list = subject.get_attributes_for_oid(oid) |
| |
| if len(attribute_list) == 1: |
| return int(attribute_list[0].value, 16) |
| |
| return None |
| |
| |
| def extract_fallback_tag_from_common_name(cn, marker): |
| val_len = 4 |
| start_idx = cn.find(marker) |
| |
| if start_idx != -1: |
| val_start_idx = start_idx + len(marker) |
| val = cn[val_start_idx:val_start_idx + val_len] |
| return int(val, 16) if len(val) == 4 else None |
| |
| return None |
| |
| |
| def parse_vid_pid_from_distinguished_name(distinguished_name): |
| # VID/PID encoded using Matter specific RDNs |
| vid = extract_single_integer_attribute(distinguished_name, OID_VENDOR_ID) |
| pid = extract_single_integer_attribute(distinguished_name, OID_PRODUCT_ID) |
| |
| # Fallback method to get the VID/PID, encoded in CN as "Mvid:FFFF Mpid:1234" |
| if vid is None and pid is None: |
| cn = distinguished_name.get_attributes_for_oid(NameOID.COMMON_NAME)[0].value |
| vid = extract_fallback_tag_from_common_name(cn, 'Mvid:') |
| pid = extract_fallback_tag_from_common_name(cn, 'Mpid:') |
| |
| return vid, pid |
| |
| |
| def get_akid(cert: x509.Certificate) -> str: |
| return cert.extensions.get_extension_for_oid(x509.OID_AUTHORITY_KEY_IDENTIFIER).value.key_identifier.hex().upper() |
| |
| |
| def get_skid(cert: x509.Certificate) -> str: |
| return cert.extensions.get_extension_for_oid(x509.OID_SUBJECT_KEY_IDENTIFIER).value.key_identifier.hex().upper() |
| |
| |
| def verify_cert(cert: x509.Certificate, root: x509.Certificate) -> CertVerificationResult: |
| ''' |
| Verifies if the cert is signed by root. |
| ''' |
| try: |
| cert_akid = get_akid(cert) |
| except ExtensionNotFound: |
| return CertVerificationResult.AKID_NOT_FOUND |
| try: |
| root_skid = get_skid(root) |
| except ExtensionNotFound: |
| return CertVerificationResult.SKID_NOT_FOUND |
| |
| if cert_akid != root_skid: |
| return CertVerificationResult.AKID_MISMATCH |
| |
| if cert.issuer != root.subject: |
| return CertVerificationResult.ISSUER_MISMATCH |
| |
| # public_key().verify() do not return anything if signature is valid, |
| # will raise an exception if signature is invalid |
| try: |
| root.public_key().verify(cert.signature, cert.tbs_certificate_bytes, ec.ECDSA(cert.signature_hash_algorithm)) |
| except Exception: |
| return CertVerificationResult.SIGNATURE_VERIFICATION_FAILED |
| |
| return CertVerificationResult.SUCCESS |
| |
| |
| def is_self_signed_certificate(cert: x509.Certificate) -> bool: |
| result = verify_cert(cert, cert) |
| if result == CertVerificationResult.SUCCESS: |
| return True |
| log.debug("Certificate with subject: %s is not a valid self-signed certificate. Result: %s", |
| cert.subject.rfc4514_string(), result.name) |
| return False |
| |
| |
| # delegator is optional so can be None, but crl_signer and paa has to be present |
| def validate_cert_chain(crl_signer: x509.Certificate, crl_signer_delegator: x509.Certificate, paa: x509.Certificate): |
| ''' |
| There could be four scenarios: |
| 1. CRL Signer is PAA itself, hence its self-signed certificate |
| 2. CRL Signer is PAI certificate, and we can validate (crl_signer -> paa) chain |
| 3. CRL Signer delegator is PAA, and we can validate (crl_signer -> crl_signer_delegator(paa) -> paa) chain |
| 4. CRL Signer delegator is PAI, and we can validate (crl_signer -> crl_signer_delegator -> paa) chain |
| ''' |
| |
| if crl_signer_delegator: |
| result_signer = verify_cert(crl_signer, crl_signer_delegator) |
| if result_signer != CertVerificationResult.SUCCESS: |
| log.debug("Cannot verify certificate subject: %s issued by certificate subject: %s. Result: %s", |
| crl_signer.subject.rfc4514_string(), crl_signer_delegator.subject.rfc4514_string(), result_signer.name) |
| return False |
| |
| result_delegator = verify_cert(crl_signer_delegator, paa) |
| if result_delegator != CertVerificationResult.SUCCESS: |
| log.debug("Cannot verify certificate subject: %s issued by certificate subject: %s. Result: %s", |
| crl_signer_delegator.subject.rfc4514_string(), paa.subject.rfc4514_string(), result_delegator.name) |
| return False |
| return True |
| result = verify_cert(crl_signer, paa) |
| if result != CertVerificationResult.SUCCESS: |
| log.debug("Cannot verify certificate subject: %s issued by certificate subject: %s. Result: %s", |
| crl_signer.subject.rfc4514_string(), paa.subject.rfc4514_string(), result.name) |
| return False |
| return True |
| |
| |
| def validate_vid_pid(revocation_point: RevocationPoint, crl_signer_certificate: x509.Certificate, crl_signer_delegator_certificate: x509.Certificate) -> bool: |
| crl_signer_vid, crl_signer_pid = parse_vid_pid_from_distinguished_name(crl_signer_certificate.subject) |
| log.debug("vid: %s)", revocation_point.vid) |
| log.debug("crl_signer_vid: %s)", crl_signer_vid) |
| if revocation_point.isPAA: |
| if crl_signer_vid is not None: |
| if revocation_point.vid != crl_signer_vid: |
| log.warning("VID in CRL Signer Certificate does not match with VID in revocation point, continue...") |
| return False |
| else: |
| vid_to_match = crl_signer_vid |
| pid_to_match = crl_signer_pid |
| |
| # if the CRL Signer is delegated then match the VID and PID of the CRL Signer Delegator |
| if crl_signer_delegator_certificate: |
| vid_to_match, pid_to_match = parse_vid_pid_from_distinguished_name(crl_signer_delegator_certificate.subject) |
| log.debug("vid_to_match: %s)", vid_to_match) |
| log.debug("pid_to_match: %s)", pid_to_match) |
| if vid_to_match is None or revocation_point.vid != vid_to_match: |
| log.warning("VID in CRL Signer Certificate does not match with VID in revocation point, continue...") |
| return False |
| |
| if pid_to_match is not None: |
| if revocation_point.pid != pid_to_match: |
| log.warning("PID in CRL Signer Certificate does not match with PID in revocation point, continue...") |
| return False |
| |
| return True |
| |
| |
| def generate_revocation_set_from_crl(crl_file: x509.CertificateRevocationList, |
| crl_signer_certificate: x509.Certificate, |
| certificate_authority_name: x509.Name, |
| certificate_akid_hex: str, |
| crl_signer_delegator_cert: x509.Certificate) -> RevocationSet: |
| """Generate a revocation set from a CRL file. |
| |
| Args: |
| crl_file: The CRL object containing revoked certificates |
| crl_signer_certificate: The certificate object used to sign the CRL |
| certificate_authority_name: x509.Name of the issuer |
| certificate_akid_hex: Hex encoded Authority Key Identifier |
| crl_signer_delegator_cert: crl signer delegator certificate object |
| |
| Returns: |
| RevocationSet containing the revocation set data with fields: |
| - type: "revocation_set" |
| - issuer_subject_key_id: Authority Key Identifier (hex) |
| - issuer_name: Issuer name (base64) |
| - revoked_serial_numbers: List of revoked serial numbers |
| - crl_signer_cert: CRL signer certificate (base64 DER) |
| - crl_signer_delegator: Optional delegator certificate (base64 DER) |
| """ |
| serialnumber_list = [] |
| |
| for revoked_cert in crl_file: |
| try: |
| cert_issuer_entry_ext = revoked_cert.extensions.get_extension_for_oid(x509.CRLEntryExtensionOID.CERTIFICATE_ISSUER) |
| revoked_cert_issuer = cert_issuer_entry_ext.value.get_values_for_type(x509.DirectoryName)[0] |
| if revoked_cert_issuer is not None: |
| # check if this really are the same thing |
| if revoked_cert_issuer != x509.DirectoryName(certificate_authority_name).value: |
| log.warning("CRL entry issuer is not CRL File Issuer, continue...") |
| continue |
| except Exception: |
| pass |
| |
| # Ensure the serial number is always a 2-byte aligned hex string. |
| # TestDACRevocationDelegateImpl encodes the serial number as an even-length hex string |
| # using BytesToHex in src/lib/support/BytesToHex.cpp. |
| # As the primary consumer of this data, we should use the same here. |
| serialnumber = '{:02X}'.format(revoked_cert.serial_number) |
| serialnumber = serialnumber if len(serialnumber) % 2 == 0 else '0' + serialnumber |
| serialnumber_list.append(serialnumber) |
| |
| entry = RevocationSet( |
| type='revocation_set', |
| issuer_subject_key_id=certificate_akid_hex, |
| issuer_name=get_b64_name(certificate_authority_name), |
| revoked_serial_numbers=serialnumber_list, |
| crl_signer_cert=base64.b64encode(crl_signer_certificate.public_bytes(serialization.Encoding.DER)).decode('utf-8'), |
| ) |
| |
| if crl_signer_delegator_cert: |
| entry.crl_signer_delegator = base64.b64encode( |
| crl_signer_delegator_cert.public_bytes(serialization.Encoding.DER)).decode('utf-8') |
| |
| return entry |
| |
| |
| # This is implemented as per point (9) in 6.2.4.1. Conceptual algorithm for revocation set construction |
| def get_certificate_authority_details(crl_signer_certificate: x509.Certificate, |
| crl_signer_delegator_cert: x509.Certificate, |
| paa_certificate_object: x509.Certificate, |
| is_paa: bool) -> tuple[x509.Name, str]: |
| """Get certificate authority name and AKID based on certificate hierarchy. |
| |
| Args: |
| crl_signer_certificate: The CRL signer certificate |
| crl_signer_delegator_cert: Optional delegator certificate |
| paa_certificate_object: Optional PAA certificate |
| is_paa: Whether this is a PAA certificate |
| |
| Returns: |
| tuple[str, str]: (certificate_authority_name, certificate_akid_hex) |
| """ |
| if is_paa and not is_self_signed_certificate(crl_signer_certificate): |
| cert_for_details = paa_certificate_object |
| log.debug("Using PAA certificate for details") |
| elif crl_signer_delegator_cert: |
| cert_for_details = crl_signer_delegator_cert |
| log.debug("Using CRL Signer Delegator certificate for details") |
| else: |
| cert_for_details = crl_signer_certificate |
| log.debug("Using CRL Signer certificate for details") |
| |
| certificate_authority_name = cert_for_details.subject |
| try: |
| certificate_akid = get_skid(cert_for_details) |
| log.debug("Certificate Authority Name: %s", certificate_authority_name) |
| log.debug("Certificate AKID: %s", certificate_akid) |
| |
| return certificate_authority_name, certificate_akid |
| except ExtensionNotFound: |
| log.warning("Certificate SKID not found in authoarity certificate.") |
| |
| |
| def get_b64_name(name: x509.name.Name) -> str: |
| ''' |
| Get base64 encoded name |
| ''' |
| return base64.b64encode(name.public_bytes()).decode('utf-8') |
| |
| |
| def fetch_crl_from_url(url: str, timeout: int) -> x509.CertificateRevocationList: |
| log.debug("Fetching CRL from %s", url) |
| |
| try: |
| r = requests.get(url, timeout=timeout) |
| log.debug("Fetched CRL: %s", r.content) |
| return x509.load_der_x509_crl(r.content) |
| except Exception as e: |
| log.error("Failed to fetch a valid CRL: %s", e) |
| |
| |
| class DclClientInterface: |
| ''' |
| An interface for interacting with DCLD. |
| ''' |
| |
| def send_get_request(self, url: str) -> dict: |
| ''' |
| Send a GET request for a json object. |
| ''' |
| try: |
| return requests.get(url).json() |
| except Exception as e: |
| log.error("Failed to fetch %s: %s", url, e) |
| return None |
| |
| def get_revocation_points(self) -> list[RevocationPoint]: |
| ''' |
| Get revocation points from DCL |
| |
| Returns |
| ------- |
| list[RevocationPoint] |
| List of revocation points |
| ''' |
| raise NotImplementedError |
| |
| def get_revocation_points_by_skid(self, issuer_subject_key_id) -> list[RevocationPoint]: |
| ''' |
| Get revocation points by subject key ID |
| |
| Parameters |
| ---------- |
| issuer_subject_key_id: str |
| Subject key ID |
| |
| Returns |
| ------- |
| list[RevocationPoint] |
| List of revocation points |
| ''' |
| raise NotImplementedError |
| |
| def get_approved_certificate(self, subject_name: x509.name.Name, skid_hex: str) -> tuple[bool, x509.Certificate]: |
| ''' |
| Get certificate from DCL. |
| ''' |
| raise NotImplementedError |
| |
| def get_only_approved_certificate(self, response: dict, skid_hex: str) -> tuple[bool, x509.Certificate | None]: |
| ''' |
| Get only approved certificate from DCL resposne. |
| ''' |
| if response is None or not response.get("approvedCertificates", {}).get("certs", []): |
| raise requests.exception.NotFound(f"No certificate found for {skid_hex}") |
| if len(response["approvedCertificates"]["certs"]) > 1: |
| raise ValueError(f"Multiple certificates found for {skid_hex}") |
| issuer_certificate = x509.load_pem_x509_certificate(bytes(response["approvedCertificates"]["certs"][0]["pemCert"], "utf-8")) |
| return response["approvedCertificates"]["certs"][0]["isRoot"], issuer_certificate |
| |
| def get_paa_cert(self, initial_cert: x509.Certificate) -> x509.Certificate | None: |
| ''' |
| Get the PAA certificate for the CRL Signer Certificate. |
| ''' |
| issuer_name = initial_cert.issuer |
| try: |
| akid = get_akid(initial_cert) |
| except ExtensionNotFound: |
| log.warning('Certificate AKID not found.') |
| return None |
| paa_certificate = None |
| while not paa_certificate: |
| try: |
| is_root, issuer_certificate = self.get_approved_certificate(issuer_name, akid) |
| if is_root: |
| paa_certificate = issuer_certificate |
| break |
| |
| except Exception as e: |
| log.error("Failed to get PAA certificate: %s", e) |
| return None |
| log.debug("issuer_name: %s", issuer_certificate.subject.rfc4514_string()) |
| issuer_name = issuer_certificate.issuer |
| try: |
| akid = get_akid(issuer_certificate) |
| except ExtensionNotFound: |
| log.warning('Issuer Certificate AKID not found.') |
| log.debug("akid: %s", akid) |
| if paa_certificate is None: |
| log.warning("PAA Certificate not found, continue...") |
| return paa_certificate |
| |
| def get_crl_file(self, |
| revocation_point: RevocationPoint, |
| crl_signer_certificate: x509.Certificate) -> x509.CertificateRevocationList: |
| """Obtain the CRL.""" |
| try: |
| r = requests.get(revocation_point.dataURL, timeout=5) |
| log.debug("Fetched CRL: %s", r.content) |
| return x509.load_der_x509_crl(r.content) |
| except Exception: |
| log.warning("Failed to fetch a valid CRL for': %s", crl_signer_certificate.subject.rfc4514_string()) |
| |
| def get_formatted_hex_skid(self, skid_hex: str) -> str: |
| return ':'.join([skid_hex[i:i+2] for i in range(0, len(skid_hex), 2)]) |
| |
| |
| class NodeDclClient(DclClientInterface): |
| ''' |
| A client for interacting with DCLD using command line interface (CLI). |
| ''' |
| |
| def __init__(self, dcld_exe: str, use_test_net: bool): |
| ''' |
| Initialize the client. |
| |
| dcld_exe: str |
| Path to `dcld` executable. |
| use_test_net: bool |
| Indicates if the client should use TestNet or MainNet URL with dcld executable. |
| ''' |
| |
| self.dcld_exe = dcld_exe |
| self.use_test_net = use_test_net |
| |
| 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 + ([] if self.use_test_net else ['--node', PRODUCTION_NODE_URL]) |
| |
| 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. |
| ''' |
| |
| # Set the output as JSON |
| subprocess.Popen([self.dcld_exe, 'config', 'output', 'json']) |
| |
| 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) -> list[RevocationPoint]: |
| ''' |
| Get revocation points from DCL. |
| |
| Returns |
| ------- |
| list[RevocationPoint] |
| List of revocation points. |
| ''' |
| |
| response = self.get_dcld_cmd_output_json(['query', 'pki', 'all-revocation-points']) |
| return [RevocationPoint(**r) for r in response["PkiRevocationDistributionPoint"]] |
| |
| def get_revocation_points_by_skid(self, issuer_subject_key_id) -> list[RevocationPoint]: |
| ''' |
| Get revocation points by subject key ID. |
| |
| Parameters |
| ---------- |
| issuer_subject_key_id: str |
| Subject key ID. |
| |
| Returns |
| ------- |
| list[RevocationPoint] |
| List of revocation points. |
| ''' |
| |
| response = self.get_dcld_cmd_output_json(['query', 'pki', 'revocation-points', |
| '--issuer-subject-key-id', issuer_subject_key_id]) |
| log.debug("Response revocation points: %s", response) |
| return [RevocationPoint(**r) for r in response["pkiRevocationDistributionPointsByIssuerSubjectKeyID"]["points"]] |
| |
| def get_approved_certificate(self, subject_name: x509.name.Name, skid_hex: str) -> tuple[bool, x509.Certificate]: |
| ''' |
| Get certificate from DCL. |
| |
| Parameters |
| ---------- |
| subject_name: x509.name.Name |
| Subject Name object. |
| skid_hex: str |
| Subject Key ID in hex format. |
| |
| Returns |
| ------- |
| tuple[bool, x509.Certificate] |
| Tuple of is_paa and the certificate from the DCL. |
| ''' |
| subject_name_b64 = get_b64_name(subject_name) |
| query_cmd_list = ['query', 'pki', 'x509-cert', '-u', subject_name_b64, '-k', skid_hex] |
| log.debug("Fetching issuer from dcl query%s", ' '.join(query_cmd_list)) |
| response = self.get_dcld_cmd_output_json(query_cmd_list) |
| return self.get_only_approved_certificate(response, skid_hex) |
| |
| |
| class RestDclClient(DclClientInterface): |
| ''' |
| A client for interacting with DCLD using the REST API. |
| ''' |
| |
| def __init__(self, use_test_net: bool): |
| ''' |
| Initialize the client. |
| |
| use_test_net: bool |
| Indicates if the client should use TestNet or MainNet REST API URL. |
| ''' |
| self.rest_node_url = TEST_NODE_URL_REST if use_test_net else PRODUCTION_NODE_URL_REST |
| |
| def get_revocation_points(self) -> list[RevocationPoint]: |
| ''' |
| Get revocation points from DCL. |
| |
| Returns |
| ------- |
| list[RevocationPoint] |
| List of revocation points. |
| ''' |
| |
| response = self.send_get_request(f"{self.rest_node_url}/dcl/pki/revocation-points") |
| return [RevocationPoint(**r) for r in response["PkiRevocationDistributionPoint"]] |
| |
| def get_revocation_points_by_skid(self, issuer_subject_key_id) -> list[RevocationPoint]: |
| ''' |
| Get revocation points by subject key ID. |
| |
| Parameters |
| ---------- |
| issuer_subject_key_id: str |
| Subject key ID. |
| |
| Returns |
| ------- |
| list[RevocationPoint] |
| List of revocation points. |
| ''' |
| |
| response = self.send_get_request(f"{self.rest_node_url}/dcl/pki/revocation-points/{issuer_subject_key_id}") |
| return [RevocationPoint(**r) for r in response["pkiRevocationDistributionPointsByIssuerSubjectKeyID"]["points"]] |
| |
| def get_approved_certificate(self, subject_name: x509.name.Name, skid_hex: str) -> tuple[bool, x509.Certificate]: |
| ''' |
| Get certificate from DCL. |
| |
| Parameters |
| ---------- |
| subject_name: x509.name.Name |
| Subject Name object. |
| skid_hex: str |
| Subject Key ID in hex format. |
| |
| Returns |
| ------- |
| tuple[bool, x509.Certificate] |
| Tuple of is_paa and the certificate from the DCL. |
| ''' |
| log.debug("Fetching issuer from:%s/dcl/pki/certificates/%s/%s", |
| self.rest_node_url, get_b64_name(subject_name), self.get_formatted_hex_skid(skid_hex)) |
| response = self.send_get_request( |
| f"{self.rest_node_url}/dcl/pki/certificates/{get_b64_name(subject_name)}/{self.get_formatted_hex_skid(skid_hex)}") |
| log.debug("Response certificate: %s", response) |
| return self.get_only_approved_certificate(response, skid_hex) |
| |
| |
| class LocalFilesDclClient(DclClientInterface): |
| ''' |
| A client for interacting with local DLCD response data. |
| ''' |
| |
| def __init__(self, crls: [], dcl_certificates: [], revocation_points_response_file: str): |
| ''' |
| Initialize the client. |
| |
| Parameters |
| ---------- |
| crls: list |
| List of CRL files. |
| dcl_certificates: list |
| List of certificate files. |
| revocation_points_response_file: str |
| Path to the get-revocation-points response json file. |
| ''' |
| |
| log.debug("Loading certificates from %s", dcl_certificates) |
| log.debug("Loading crls from %s", crls) |
| log.debug("Loading revocation points response from %s", revocation_points_response_file) |
| self.crls = self.get_crls(crls) |
| self.revocation_points = [RevocationPoint(**r) |
| for r in json.load(revocation_points_response_file)["PkiRevocationDistributionPoint"]] |
| self.authoritative_certs = self.get_authoritative_certificates(dcl_certificates) |
| |
| def get_lookup_key(self, certificate: x509.Certificate) -> str: |
| ''' |
| Get key used in this class to lookup certificates. |
| |
| Parameters |
| ---------- |
| certificate: x509.Certificate |
| Certificate object. |
| |
| Returns |
| ------- |
| str: |
| lookup key derived from the certificate. |
| ''' |
| base64_name = get_b64_name(certificate.subject) |
| try: |
| skid = get_skid(certificate) |
| return self.format_lookup_key(base64_name, skid) |
| except ExtensionNotFound: |
| log.warning("CertificateSKID not found, continue...") |
| |
| def format_lookup_key(self, base64_name: str, skid_hex: str) -> str: |
| ''' |
| Get formatted key used in this class to lookup certificates. |
| |
| Parameters |
| ---------- |
| base64_name: str |
| Base64 encoded subject name. |
| skid_hex: str |
| Subject Key ID in hex format. |
| |
| Returns |
| ------- |
| str: |
| Key used in this class to lookup certificates. |
| ''' |
| delimiter = '/' |
| skid_hex_formatted = self.get_formatted_hex_skid(skid_hex) |
| return delimiter.join([base64_name, skid_hex_formatted]) |
| |
| def get_crls(self, unread_crls: []) -> list[x509.CertificateRevocationList]: |
| ''' |
| Get CRLs from list of files. |
| |
| Parameters |
| ---------- |
| unread_crls: list |
| List of CRL files. |
| |
| Returns |
| ------- |
| list[x509.CertificateRevocationList] |
| List of CRLs. |
| ''' |
| crls = [] |
| for file in unread_crls: |
| crl_content = file.read() |
| crl_file = x509.load_der_x509_crl(crl_content) |
| crls.append(crl_file) |
| return crls |
| |
| def get_authoritative_certificates(self, dcl_certificates: []) -> dict[str, x509.Certificate]: |
| ''' |
| Get certificates from revocation points response file and list of provided dcl certificates. |
| |
| Parameters |
| ---------- |
| dcl_certificates: list |
| List of certificate files. |
| |
| Returns |
| ------- |
| dict[str, x509.CertificateRevocationList] |
| Dictionary of certificates, keyed by lookup key. |
| ''' |
| certificates = {} |
| log.debug("Loading certificates from %s", dcl_certificates) |
| if dcl_certificates: |
| for file in dcl_certificates: |
| log.debug("Loading certificate from %s", file) |
| # with open(file, "r") as f: |
| certificate = x509.load_pem_x509_certificate(file.read()) |
| certificates[self.get_lookup_key(certificate)] = certificate |
| |
| log.debug("Loading certificates from revocation_points_response file.") |
| for point in self.revocation_points: |
| if point.crlSignerDelegator: |
| certificate = x509.load_pem_x509_certificate(bytes(point.crlSignerDelegator, 'utf-8')) |
| certificates[self.get_lookup_key(certificate)] = certificate |
| elif point.crlSignerCertificate: |
| certificate = x509.load_pem_x509_certificate(bytes(point.crlSignerCertificate, 'utf-8')) |
| certificates[self.get_lookup_key(certificate)] = certificate |
| return certificates |
| |
| def get_revocation_points(self) -> list[RevocationPoint]: |
| ''' |
| Get revocation points from DCL. |
| |
| Returns |
| ------- |
| list[RevocationPoint] |
| List of revocation points. |
| ''' |
| return self.revocation_points |
| |
| def get_revocation_points_by_skid(self, issuer_subject_key_id) -> list[RevocationPoint]: |
| ''' |
| Get revocation points by subject key ID |
| |
| Parameters |
| ---------- |
| issuer_subject_key_id: str |
| Subject key ID. |
| |
| Returns |
| ------- |
| list[RevocationPoint] |
| List of revocation points with the same issuer subject key ID. |
| ''' |
| same_issuer_points = [] |
| for point in self.revocation_points: |
| if point.issuerSubjectKeyID == issuer_subject_key_id: |
| same_issuer_points.append(point) |
| return same_issuer_points |
| |
| def get_approved_certificate(self, subject_name: x509.name.Name, skid_hex: str) -> tuple[bool, x509.Certificate]: |
| ''' |
| Get certificate from DCL |
| |
| Parameters |
| ---------- |
| subject_name: x509.name.Name |
| Subject name object. |
| |
| skid_hex: str |
| Subject Key ID in hex format. |
| |
| Returns |
| ------- |
| tuple[bool, x509.Certificate] |
| Tuple of is_paa and the certificate from the DCL. |
| ''' |
| lookup_key = self.format_lookup_key(get_b64_name(subject_name), skid_hex) |
| if lookup_key in self.authoritative_certs: |
| return is_self_signed_certificate(self.authoritative_certs[lookup_key]), self.authoritative_certs[lookup_key] |
| return False, None |
| |
| def get_crl_file(self, |
| unused_revocation_point: RevocationPoint, |
| crl_signer_certificate: x509.Certificate) -> x509.CertificateRevocationList: |
| ''' |
| Obtain the CRL. |
| |
| Parameters |
| ---------- |
| unused_revocation_point: RevocationPoint |
| Revocation point. Not used. |
| |
| crl_signer_certificate: x509.Certificate |
| Crl signer certificate. |
| |
| Returns |
| ------- |
| x509.CertificateRevocationList |
| CRL signed by the CRL signer certificate. |
| ''' |
| for crl in self.crls: |
| if crl.issuer.public_bytes() == crl_signer_certificate.subject.public_bytes(): |
| log.debug("Found CRL for issuer: %s", crl.issuer.rfc4514_string()) |
| return crl |
| return None |
| |
| |
| @click.group() |
| def cli(): |
| pass |
| |
| |
| @cli.command('from-dcl') |
| @click.help_option('-h', '--help') |
| @optgroup.group('Input data sources', cls=RequiredMutuallyExclusiveOptionGroup) |
| @optgroup.option('--use-main-net-dcld', type=str, default='', metavar='PATH', help="Location of `dcld` binary, to use `dcld` for mirroring MainNet.") |
| @optgroup.option('--use-test-net-dcld', type=str, default='', metavar='PATH', help="Location of `dcld` binary, to use `dcld` for mirroring TestNet.") |
| @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.option('--use-local-data', is_flag=True, type=bool, help="Fake response directory: see \" DATA_DIR/") |
| @optgroup.group('Required arguments if use-local-data is used', cls=AllOptionGroup) |
| @optgroup.option('--certificates', type=click.File('rb'), multiple=True, help='Paths to PEM formated certificates (i.e. PAA) in DCL but missing from the revocation-points-response file.') |
| @optgroup.option('--crls', type=click.File('rb'), multiple=True, help='Paths to the crl der files') |
| @optgroup.option('--revocation-points-response', type=click.File('rb'), help='Path to the get-revocation-points response json file.') |
| @optgroup.group('Optional output arguments') |
| @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 from_dcl(use_main_net_dcld: str, use_test_net_dcld: str, use_main_net_http: bool, use_test_net_http: bool, use_local_data: bool, revocation_points_response: str, crls: [], certificates: [], output: str, log_level: str): |
| """Generate revocation set from DCL using generation algorithm from Matter Spec section 6.2.4.1.""" |
| logging.basicConfig( |
| level=log_level, |
| format='%(asctime)s %(name)s %(levelname)-7s %(message)s', |
| datefmt='%Y-%m-%d %H:%M:%S' |
| ) |
| |
| if use_local_data: |
| dcld_client = LocalFilesDclClient(crls, certificates, revocation_points_response) |
| elif use_main_net_http or use_test_net_http: |
| dcld_client = RestDclClient(bool(use_test_net_http)) |
| else: |
| dcld_client = NodeDclClient(use_main_net_dcld or use_test_net_dcld, bool(use_test_net_dcld)) |
| |
| revocation_point_list = dcld_client.get_revocation_points() |
| revocation_set = [] |
| for revocation_point in revocation_point_list: |
| # 1. Validate Revocation Type |
| if revocation_point.revocationType != RevocationType.CRL.value: |
| log.warning("Revocation Type is not CRL, continue...") |
| continue |
| |
| # 2. Parse the certificate |
| try: |
| crl_signer_certificate = x509.load_pem_x509_certificate(bytes(revocation_point.crlSignerCertificate, 'utf-8')) |
| except Exception: |
| log.warning("CRL Signer Certificate is not valid, continue...") |
| continue |
| |
| # Parse the crl signer delegator |
| crl_signer_delegator_cert = None |
| if revocation_point.crlSignerDelegator: |
| crl_signer_delegator_cert_pem = revocation_point.crlSignerDelegator |
| log.debug("CRLSignerDelegator: %s", crl_signer_delegator_cert_pem) |
| try: |
| crl_signer_delegator_cert = x509.load_pem_x509_certificate(bytes(crl_signer_delegator_cert_pem, 'utf-8')) |
| except Exception: |
| log.warning("CRL Signer Delegator Certificate not found...") |
| |
| # 3. and 4. Validate VID/PID |
| if not validate_vid_pid(revocation_point, crl_signer_certificate, crl_signer_delegator_cert): |
| log.warning("Failed to validate VID/PID, continue...") |
| continue |
| |
| # 5. Validate the certification path containing CRLSignerCertificate. |
| paa_certificate_object = dcld_client.get_paa_cert(crl_signer_certificate) |
| if paa_certificate_object is None: |
| log.warning("PAA Certificate not found, continue...") |
| continue |
| |
| if validate_cert_chain(crl_signer_certificate, crl_signer_delegator_cert, paa_certificate_object) is False: |
| log.warning("Failed to validate CRL Signer Certificate chain, continue...") |
| continue |
| |
| # 6. Obtain the CRL |
| crl_file = dcld_client.get_crl_file(revocation_point, crl_signer_certificate) |
| if crl_file is None: |
| log.warning("CRL file not found for revocation point, continue...") |
| continue |
| |
| # 7. Perform CRL File Validation |
| # a. |
| try: |
| crl_signer_skid = get_skid(crl_signer_certificate) |
| except ExtensionNotFound: |
| log.warning("CRL Signer SKID not found, continue...") |
| continue |
| try: |
| crl_akid = get_akid(crl_file) |
| except ExtensionNotFound: |
| log.warning("CRL AKID is not found, continue...") |
| continue |
| if crl_akid != crl_signer_skid: |
| log.warning("CRL AKID is not CRL Signer SKID, continue...") |
| continue |
| |
| # b. |
| same_issuer_points = dcld_client.get_revocation_points_by_skid(crl_akid) |
| count_with_matching_vid_issuer_skid = sum(item.vid == revocation_point.vid for item in same_issuer_points) |
| |
| if count_with_matching_vid_issuer_skid > 1: |
| try: |
| issuing_distribution_point = crl_file.extensions.get_extension_for_oid( |
| x509.oid.ExtensionOID.ISSUING_DISTRIBUTION_POINT).value |
| except Exception: |
| log.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: |
| log.warning("CRL Issuing Distribution Point URI is not CRL URL, continue...") |
| continue |
| else: |
| log.warning("CRL Issuing Distribution Point URI is not CRL URL, continue...") |
| continue |
| |
| # TODO: 8. Validate CRL as per Section 6.3 of RFC 5280 |
| |
| # 9. Decide on certificate authority name and AKID |
| certificate_authority_name, certificate_akid_hex = get_certificate_authority_details( |
| crl_signer_certificate, crl_signer_delegator_cert, paa_certificate_object, revocation_point.isPAA) |
| |
| # validate issuer skid matchces with the one in revocation points |
| log.debug("revocation_point.issuerSubjectKeyID: %s", revocation_point.issuerSubjectKeyID) |
| |
| if revocation_point.issuerSubjectKeyID != certificate_akid_hex: |
| log.warning("CRL Issuer Subject Key ID is not CRL Signer Subject Key ID, continue...") |
| continue |
| |
| # 10. Iterate through the Revoked Certificates List |
| entry = generate_revocation_set_from_crl(crl_file, crl_signer_certificate, |
| certificate_authority_name, certificate_akid_hex, crl_signer_delegator_cert) |
| log.debug("Entry to append: %s", entry) |
| revocation_set.append(entry) |
| |
| with open(output, 'w+') as outfile: |
| json.dump([revocation.asDict() for revocation in revocation_set], outfile, indent=4) |
| |
| |
| class TestRevocationSetGeneration(unittest.TestCase): |
| """Test class for revocation set generation""" |
| |
| def setUp(self): |
| # Get the directory containing this file |
| self.test_base_dir = os.path.dirname(os.path.abspath(__file__)) |
| |
| def get_test_file_path(self, filename): |
| return os.path.join(self.test_base_dir, 'test', filename) |
| |
| def get_expected_revocation_set(self, idx): |
| with open(os.path.join(self.test_base_dir, 'test/revoked-attestation-certificates/revocation-sets/revocation-set.json')) as f: |
| return RevocationSet(**json.load(f)[idx]) |
| |
| def compare_revocation_sets(self, generated_set, expected): |
| # Compare required fields |
| self.assertEqual(generated_set.type, expected.type) |
| self.assertEqual(generated_set.issuer_subject_key_id, expected.issuer_subject_key_id) |
| self.assertEqual(generated_set.issuer_name, expected.issuer_name) |
| self.assertEqual(set(generated_set.revoked_serial_numbers), set(expected.revoked_serial_numbers)) |
| self.assertEqual(generated_set.crl_signer_cert, expected.crl_signer_cert) |
| |
| # Compare optional fields if present in either set |
| if generated_set.crl_signer_delegator or expected.crl_signer_delegator: |
| self.assertEqual(generated_set.crl_signer_delegator, expected.crl_signer_delegator, |
| f'CRL signer delegator certificates do not match, expected: {expected.crl_signer_delegator}, actual: {generated_set.crl_signer_delegator}') |
| |
| def test_paa_revocation_set(self): |
| """Test generation of PAA revocation set""" |
| with open(self.get_test_file_path('revoked-attestation-certificates/Chip-Test-PAA-FFF1-CRL.pem'), 'rb') as f: |
| crl = x509.load_pem_x509_crl(f.read()) |
| with open(self.get_test_file_path('revoked-attestation-certificates/Chip-Test-PAA-FFF1-Cert.pem'), 'rb') as f: |
| crl_signer = x509.load_pem_x509_certificate(f.read()) |
| |
| ca_name_b64, ca_akid_hex = get_certificate_authority_details( |
| crl_signer, None, None, True) |
| revocation_set = generate_revocation_set_from_crl( |
| crl, crl_signer, ca_name_b64, ca_akid_hex, None) |
| |
| self.compare_revocation_sets(revocation_set, self.get_expected_revocation_set(0)) |
| |
| def test_pai_revocation_set(self): |
| """Test generation of PAI revocation set""" |
| with open(self.get_test_file_path('revoked-attestation-certificates/Matter-Development-PAI-FFF1-noPID-CRL.pem'), 'rb') as f: |
| crl = x509.load_pem_x509_crl(f.read()) |
| with open(self.get_test_file_path('revoked-attestation-certificates/Matter-Development-PAI-FFF1-noPID-Cert.pem'), 'rb') as f: |
| crl_signer = x509.load_pem_x509_certificate(f.read()) |
| with open(self.get_test_file_path('revoked-attestation-certificates/Chip-Test-PAA-FFF1-Cert.pem'), 'rb') as f: |
| paa = x509.load_pem_x509_certificate(f.read()) |
| |
| ca_name_b64, ca_akid_hex = get_certificate_authority_details( |
| crl_signer, None, paa, False) |
| revocation_set = generate_revocation_set_from_crl( |
| crl, crl_signer, ca_name_b64, ca_akid_hex, None) |
| |
| self.compare_revocation_sets(revocation_set, self.get_expected_revocation_set(1)) |
| |
| def test_revoked_pai_revocation_set(self): |
| """Test generation of revocation set of revoked PAI""" |
| with open(self.get_test_file_path('revoked-attestation-certificates/Chip-Test-PAI-FFF1-noPID-Revoked-CRL.pem'), 'rb') as f: |
| crl = x509.load_pem_x509_crl(f.read()) |
| with open(self.get_test_file_path('revoked-attestation-certificates/Chip-Test-PAI-FFF1-noPID-Revoked-Cert.pem'), 'rb') as f: |
| crl_signer = x509.load_pem_x509_certificate(f.read()) |
| with open(self.get_test_file_path('revoked-attestation-certificates/Chip-Test-PAA-FFF1-Cert.pem'), 'rb') as f: |
| paa = x509.load_pem_x509_certificate(f.read()) |
| |
| ca_name_b64, ca_akid_hex = get_certificate_authority_details( |
| crl_signer, None, paa, False) |
| revocation_set = generate_revocation_set_from_crl( |
| crl, crl_signer, ca_name_b64, ca_akid_hex, None) |
| |
| self.compare_revocation_sets(revocation_set, self.get_expected_revocation_set(2)) |
| |
| |
| if __name__ == "__main__": |
| if len(sys.argv) > 1 and sys.argv[1] == 'test': |
| # Remove the 'test' argument and run tests |
| sys.argv.pop(1) |
| unittest.main() |
| elif len(sys.argv) == 1: |
| cli.main(['--help']) |
| else: |
| cli() |
