From d053061d5a988674925ffdaf5624ef049fb6b9ef Mon Sep 17 00:00:00 2001 From: David Stainton Date: Sat, 4 Jul 2026 21:37:29 +0000 Subject: [PATCH] Remove the SACK write_stream/read_stream API Mirrors the katzenpost daemon removal in PR #1053: the windowed selective-ack streaming let a courier link the boxes of one transfer, undermining the very unlinkability it exists to preserve. Drop write_stream/read_stream from the Rust and Python clients and the --sack option from pigeonhole-cp (its default copy and per-box streaming paths are untouched), and repin the docker integration to the katzenpost commit that removed the daemon side. --- .github/workflows/test-integration-docker.yml | 2 +- katzenpost_thinclient/__init__.py | 4 - katzenpost_thinclient/core.py | 4 +- katzenpost_thinclient/pigeonhole.py | 79 ------- src/bin/pigeonhole_cp.rs | 216 +----------------- src/error.rs | 4 +- src/pigeonhole.rs | 162 ------------- 7 files changed, 16 insertions(+), 455 deletions(-) diff --git a/.github/workflows/test-integration-docker.yml b/.github/workflows/test-integration-docker.yml index 7d97e7a..b4b21d0 100644 --- a/.github/workflows/test-integration-docker.yml +++ b/.github/workflows/test-integration-docker.yml @@ -22,7 +22,7 @@ jobs: uses: actions/checkout@v4 with: repository: katzenpost/katzenpost - ref: 86d4dbc5f11d9162d52b371708f41c937566cf7f # courier/replica error-code disambiguation + ref: cdb88125ec6d3c54cc9f5413af39967edb9a51b4 # PR #1053: Poisson-gate ARQ follow-ups + remove SACK path: katzenpost - name: Set up Docker Buildx diff --git a/katzenpost_thinclient/__init__.py b/katzenpost_thinclient/__init__.py index 7c7ccd8..78cbd4e 100644 --- a/katzenpost_thinclient/__init__.py +++ b/katzenpost_thinclient/__init__.py @@ -150,8 +150,6 @@ async def main(): start_resending_encrypted_message, start_resending_encrypted_message_return_box_exists, start_resending_encrypted_message_no_retry, - write_stream, - read_stream, cancel_resending_encrypted_message, next_message_box_index, get_message_box_index_counter, @@ -187,8 +185,6 @@ async def main(): ThinClient.start_resending_encrypted_message = start_resending_encrypted_message ThinClient.start_resending_encrypted_message_return_box_exists = start_resending_encrypted_message_return_box_exists ThinClient.start_resending_encrypted_message_no_retry = start_resending_encrypted_message_no_retry -ThinClient.write_stream = write_stream -ThinClient.read_stream = read_stream ThinClient.cancel_resending_encrypted_message = cancel_resending_encrypted_message ThinClient.next_message_box_index = next_message_box_index ThinClient.get_message_box_index_counter = get_message_box_index_counter diff --git a/katzenpost_thinclient/core.py b/katzenpost_thinclient/core.py index fd0b88f..c043ad9 100644 --- a/katzenpost_thinclient/core.py +++ b/katzenpost_thinclient/core.py @@ -240,8 +240,8 @@ def __init__(self, replica_error_code: int = 0, failed_envelope_index: int = 0) class PayloadTooLargeError(Exception): - """A WriteStream plaintext or a ReadStream result exceeded the daemon's - configured maximum stream payload size.""" + """A request's payload exceeded the daemon's configured maximum payload + size.""" pass diff --git a/katzenpost_thinclient/pigeonhole.py b/katzenpost_thinclient/pigeonhole.py index 3c4ada5..2271133 100644 --- a/katzenpost_thinclient/pigeonhole.py +++ b/katzenpost_thinclient/pigeonhole.py @@ -404,85 +404,6 @@ async def start_resending_encrypted_message( ) -async def write_stream(self, write_cap, start_index, payload, window=0): - """ - Writes a whole payload, of any size, to a channel using the daemon's - windowed selective-ack (SACK) ARQ. The daemon splits the payload into as - many BACAP boxes as it spans and keeps up to ``window`` boxes in flight at - once, retransmitting only those whose acknowledgements time out, so a - multi-box payload is no longer serialised one round trip per box. A - ``window`` of zero asks the daemon to choose a default. - - The daemon does all chunking and encryption; the caller supplies only the - cleartext payload, the write capability, and the start index. - - Args: - write_cap: Write capability for the destination channel. - start_index: Message box index of the first box written. - payload: Cleartext payload to write. - window: Maximum boxes in flight at once (0 = daemon default). - - Returns: - The message box index immediately after the last box written. - """ - query_id = self.new_query_id() - request = { - "write_stream": { - "query_id": query_id, - "write_cap": write_cap, - "start_index": start_index, - "payload": payload, - "window": window, - } - } - reply = await self._send_and_wait(query_id=query_id, request=request) - error_code = reply.get("error_code", 0) - if error_code != THIN_CLIENT_SUCCESS: - exc = error_code_to_exception(error_code) - if exc: - raise exc - raise Exception(f"write_stream failed: {thin_client_error_to_string(error_code)}") - return reply.get("next_message_box_index") - - -async def read_stream(self, read_cap, start_index, box_count, window=0): - """ - Reads ``box_count`` sequential boxes from a channel using the daemon's - windowed selective-ack (SACK) ARQ, the read counterpart of - ``write_stream``. The daemon keeps up to ``window`` boxes in flight, - decrypts each, and reassembles them in order. A ``window`` of zero asks the - daemon to choose a default. - - Args: - read_cap: Read capability for the source channel. - start_index: Message box index of the first box read. - box_count: Number of sequential boxes to read. - window: Maximum boxes in flight at once (0 = daemon default). - - Returns: - A tuple ``(payload, next_message_box_index)``: the concatenation of the - decrypted boxes in order, and the index immediately after the last box. - """ - query_id = self.new_query_id() - request = { - "read_stream": { - "query_id": query_id, - "read_cap": read_cap, - "start_index": start_index, - "box_count": box_count, - "window": window, - } - } - reply = await self._send_and_wait(query_id=query_id, request=request) - error_code = reply.get("error_code", 0) - if error_code != THIN_CLIENT_SUCCESS: - exc = error_code_to_exception(error_code) - if exc: - raise exc - raise Exception(f"read_stream failed: {thin_client_error_to_string(error_code)}") - return reply.get("payload", b""), reply.get("next_message_box_index") - - async def start_resending_encrypted_message_return_box_exists( self, read_cap: "bytes|None", diff --git a/src/bin/pigeonhole_cp.rs b/src/bin/pigeonhole_cp.rs index 2d31e76..5f41390 100644 --- a/src/bin/pigeonhole_cp.rs +++ b/src/bin/pigeonhole_cp.rs @@ -19,12 +19,6 @@ use katzenpost_thin_client::{Config, ThinClient}; const MAX_NAME_LEN: usize = 255; -/// Number of boxes the SACK paths read or write per `read_stream` / -/// `write_stream` call. The windowed ARQ keeps this many boxes in flight at -/// once, and it bounds the memory either SACK direction holds: a block, not -/// the whole file. -const SACK_BLOCK_BOXES: usize = 10; - #[derive(Debug, thiserror::Error)] pub enum FileNameError { #[error("path has no file name component")] @@ -141,14 +135,6 @@ enum Commands { /// --no-copy streams the file a box at a time and has no such limit. #[arg(long)] no_copy: bool, - - /// Use the windowed SACK ARQ to write, keeping a block of boxes in - /// flight at once instead of the default per-box stop-and-wait. The - /// window is computed automatically by the daemon from the PKI - /// document (routing layers and Mu). It streams the file a block at - /// a time, so it does not load the whole file into memory. - #[arg(long)] - sack: bool, }, /// Read from a Pigeonhole channel and write to a file @@ -168,13 +154,6 @@ enum Commands { /// Output directory (file name comes from the FileMetaData header) #[arg(short, long)] dest_dir: PathBuf, - - /// Use the windowed SACK ARQ to read the payload, keeping many - /// boxes in flight at once instead of reading one box per round - /// trip. The window is computed automatically by the daemon from - /// the PKI document (routing layers and Mu). - #[arg(long)] - sack: bool, }, } @@ -185,11 +164,11 @@ async fn main() -> Result<(), Box> { match cli.command { Commands::Genkey { config } => run_genkey(config).await, - Commands::Send { config, write_cap, index, file, no_copy, sack } => { - run_send(config, write_cap, index, file, !no_copy, sack).await + Commands::Send { config, write_cap, index, file, no_copy } => { + run_send(config, write_cap, index, file, !no_copy).await } - Commands::Receive { config, read_cap, index, dest_dir, sack } => { - run_receive(config, read_cap, index, dest_dir, sack).await + Commands::Receive { config, read_cap, index, dest_dir } => { + run_receive(config, read_cap, index, dest_dir).await } } } @@ -282,16 +261,13 @@ async fn run_genkey(config: PathBuf) -> Result<(), Box> { /// channel and dispatches its contents to the destination atomically (but /// buffers the whole file). `--no-copy` opts out, writing each box to the /// destination directly via per-box ARQ, reading the file one box at a -/// time so it never holds the whole file in memory. `--sack` opts into the -/// windowed ARQ, which streams the file a block of boxes at a time and so -/// also avoids loading it whole. +/// time so it never holds the whole file in memory. async fn run_send( config: PathBuf, write_cap_b64: String, next_index_b64: String, input_file: PathBuf, copy: bool, - sack: bool, ) -> Result<(), Box> { let write_cap = BASE64.decode(&write_cap_b64)?; let next_index = BASE64.decode(&next_index_b64)?; @@ -309,117 +285,12 @@ async fn run_send( // The default uses the atomic courier Copy command (which buffers the // whole file). --no-copy streams each box directly and never loads the - // whole file; --sack chooses the windowed ARQ, which streams the file a - // block of boxes at a time. - match (copy, sack) { - (false, false) => send_direct(&pigeonhole, &write_cap, &next_index, &input_file, total_len, &header).await, - (false, true) => send_sack(&pigeonhole, &write_cap, &next_index, &input_file, total_len, &header).await, - (true, false) => send_copy(&pigeonhole, &write_cap, &next_index, &input_file, total_len, &header).await, - (true, true) => send_sack_copy(&pigeonhole, &write_cap, &next_index, &input_file, total_len, &header).await, - } -} - -/// SACK + COPY path: stage the payload into a temporary channel using the -/// windowed SACK ARQ, then issue the courier Copy command to dispatch it to -/// the destination atomically. The temporary channel is just a BACAP byte -/// stream of copy-stream elements, so filling it with `write_stream` produces -/// the same stream the per-box `send_copy` builds, only windowed. -async fn send_sack_copy( - pigeonhole: &PigeonholeClient, - dest_write_cap: &[u8], - dest_index: &[u8], - input_file: &Path, - total_len: u64, - header: &[u8], -) -> Result<(), Box> { - let mut payload = Vec::with_capacity(header.len() + total_len as usize); - payload.extend_from_slice(header); - File::open(input_file)?.read_to_end(&mut payload)?; - - let client = pigeonhole.thin_client(); - - // A fresh temporary channel to stage the copy stream. - let mut seed = [0u8; 32]; - rand::thread_rng().fill_bytes(&mut seed); - let kp = client.new_keypair(&seed).await?; - - // Encode the payload into copy-stream elements addressed at the - // destination, then concatenate them into the byte stream the temp - // channel must carry. - let result = client - .create_courier_envelopes_from_payload(&payload, dest_write_cap, dest_index, true, true) - .await?; - let element_count = result.envelopes.len(); - let mut stream = Vec::new(); - for chunk in &result.envelopes { - stream.extend_from_slice(chunk); + // whole file. + if copy { + send_copy(&pigeonhole, &write_cap, &next_index, &input_file, total_len, &header).await + } else { + send_direct(&pigeonhole, &write_cap, &next_index, &input_file, total_len, &header).await } - - let box_payload_size = client.pigeonhole_geometry().max_plaintext_payload_length; - let boxes = stream.len().div_ceil(box_payload_size - 4); - - let start = std::time::Instant::now(); - client - .write_stream(&kp.write_cap, &kp.first_message_index, &stream, 0) - .await?; - client - .start_resending_copy_command(&kp.write_cap, None, None) - .await?; - print_throughput("sack-copy", total_len, boxes, start.elapsed()); - println!("(staged {} copy-stream elements via temp channel, then COPY)", element_count); - Ok(()) -} - -/// SACK path: write the file with the daemon's windowed selective-ack ARQ, -/// a block of `SACK_BLOCK_BOXES` boxes at a time. Each `write_stream` call -/// keeps that block's boxes in flight at once, and reading the file block by -/// block means we never hold more than a block in memory. Every block but -/// the last is an exact multiple of `per_box`, so the box boundaries match -/// what a single whole-file `write_stream` would have produced, and the -/// `receive` side reads back the same stream either way. -async fn send_sack( - pigeonhole: &PigeonholeClient, - write_cap: &[u8], - next_index: &[u8], - input_file: &Path, - total_len: u64, - header: &[u8], -) -> Result<(), Box> { - let per_box = pigeonhole - .thin_client() - .pigeonhole_geometry() - .max_plaintext_payload_length - - 4; - let block_len = per_box * SACK_BLOCK_BOXES; - - let mut input_reader = BufReader::new(File::open(input_file)?); - let mut index = next_index.to_vec(); - let mut block: Vec = Vec::with_capacity(block_len); - block.extend_from_slice(header); - let mut boxes = 0usize; - - let start = std::time::Instant::now(); - loop { - let want = block_len - block.len(); - let mut buf = vec![0u8; want]; - let n = read_fill(&mut input_reader, &mut buf)?; - block.extend_from_slice(&buf[..n]); - let eof = n < want; - if block.is_empty() { - break; - } - index = pigeonhole - .thin_client() - .write_stream(write_cap, &index, &block, 0) - .await?; - boxes += block.len().div_ceil(per_box); - block.clear(); - if eof { - break; - } - } - print_throughput("sack", total_len, boxes, start.elapsed()); - Ok(()) } /// Direct path: write each box to the destination via per-box ARQ. @@ -519,7 +390,7 @@ async fn send_copy( /// Print a transfer's throughput: total bytes and boxes, the wall time /// spent in the send, and the derived boxes/sec and bytes/sec. Used to -/// compare ARQ strategies (per-box, copy, SACK) on equal footing. +/// compare ARQ strategies (per-box, copy) on equal footing. fn print_throughput(mode: &str, bytes: u64, boxes: usize, elapsed: std::time::Duration) { let secs = elapsed.as_secs_f64(); let boxes_per_sec = if secs > 0.0 { boxes as f64 / secs } else { 0.0 }; @@ -559,17 +430,12 @@ async fn run_receive( read_cap_b64: String, next_index_b64: String, dest_dir: PathBuf, - sack: bool, ) -> Result<(), Box> { let read_cap = BASE64.decode(&read_cap_b64)?; let next_index = BASE64.decode(&next_index_b64)?; let client = init_client(config).await?; - if sack { - return run_receive_sack(client, &read_cap, &next_index, dest_dir).await; - } - let pigeonhole = PigeonholeClient::new_in_memory(client.clone())?; let mut reader = pigeonhole.load_read_channel("pigeonhole-cp", &read_cap, &next_index)?; @@ -616,63 +482,3 @@ async fn run_receive( ); Ok(()) } - -/// SACK receive: the windowed counterpart to `run_receive`. It first reads -/// box zero to recover the file size from its `FileMetaData` header and -/// compute how many boxes the payload spans (the daemon chunks a write at -/// `MaxPlaintextPayloadLength - 4` per box), then reads the remainder a -/// block of `SACK_BLOCK_BOXES` boxes at a time, writing each block straight -/// to the temp file so we never hold more than a block in memory. The daemon -/// computes the window itself from the PKI document. -async fn run_receive_sack( - client: Arc, - read_cap: &[u8], - start_index: &[u8], - dest_dir: PathBuf, -) -> Result<(), Box> { - let start = std::time::Instant::now(); - - let (first, mut index) = client.read_stream(read_cap, start_index, 1, 0).await?; - let mut deserializer = serde_cbor::Deserializer::from_slice(&first); - let metadata = FileMetaData::deserialize(&mut deserializer)?; - let header_end = deserializer.byte_offset(); - - let per_box = client.pigeonhole_geometry().max_plaintext_payload_length - 4; - let total_payload_len = header_end + metadata.size as usize; - let total_boxes = total_payload_len.div_ceil(per_box); - - let final_path = sanitize_for_receive(&metadata.name, &dest_dir)?; - let parent = final_path - .parent() - .expect("sanitize_for_receive guarantees a parent"); - let mut tmp = NamedTempFile::new_in(parent)?; - - let mut remaining = metadata.size; - let first_file = &first[header_end..]; - let take = (first_file.len() as u64).min(remaining) as usize; - tmp.write_all(&first_file[..take])?; - remaining -= take as u64; - - let mut boxes_read = 1usize; - while boxes_read < total_boxes { - let want = (total_boxes - boxes_read).min(SACK_BLOCK_BOXES); - let (chunk, next) = client.read_stream(read_cap, &index, want as u32, 0).await?; - index = next; - let take = (chunk.len() as u64).min(remaining) as usize; - tmp.write_all(&chunk[..take])?; - remaining -= take as u64; - boxes_read += want; - } - - tmp.as_file().sync_all()?; - tmp.persist_noclobber(&final_path).map_err(|e| e.error)?; - - println!( - "received {} bytes in {} box(es) (sack) in {:.3}s -> {}", - metadata.size, - total_boxes, - start.elapsed().as_secs_f64(), - final_path.display() - ); - Ok(()) -} diff --git a/src/error.rs b/src/error.rs index f559913..6ca1654 100644 --- a/src/error.rs +++ b/src/error.rs @@ -71,8 +71,8 @@ pub enum ThinClientError { failed_envelope_index: u64, }, - /// A WriteStream plaintext or a ReadStream result exceeded the daemon's - /// configured maximum stream payload size (error code 27). + /// A request's payload exceeded the daemon's configured maximum payload + /// size (error code 27). PayloadTooLarge, /// A Contact Voucher payload did not hash to the Voucher token handed diff --git a/src/pigeonhole.rs b/src/pigeonhole.rs index c58a269..cdca65e 100644 --- a/src/pigeonhole.rs +++ b/src/pigeonhole.rs @@ -180,57 +180,6 @@ pub struct StartResendingResult { pub courier_queue_id: Option>, } -/// Request to write a whole multi-box payload via the windowed SACK ARQ. -#[derive(Debug, Clone, serde::Serialize)] -struct WriteStreamRequest { - #[serde(with = "serde_bytes")] - query_id: Vec, - #[serde(with = "serde_bytes")] - write_cap: Vec, - #[serde(with = "serde_bytes")] - start_index: Vec, - #[serde(with = "serde_bytes")] - payload: Vec, - window: i64, -} - -/// Reply to a WriteStream request. -#[derive(Debug, Clone, serde::Deserialize)] -struct WriteStreamReply { - #[serde(default, with = "optional_bytes")] - next_message_box_index: Option>, - #[serde(default)] - error_code: u8, - #[serde(default)] - box_count: u32, -} - -/// Request to read many sequential boxes via the windowed SACK ARQ. -#[derive(Debug, Clone, serde::Serialize)] -struct ReadStreamRequest { - #[serde(with = "serde_bytes")] - query_id: Vec, - #[serde(with = "serde_bytes")] - read_cap: Vec, - #[serde(with = "serde_bytes")] - start_index: Vec, - box_count: u32, - window: i64, -} - -/// Reply to a ReadStream request. -#[derive(Debug, Clone, serde::Deserialize)] -struct ReadStreamReply { - #[serde(default, with = "optional_bytes")] - payload: Option>, - #[serde(default, with = "optional_bytes")] - next_message_box_index: Option>, - #[serde(default)] - error_code: u8, - #[serde(default)] - box_count: u32, -} - /// Request to cancel resending an encrypted message. #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] struct CancelResendingEncryptedMessageRequest { @@ -1051,87 +1000,6 @@ impl ThinClient { }) } - /// Writes a whole payload, of any size, to a channel using the daemon's - /// windowed selective-ack (SACK) ARQ. The daemon splits the payload into - /// as many BACAP boxes as it spans and keeps up to `window` boxes in - /// flight at once, retransmitting only those whose acknowledgements time - /// out, so a multi-box payload is no longer serialised one round trip per - /// box. A `window` of zero asks the daemon to choose a default. - /// - /// The daemon does all chunking and encryption; the caller supplies only - /// the cleartext payload, the write capability, and the start index. - /// Returns the message box index immediately after the last box written. - pub async fn write_stream( - &self, - write_cap: &[u8], - start_index: &[u8], - payload: &[u8], - window: i64, - ) -> Result, ThinClientError> { - let query_id = Self::new_query_id(); - let request_inner = WriteStreamRequest { - query_id: query_id.clone(), - write_cap: write_cap.to_vec(), - start_index: start_index.to_vec(), - payload: payload.to_vec(), - window, - }; - let request_value = - serde_cbor::value::to_value(&request_inner).map_err(|e| ThinClientError::CborError(e))?; - let mut request = BTreeMap::new(); - request.insert(Value::Text("write_stream".to_string()), request_value); - - let reply_map = self.send_and_wait_direct(query_id, request).await?; - let reply: WriteStreamReply = serde_cbor::value::from_value(Value::Map(reply_map)) - .map_err(|e| ThinClientError::CborError(e))?; - - debug!("write_stream: received reply, error_code={}, boxes={}", reply.error_code, reply.box_count); - if reply.error_code != 0 { - return Err(error_code_to_error(reply.error_code)); - } - Ok(reply.next_message_box_index.unwrap_or_default()) - } - - /// Reads `box_count` sequential boxes from a channel using the daemon's - /// windowed selective-ack (SACK) ARQ, the read counterpart of - /// `write_stream`. The daemon keeps up to `window` boxes in flight, - /// decrypts each, and reassembles them in order. A `window` of zero asks - /// the daemon to choose a default. Returns the concatenated payload and the - /// message box index immediately after the last box read. - pub async fn read_stream( - &self, - read_cap: &[u8], - start_index: &[u8], - box_count: u32, - window: i64, - ) -> Result<(Vec, Vec), ThinClientError> { - let query_id = Self::new_query_id(); - let request_inner = ReadStreamRequest { - query_id: query_id.clone(), - read_cap: read_cap.to_vec(), - start_index: start_index.to_vec(), - box_count, - window, - }; - let request_value = - serde_cbor::value::to_value(&request_inner).map_err(|e| ThinClientError::CborError(e))?; - let mut request = BTreeMap::new(); - request.insert(Value::Text("read_stream".to_string()), request_value); - - let reply_map = self.send_and_wait_direct(query_id, request).await?; - let reply: ReadStreamReply = serde_cbor::value::from_value(Value::Map(reply_map)) - .map_err(|e| ThinClientError::CborError(e))?; - - debug!("read_stream: received reply, error_code={}, boxes={}", reply.error_code, reply.box_count); - if reply.error_code != 0 { - return Err(error_code_to_error(reply.error_code)); - } - Ok(( - reply.payload.unwrap_or_default(), - reply.next_message_box_index.unwrap_or_default(), - )) - } - /// Cancels ARQ resending for an encrypted message. /// /// This method stops the automatic repeat request for a previously started @@ -1844,36 +1712,6 @@ mod sack_request_tests { // The daemon decodes these requests by CBOR field name; a rename here that // drifts from the Go `cbor:"..."` tags would silently break interop, so we // pin the wire field names. - #[test] - fn write_stream_request_field_names() { - let req = WriteStreamRequest { - query_id: vec![1, 2, 3], - write_cap: vec![4, 5], - start_index: vec![6, 7], - payload: vec![8, 9], - window: 16, - }; - let keys = map_keys(serde_cbor::value::to_value(&req).unwrap()); - for expected in ["query_id", "write_cap", "start_index", "payload", "window"] { - assert!(keys.iter().any(|k| k == expected), "missing field {expected}, got {keys:?}"); - } - } - - #[test] - fn read_stream_request_field_names() { - let req = ReadStreamRequest { - query_id: vec![1], - read_cap: vec![2], - start_index: vec![3], - box_count: 4, - window: 0, - }; - let keys = map_keys(serde_cbor::value::to_value(&req).unwrap()); - for expected in ["query_id", "read_cap", "start_index", "box_count", "window"] { - assert!(keys.iter().any(|k| k == expected), "missing field {expected}, got {keys:?}"); - } - } - #[test] fn voucher_mint_request_field_names() { let req = VoucherMintRequest {