From 0d84bf6f6a5708bc5521910444f36ad6f6380cea Mon Sep 17 00:00:00 2001 From: KiriAky 107 Date: Wed, 9 Sep 2026 05:57:36 +0800 Subject: [PATCH] =?UTF-8?q?perf(sync):=20=E9=99=90=E5=88=B6=E6=9A=82?= =?UTF-8?q?=E5=AD=98=E9=AA=8C=E8=AF=81=E5=B9=B6=E5=A4=8D=E7=94=A8=E5=B7=B2?= =?UTF-8?q?=E9=AA=8C=E8=AF=81=E4=B8=8A=E4=BC=A0=E5=8F=A5=E6=9F=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- frontend/src-tauri/src/payloads.rs | 114 +++++++++++++++++++++++++- frontend/src-tauri/src/sync_client.rs | 25 ++---- frontend/src-tauri/src/sync_inbox.rs | 11 +-- 3 files changed, 117 insertions(+), 33 deletions(-) diff --git a/frontend/src-tauri/src/payloads.rs b/frontend/src-tauri/src/payloads.rs index 7a71a22..571fff1 100644 --- a/frontend/src-tauri/src/payloads.rs +++ b/frontend/src-tauri/src/payloads.rs @@ -4,7 +4,7 @@ use rusqlite::{params, OptionalExtension}; use sha2::{Digest, Sha256}; use std::{ fs, - io::{Read, Write}, + io::{Read, Seek, SeekFrom, Write}, path::Path, }; impl Workspace { @@ -59,6 +59,11 @@ impl Workspace { } } pub(crate) fn verify(path: &Path, digest: &str, size: u64) -> Result<()> { + open_verified(path, digest, size).map(drop) +} +/// Return the verified handle, rewound for use by a streaming caller. +/// Path containment is the caller's responsibility; this is not a sandbox opener. +pub(crate) fn open_verified(path: &Path, digest: &str, size: u64) -> Result { let meta = fs::symlink_metadata(path)?; if meta.file_type().is_symlink() || !meta.is_file() || meta.len() != size { return Err(HostError::new("SYNC_SPOOL_CORRUPT")); @@ -71,17 +76,118 @@ pub(crate) fn verify(path: &Path, digest: &str, size: u64) -> Result<()> { } } let mut stream = fs::File::open(path)?; + if stream.metadata()?.len() != size { + return Err(HostError::new("SYNC_SPOOL_CORRUPT")); + } + verify_reader(&mut stream, digest, size)?; + stream.seek(SeekFrom::Start(0))?; + Ok(stream) +} +const VERIFY_BUFFER_BYTES: usize = 64 * 1024; +fn verify_reader(stream: &mut impl Read, digest: &str, size: u64) -> Result<()> { let mut hasher = Sha256::new(); - let mut buffer = vec![0; 1024 * 1024]; + let mut buffer = vec![0; VERIFY_BUFFER_BYTES]; + let mut length = 0u64; loop { - let count = stream.read(&mut buffer)?; + // Even if a file grows after metadata inspection, consume at most the + // declared payload plus one byte, never an unbounded changing stream. + let limit = size + .saturating_sub(length) + .saturating_add(1) + .min(buffer.len() as u64) as usize; + let count = stream.read(&mut buffer[..limit])?; if count == 0 { break; } + length += count as u64; + if length > size { + return Err(HostError::new("SYNC_SPOOL_CORRUPT")); + } hasher.update(&buffer[..count]); } - if format!("{:x}", hasher.finalize()) != digest { + if length != size || format!("{:x}", hasher.finalize()) != digest { return Err(HostError::new("SYNC_SPOOL_CORRUPT")); } Ok(()) } + +#[cfg(test)] +mod tests { + use super::*; + struct Generated { + remaining: u64, + consumed: u64, + max_request: usize, + } + impl Read for Generated { + fn read(&mut self, buffer: &mut [u8]) -> std::io::Result { + self.max_request = self.max_request.max(buffer.len()); + let count = self.remaining.min(buffer.len() as u64) as usize; + buffer[..count].fill(7); + self.remaining -= count as u64; + self.consumed += count as u64; + Ok(count) + } + } + #[test] + fn hundred_mib_verification_is_bounded_and_rejects_growth_or_truncation() { + let size = 100 * 1024 * 1024u64; + let block = vec![7u8; VERIFY_BUFFER_BYTES]; + let mut hasher = Sha256::new(); + for _ in 0..size / block.len() as u64 { + hasher.update(&block); + } + let digest = format!("{:x}", hasher.finalize()); + let mut stream = Generated { + remaining: size, + consumed: 0, + max_request: 0, + }; + verify_reader(&mut stream, &digest, size).unwrap(); + assert_eq!(stream.consumed, size); + assert_eq!(stream.max_request, VERIFY_BUFFER_BYTES); + let mut growing = Generated { + remaining: u64::MAX, + consumed: 0, + max_request: 0, + }; + assert_eq!( + verify_reader(&mut growing, &hash(&[7; 8]), 8) + .unwrap_err() + .code, + "SYNC_SPOOL_CORRUPT" + ); + assert_eq!(growing.consumed, 9); + assert_eq!( + verify_reader(&mut &[7; 7][..], &hash(&[7; 8]), 8) + .unwrap_err() + .code, + "SYNC_SPOOL_CORRUPT" + ); + assert_eq!( + verify_reader(&mut &[8; 8][..], &hash(&[7; 8]), 8) + .unwrap_err() + .code, + "SYNC_SPOOL_CORRUPT" + ); + verify_reader(&mut &[][..], &hash(&[]), 0).unwrap(); + } + #[test] + fn verified_file_is_rewound_and_corruption_is_rejected() { + let temp = tempfile::tempdir().unwrap(); + let path = temp.path().join("payload"); + fs::write(&path, b"verified payload").unwrap(); + let mut file = open_verified(&path, &hash(b"verified payload"), 16).unwrap(); + let mut bytes = Vec::new(); + file.read_to_end(&mut bytes).unwrap(); + assert_eq!(bytes, b"verified payload"); + drop(file); + fs::write(&path, b"modified payload").unwrap(); + assert_eq!( + verify(&path, &hash(b"verified payload"), 16) + .unwrap_err() + .code, + "SYNC_SPOOL_CORRUPT" + ); + } +} diff --git a/frontend/src-tauri/src/sync_client.rs b/frontend/src-tauri/src/sync_client.rs index 13b0ec2..586c8e5 100644 --- a/frontend/src-tauri/src/sync_client.rs +++ b/frontend/src-tauri/src/sync_client.rs @@ -449,11 +449,9 @@ impl SyncClient { ws.sync_spool(digest) })?; if target.exists() { - let bytes = std::fs::read(&target)?; - if bytes.len() as i64 == revision.size && crate::workspace::hash(&bytes) == digest { - return Ok(()); - } - return Err(SyncError::new("SYNC_SPOOL_CORRUPT")); + crate::payloads::verify(&target, digest, revision.size as u64)?; + workspace.access(|ws| ws.check_binding(&binding.id))?; + return Ok(()); } let url = self .endpoint @@ -511,22 +509,9 @@ impl SyncClient { job: &Job, ) -> Result<()> { let path = workspace.access(|ws| ws.sync_spool(&job.hash))?; - let mut file = std::fs::File::open(path)?; - if file.metadata()?.len() != job.size as u64 { - return Err(SyncError::new("SYNC_SPOOL_CORRUPT")); - } - let mut hasher = Sha256::new(); + let mut file = crate::payloads::open_verified(&path, &job.hash, job.size as u64)?; + workspace.access(|ws| ws.check_binding(&binding.id))?; let mut buffer = vec![0u8; 1048576]; - loop { - let count = file.read(&mut buffer)?; - if count == 0 { - break; - } - hasher.update(&buffer[..count]); - } - if format!("{:x}", hasher.finalize()) != job.hash { - return Err(SyncError::new("SYNC_SPOOL_CORRUPT")); - } let base = format!("sync/v1/vaults/{}/uploads", binding.remote_vault); let mut upload_id = job.upload_id.clone(); let mut offset = 0; diff --git a/frontend/src-tauri/src/sync_inbox.rs b/frontend/src-tauri/src/sync_inbox.rs index f2da630..fb7a61d 100644 --- a/frontend/src-tauri/src/sync_inbox.rs +++ b/frontend/src-tauri/src/sync_inbox.rs @@ -115,11 +115,7 @@ impl Workspace { let digest = hash(bytes); let path = self.sync_spool(&digest)?; if path.exists() { - if fs::symlink_metadata(&path)?.file_type().is_symlink() - || hash(&fs::read(&path)?) != digest - { - return Err(HostError::new("SYNC_SPOOL_CORRUPT")); - } + crate::payloads::verify(&path, &digest, bytes.len() as u64)?; } else { let mut temp = tempfile::NamedTempFile::new_in(path.parent().unwrap())?; temp.write_all(bytes)?; @@ -150,10 +146,7 @@ impl Workspace { return Err(HostError::new("SYNC_CURSOR_INVALID")); } if let Some(digest) = &revision.hash { - let bytes = fs::read(self.sync_spool(digest)?)?; - if bytes.len() as i64 != revision.size || hash(&bytes) != *digest { - return Err(HostError::new("SYNC_SPOOL_CORRUPT")); - } + crate::payloads::verify(&self.sync_spool(digest)?, digest, revision.size as u64)?; } let encoded = serde_json::to_string(revision).map_err(|_| HostError::new("SYNC_RESPONSE_INVALID"))?;