Add flag for test-net DCL (#786)

Author: agners

matter_server/server/__main__.py

@@ -98,6 +98,12 @@
     default=None,
     help="Directory where PAA root certificates are stored.",
 )
+parser.add_argument(
+    "--enable-test-net-dcl",
+    type=bool,
+    default=False,
+    help="Enable PAA root certificates and other device information from test-net DCL.",
+)
 
 args = parser.parse_args()
 
@@ -181,6 +187,7 @@ def main() -> None:
         args.listen_address,
         args.primary_interface,
         args.paa_root_cert_dir,
+        args.enable_test_net_dcl,
     )
 
     async def handle_stop(loop: asyncio.AbstractEventLoop) -> None:

matter_server/server/helpers/paa_certificates.py

@@ -149,7 +149,9 @@ async def fetch_dcl_certificates(
 # are correctly captured
 
 
-async def fetch_git_certificates(paa_root_cert_dir: Path) -> int:
+async def fetch_git_certificates(
+    paa_root_cert_dir: Path, prefix: str | None = None
+) -> int:
     """Fetch Git PAA Certificates."""
     fetch_count = 0
     LOGGER.info("Fetching the latest PAA root certificates from Git.")
@@ -163,6 +165,8 @@ async def fetch_git_certificates(paa_root_cert_dir: Path) -> int:
                 git_certs = {item["name"].split(".")[0] for item in contents}
             # Fetch certificates
             for cert in git_certs:
+                if prefix and not cert.startswith(prefix):
+                    continue
                 async with http_session.get(f"{GIT_URL}/{cert}.pem") as response:
                     certificate = await response.text()
                 if await write_paa_root_cert(
@@ -238,6 +242,11 @@ def _check_paa_root_dir(
 
     if fetch_test_certificates:
         total_fetch_count += await fetch_git_certificates(paa_root_cert_dir)
+    else:
+        # Treat the Chip-Test certificates as production, we use them in our examples
+        total_fetch_count += await fetch_git_certificates(
+            paa_root_cert_dir, "Chip-Test"
+        )
 
     await loop.run_in_executor(None, paa_root_cert_dir_version.write_text, "1")
 

matter_server/server/server.py

@@ -108,6 +108,7 @@ def __init__(
         listen_addresses: list[str] | None = None,
         primary_interface: str | None = None,
         paa_root_cert_dir: Path | None = None,
+        enable_test_net_dcl: bool = False,
     ) -> None:
         """Initialize the Matter Server."""
         self.storage_path = storage_path
@@ -120,6 +121,7 @@ def __init__(
             self.paa_root_cert_dir = DEFAULT_PAA_ROOT_CERTS_DIR
         else:
             self.paa_root_cert_dir = Path(paa_root_cert_dir).absolute()
+        self.enable_test_net_dcl = enable_test_net_dcl
         self.logger = logging.getLogger(__name__)
         self.app = web.Application()
         self.loop: asyncio.AbstractEventLoop | None = None
@@ -156,7 +158,11 @@ async def start(self) -> None:
 
         # (re)fetch all PAA certificates once at startup
         # NOTE: this must be done before initializing the controller
-        await fetch_certificates(self.paa_root_cert_dir)
+        await fetch_certificates(
+            self.paa_root_cert_dir,
+            fetch_test_certificates=self.enable_test_net_dcl,
+            fetch_production_certificates=True,
+        )
 
         # Initialize our (intermediate) device controller which keeps track
         # of Matter devices and their subscriptions.