fix(updater): harden DMG downloads

updater/src/app.rs

@@ -11,7 +11,6 @@ use crate::{
 };
 use anyhow::{Context, Result};
 use chrono::{Duration as ChronoDuration, Utc};
-use reqwest::Client;
 use serde::Deserialize;
 use std::{
     ffi::OsString,
@@ -933,7 +932,7 @@ async fn run_check_cycle(
         return Ok(());
     };
 
-    let client = Client::builder().build()?;
+    let client = upstream::http_client()?;
 
     sync_runtime_state(config, state);
     state.status = UpdateStatus::CheckingUpstream;

updater/src/upstream.rs

@@ -5,8 +5,35 @@ use chrono::{DateTime, Utc};
 use futures_util::StreamExt;
 use reqwest::{header, Client};
 use sha2::{Digest, Sha256};
-use std::path::{Path, PathBuf};
-use tokio::{fs::File, io::AsyncWriteExt};
+use std::time::Duration;
+use std::{
+    path::{Path, PathBuf},
+    sync::atomic::{AtomicU64, Ordering},
+};
+use tokio::{fs::OpenOptions, io::AsyncWriteExt};
+
+const HTTP_CONNECT_TIMEOUT: Duration = Duration::from_secs(30);
+const HTTP_READ_TIMEOUT: Duration = Duration::from_secs(60);
+const DOWNLOAD_TEMP_PREFIX: &str = ".Codex.dmg.download-";
+static DOWNLOAD_TEMP_COUNTER: AtomicU64 = AtomicU64::new(0);
+
+struct DownloadTempFile {
+    path: PathBuf,
+}
+
+impl DownloadTempFile {
+    fn commit(mut self) {
+        self.path = PathBuf::new();
+    }
+}
+
+impl Drop for DownloadTempFile {
+    fn drop(&mut self) {
+        if !self.path.as_os_str().is_empty() {
+            let _ = std::fs::remove_file(&self.path);
+        }
+    }
+}
 
 #[derive(Debug, Clone, PartialEq, Eq)]
 /// Selected HTTP metadata used to detect upstream DMG changes.
@@ -25,6 +52,15 @@ pub struct DownloadedDmg {
     pub candidate_version: String,
 }
 
+/// Builds the HTTP client shared by upstream metadata and DMG requests.
+pub fn http_client() -> Result<Client> {
+    Client::builder()
+        .connect_timeout(HTTP_CONNECT_TIMEOUT)
+        .read_timeout(HTTP_READ_TIMEOUT)
+        .build()
+        .context("Failed to build upstream HTTP client")
+}
+
 /// Fetches the upstream DMG headers used to detect candidate updates.
 pub async fn fetch_remote_metadata(client: &Client, dmg_url: &str) -> Result<RemoteMetadata> {
     let response = client
@@ -80,11 +116,6 @@ pub async fn download_dmg(
         .await
         .with_context(|| format!("Failed to create {}", destination_dir.display()))?;
 
-    let destination = destination_dir.join("Codex.dmg");
-    let mut file = File::create(&destination)
-        .await
-        .with_context(|| format!("Failed to create {}", destination.display()))?;
-
     let response = client
         .get(dmg_url)
         .send()
@@ -93,20 +124,27 @@ pub async fn download_dmg(
         .error_for_status()
         .with_context(|| format!("GET request for {dmg_url} returned an error status"))?;
 
+    let destination = destination_dir.join("Codex.dmg");
+    let (temp, mut file) = create_download_temp(destination_dir).await?;
+
     let mut hasher = Sha256::new();
     let mut stream = response.bytes_stream();
 
     while let Some(chunk) = stream.next().await {
         let chunk = chunk.with_context(|| format!("Failed downloading {dmg_url}"))?;
         file.write_all(&chunk)
             .await
-            .with_context(|| format!("Failed writing {}", destination.display()))?;
+            .with_context(|| format!("Failed writing {}", temp.path.display()))?;
         hasher.update(&chunk);
     }
 
     file.flush()
         .await
-        .with_context(|| format!("Failed flushing {}", destination.display()))?;
+        .with_context(|| format!("Failed flushing {}", temp.path.display()))?;
+    file.sync_all()
+        .await
+        .with_context(|| format!("Failed syncing {}", temp.path.display()))?;
+    drop(file);
 
     let sha256 = hasher
         .finalize()
@@ -115,13 +153,47 @@ pub async fn download_dmg(
         .collect::<String>();
     let candidate_version = derive_candidate_version(&sha256, version_timestamp)?;
 
+    tokio::fs::rename(&temp.path, &destination)
+        .await
+        .with_context(|| {
+            format!(
+                "Failed to atomically replace {} with completed download",
+                destination.display()
+            )
+        })?;
+    temp.commit();
+
     Ok(DownloadedDmg {
         path: destination,
         sha256,
         candidate_version,
     })
 }
 
+async fn create_download_temp(
+    destination_dir: &Path,
+) -> Result<(DownloadTempFile, tokio::fs::File)> {
+    loop {
+        let id = DOWNLOAD_TEMP_COUNTER.fetch_add(1, Ordering::Relaxed);
+        let path = destination_dir.join(format!(
+            "{DOWNLOAD_TEMP_PREFIX}{}-{id}.tmp",
+            std::process::id()
+        ));
+        match OpenOptions::new()
+            .write(true)
+            .create_new(true)
+            .open(&path)
+            .await
+        {
+            Ok(file) => return Ok((DownloadTempFile { path }, file)),
+            Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => continue,
+            Err(error) => {
+                return Err(error).with_context(|| format!("Failed to create {}", path.display()))
+            }
+        }
+    }
+}
+
 /// Derives a local package version from the DMG hash and download timestamp.
 pub fn derive_candidate_version(sha256: &str, timestamp: DateTime<Utc>) -> Result<String> {
     let short_hash = sha256
@@ -139,6 +211,12 @@ mod tests {
     use super::*;
     use anyhow::Result;
     use chrono::TimeZone;
+    use std::{
+        io::{Read, Write},
+        net::TcpListener,
+        thread,
+        time::Duration,
+    };
     use tempfile::tempdir;
     use wiremock::{
         matchers::{method, path},
@@ -184,6 +262,7 @@ mod tests {
 
         let client = Client::builder().build()?;
         let temp = tempdir()?;
+        std::fs::write(temp.path().join("Codex.dmg"), b"old-valid-cache")?;
         let downloaded = download_dmg(
             &client,
             &format!("{}/Codex.dmg", server.uri()),
@@ -198,6 +277,99 @@ mod tests {
             "678cd508ffe0071e217020a7a4eecbebe25362c022ac78c13a5ae87b7a3a0c92"
         );
         assert_eq!(downloaded.candidate_version, "2026.03.24.120000+678cd508");
+        assert_eq!(std::fs::read(&downloaded.path)?, body);
+        assert_no_download_temps(temp.path())?;
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn http_error_preserves_existing_dmg() -> Result<()> {
+        let server = MockServer::start().await;
+        Mock::given(method("GET"))
+            .and(path("/Codex.dmg"))
+            .respond_with(ResponseTemplate::new(503))
+            .mount(&server)
+            .await;
+
+        let temp = tempdir()?;
+        let destination = temp.path().join("Codex.dmg");
+        std::fs::write(&destination, b"old-valid-cache")?;
+        let error = download_dmg(
+            &Client::builder().build()?,
+            &format!("{}/Codex.dmg", server.uri()),
+            temp.path(),
+            Utc::now(),
+        )
+        .await
+        .expect_err("error response should fail");
+
+        assert!(error.to_string().contains("returned an error status"));
+        assert_eq!(std::fs::read(destination)?, b"old-valid-cache");
+        assert_no_download_temps(temp.path())?;
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn interrupted_body_preserves_existing_dmg_and_cleans_temp() -> Result<()> {
+        let listener = TcpListener::bind("127.0.0.1:0")?;
+        let address = listener.local_addr()?;
+        let server = thread::spawn(move || -> std::io::Result<()> {
+            let (mut connection, _) = listener.accept()?;
+            let mut request = [0_u8; 1024];
+            let mut request_len = 0;
+            while request_len < request.len() {
+                let read = connection.read(&mut request[request_len..])?;
+                if read == 0 {
+                    break;
+                }
+                request_len += read;
+                if request[..request_len]
+                    .windows(4)
+                    .any(|window| window == b"\r\n\r\n")
+                {
+                    break;
+                }
+            }
+            connection.write_all(
+                b"HTTP/1.1 200 OK\r\nContent-Length: 100\r\nConnection: close\r\n\r\n",
+            )?;
+            connection.flush()?;
+            thread::sleep(Duration::from_millis(100));
+            connection.write_all(b"partial")?;
+            connection.flush()?;
+            thread::sleep(Duration::from_millis(100));
+            Ok(())
+        });
+
+        let temp = tempdir()?;
+        let destination = temp.path().join("Codex.dmg");
+        std::fs::write(&destination, b"old-valid-cache")?;
+        let result = download_dmg(
+            &Client::builder().build()?,
+            &format!("http://{address}/Codex.dmg"),
+            temp.path(),
+            Utc::now(),
+        )
+        .await;
+        server.join().expect("test server thread panicked")?;
+
+        assert!(result.is_err(), "truncated body should fail");
+        assert_eq!(std::fs::read(destination)?, b"old-valid-cache");
+        assert_no_download_temps(temp.path())?;
+        Ok(())
+    }
+
+    fn assert_no_download_temps(directory: &Path) -> Result<()> {
+        let leftovers = std::fs::read_dir(directory)?
+            .filter_map(|entry| entry.ok())
+            .filter(|entry| {
+                entry
+                    .file_name()
+                    .to_string_lossy()
+                    .starts_with(DOWNLOAD_TEMP_PREFIX)
+            })
+            .collect::<Vec<_>>();
+        assert!(leftovers.is_empty(), "temporary downloads remain");
         Ok(())
     }
 

updater/src/wrapper_apply.rs

@@ -18,6 +18,7 @@ use serde_json::Value;
 use std::{
     collections::HashSet,
     fs,
+    io::{BufReader, Read},
     os::unix::fs::{self as unix_fs, PermissionsExt},
     path::{Path, PathBuf},
     process::Command,
@@ -489,7 +490,7 @@ async fn cached_or_downloaded_dmg(
         }
     }
 
-    let client = reqwest::Client::builder().build()?;
+    let client = upstream::http_client()?;
     let downloads_dir = config.workspace_root.join("downloads");
     let downloaded =
         upstream::download_dmg(&client, &config.dmg_url, &downloads_dir, chrono::Utc::now())
@@ -504,10 +505,20 @@ async fn cached_or_downloaded_dmg(
 /// contents, matching the DMG update path's scheme.
 fn derive_package_version(dmg_path: &Path) -> Result<String> {
     use sha2::{Digest, Sha256};
-    let bytes = std::fs::read(dmg_path)
-        .with_context(|| format!("Failed to read {}", dmg_path.display()))?;
+    let file = fs::File::open(dmg_path)
+        .with_context(|| format!("Failed to open {}", dmg_path.display()))?;
+    let mut reader = BufReader::new(file);
     let mut hasher = Sha256::new();
-    hasher.update(&bytes);
+    let mut buffer = [0_u8; 64 * 1024];
+    loop {
+        let bytes_read = reader
+            .read(&mut buffer)
+            .with_context(|| format!("Failed to read {}", dmg_path.display()))?;
+        if bytes_read == 0 {
+            break;
+        }
+        hasher.update(&buffer[..bytes_read]);
+    }
     let sha = hasher
         .finalize()
         .iter()
@@ -581,6 +592,18 @@ mod tests {
         }
     }
 
+    #[test]
+    fn derives_package_version_with_streamed_dmg_hash() -> Result<()> {
+        let root = tempdir()?;
+        let dmg = root.path().join("Codex.dmg");
+        std::fs::write(&dmg, b"codex-dmg-test-payload")?;
+
+        let version = derive_package_version(&dmg)?;
+
+        assert!(version.ends_with("+678cd508"));
+        Ok(())
+    }
+
     fn write_local_feature(root: &Path, id: &str) {
         let feature_dir = root.join("builder/linux-features/local").join(id);
         std::fs::create_dir_all(feature_dir.join("nested")).unwrap();