|
18 | 18 |
|
19 | 19 | from matter_server.server.const import PAA_ROOT_CERTS_DIR
|
20 | 20 |
|
| 21 | +# Git repo details |
| 22 | +OWNER = "project-chip" |
| 23 | +REPO = "connectedhomeip" |
| 24 | +PATH = "credentials/development/paa-root-certs" |
| 25 | + |
21 | 26 | LOGGER = logging.getLogger(__name__)
|
22 | 27 | PRODUCTION_URL = "https://on.dcl.csa-iot.org"
|
23 | 28 | TEST_URL = "https://on.test-net.dcl.csa-iot.org"
|
24 |
| -GIT_URL = "https://github.com/project-chip/connectedhomeip/raw/master/credentials/development/paa-root-certs" # pylint: disable=line-too-long |
25 |
| -GIT_CERTS = [ |
26 |
| - "Chip-Test-PAA-FFF1-Cert", |
27 |
| - "Chip-Test-PAA-NoVID-Cert", |
28 |
| -] |
| 29 | +GIT_URL = f"https://raw.githubusercontent.com/{OWNER}/{REPO}/master/{PATH}" |
| 30 | + |
| 31 | + |
29 | 32 | LAST_CERT_IDS: set[str] = set()
|
30 | 33 |
|
31 | 34 |
|
@@ -66,13 +69,13 @@ async def fetch_dcl_certificates(
|
66 | 69 | base_urls = set()
|
67 | 70 | # determine which url's need to be queried.
|
68 | 71 | # if we're going to fetch both prod and test, do test first
|
69 |
| - # so any duplicates will be overwritten/preferred by the production version |
| 72 | + # so any duplicates will be overwritten/preferred by the production version. |
| 73 | + |
70 | 74 | # NOTE: While Matter is in BETA we fetch the test certificates by default
|
71 | 75 | if fetch_test_certificates:
|
72 | 76 | base_urls.add(TEST_URL)
|
73 | 77 | if fetch_production_certificates:
|
74 | 78 | base_urls.add(PRODUCTION_URL)
|
75 |
| - |
76 | 79 | try:
|
77 | 80 | async with ClientSession(raise_for_status=True) as http_session:
|
78 | 81 | for url_base in base_urls:
|
@@ -113,16 +116,27 @@ async def fetch_dcl_certificates(
|
113 | 116 | return fetch_count
|
114 | 117 |
|
115 | 118 |
|
| 119 | +# Manufacturers release test certificates through the SDK (Git) as a part |
| 120 | +# of their standard product release workflow. This will ensure those certs |
| 121 | +# are correctly captured |
| 122 | + |
| 123 | + |
116 | 124 | async def fetch_git_certificates() -> int:
|
117 | 125 | """Fetch Git PAA Certificates."""
|
118 | 126 | fetch_count = 0
|
119 | 127 | LOGGER.info("Fetching the latest PAA root certificates from Git.")
|
| 128 | + |
120 | 129 | try:
|
121 | 130 | async with ClientSession(raise_for_status=True) as http_session:
|
122 |
| - for cert in GIT_CERTS: |
| 131 | + # Fetch directory contents and filter out extension |
| 132 | + api_url = f"https://api.github.com/repos/{OWNER}/{REPO}/contents/{PATH}" |
| 133 | + async with http_session.get(api_url, timeout=20) as response: |
| 134 | + contents = await response.json() |
| 135 | + git_certs = {item["name"].split(".")[0] for item in contents} |
| 136 | + # Fetch certificates |
| 137 | + for cert in git_certs: |
123 | 138 | if cert in LAST_CERT_IDS:
|
124 | 139 | continue
|
125 |
| - |
126 | 140 | async with http_session.get(f"{GIT_URL}/{cert}.pem") as response:
|
127 | 141 | certificate = await response.text()
|
128 | 142 | await write_paa_root_cert(certificate, cert)
|
|
0 commit comments