diff --git a/src/apt/release.rs b/src/apt/release.rs index 0606fd7..bf071f2 100644 --- a/src/apt/release.rs +++ b/src/apt/release.rs @@ -137,7 +137,7 @@ impl ChecksumKind { } /// Human-readable algorithm name, for error messages - fn name(self) -> &'static str { + pub fn name(self) -> &'static str { match self { ChecksumKind::Md5 => "MD5", ChecksumKind::Sha1 => "SHA-1", @@ -782,7 +782,7 @@ pub async fn verify_suite( /// GET a URL, returning None on any HTTP error, non-success status, or body /// read failure (Release files are probed, so absence is a normal outcome) async fn fetch_optional(url: &str) -> Option> { - match reqwest::get(url).await { + match crate::distro_info::http_get_retried(url).await { Ok(response) if response.status().is_success() => match response.bytes().await { Ok(bytes) => return Some(bytes.to_vec()), Err(e) => debug!("Reading the body of '{url}' failed: {e}"), @@ -811,7 +811,7 @@ async fn fetch_keyring_cached(url: &str) -> Result, Box &'static reqwest::Client { &HTTP_CLIENT } +/// GET `url` with bounded retries on transient transport errors (a pooled +/// keep-alive connection closed by the remote, a momentary network hiccup, +/// ...): these always succeed again on a fresh connection, and mirrors are +/// busy enough that unguarded single attempts make bulk operations flaky. +/// +/// The response status is not inspected: 404s and the like are meaningful +/// answers, not transport failures. +pub(crate) async fn http_get_retried(url: &str) -> reqwest::Result { + http_get_retried_with_timeout(url, None).await +} + +/// [`http_get_retried`] with a per-request timeout override, for large +/// streaming downloads that exceed the shared client's total timeout +pub(crate) async fn http_get_retried_with_timeout( + url: &str, + timeout: Option, +) -> reqwest::Result { + const ATTEMPTS: u32 = 3; + let mut last_error: Option = None; + for attempt in 0..ATTEMPTS { + let mut request = http_client().get(url); + if let Some(timeout) = timeout { + request = request.timeout(timeout); + } + match request.send().await { + Ok(response) => return Ok(response), + Err(e) => { + if attempt + 1 < ATTEMPTS { + log::debug!( + "GET '{url}' failed (attempt {}/{}, retrying): {}", + attempt + 1, + ATTEMPTS, + e + ); + tokio::time::sleep(Duration::from_millis(300 * (u64::from(attempt) + 1))).await; + } + last_error = Some(e); + } + } + } + Err(last_error.expect("at least one attempt was made")) +} + /// Parse an optional '%Y-%m-%d' date from a CSV cell, warning instead of /// panicking on invalid remote data fn parse_optional_date(value: Option<&str>, series: &str, field: &str) -> Option { diff --git a/src/package_info.rs b/src/package_info.rs index 03cfce4..5a743c3 100644 --- a/src/package_info.rs +++ b/src/package_info.rs @@ -387,30 +387,41 @@ async fn get( debug!("Fetching sources from: {}", url); - let response = match crate::distro_info::http_client().get(&url).send().await { - Ok(resp) => resp, + let compressed_data = match fetch_index_bytes(&url).await { + Ok(data) => data, Err(e) => { - debug!("Failed to fetch {}: {}", url, e); - fetch_errors.push(format!("{suite}/{component}: {}", e)); + debug!("Failed to fetch {url}: {e}"); + fetch_errors.push(format!("{suite}/{component}: {e}")); continue; } }; - if !response.status().is_success() { - debug!("Failed to fetch {}: status {}", url, response.status()); - fetch_errors.push(format!("{suite}/{component}: HTTP {}", response.status())); - continue; - } - - let compressed_data = response.bytes().await?; - // The index must match the checksums listed in the signed Release // file: this is what closes the 'substituted index with matching // artifact checksums' man-in-the-middle attack let suite_rel_path = format!("{component}/source/Sources.gz"); - verified - .verify_file(&suite_rel_path, &compressed_data) - .map_err(release::VerifyError)?; + let compressed_data = match verified.verify_file(&suite_rel_path, &compressed_data) { + Ok(()) => compressed_data, + Err(verify_error) => { + // Busy mirrors and CDNs can serve an index generation + // slightly older or newer than the Release file fetched + // moments before. Pin the exact generation listed in the + // Release file via Debian's by-hash mechanism before + // failing. + match fetch_index_by_hash(&url, &verified, &suite_rel_path).await { + Ok(pinned) => { + debug!( + "index at '{url}' did not match the Release file: used the by-hash copy" + ); + pinned + } + Err(by_hash_error) => { + debug!("by-hash fetch of '{}' failed: {}", url, by_hash_error); + return Err(release::VerifyError(verify_error).into()); + } + } + } + }; debug!( "Downloaded Sources.gz for {}/{}/{}", @@ -577,10 +588,86 @@ async fn get_flat_repo_series(base_url: &str) -> Result> Err(format!("No Codename or Suite field in Release file at '{url}'").into()) } +/// Fetch an index pinned to the exact generation listed in the Release file +/// +/// Uses Debian's by-hash mechanism, supported by the main archives (Debian, +/// Ubuntu, PPAs): the server serves the index generation matching the +/// Release file instead of whatever the mirror/CDN currently holds. The +/// returned content is verified again, so a by-hash-incapable mirror can +/// only cause a 404 here, never a silent mismatch. +async fn fetch_index_by_hash( + index_url: &str, + verified: &VerifiedRelease, + rel_path: &str, +) -> Result, Box> { + let entry = verified.hash_for(rel_path).ok_or_else(|| { + format!("'{rel_path}' is not listed in the Release file: cannot fetch it by hash") + })?; + // By-hash URLs use the Release field names (MD5Sum, SHA1, SHA256, ...) + let algo = match entry.kind { + release::ChecksumKind::Md5 => "MD5Sum", + release::ChecksumKind::Sha1 => "SHA1", + release::ChecksumKind::Sha256 => "SHA256", + release::ChecksumKind::Sha512 => "SHA512", + }; + // The by-hash layout replaces the file name with the hash reference: + // 'main/source/Sources.gz' is served at 'main/source/by-hash/SHA256/' + let (dir, _) = index_url + .rsplit_once('/') + .ok_or_else(|| format!("'{index_url}' has no parent directory"))?; + let url = format!("{dir}/by-hash/{algo}/{}", entry.hash); + + let response = crate::distro_info::http_get_retried(&url).await?; + if !response.status().is_success() { + return Err(format!("HTTP {} for '{url}'", response.status()).into()); + } + let data = response.bytes().await?.to_vec(); + verified.verify_file(rel_path, &data)?; + Ok(data) +} + +/// Fetch the compressed sources index at `url` +/// +/// Transient transport failures and truncated bodies are retried; a +/// non-success HTTP status is a meaningful answer (e.g. a component that +/// does not exist in the suite) and is reported without retrying. +async fn fetch_index_bytes(url: &str) -> Result, String> { + const ATTEMPTS: u32 = 3; + let mut last_error = String::new(); + for attempt in 1..=ATTEMPTS { + match crate::distro_info::http_get_retried(url).await { + Err(e) => { + last_error = e.to_string(); + log::debug!("fetch of '{url}' failed (attempt {attempt}/{ATTEMPTS}): {last_error}"); + tokio::time::sleep(std::time::Duration::from_millis(300 * u64::from(attempt))).await; + } + Ok(response) => { + if !response.status().is_success() { + return Err(format!("HTTP {}", response.status())); + } + match response.bytes().await { + Ok(bytes) => return Ok(bytes.to_vec()), + Err(e) => { + // Truncated or corrupted body: retry from scratch + last_error = e.to_string(); + log::debug!( + "reading the body of '{url}' failed (attempt {attempt}/{ATTEMPTS}): {last_error}" + ); + tokio::time::sleep( + std::time::Duration::from_millis(300 * u64::from(attempt)), + ) + .await; + } + } + } + } + } + Err(last_error) +} + /// Fetch the sources index of a flat repository /// -/// Flat repositories are free to serve any compressed variant of the index -/// (or an uncompressed one), so try the usual candidates in turn. When the +/// Flat repositories are free to serve any compressed variant of the index/// (or an uncompressed one), so try the usual candidates in turn. When the /// repository published a Release file, each index candidate is /// checksum-verified against it (failing on mismatch, since the artifact /// hashes would come from the index itself). @@ -592,7 +679,7 @@ async fn get_flat_repo_sources( let mut errors = Vec::new(); for name in ["Sources.xz", "Sources.gz", "Sources"] { let url = format!("{base}/{name}"); - match crate::distro_info::http_client().get(&url).send().await { + match crate::distro_info::http_get_retried(&url).await { Ok(response) if response.status().is_success() => { let data = response.bytes().await?.to_vec(); diff --git a/src/pull.rs b/src/pull.rs index 627e397..7ba7d0a 100644 --- a/src/pull.rs +++ b/src/pull.rs @@ -309,23 +309,53 @@ async fn download_file_checksum( algo: crate::package_info::ChecksumAlgo, target_dir: &Path, progress: ProgressCallback<'_>, +) -> Result<(), Box> { + // Archive mirrors and CDNs are busy enough that single attempts fail + // spuriously (dropped connections, truncated bodies, index generations + // momentarily out of sync): retry the whole download a few times before + // reporting the last failure. + const ATTEMPTS: u32 = 3; + let mut last_error: Box = String::new().into(); + for attempt in 1..=ATTEMPTS { + match download_file_checksum_once(url, checksum, algo, target_dir, progress).await { + Ok(()) => return Ok(()), + Err(e) => { + log::warn!( + "download of '{url}' failed (attempt {attempt}/{ATTEMPTS}): {e}" + ); + last_error = e; + tokio::time::sleep(std::time::Duration::from_millis(500 * u64::from(attempt))).await; + } + } + } + Err(format!( + "downloading '{url}' failed after {ATTEMPTS} attempts: {last_error}" + ) + .into()) +} + +/// One download attempt of [`download_file_checksum`], verifying the +/// content length and the expected checksum +async fn download_file_checksum_once( + url: &str, + checksum: &str, + algo: crate::package_info::ChecksumAlgo, + target_dir: &Path, + progress: ProgressCallback<'_>, ) -> Result<(), Box> { // Download with the shared client (connect timeout). Large orig tarballs // can legitimately take longer than the client's default total timeout, // so use a generous per-request timeout for streaming downloads - let response = crate::distro_info::http_client() - .get(url) - .timeout(std::time::Duration::from_secs(30 * 60)) - .send() - .await?; + let response = crate::distro_info::http_get_retried_with_timeout( + url, + Some(std::time::Duration::from_secs(30 * 60)), + ) + .await + .map_err(|e| Box::new(e) as Box)?; if !response.status().is_success() { return Err(format!("Failed to download '{}' : {}", url, response.status()).into()); } - - let total_size = response - .content_length() - .ok_or(format!("Failed to get content length from '{}'", url))?; - let mut index = 0; + let total_size = response.content_length(); // Target file: extract file name from URL let filename = Path::new(url) @@ -341,18 +371,30 @@ async fn download_file_checksum( let mut stream = response.bytes_stream(); // Accumulate the downloaded bytes so we can compute the final digest with the // correct algorithm once the download is complete. - let mut buffer: Vec = Vec::with_capacity(total_size as usize); + let mut buffer: Vec = Vec::with_capacity(total_size.unwrap_or(0) as usize); while let Some(item) = stream.next().await { let chunk = item?; file.write_all(&chunk)?; buffer.extend_from_slice(&chunk); - if let Some(cb) = progress { - index = min(index + chunk.len(), total_size as usize); - cb("", "Downloading...", index, total_size as usize); + if let (Some(cb), Some(total)) = (progress, total_size) { + let index = min(buffer.len(), total as usize); + cb("", "Downloading...", index, total as usize); } } + // A dropped connection can end the stream early: never hand a truncated + // file to the checksum check (its mismatch message would hide the cause) + if let Some(total) = total_size + && buffer.len() != total as usize + { + return Err(format!( + "incomplete download from '{url}': got {} of {total} bytes", + buffer.len() + ) + .into()); + } + // Verify checksum using the algorithm specified for this file let calculated_checksum = algo.hex_digest(&buffer); if calculated_checksum != checksum {