Helper: Load system certificates using rustls-native-certs This uses platform-specific methods: - Linux: openssl-probe to find certificate files - macOS: Keychain API - Windows: System certificate store (ROOT + CA stores)
(
&self,
store: &mut rustls::RootCertStore,
vm: &VirtualMachine,
)
| 1366 | /// - macOS: Keychain API |
| 1367 | /// - Windows: System certificate store (ROOT + CA stores) |
| 1368 | fn load_system_certificates( |
| 1369 | &self, |
| 1370 | store: &mut rustls::RootCertStore, |
| 1371 | vm: &VirtualMachine, |
| 1372 | ) -> PyResult<()> { |
| 1373 | #[cfg(windows)] |
| 1374 | { |
| 1375 | // Windows: Use schannel to load from both ROOT and CA stores |
| 1376 | use schannel::cert_store::CertStore; |
| 1377 | |
| 1378 | let store_names = ["ROOT", "CA"]; |
| 1379 | let open_fns = [CertStore::open_current_user, CertStore::open_local_machine]; |
| 1380 | |
| 1381 | for store_name in store_names { |
| 1382 | for open_fn in &open_fns { |
| 1383 | if let Ok(cert_store) = open_fn(store_name) { |
| 1384 | for cert_ctx in cert_store.certs() { |
| 1385 | let der_bytes = cert_ctx.to_der(); |
| 1386 | let cert = |
| 1387 | rustls::pki_types::CertificateDer::from(der_bytes.to_vec()); |
| 1388 | let is_ca = cert::is_ca_certificate(cert.as_ref()); |
| 1389 | if store.add(cert).is_ok() { |
| 1390 | *self.x509_cert_count.write() += 1; |
| 1391 | if is_ca { |
| 1392 | *self.ca_cert_count.write() += 1; |
| 1393 | } |
| 1394 | } |
| 1395 | } |
| 1396 | } |
| 1397 | } |
| 1398 | } |
| 1399 | |
| 1400 | if *self.x509_cert_count.read() == 0 { |
| 1401 | return Err(vm.new_os_error("Failed to load certificates from Windows store")); |
| 1402 | } |
| 1403 | |
| 1404 | Ok(()) |
| 1405 | } |
| 1406 | |
| 1407 | #[cfg(not(windows))] |
| 1408 | { |
| 1409 | let result = rustls_native_certs::load_native_certs(); |
| 1410 | |
| 1411 | // Load successfully found certificates |
| 1412 | for cert in result.certs { |
| 1413 | let is_ca = cert::is_ca_certificate(cert.as_ref()); |
| 1414 | if store.add(cert).is_ok() { |
| 1415 | *self.x509_cert_count.write() += 1; |
| 1416 | if is_ca { |
| 1417 | *self.ca_cert_count.write() += 1; |
| 1418 | } |
| 1419 | } |
| 1420 | } |
| 1421 | |
| 1422 | // If there were errors but some certs loaded, just continue |
| 1423 | // If NO certs loaded and there were errors, report the first error |
| 1424 | if *self.x509_cert_count.read() == 0 && !result.errors.is_empty() { |
| 1425 | return Err(vm.new_os_error(format!( |
no test coverage detected