fix(download): verify etags for resumed requests

Signed-off-by: Marcel Guzik <[email protected]>

Author: Bravo555

Cargo.lock

@@ -13,6 +13,7 @@ repository = { workspace = true }
 anyhow = { workspace = true, features = ["backtrace"] }
 backoff = { workspace = true }
 certificate = { workspace = true, features = ["reqwest"] }
+http = { workspace = true }
 hyper = { workspace = true }
 log = { workspace = true }
 nix = { workspace = true }

crates/common/download/Cargo.toml

@@ -1,4 +1,5 @@
 pub mod partial_response;
+use crate::download::partial_response::PartialResponse;
 use crate::error::DownloadError;
 use crate::error::ErrContext;
 use anyhow::anyhow;
@@ -187,7 +188,7 @@ impl Downloader {
 
     /// If interrupted, continues ongoing download.
     ///
-    /// If the server supports is, a range request is used to download only the
+    /// If the server supports it, a range request is used to download only the
     /// remaining range of the file. Otherwise, progress is restarted and we
     /// download full range of the file again.
     async fn download_continue(
@@ -197,9 +198,17 @@ impl Downloader {
         mut prev_response: Response,
     ) -> Result<(), DownloadError> {
         loop {
-            let offset = next_request_offset(&prev_response, file);
-            let mut response = self.request_range_from(url, offset).await?;
-            let offset = partial_response::response_range_start(&response)?;
+            let request_offset = next_request_offset(&prev_response, file)?;
+            let mut response = self.request_range_from(url, request_offset).await?;
+            let offset = match partial_response::response_range_start(&response, &prev_response)? {
+                PartialResponse::CompleteContent => 0,
+                PartialResponse::PartialContent(pos) => pos,
+                PartialResponse::ResourceModified => {
+                    file.seek(SeekFrom::Start(0))
+                        .context("failed to seek in file".to_string())?;
+                    continue;
+                }
+            };
 
             if offset != 0 {
                 info!("Resuming file download at position={offset}");
@@ -211,9 +220,7 @@ impl Downloader {
                 Ok(()) => break,
 
                 Err(SaveChunksError::Network(err)) => {
-                    prev_response = response;
                     warn!("Error while downloading response: {err}.\nRetrying...");
-                    continue;
                 }
 
                 Err(SaveChunksError::Io(err)) => {
@@ -223,6 +230,7 @@ impl Downloader {
                     })
                 }
             }
+            prev_response = response;
         }
 
         Ok(())
@@ -424,20 +432,22 @@ pub fn try_pre_allocate_space(
     Ok(())
 }
 
-fn next_request_offset(prev_response: &Response, file: &mut File) -> u64 {
+fn next_request_offset(prev_response: &Response, file: &mut File) -> Result<u64, DownloadError> {
     use hyper::header;
     use hyper::StatusCode;
     use std::io::Seek;
-    let pos = file.stream_position().unwrap();
+    let pos = file
+        .stream_position()
+        .context("failed to get cursor position".to_string())?;
     if prev_response.status() == StatusCode::PARTIAL_CONTENT
         || prev_response
             .headers()
             .get(header::ACCEPT_RANGES)
             .is_some_and(|unit| unit == "bytes")
     {
-        pos
+        Ok(pos)
     } else {
-        0
+        Ok(0)
     }
 }
 

crates/common/download/src/download.rs

@@ -2,35 +2,98 @@
 
 use reqwest::header;
 use reqwest::header::HeaderValue;
+use reqwest::Response;
 use reqwest::StatusCode;
 
+pub(super) enum PartialResponse {
+    /// Server returned a partial content response starting at given position
+    PartialContent(u64),
+
+    /// Server returned regular OK response, resume writing at 0
+    CompleteContent,
+
+    /// Server returned partial content but resource was modified, request needs to be retried
+    ResourceModified,
+}
+
 /// Returns the position of the partial response in the resource.
 ///
 /// When making a partial request, a server can return a different range than
 /// the one we asked for, in which case we need to extract the position from the
 /// Content-Range header. The server could also just ignore the Range header and
 /// respond with 200 OK, in which case we need to download the entire resource
 /// all over again.
-pub fn response_range_start(response: &reqwest::Response) -> Result<u64, InvalidResponseError> {
-    let chunk_pos = match response.status() {
+pub(super) fn response_range_start(
+    response: &reqwest::Response,
+    prev_response: &Response,
+) -> Result<PartialResponse, InvalidResponseError> {
+    match response.status() {
         // Complete response, seek to the beginning of the file
-        StatusCode::OK => 0,
+        StatusCode::OK => Ok(PartialResponse::CompleteContent),
 
         // Partial response, the range might be different from what we
         // requested, so we need to parse it. Because we only request a single
         // range from the current position to the end of the document, we can
         // ignore multipart/byteranges media type.
-        StatusCode::PARTIAL_CONTENT => partial_response_start_range(response)?,
+        StatusCode::PARTIAL_CONTENT => {
+            if was_resource_modified(response, prev_response) {
+                return Ok(PartialResponse::ResourceModified);
+            }
+            let pos = partial_response_start_range(response)?;
+            Ok(PartialResponse::PartialContent(pos))
+        }
 
         // We don't expect to receive any other 200-299 status code, but if we
         // do, treat it the same as OK
-        status_code if status_code.is_success() => 0,
+        status_code if status_code.is_success() => Ok(PartialResponse::CompleteContent),
+
+        status_code => Err(InvalidResponseError::UnexpectedStatus(status_code)),
+    }
+}
+
+/// Checks if the resource was modified between the current and previous response.
+///
+/// If the resource was updated, we should restart download and request full range of the new
+/// resource. Otherwise, a partial request can be used to resume the download.
+fn was_resource_modified(response: &Response, prev_response: &Response) -> bool {
+    if response.status() != hyper::StatusCode::PARTIAL_CONTENT {
+        // not using a partial request, don't care if it's modified or not
+        return false;
+    }
+
+    // etags in current and previous request must match
+    let etag = response
+        .headers()
+        .get(header::ETAG)
+        .and_then(|h| h.to_str().ok());
+    let prev_etag = prev_response
+        .headers()
+        .get(header::ETAG)
+        .and_then(|h| h.to_str().ok());
 
-        status_code => {
-            return Err(InvalidResponseError::UnexpectedStatus(status_code));
+    match (etag, prev_etag) {
+        (None, None) => {
+            // no etags in either request, assume resource is unchanged
+            false
         }
-    };
-    Ok(chunk_pos)
+        (None, Some(_)) | (Some(_), None) => {
+            // previous request didn't have etag and this does or vice versa, abort
+            true
+        }
+        (Some(etag), Some(prev_etag)) => {
+            // Examples:
+            // ETag: "xyzzy"
+            // ETag: W/"xyzzy"
+            // ETag: ""
+            if etag.starts_with("W/") {
+                // validator is weak, but in range requests tags must match using strong comparison
+                // https://www.rfc-editor.org/rfc/rfc9110#entity.tag.comparison
+                return true;
+            }
+
+            etag != prev_etag
+        }
+    }
 }
 
 #[derive(Debug, thiserror::Error)]
@@ -88,3 +151,31 @@ pub struct ContentRangeParseError {
     reason: &'static str,
     value: header::HeaderValue,
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test_case::test_case(Some(r#""xyzzy""#), Some(r#""xyzzy""#), false)]
+    #[test_case::test_case(Some(r#"W/"xyzzy""#), Some(r#""xyzzy""#), true)]
+    #[test_case::test_case(Some(r#""xyzzy""#), Some(r#"W/"xyzzy""#), true)]
+    #[test_case::test_case(Some(r#""xyzzy1""#), Some(r#""xyzzy2""#), true)]
+    #[test_case::test_case(None, None, false)]
+    #[test_case::test_case(Some(r#""xyzzy1""#), None, true)]
+    #[test_case::test_case(None, Some(r#""xyzzy2""#), true)]
+    fn verifies_etags(etag1: Option<&'static str>, etag2: Option<&'static str>, modified: bool) {
+        let mut response1 = http::Response::builder().status(StatusCode::PARTIAL_CONTENT);
+        if let Some(etag) = etag1 {
+            response1 = response1.header(http::header::ETAG, etag);
+        }
+        let response1 = response1.body("").unwrap().into();
+
+        let mut response2 = http::Response::builder().status(StatusCode::PARTIAL_CONTENT);
+        if let Some(etag) = etag2 {
+            response2 = response2.header(http::header::ETAG, etag);
+        }
+        let response2 = response2.body("").unwrap().into();
+
+        assert_eq!(was_resource_modified(&response1, &response2), modified);
+    }
+}

crates/common/download/src/download/partial_response.rs

@@ -102,3 +102,109 @@ async fn resume_download_when_disconnected() {
 
     server_task.abort();
 }
+
+/// If after retrying ETag of the resource is different, we should download it
+/// from scratch again.
+#[tokio::test]
+async fn resume_download_with_etag_changed() {
+    let file_v1 = "AAAABBBBCCCCDDDD";
+    let file_v2 = "XXXXYYYYZZZZWWWW";
+    let chunk_size = 4;
+
+    let listener = TcpListener::bind("localhost:0").await.unwrap();
+    let port = listener.local_addr().unwrap().port();
+
+    // server should do 3 things:
+    // - only send a portion of the first request
+    // - after the first request update the resource
+    // - serve 2nd request normally (but we expect it to be a range request)
+    let server_task = tokio::spawn(async move {
+        let mut request_count = 0;
+        while let Ok((mut stream, _addr)) = listener.accept().await {
+            let response_task = async move {
+                let (reader, mut writer) = stream.split();
+                let mut lines = BufReader::new(reader).lines();
+                let mut range: Option<std::ops::Range<usize>> = None;
+
+                // We got an HTTP request, read the lines of the request
+                while let Ok(Some(line)) = lines.next_line().await {
+                    if line.to_ascii_lowercase().contains("range:") {
+                        let (_, bytes) = line.split_once('=').unwrap();
+                        let (start, end) = bytes.split_once('-').unwrap();
+                        let start = start.parse().unwrap_or(0);
+                        let end = end.parse().unwrap_or(file_v2.len());
+                        range = Some(start..end)
+                    }
+                    // On `\r\n\r\n` (empty line) stop reading the request
+                    // and start responding
+                    if line.is_empty() {
+                        break;
+                    }
+                }
+
+                let file = if request_count == 0 { file_v1 } else { file_v2 };
+                let etag = if request_count == 0 {
+                    "v1-initial"
+                } else {
+                    "v2-changed"
+                };
+
+                if let Some(range) = range {
+                    // Return range for both first and subsequent requests
+                    let start = range.start;
+                    let end = range.end.min(file.len());
+                    let header = format!(
+                        "HTTP/1.1 206 Partial Content\r\n\
+                        transfer-encoding: chunked\r\n\
+                        connection: close\r\n\
+                        content-type: application/octet-stream\r\n\
+                        content-range: bytes {start}-{end}/*\r\n\
+                        accept-ranges: bytes\r\n\
+                        etag: \"{etag}\"\r\n"
+                    );
+                    let body = &file[start..end];
+                    let size = body.len();
+                    let msg = format!("{header}\r\n{size:x}\r\n{body}\r\n0\r\n\r\n");
+                    writer.write_all(msg.as_bytes()).await.unwrap();
+                    writer.flush().await.unwrap();
+                } else {
+                    let header = format!(
+                        "HTTP/1.1 200 OK\r\n\
+                        transfer-encoding: chunked\r\n\
+                        connection: close\r\n\
+                        content-type: application/octet-stream\r\n\
+                        accept-ranges: bytes\r\n\
+                        etag: \"{etag}\"\r\n"
+                    );
+
+                    let body = &file[0..chunk_size];
+                    let size = body.len();
+                    let msg = format!("{header}\r\n{size:x}\r\n{body}\r\n");
+                    writer.write_all(msg.as_bytes()).await.unwrap();
+                    writer.flush().await.unwrap();
+                    // Connection drops here without sending final chunk
+                }
+            };
+            request_count += 1;
+            tokio::spawn(response_task);
+        }
+    });
+
+    // Wait until task binds a listener on the TCP port
+    tokio::time::sleep(std::time::Duration::from_millis(50)).await;
+
+    let tmpdir = TempDir::new().unwrap();
+    let target_path = tmpdir.path().join("partial_download_etag");
+
+    let downloader = Downloader::new(target_path, None, CloudHttpConfig::test_value());
+    let url = DownloadInfo::new(&format!("http://localhost:{port}/"));
+
+    downloader.download(&url).await.unwrap();
+    let saved_file = std::fs::read_to_string(downloader.filename()).unwrap();
+    // Should have the complete new file content since ETag changed
+    assert_eq!(saved_file, file_v2);
+
+    downloader.cleanup().await.unwrap();
+
+    server_task.abort();
+}