fix(download): verify etags for resumed requests

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

Author: Bravo555

crates/common/download/src/download.rs

@@ -5,6 +5,7 @@ use anyhow::anyhow;
 use backoff::future::retry_notify;
 use backoff::ExponentialBackoff;
 use certificate::CloudHttpConfig;
+use hyper::header;
 use log::debug;
 use log::info;
 use log::warn;
@@ -199,6 +200,12 @@ impl Downloader {
         loop {
             let offset = next_request_offset(&prev_response, file);
             let mut response = self.request_range_from(url, offset).await?;
+            // etags in current and previous request must match to resume
+            if !etags_matching(&prev_response, &response) {
+                file.seek(SeekFrom::Start(0)).unwrap();
+                prev_response = response;
+                continue;
+            }
             let offset = partial_response::response_range_start(&response)?;
 
             if offset != 0 {
@@ -211,9 +218,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 +228,7 @@ impl Downloader {
                     })
                 }
             }
+            prev_response = response;
         }
 
         Ok(())
@@ -352,6 +358,37 @@ impl Downloader {
     }
 }
 
+fn etags_matching(prev_response: &Response, response: &Response) -> bool {
+    if let Some(etag) = response
+        .headers()
+        .get(header::ETAG)
+        .and_then(|h| h.to_str().ok())
+    {
+        // TODO: handle optional backslashes
+        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#section-13.1.5-12.1
+            return false;
+        }
+
+        let Some(prev_etag) = prev_response
+            .headers()
+            .get(header::ETAG)
+            .and_then(|h| h.to_str().ok())
+        else {
+            // previous request didn't have etag and this does or vice versa, abort
+            return false;
+        };
+
+        if etag != prev_etag {
+            // etags don't match, abort
+            return false;
+        }
+    }
+    // etags match
+    true
+}
+
 /// Decides whether HTTP request error is retryable.
 fn reqwest_err_to_backoff(err: reqwest::Error) -> backoff::Error<reqwest::Error> {
     if err.is_timeout() || err.is_connect() {

crates/common/download/src/download/tests/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();
+}