Fetch certificates only once a day (#611)

matter_server/server/device_controller.py

@@ -126,6 +126,7 @@ async def initialize(self) -> None:
         # (re)fetch all PAA certificates once at startup
         # NOTE: this must be done before initializing the controller
         await fetch_certificates()
+
         # Instantiate the underlying ChipDeviceController instance on the Fabric
         self.chip_controller = self.server.stack.fabric_admin.NewController(
             paaTrustStorePath=str(PAA_ROOT_CERTS_DIR)

matter_server/server/helpers/paa_certificates.py

@@ -8,6 +8,7 @@
 """
 
 import asyncio
+from datetime import UTC, datetime, timedelta
 import logging
 from os import makedirs
 import re
@@ -62,9 +63,6 @@ async def fetch_dcl_certificates(
 ) -> int:
     """Fetch DCL PAA Certificates."""
     LOGGER.info("Fetching the latest PAA root certificates from DCL.")
-    if not PAA_ROOT_CERTS_DIR.is_dir():
-        loop = asyncio.get_running_loop()
-        await loop.run_in_executor(None, makedirs, PAA_ROOT_CERTS_DIR)
     fetch_count: int = 0
     base_urls = set()
     # determine which url's need to be queried.
@@ -152,11 +150,30 @@ async def fetch_git_certificates() -> int:
     return fetch_count
 
 
+async def _get_certificate_age() -> datetime:
+    """Get last time PAA Certificates have been fetched."""
+    loop = asyncio.get_running_loop()
+    stat = await loop.run_in_executor(None, PAA_ROOT_CERTS_DIR.stat)
+    return datetime.fromtimestamp(stat.st_mtime, tz=UTC)
+
+
 async def fetch_certificates(
     fetch_test_certificates: bool = True,
     fetch_production_certificates: bool = True,
 ) -> int:
     """Fetch PAA Certificates."""
+    loop = asyncio.get_running_loop()
+
+    if not PAA_ROOT_CERTS_DIR.is_dir():
+        await loop.run_in_executor(None, makedirs, PAA_ROOT_CERTS_DIR)
+    else:
+        stat = await loop.run_in_executor(None, PAA_ROOT_CERTS_DIR.stat)
+        last_fetch = datetime.fromtimestamp(stat.st_mtime, tz=UTC)
+        if last_fetch > datetime.now(tz=UTC) - timedelta(days=1):
+            LOGGER.info(
+                "Skip fetching certificates (already fetched within the last 24h)."
+            )
+            return 0
 
     fetch_count = await fetch_dcl_certificates(
         fetch_test_certificates=fetch_test_certificates,
@@ -166,4 +183,6 @@ async def fetch_certificates(
     if fetch_test_certificates:
         fetch_count += await fetch_git_certificates()
 
+    await loop.run_in_executor(None, PAA_ROOT_CERTS_DIR.touch)
+
     return fetch_count