From 28a97f675ec9c2358659c2830bd60613fcdf3d41 Mon Sep 17 00:00:00 2001 From: Donald Knuth Date: Sat, 3 Sep 2022 14:58:48 -0400 Subject: [PATCH 1/3] Resume does not work --- src/client.rs | 31 +++++++++++++++++++++---------- src/connection.rs | 21 ++++++++++++++++++--- src/file.rs | 43 ++++++++++++++++++++++++++++++------------- src/message.rs | 1 + src/ui.rs | 2 +- 5 files changed, 71 insertions(+), 27 deletions(-) diff --git a/src/client.rs b/src/client.rs index 61e6fc2..052d405 100644 --- a/src/client.rs +++ b/src/client.rs @@ -6,7 +6,6 @@ use crate::ui::prompt_user_for_file_confirmation; use anyhow::{anyhow, Result}; -use std::ffi::OsStr; use std::path::PathBuf; use tokio::fs::File; @@ -46,7 +45,7 @@ pub async fn receive(password: &String) -> Result<()> { // Wait for offered files, respond with desired files let desired_files = request_specific_files(&mut connection).await?; // Create files - let std_file_handles = create_files(desired_files).await?; + let std_file_handles = create_or_find_files(desired_files).await?; // Download them connection.download_files(std_file_handles).await?; return Ok(()); @@ -57,7 +56,10 @@ pub async fn offer_files( file_handles: &Vec, ) -> Result> { // Collect file offer - let files = file_handles.iter().map(|fh| fh.to_file_offer()).collect(); + let mut files = vec![]; + for handle in file_handles { + files.push(handle.to_file_offer()?); + } let msg = Message::FileOffer(FileOfferPayload { files }); // Send file offer conn.send_msg(msg).await?; @@ -92,15 +94,24 @@ pub async fn request_specific_files(conn: &mut Connection) -> Result) -> Result> { +pub async fn create_or_find_files(desired_files: Vec) -> Result> { let mut v = Vec::new(); for desired_file in desired_files { - let filename = desired_file - .path - .file_name() - .unwrap_or(OsStr::new("random.txt")); - let file = File::create(filename).await?; - let std_file_handle = StdFileHandle::new(desired_file.id, file, 0).await?; + let mut filename = desired_file.path; + filename.push_str(".part"); + let file = match File::open(filename.clone()).await { + Ok(file) => { + println!( + "File {:?} already exists. Attempting to resume download.", + filename + ); + file + } + Err(_) => File::create(filename).await?, + }; + let metadata = file.metadata().await?; + let std_file_handle = + StdFileHandle::new(desired_file.id, file, metadata.len(), desired_file.size).await?; v.push(std_file_handle) } return Ok(v); diff --git a/src/connection.rs b/src/connection.rs index 80f5df4..6b91c36 100644 --- a/src/connection.rs +++ b/src/connection.rs @@ -77,6 +77,7 @@ impl Connection { Err(e) => return Err(anyhow!(e.to_string())), } } + self.send_msg(Message::FileTransferComplete).await?; let elapsed = before.elapsed(); let mb_sent = bytes_sent / 1_048_576; println!( @@ -105,24 +106,38 @@ impl Connection { } pub async fn download_file(&mut self, handle: StdFileHandle) -> Result<()> { + let clone = handle.file.try_clone()?; let mut decoder = GzDecoder::new(handle.file); loop { let msg = self.await_msg().await?; match msg { Message::FileTransfer(payload) => { + println!("In download"); if payload.chunk_header.id != handle.id { return Err(anyhow!("Wrong file")); } - if payload.chunk.len() == 0 { - break; - } decoder.write_all(&payload.chunk[..])? } + Message::FileTransferComplete => { + break; + } _ => return Err(anyhow!("Expecting file transfer message")), } } decoder.finish()?; println!("Done downloading file."); + Connection::check_and_finish_download(clone, handle.size).await?; Ok(()) } + + pub async fn check_and_finish_download(file: std::fs::File, size: u64) -> Result<()> { + let metadata = file.metadata()?; + if metadata.len() == size { + println!("File looks good."); + return Ok(()); + } + return Err(anyhow!( + "Downloaded file does not match expected size. Try again" + )); + } } diff --git a/src/file.rs b/src/file.rs index c63e094..dec835e 100644 --- a/src/file.rs +++ b/src/file.rs @@ -1,4 +1,4 @@ -use anyhow::Result; +use anyhow::{anyhow, Result}; use futures::future::try_join_all; use serde::{Deserialize, Serialize}; @@ -18,7 +18,7 @@ pub struct ChunkHeader { #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub struct FileOffer { pub id: u8, - pub path: PathBuf, + pub path: String, pub size: u64, } @@ -26,16 +26,21 @@ pub struct StdFileHandle { pub id: u8, pub file: std::fs::File, pub start: u64, + pub size: u64, } impl StdFileHandle { - pub async fn new(id: u8, file: File, start: u64) -> Result { - let mut std_file = file.into_std().await; - std_file.seek(SeekFrom::Start(start))?; + pub async fn new(id: u8, file: File, start: u64, size: u64) -> Result { + let mut file = file.into_std().await; + file.seek(SeekFrom::Start(start))?; + if start != 0 { + println!("Seeking to {:?}", start); + } Ok(StdFileHandle { - id: id, - file: std_file, - start: start, + id, + file, + start, + size, }) } } @@ -80,15 +85,15 @@ impl FileHandle { } async fn to_std(self, chunk_header: &ChunkHeader) -> Result { - StdFileHandle::new(self.id, self.file, chunk_header.start).await + StdFileHandle::new(self.id, self.file, chunk_header.start, self.md.len()).await } - pub fn to_file_offer(&self) -> FileOffer { - FileOffer { + pub fn to_file_offer(&self) -> Result { + Ok(FileOffer { id: self.id, - path: self.path.clone(), + path: pathbuf_to_string(&self.path)?, size: self.md.len(), - } + }) } pub async fn get_file_handles(file_paths: &Vec) -> Result> { @@ -115,3 +120,15 @@ pub fn to_size_string(size: u64) -> String { result } + +pub fn pathbuf_to_string(path: &PathBuf) -> Result { + let filename = match path.file_name() { + Some(s) => s, + None => return Err(anyhow!("Could not get filename from file offer.")), + }; + let filename = filename.to_os_string(); + match filename.into_string() { + Ok(s) => Ok(s), + Err(e) => Err(anyhow!("Error converting {:?} to String", path)), + } +} diff --git a/src/message.rs b/src/message.rs index 0e27365..2bc6f75 100644 --- a/src/message.rs +++ b/src/message.rs @@ -11,6 +11,7 @@ pub enum Message { FileOffer(FileOfferPayload), FileRequest(FileRequestPayload), FileTransfer(FileTransferPayload), + FileTransferComplete, } #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] diff --git a/src/ui.rs b/src/ui.rs index 9b94a41..eadc95e 100644 --- a/src/ui.rs +++ b/src/ui.rs @@ -26,7 +26,7 @@ pub async fn prompt_user_input( stdin: &mut FramedRead, file_offer: &FileOffer, ) -> Option { - let prompt_name = file_offer.path.file_name().unwrap(); + let prompt_name = &file_offer.path; println!( "Accept {:?}? ({:?}). (Y/n)", prompt_name, From 02f5fca5655cc24bf0cefadb86ca0378e47d5b88 Mon Sep 17 00:00:00 2001 From: Donald Knuth Date: Sat, 3 Sep 2022 15:12:47 -0400 Subject: [PATCH 2/3] Still broken --- src/client.rs | 18 +++++++++++------- src/file.rs | 1 + 2 files changed, 12 insertions(+), 7 deletions(-) diff --git a/src/client.rs b/src/client.rs index 052d405..8a023c5 100644 --- a/src/client.rs +++ b/src/client.rs @@ -43,9 +43,7 @@ pub async fn receive(password: &String) -> Result<()> { let (socket, key) = handshake.negotiate(socket, s1).await?; let mut connection = Connection::new(socket, key); // Wait for offered files, respond with desired files - let desired_files = request_specific_files(&mut connection).await?; - // Create files - let std_file_handles = create_or_find_files(desired_files).await?; + let std_file_handles = request_specific_files(&mut connection).await?; // Download them connection.download_files(std_file_handles).await?; return Ok(()); @@ -72,7 +70,7 @@ pub async fn offer_files( } } -pub async fn request_specific_files(conn: &mut Connection) -> Result> { +pub async fn request_specific_files(conn: &mut Connection) -> Result> { // Wait for offer message let offer_message = conn.await_msg().await?; let offered_files: Vec = match offer_message { @@ -81,17 +79,18 @@ pub async fn request_specific_files(conn: &mut Connection) -> Result) -> Result> { @@ -110,6 +109,11 @@ pub async fn create_or_find_files(desired_files: Vec) -> Result File::create(filename).await?, }; let metadata = file.metadata().await?; + println!( + "Current len: {:?}, Full Size: {:?}", + metadata.len(), + desired_file.size + ); let std_file_handle = StdFileHandle::new(desired_file.id, file, metadata.len(), desired_file.size).await?; v.push(std_file_handle) diff --git a/src/file.rs b/src/file.rs index dec835e..43ed94f 100644 --- a/src/file.rs +++ b/src/file.rs @@ -85,6 +85,7 @@ impl FileHandle { } async fn to_std(self, chunk_header: &ChunkHeader) -> Result { + println!("{:?} requested start?", chunk_header.start); StdFileHandle::new(self.id, self.file, chunk_header.start, self.md.len()).await } From 1c321301fd9f42eb4d78bf93600a89d2af4c3494 Mon Sep 17 00:00:00 2001 From: Donald Knuth Date: Sat, 3 Sep 2022 15:34:14 -0400 Subject: [PATCH 3/3] Cleanup --- README.md | 23 ++++++++++++++++------- src/client.rs | 19 +++++++++++++------ src/connection.rs | 18 +++++++++++------- src/file.rs | 25 ++++++++++++++++++------- src/handshake.rs | 2 +- 5 files changed, 59 insertions(+), 28 deletions(-) diff --git a/README.md b/README.md index 97d7408..843f9c6 100644 --- a/README.md +++ b/README.md @@ -2,11 +2,22 @@ `ruck` is a command line tool used for hosting relay servers and sending end-to-end encrypted files between clients. It was heavily inspired by [croc](https://github.com/schollz/croc), one of the easiest ways to send files between peers. This document describes the protocol `ruck` uses to support this functionality. -### Version +## Usage -This document refers to version `0.1.0` of `ruck` as defined by the `Cargo.toml` file. +```bash +## tab 1 +cargo run relay # this starts the server -## Server +## tab 2 +cargo run send /path/to/file1.md /path/to/file2.md supersecretpassword + +## tab 3 +cargo run receive supersecretpassword +``` + +## Protocol + +### Server The server in `ruck` exposes a TCP port. Its only functions are to staple connections and shuttle bytes between stapled connections. @@ -19,7 +30,7 @@ The time out is set to remove idle connections. The server does nothing else with the bytes, so the clients are free to end-to-end encrypt their messages. For this reason, updates to the `ruck` protocol do not typically necessitate server redeployments. -## Client +### Client There are two types of clients - `send` and `receive` clients. Out of band, the clients agree on a relay server and password, from which they can derive the 32 byte identifier used by the server to staple their connections. @@ -29,7 +40,5 @@ Once the handshake is complete, `send` and `receive` negotiate and exchange file - `send` offers a list of files and waits. - `receive` specifies which bytes it wants from these files. -- `send` sends the specified bytes and waits. -- `receive` sends heartbeats with progress updates. -- `send` hangs up once the heartbeats stop or received a successful heartbeat. +- `send` sends the specified bytes, then a completion message and hangs up. - `receive` hangs up once the downloads are complete. diff --git a/src/client.rs b/src/client.rs index 8a023c5..461c530 100644 --- a/src/client.rs +++ b/src/client.rs @@ -96,8 +96,8 @@ pub async fn request_specific_files(conn: &mut Connection) -> Result) -> Result> { let mut v = Vec::new(); for desired_file in desired_files { - let mut filename = desired_file.path; - filename.push_str(".part"); + let filename = desired_file.path; + // filename.push_str(".part"); let file = match File::open(filename.clone()).await { Ok(file) => { println!( @@ -106,16 +106,23 @@ pub async fn create_or_find_files(desired_files: Vec) -> Result File::create(filename).await?, + Err(_) => File::create(&filename).await?, }; let metadata = file.metadata().await?; println!( - "Current len: {:?}, Full Size: {:?}", + "File: {:?}. Current len: {:?}, Full Size: {:?}", + filename.clone(), metadata.len(), desired_file.size ); - let std_file_handle = - StdFileHandle::new(desired_file.id, file, metadata.len(), desired_file.size).await?; + let std_file_handle = StdFileHandle::new( + desired_file.id, + filename, + file, + metadata.len(), + desired_file.size, + ) + .await?; v.push(std_file_handle) } return Ok(v); diff --git a/src/connection.rs b/src/connection.rs index 6b91c36..a02caa4 100644 --- a/src/connection.rs +++ b/src/connection.rs @@ -81,12 +81,13 @@ impl Connection { let elapsed = before.elapsed(); let mb_sent = bytes_sent / 1_048_576; println!( - "{:?} mb sent, {:?} iterations. {:?} total time, {:?} avg per iteration, {:?} avg mb/sec", + "{:?}: {:?} mb sent (compressed), {:?} iterations. {:?} total time, {:?} avg per iteration, {:?} avg mb/sec", + handle.name, mb_sent, count, elapsed, elapsed / count, - mb_sent / elapsed.as_secs() + 1000 * mb_sent as u128 / elapsed.as_millis() ); Ok(()) } @@ -112,7 +113,6 @@ impl Connection { let msg = self.await_msg().await?; match msg { Message::FileTransfer(payload) => { - println!("In download"); if payload.chunk_header.id != handle.id { return Err(anyhow!("Wrong file")); } @@ -125,15 +125,19 @@ impl Connection { } } decoder.finish()?; - println!("Done downloading file."); - Connection::check_and_finish_download(clone, handle.size).await?; + println!("Done downloading {:?}.", handle.name); + Connection::check_and_finish_download(clone, handle.name, handle.size).await?; Ok(()) } - pub async fn check_and_finish_download(file: std::fs::File, size: u64) -> Result<()> { + pub async fn check_and_finish_download( + file: std::fs::File, + filename: String, + size: u64, + ) -> Result<()> { let metadata = file.metadata()?; if metadata.len() == size { - println!("File looks good."); + println!("OK: downloaded {:?} matches advertised size.", filename); return Ok(()); } return Err(anyhow!( diff --git a/src/file.rs b/src/file.rs index 43ed94f..ba47a6f 100644 --- a/src/file.rs +++ b/src/file.rs @@ -24,20 +24,25 @@ pub struct FileOffer { pub struct StdFileHandle { pub id: u8, + pub name: String, pub file: std::fs::File, pub start: u64, pub size: u64, } impl StdFileHandle { - pub async fn new(id: u8, file: File, start: u64, size: u64) -> Result { + pub async fn new( + id: u8, + name: String, + file: File, + start: u64, + size: u64, + ) -> Result { let mut file = file.into_std().await; file.seek(SeekFrom::Start(start))?; - if start != 0 { - println!("Seeking to {:?}", start); - } Ok(StdFileHandle { id, + name, file, start, size, @@ -85,8 +90,14 @@ impl FileHandle { } async fn to_std(self, chunk_header: &ChunkHeader) -> Result { - println!("{:?} requested start?", chunk_header.start); - StdFileHandle::new(self.id, self.file, chunk_header.start, self.md.len()).await + StdFileHandle::new( + self.id, + pathbuf_to_string(&self.path)?, + self.file, + chunk_header.start, + self.md.len(), + ) + .await } pub fn to_file_offer(&self) -> Result { @@ -130,6 +141,6 @@ pub fn pathbuf_to_string(path: &PathBuf) -> Result { let filename = filename.to_os_string(); match filename.into_string() { Ok(s) => Ok(s), - Err(e) => Err(anyhow!("Error converting {:?} to String", path)), + Err(_) => Err(anyhow!("Error converting {:?} to String", path)), } } diff --git a/src/handshake.rs b/src/handshake.rs index 1b1b409..c695d29 100644 --- a/src/handshake.rs +++ b/src/handshake.rs @@ -49,7 +49,7 @@ impl Handshake { ) -> Result<(TcpStream, Vec)> { let mut socket = socket; let bytes = &self.to_bytes(); - println!("client - sending handshake msg= {:?}", &bytes); + // println!("client - sending handshake msg= {:?}", &bytes); socket.write_all(&bytes).await?; let mut buffer = [0; HANDSHAKE_MSG_SIZE]; let n = socket.read_exact(&mut buffer).await?;