Automatically retrieve PAA certificates for device attestation (#240)

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>

Author: marcelveldt

matter_server/server/device_controller.py

@@ -7,13 +7,16 @@
 from datetime import datetime
 from functools import partial
 import logging
-from typing import TYPE_CHECKING, Any, Callable, Deque, Type, TypeVar, cast
+import pathlib
+from typing import TYPE_CHECKING, Any, Callable, Deque, Final, Type, TypeVar, cast
 
 from chip.ChipDeviceCtrl import CommissionableNode
 from chip.clusters import Attribute, Objects as Clusters
 from chip.clusters.ClusterObjects import ALL_CLUSTERS, Cluster
 from chip.exceptions import ChipStackError
 
+from matter_server.server.helpers.paa_certificates import fetch_certificates
+
 from ..common.const import SCHEMA_VERSION
 from ..common.errors import (
     NodeCommissionFailed,
@@ -31,6 +34,7 @@
 
 if TYPE_CHECKING:
     from .server import MatterServer
+    from chip.ChipDeviceCtrl import ChipDeviceController
 
 _T = TypeVar("_T")
 
@@ -39,18 +43,34 @@
 
 LOGGER = logging.getLogger(__name__)
 
+# the paa-root-certs path is hardcoded in the sdk at this time
+# and always uses the development subfolder
+# regardless of anything you pass into instantiating the controller
+# revisit this once matter 1.1 is released
+PAA_ROOT_CERTS_DIR: Final[pathlib.Path] = (
+    pathlib.Path(__file__)
+    .parent.resolve()
+    .parent.resolve()
+    .parent.resolve()
+    .joinpath("credentials/development/paa-root-certs")
+)
+
 
 class MatterDeviceController:
     """Class that manages the Matter devices."""
 
+    chip_controller: ChipDeviceController | None
+
     def __init__(
         self,
         server: MatterServer,
     ):
         """Initialize the device controller."""
         self.server = server
         # Instantiate the underlying ChipDeviceController instance on the Fabric
-        self.chip_controller = server.stack.fabric_admin.NewController()
+        if not PAA_ROOT_CERTS_DIR.is_dir():
+            raise RuntimeError("PAA certificates directory not found")
+
         # we keep the last events in memory so we can include them in the diagnostics dump
         self.event_history: Deque[Attribute.EventReadResult] = deque(maxlen=25)
         self._subscriptions: dict[int, Attribute.SubscriptionTransaction] = {}
@@ -60,13 +80,14 @@ def __init__(
         self.compressed_fabric_id: int | None = None
         self._interview_task: asyncio.Task | None = None
 
-    @property
-    def fabric_id(self) -> int:
-        """Return Fabric ID."""
-        return cast(int, self.chip_controller.fabricId)
-
     async def initialize(self) -> None:
         """Async initialize of controller."""
+        # (re)fetch all PAA certificates once at startup
+        # NOTE: this must be done before initializing the controller
+        await fetch_certificates(PAA_ROOT_CERTS_DIR)
+        self.chip_controller = self.server.stack.fabric_admin.NewController(
+            paaTrustStorePath=str(PAA_ROOT_CERTS_DIR)
+        )
         self.compressed_fabric_id = await self._call_sdk(
             self.chip_controller.GetCompressedFabricId
         )
@@ -126,6 +147,9 @@ async def commission_with_code(self, code: str) -> MatterNodeData:
 
         Returns full NodeInfo once complete.
         """
+        # perform a quick delta sync of certificates to make sure
+        # we have the latest paa root certs
+        await fetch_certificates(PAA_ROOT_CERTS_DIR)
         node_id = self._get_next_node_id()
 
         success = await self._call_sdk(
@@ -160,6 +184,12 @@ async def commission_on_network(
         a string or None depending on the actual type of selected filter.
         Returns full NodeInfo once complete.
         """
+        # perform a quick delta sync of certificates to make sure
+        # we have the latest paa root certs
+        # NOTE: Its not very clear if the newly fetched certificates can be used without
+        # restarting the device controller
+        await fetch_certificates(PAA_ROOT_CERTS_DIR)
+
         node_id = self._get_next_node_id()
 
         success = await self._call_sdk(

matter_server/server/helpers/__init__.py

@@ -0,0 +1 @@
+"""Helpers/utils for the Matter Server."""

matter_server/server/helpers/paa_certificates.py

@@ -0,0 +1,107 @@
+"""
+Utils to fetch CHIP Development Product Attestation Authority (PAA) certificates from DCL.
+
+This is based on the original script from project-chip here:
+https://github.com/project-chip/connectedhomeip/edit/master/credentials/fetch-paa-certs-from-dcl.py
+
+All rights reserved.
+"""
+
+import asyncio
+import logging
+import pathlib
+import re
+
+from aiohttp import ClientSession, ClientError
+from cryptography import x509
+from cryptography.hazmat.primitives import serialization
+
+LOGGER = logging.getLogger(__name__)
+PRODUCTION_URL = "https://on.dcl.csa-iot.org"
+TEST_URL = "https://on.test-net.dcl.csa-iot.org"
+
+LAST_CERT_IDS: set[str] = set()
+
+
+async def write_paa_root_cert(
+    certificate: str, subject: str, root_path: pathlib.Path
+) -> None:
+    """Write certificate from string to file."""
+
+    def _write() -> None:
+        filename_base = "dcld_mirror_" + re.sub(
+            "[^a-zA-Z0-9_-]", "", re.sub("[=, ]", "_", subject)
+        )
+        filepath_base = root_path.joinpath(filename_base)
+        # handle PEM certificate file
+        file_path_pem = f"{filepath_base}.pem"
+        LOGGER.debug("Writing certificate %s", file_path_pem)
+        with open(file_path_pem, "w+", encoding="utf-8") as outfile:
+            outfile.write(certificate)
+        # handle DER certificate file (converted from PEM)
+        pem_certificate = x509.load_pem_x509_certificate(certificate.encode())
+        file_path_der = f"{filepath_base}.der"
+        LOGGER.debug("Writing certificate %s", file_path_der)
+        with open(file_path_der, "wb+") as outfile:
+            der_certificate = pem_certificate.public_bytes(serialization.Encoding.DER)
+            outfile.write(der_certificate)
+
+    return await asyncio.get_running_loop().run_in_executor(None, _write)
+
+
+async def fetch_certificates(
+    paa_trust_store_path: pathlib.Path,
+    fetch_test_certificates: bool = True,
+    fetch_production_certificates: bool = True,
+) -> int:
+    """Fetch PAA Certificates."""
+    LOGGER.info("Fetching the latest PAA root certificates from DCL.")
+    fetch_count: int = 0
+    base_urls = set()
+    # determine which url's need to be queried.
+    # if we're going to fetch both prod and test, do test first
+    # so any duplicates will be overwritten/preferred by the production version
+    # NOTE: While Matter is in BETA we fetch the test certificates by default
+    if fetch_test_certificates:
+        base_urls.add(TEST_URL)
+    if fetch_production_certificates:
+        base_urls.add(PRODUCTION_URL)
+
+    try:
+        async with ClientSession(raise_for_status=True) as http_session:
+            for url_base in base_urls:
+                # fetch the paa certificates list
+                async with http_session.get(
+                    f"{url_base}/dcl/pki/root-certificates"
+                ) as response:
+                    result = await response.json()
+                paa_list = result["approvedRootCertificates"]["certs"]
+                # grab each certificate
+                for paa in paa_list:
+                    # do not fetch a certificate if we already fetched it
+                    if paa["subjectKeyId"] in LAST_CERT_IDS:
+                        continue
+                    async with http_session.get(
+                        f"{url_base}/dcl/pki/certificates/{paa['subject']}/{paa['subjectKeyId']}"
+                    ) as response:
+                        result = await response.json()
+
+                    certificate_data: dict = result["approvedCertificates"]["certs"][0]
+                    certificate: str = certificate_data["pemCert"]
+                    subject = certificate_data["subjectAsText"]
+                    certificate = certificate.rstrip("\n")
+
+                    await write_paa_root_cert(
+                        certificate,
+                        subject,
+                        paa_trust_store_path,
+                    )
+                    LAST_CERT_IDS.add(paa["subjectKeyId"])
+                    fetch_count += 1
+    except ClientError as err:
+        LOGGER.warning(
+            "Fetching latest certificates failed: error %s", err, exc_info=err
+        )
+    else:
+        LOGGER.info("Fetched %s PAA root certificates from DCL.", fetch_count)
+    return fetch_count

matter_server/server/server.py

@@ -137,7 +137,7 @@ def get_info(self) -> ServerInfoMessage:
         """Return (version)info of the Matter Server."""
         assert self.device_controller.compressed_fabric_id is not None
         return ServerInfoMessage(
-            fabric_id=self.device_controller.fabric_id,
+            fabric_id=self.fabric_id,
             compressed_fabric_id=self.device_controller.compressed_fabric_id,
             schema_version=SCHEMA_VERSION,
             min_supported_schema_version=MIN_SCHEMA_VERSION,

pyproject.toml

@@ -31,7 +31,8 @@ dependencies = [
 
 [project.optional-dependencies]
 server = [
-  "home-assistant-chip-core==2023.2.2"
+  "home-assistant-chip-core==2023.2.2",
+  "cryptography==39.0.1"
 ]
 test = [
   "black==23.1.0",