diff --git a/dstack/Cargo.lock b/dstack/Cargo.lock index e4156c67a..f800d2331 100644 --- a/dstack/Cargo.lock +++ b/dstack/Cargo.lock @@ -4757,7 +4757,7 @@ dependencies = [ "nsm-attest", "p384", "pem", - "reqwest", + "pki-fetch", "rustls-pki-types", "serde", "sha2 0.10.9", @@ -5412,6 +5412,18 @@ dependencies = [ "spki", ] +[[package]] +name = "pki-fetch" +version = "0.6.0" +dependencies = [ + "anyhow", + "pem", + "reqwest", + "tokio", + "tracing", + "x509-parser 0.16.0", +] + [[package]] name = "poly1305" version = "0.8.0" @@ -8104,13 +8116,12 @@ dependencies = [ "nom", "p256", "pem", - "reqwest", + "pki-fetch", "rsa", "rustls-pki-types", "serde", "serde_json", "sha2 0.10.9", - "tokio", "tpm-types", "tracing", "x509-parser 0.16.0", diff --git a/dstack/Cargo.toml b/dstack/Cargo.toml index 61cc8e56f..7d7cbaf71 100644 --- a/dstack/Cargo.toml +++ b/dstack/Cargo.toml @@ -73,6 +73,7 @@ members = [ "crates/build-info", "crates/mock-attestation", "crates/qemu-acpi", + "crates/pki-fetch", ] # Vendored third-party crates are path dependencies but deliberately not members: # `--all-features` applies to members, and ktls declares its `ring` and @@ -95,6 +96,7 @@ dstack-volume = { path = "crates/dstack-volume" } dstack-api-auth = { path = "crates/api-auth" } dstack-build-info = { path = "crates/build-info" } qemu-acpi = { path = "crates/qemu-acpi" } +pki-fetch = { path = "crates/pki-fetch" } cc-eventlog = { path = "cc-eventlog" } supervisor = { path = "supervisor" } supervisor-client = { path = "supervisor/client" } diff --git a/dstack/crates/mock-attestation/src/server.rs b/dstack/crates/mock-attestation/src/server.rs index 35d0216a0..c251f8c38 100644 --- a/dstack/crates/mock-attestation/src/server.rs +++ b/dstack/crates/mock-attestation/src/server.rs @@ -369,18 +369,7 @@ mod tests { let task = tokio::spawn(serve_listener(listener, state.clone())); let quote = state.tpm.attest(&[0x42; 32]).unwrap(); let root = state.tpm.root_ca_pem(); - let q = quote.clone(); - let collateral = tokio::task::spawn_blocking(move || { - let runtime = tokio::runtime::Builder::new_current_thread() - .enable_all() - .build() - .unwrap(); - runtime - .block_on(tpm_qvl::get_collateral(&q, &root)) - .unwrap() - }) - .await - .unwrap(); + let collateral = tpm_qvl::get_collateral("e, &root).await.unwrap(); tpm_qvl::QuoteVerifier::new(state.tpm.root_ca_pem()) .verify("e, &collateral) .unwrap(); diff --git a/dstack/crates/pki-fetch/Cargo.toml b/dstack/crates/pki-fetch/Cargo.toml new file mode 100644 index 000000000..b7d47b386 --- /dev/null +++ b/dstack/crates/pki-fetch/Cargo.toml @@ -0,0 +1,21 @@ +# SPDX-FileCopyrightText: © 2026 Phala Network +# +# SPDX-License-Identifier: Apache-2.0 + +[package] +name = "pki-fetch" +version.workspace = true +authors.workspace = true +edition.workspace = true +license.workspace = true +description = "Bounded download of issuer certificates and CRLs named by unverified certificates" + +[dependencies] +anyhow.workspace = true +pem.workspace = true +reqwest.workspace = true +tracing.workspace = true +x509-parser.workspace = true + +[dev-dependencies] +tokio = { workspace = true, features = ["macros", "net", "rt", "time"] } diff --git a/dstack/crates/pki-fetch/src/lib.rs b/dstack/crates/pki-fetch/src/lib.rs new file mode 100644 index 000000000..e5cc2b564 --- /dev/null +++ b/dstack/crates/pki-fetch/src/lib.rs @@ -0,0 +1,218 @@ +// SPDX-FileCopyrightText: © 2026 Phala Network +// +// SPDX-License-Identifier: Apache-2.0 + +//! Bounded download of PKI collateral (issuer certificates and CRLs). +//! +//! Collateral URLs come from certificates that have not been verified yet, so +//! whoever supplies the certificate chooses them. The endpoints are untrusted +//! transport: everything they serve still has to chain to a pinned root CA. +//! What they must not decide is how much work one verification does, so a +//! [`Fetcher`] spends a fixed budget of time, requests and bytes. + +use std::time::{Duration, Instant}; + +use anyhow::{bail, ensure, Context, Result}; +use tracing::{debug, warn}; +use x509_parser::{ + extensions::{DistributionPointName, GeneralName, ParsedExtension}, + prelude::*, +}; + +const TOTAL_TIMEOUT: Duration = Duration::from_secs(60); +/// Lets a hanging endpoint fall through to the next one within the budget. +const REQUEST_TIMEOUT: Duration = Duration::from_secs(30); +const MAX_REQUESTS: usize = 16; +const MAX_BYTES: usize = 4 * 1024 * 1024; + +/// Downloads collateral for one verification within a fixed budget. +pub struct Fetcher { + client: reqwest::Client, + deadline: Instant, + requests_left: usize, + bytes_left: usize, +} + +impl Fetcher { + pub fn new() -> Result { + Ok(Self { + client: reqwest::Client::builder() + .build() + .context("failed to build HTTP client")?, + deadline: Instant::now() + TOTAL_TIMEOUT, + requests_left: MAX_REQUESTS, + bytes_left: MAX_BYTES, + }) + } + + /// GET `url`, charging the request, its time and its body to the budget. + pub async fn download(&mut self, url: &str) -> Result> { + ensure!( + self.requests_left > 0, + "collateral fetch exceeded {MAX_REQUESTS} requests" + ); + self.requests_left -= 1; + let remaining = self.deadline.saturating_duration_since(Instant::now()); + ensure!( + !remaining.is_zero(), + "collateral fetch exceeded {TOTAL_TIMEOUT:?}" + ); + debug!("downloading {url}"); + let mut response = self + .client + .get(url) + .timeout(remaining.min(REQUEST_TIMEOUT)) + .send() + .await + .and_then(|r| r.error_for_status()) + .with_context(|| format!("failed to download {url}"))?; + let mut body = Vec::new(); + while let Some(chunk) = response + .chunk() + .await + .with_context(|| format!("failed to read response body from {url}"))? + { + ensure!( + chunk.len() <= self.bytes_left, + "collateral fetch exceeded {MAX_BYTES} bytes" + ); + self.bytes_left -= chunk.len(); + body.extend_from_slice(&chunk); + } + Ok(body) + } + + /// Download the CRLs of every certificate in `certs` that names one. + pub async fn crls(&mut self, certs: &[Vec]) -> Result>> { + let mut crls = Vec::new(); + for cert in certs { + crls.extend(self.crl(cert).await?); + } + Ok(crls) + } + + /// Download the CRL of the single root CA in `root_ca_pem`, if it names one. + pub async fn root_ca_crl(&mut self, root_ca_pem: &str) -> Result>> { + let roots = ::pem::parse_many(root_ca_pem).context("failed to parse root CA PEM")?; + let [root] = roots.as_slice() else { + bail!("expected 1 root CA, found {}", roots.len()); + }; + self.crl(root.contents()).await + } + + /// Download the CRL from the first reachable distribution point of `cert`. + async fn crl(&mut self, cert: &[u8]) -> Result>> { + let urls = crl_urls(cert)?; + if urls.is_empty() { + return Ok(None); + } + for url in &urls { + match self.download(url).await { + Ok(crl) => return Ok(Some(crl)), + Err(e) => warn!("failed to download CRL: {e:#}"), + } + } + bail!("no CRL distribution point was reachable: {urls:?}") + } +} + +fn crl_urls(cert_der: &[u8]) -> Result> { + let (_, cert) = X509Certificate::from_der(cert_der).context("failed to parse certificate")?; + let mut urls = Vec::new(); + for ext in cert.extensions() { + let ParsedExtension::CRLDistributionPoints(points) = ext.parsed_extension() else { + continue; + }; + for point in &points.points { + let Some(DistributionPointName::FullName(names)) = &point.distribution_point else { + continue; + }; + for name in names { + if let GeneralName::URI(uri) = name { + urls.push(uri.to_string()); + } + } + } + } + Ok(urls) +} + +#[cfg(test)] +mod tests { + use super::*; + use tokio::{ + io::{AsyncReadExt, AsyncWriteExt}, + net::TcpListener, + }; + + #[derive(Clone, Copy)] + enum Reply { + NotFound, + Endless, + Drip, + } + + async fn serve(reply: Reply) -> String { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("http://{}/", listener.local_addr().unwrap()); + tokio::spawn(async move { + while let Ok((mut stream, _)) = listener.accept().await { + tokio::spawn(async move { + let _ = stream.read(&mut [0; 1024]).await; + let _ = match reply { + Reply::NotFound => { + stream + .write_all(b"HTTP/1.1 404 Not Found\r\ncontent-length: 0\r\n\r\n") + .await + } + Reply::Endless => { + let _ = stream.write_all(b"HTTP/1.1 200 OK\r\n\r\n").await; + while stream.write_all(&[0; 64 * 1024]).await.is_ok() {} + Ok(()) + } + Reply::Drip => { + let _ = stream + .write_all(b"HTTP/1.1 200 OK\r\ncontent-length: 1000\r\n\r\n") + .await; + while stream.write_all(b"x").await.is_ok() { + tokio::time::sleep(Duration::from_millis(100)).await; + } + Ok(()) + } + }; + }); + } + }); + url + } + + #[tokio::test] + async fn request_count_is_bounded() { + let url = serve(Reply::NotFound).await; + let mut fetcher = Fetcher::new().unwrap(); + for _ in 0..MAX_REQUESTS { + fetcher.download(&url).await.unwrap_err(); + } + let err = fetcher.download(&url).await.unwrap_err(); + assert!(format!("{err:#}").contains("requests"), "{err:#}"); + } + + #[tokio::test] + async fn downloaded_bytes_are_bounded() { + let url = serve(Reply::Endless).await; + let err = Fetcher::new().unwrap().download(&url).await.unwrap_err(); + assert!(format!("{err:#}").contains("bytes"), "{err:#}"); + } + + #[tokio::test] + async fn total_time_is_bounded() { + let url = serve(Reply::Drip).await; + let mut fetcher = Fetcher { + deadline: Instant::now() + Duration::from_secs(1), + ..Fetcher::new().unwrap() + }; + let started = Instant::now(); + fetcher.download(&url).await.unwrap_err(); + assert!(started.elapsed() < Duration::from_secs(2)); + } +} diff --git a/dstack/nsm-qvl/Cargo.toml b/dstack/nsm-qvl/Cargo.toml index 7ef014221..13461c862 100644 --- a/dstack/nsm-qvl/Cargo.toml +++ b/dstack/nsm-qvl/Cargo.toml @@ -30,7 +30,7 @@ rustls-pki-types.workspace = true dcap-qvl-webpki = { workspace = true, features = ["alloc", "rustcrypto"] } # CRL download -reqwest = { workspace = true, features = ["rustls"] } +pki-fetch.workspace = true [dev-dependencies] nsm-attest.workspace = true diff --git a/dstack/nsm-qvl/src/collateral.rs b/dstack/nsm-qvl/src/collateral.rs index a12d73563..c41e50589 100644 --- a/dstack/nsm-qvl/src/collateral.rs +++ b/dstack/nsm-qvl/src/collateral.rs @@ -7,9 +7,9 @@ //! Extracts CRL distribution points from the device-provided cert chain and //! downloads CRLs for revocation checking, similar to dcap-qvl/tpm-qvl. -use anyhow::{bail, Context, Result}; -use tracing::{debug, warn}; -use x509_parser::{extensions::DistributionPointName, prelude::*}; +use anyhow::{Context, Result}; +use pki_fetch::Fetcher; +use tracing::debug; use crate::{ verify::verify_attestation_with_collateral, AttestationDocument, CoseSign1, NsmCollateral, @@ -31,17 +31,9 @@ pub async fn get_collateral(cose_sign1_bytes: &[u8], root_ca_pem: &str) -> Resul let doc = AttestationDocument::from_cbor(&cose.payload).context("failed to parse attestation doc")?; - let certs = build_chain_from_doc(&doc); - let crls = download_crls_for_certs(&certs).await?; - - let root_ca_crl = { - let root_ca_der = - extract_certs_webpki(root_ca_pem.as_bytes()).context("failed to parse root CA PEM")?; - if root_ca_der.len() != 1 { - bail!("expected 1 root CA, found {}", root_ca_der.len()); - } - download_crl_for_cert(&root_ca_der[0]).await? - }; + let mut fetcher = Fetcher::new()?; + let crls = fetcher.crls(&build_chain_from_doc(&doc)).await?; + let root_ca_crl = fetcher.root_ca_crl(root_ca_pem).await?; debug!( "✓ collateral fetched: {} CRL(s), root CA CRL: {}", @@ -58,104 +50,3 @@ fn build_chain_from_doc(doc: &AttestationDocument) -> Vec> { chain.extend(doc.cabundle.iter().skip(1).cloned()); chain } - -async fn download_crls_for_certs(certs: &[Vec]) -> Result>> { - debug!("downloading CRLs from device-provided cert chain..."); - - let mut crls = Vec::new(); - - for cert_der in certs { - let Some(crl) = download_crl_for_cert(cert_der) - .await - .context("failed to download CRL")? - else { - continue; - }; - crls.push(crl); - } - Ok(crls) -} - -async fn download_crl_for_cert(cert: &[u8]) -> Result>> { - let crl_urls = extract_crl_urls(cert)?; - if crl_urls.is_empty() { - debug!("no CRL Distribution Points found in certificate"); - return Ok(None); - } - - download_first_available_crl(&crl_urls).await.map(Some) -} - -async fn download_first_available_crl(urls: &[String]) -> Result> { - for url in urls { - debug!("downloading CRL from {url}"); - match download_crl(url).await { - Ok(crl) => return Ok(crl), - Err(e) => { - warn!("✗ failed to download CRL from {url}: {e:?}"); - continue; - } - } - } - bail!("failed to download CRL") -} - -fn extract_certs_webpki(cert_pem: &[u8]) -> Result>> { - let pem_items = ::pem::parse_many(cert_pem).context("failed to parse PEM")?; - let certs = pem_items - .into_iter() - .map(|pem| rustls_pki_types::CertificateDer::from(pem.into_contents())) - .collect(); - Ok(certs) -} - -async fn download_crl(url: &str) -> Result> { - debug!("downloading CRL from {url}"); - - let response = reqwest::get(url) - .await - .context(format!("failed to download CRL from {url}"))?; - - if !response.status().is_success() { - bail!("CRL download failed with status: {}", response.status()); - } - - let crl_bytes = response - .bytes() - .await - .context("failed to read CRL response body")? - .to_vec(); - - debug!("downloaded {} bytes CRL from {}", crl_bytes.len(), url); - - Ok(crl_bytes) -} - -fn extract_crl_urls(cert_der: &[u8]) -> Result> { - let (_, cert) = X509Certificate::from_der(cert_der).context("failed to parse certificate")?; - let mut crl_urls = Vec::new(); - - for ext in cert.extensions() { - let ParsedExtension::CRLDistributionPoints(crl_dist_points) = ext.parsed_extension() else { - continue; - }; - for dist_point in crl_dist_points.points.iter() { - let Some(dist_point_name) = &dist_point.distribution_point else { - continue; - }; - - let DistributionPointName::FullName(names) = dist_point_name else { - continue; - }; - for name in names.iter() { - let x509_parser::extensions::GeneralName::URI(uri) = name else { - continue; - }; - crl_urls.push(uri.to_string()); - debug!("found CRL URL: {uri}"); - } - } - } - - Ok(crl_urls) -} diff --git a/dstack/tpm-qvl/Cargo.toml b/dstack/tpm-qvl/Cargo.toml index 4276e7bb1..94f56066b 100644 --- a/dstack/tpm-qvl/Cargo.toml +++ b/dstack/tpm-qvl/Cargo.toml @@ -34,9 +34,8 @@ pem.workspace = true tpm-types.workspace = true # CRL download (optional) -reqwest = { workspace = true, features = ["blocking"], optional = true } -tokio = { workspace = true, features = ["rt"], optional = true } +pki-fetch = { workspace = true, optional = true } [features] default = ["crl-download"] -crl-download = ["reqwest", "tokio"] +crl-download = ["pki-fetch"] diff --git a/dstack/tpm-qvl/src/collateral.rs b/dstack/tpm-qvl/src/collateral.rs index 615f6dcbd..d5f229725 100644 --- a/dstack/tpm-qvl/src/collateral.rs +++ b/dstack/tpm-qvl/src/collateral.rs @@ -7,9 +7,10 @@ //! This module implements the first step of dcap-qvl architecture: //! extracting certificate chain information and downloading CRLs. -use anyhow::{bail, Context, Result}; -use tracing::{debug, warn}; -use x509_parser::{extensions::DistributionPointName, prelude::*}; +use anyhow::{Context, Result}; +use pki_fetch::Fetcher; +use tracing::debug; +use x509_parser::prelude::*; use tpm_types::TpmQuote; @@ -22,50 +23,25 @@ pub async fn get_collateral_and_verify(quote: &TpmQuote) -> Result Result { - // Collateral fetching uses synchronous (blocking) HTTP. Run it on the - // blocking pool so it never stalls (or panics on) the async runtime worker. - let ak_cert = quote.ak_cert.clone(); - let root_ca = root_ca_pem.to_string(); - tokio::task::spawn_blocking(move || get_collateral_blocking(&ak_cert, &root_ca)) - .await - .context("collateral fetch task panicked")? -} - -fn get_collateral_blocking(ak_cert_der: &[u8], root_ca_pem: &str) -> Result { debug!("fetching quote collateral (intermediate cert chain + CRLs)"); - - debug!("AK certificate (leaf) found: {} bytes", ak_cert_der.len()); - - // Build certificate chain from device (via AIA) - let chain_ders = build_cert_chain(ak_cert_der)?; - // Download CRLs from device-provided cert chain - let crls = download_crls_for_certs(&chain_ders)?; - - // Download CRL from verifier-provided root CA - let root_ca_crl = { - let root_ca_der = - extract_certs_webpki(root_ca_pem.as_bytes()).context("failed to parse root CA PEM")?; - if root_ca_der.len() != 1 { - bail!("expected 1 root CA, found {}", root_ca_der.len()); - } - download_crl_for_cert(&root_ca_der[0])? - }; - + let mut fetcher = Fetcher::new()?; + let chain_ders = build_cert_chain(&mut fetcher, "e.ak_cert).await?; + let crls = fetcher.crls(&chain_ders).await?; + let root_ca_crl = fetcher.root_ca_crl(root_ca_pem).await?; debug!( "✓ collateral fetched: {} intermediate CRL(s), root CA CRL: {}", crls.len(), if root_ca_crl.is_some() { "yes" } else { "no" } ); - let cert_chain_pem = ders_to_pem(&chain_ders)?; Ok(QuoteCollateral { - cert_chain_pem, + cert_chain_pem: ders_to_pem(&chain_ders)?, crls, root_ca_crl, }) } /// Build certificate chain by following AIA links (stops before root) -fn build_cert_chain(leaf_cert_der: &[u8]) -> Result>> { +async fn build_cert_chain(fetcher: &mut Fetcher, leaf_cert_der: &[u8]) -> Result>> { let mut chain_ders = Vec::new(); chain_ders.push(leaf_cert_der.to_vec()); let mut current_cert_der = leaf_cert_der.to_vec(); @@ -76,7 +52,7 @@ fn build_cert_chain(leaf_cert_der: &[u8]) -> Result>> { break; }; debug!("downloading parent cert from: {url}"); - let parent_der = download_cert(&url)?; + let parent_der = fetcher.download(&url).await?; // Stop if we hit a self-signed cert (root CA) if is_self_signed(&parent_der)? { debug!("found self-signed cert - stopping (root CA should be provided by verifier)"); @@ -90,49 +66,6 @@ fn build_cert_chain(leaf_cert_der: &[u8]) -> Result>> { Ok(chain_ders) } -/// Download CRLs for given certificates -fn download_crls_for_certs(certs: &[Vec]) -> Result>> { - debug!("downloading CRLs from device-provided cert chain..."); - - let mut crls = Vec::new(); - - for cert_der in certs { - let Some(crl) = download_crl_for_cert(cert_der).context("failed to download CRL")? else { - continue; - }; - crls.push(crl); - } - Ok(crls) -} - -/// Download CRL for verifier-provided root CA -fn download_crl_for_cert(cert: &[u8]) -> Result>> { - let crl_urls = extract_crl_urls(cert)?; - if crl_urls.is_empty() { - debug!("verifier root CA has no CRL DP - will skip root CA CRL check"); - return Ok(None); - } - - download_first_available_crl(&crl_urls).map(Some) -} - -/// Download first available CRL from a list of URLs -fn download_first_available_crl(urls: &[String]) -> Result> { - for url in urls { - debug!("downloading CRL from {url}"); - match download_crl(url) { - Ok(crl) => { - return Ok(crl); - } - Err(e) => { - warn!("✗ failed to download CRL from {url}: {e:?}"); - continue; - } - } - } - bail!("failed to download CRL") -} - /// Convert DER certificates to PEM format fn ders_to_pem(ders: &[Vec]) -> Result { let mut pem = String::new(); @@ -148,75 +81,6 @@ fn is_self_signed(cert_der: &[u8]) -> Result { Ok(cert.subject() == cert.issuer()) } -fn extract_certs_webpki(cert_pem: &[u8]) -> Result>> { - use ::pem::parse_many; - - let pem_items = parse_many(cert_pem).context("failed to parse PEM")?; - - let certs = pem_items - .into_iter() - .map(|pem| rustls_pki_types::CertificateDer::from(pem.into_contents())) - .collect(); - - Ok(certs) -} - -fn download_crl(url: &str) -> Result> { - debug!("downloading CRL from {url}"); - - let response = - reqwest::blocking::get(url).context(format!("failed to download CRL from {url}"))?; - - if !response.status().is_success() { - bail!("CRL download failed with status: {}", response.status()); - } - - let crl_bytes = response - .bytes() - .context("failed to read CRL response body")? - .to_vec(); - - debug!("downloaded {} bytes CRL from {}", crl_bytes.len(), url); - - Ok(crl_bytes) -} - -fn extract_crl_urls(cert_der: &[u8]) -> Result> { - use x509_parser::extensions::ParsedExtension; - - let (_, cert) = X509Certificate::from_der(cert_der).context("failed to parse certificate")?; - - let mut crl_urls = Vec::new(); - - for ext in cert.extensions() { - let ParsedExtension::CRLDistributionPoints(crl_dist_points) = ext.parsed_extension() else { - continue; - }; - for dist_point in crl_dist_points.points.iter() { - let Some(dist_point_name) = &dist_point.distribution_point else { - continue; - }; - - let DistributionPointName::FullName(names) = dist_point_name else { - continue; - }; - for name in names.iter() { - let x509_parser::extensions::GeneralName::URI(uri) = name else { - continue; - }; - crl_urls.push(uri.to_string()); - debug!("found CRL URL: {uri}"); - } - } - } - - if crl_urls.is_empty() { - debug!("no CRL Distribution Points found in certificate"); - } - - Ok(crl_urls) -} - fn extract_aia_ca_issuers(cert_der: &[u8]) -> Result> { use x509_parser::extensions::ParsedExtension; @@ -248,33 +112,6 @@ fn extract_aia_ca_issuers(cert_der: &[u8]) -> Result> { Ok(None) } -fn download_cert(url: &str) -> Result> { - debug!("downloading certificate from {url}"); - - let response = reqwest::blocking::get(url) - .context(format!("failed to download certificate from {url}"))?; - - if !response.status().is_success() { - bail!( - "certificate download failed with status: {}", - response.status() - ); - } - - let cert_bytes = response - .bytes() - .context("failed to read certificate response body")? - .to_vec(); - - debug!( - "downloaded {} bytes certificate from {}", - cert_bytes.len(), - url - ); - - Ok(cert_bytes) -} - fn der_to_pem(der: &[u8], label: &str) -> Result { use base64::Engine;