From ec9722a2d82f5258e531cac18bf0fcc2753c1289 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Wed, 28 Oct 2026 06:40:18 +0100 Subject: [PATCH] Long request ids: the Malformed refusal echoes the id; the client refuses locally (long-id-refusal-unreadable) Naming a request by the SHA-256 of its id is a wire meaning v2 does not have: no v2 client could read that reply. begin now seals Refused(Malformed) with the request's own id and drops the connection (FrameBound) only when that sealed refusal cannot fit the frame. ConnectorClient refuses an id past MAX_IDENTITY_BYTES before connecting, with the typed RequestIdTooLong error. Co-Authored-By: Claude Opus 5.5 --- crates/model-connector/src/client.rs | 33 +++++++++++- crates/model-connector/src/serve.rs | 19 ++++--- crates/model-connector/src/service.rs | 74 +++++++++++++++++---------- 3 files changed, 87 insertions(+), 39 deletions(-) diff --git a/crates/model-connector/src/client.rs b/crates/model-connector/src/client.rs index b2997ce..a1cbcb3 100644 --- a/crates/model-connector/src/client.rs +++ b/crates/model-connector/src/client.rs @@ -6,8 +6,8 @@ use zeroize::Zeroizing; use crate::contract::{TransportInvocation, TransportKey, TransportResult}; use crate::envelope::{ - ServiceError, TransportFrameError, decrypt_result, encrypt_invocation, read_transport_frame, - write_transport_frame, + MAX_IDENTITY_BYTES, ServiceError, TransportFrameError, decrypt_result, encrypt_invocation, + read_transport_frame, write_transport_frame, }; #[derive(Clone)] @@ -48,6 +48,9 @@ impl ConnectorClient { &self, invocation: &TransportInvocation, ) -> Result { + if invocation.request_id().len() > MAX_IDENTITY_BYTES { + return Err(ConnectorClientError::RequestIdTooLong); + } let request = rmp_serde::to_vec(&encrypt_invocation(invocation, &self.security)?) .map_err(|_| ConnectorClientError::Encoding)?; let connect_timeout = self @@ -119,6 +122,8 @@ pub enum ConnectorClientError { FrameSize, #[error("connector MessagePack encoding failed")] Encoding, + #[error("request id is longer than the connector identity bound")] + RequestIdTooLong, #[error(transparent)] Transport(#[from] ServiceError), } @@ -174,4 +179,28 @@ mod fixture_tests { ); server.join().unwrap(); } + + #[test] + fn an_over_long_request_id_is_refused_locally_before_any_byte_is_sent() { + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + listener.set_nonblocking(true).unwrap(); + let client = ConnectorClient::new( + listener.local_addr().unwrap(), + "epiphany-distinct-test-key", + 64 * 1024, + Some(Duration::from_secs(1)), + ) + .unwrap(); + let mut long = invocation(); + long.request.request_id = "r".repeat(MAX_IDENTITY_BYTES + 1); + assert!(matches!( + client.execute(&long), + Err(ConnectorClientError::RequestIdTooLong) + )); + assert_eq!( + listener.accept().unwrap_err().kind(), + std::io::ErrorKind::WouldBlock, + "the client must not even connect" + ); + } } diff --git a/crates/model-connector/src/serve.rs b/crates/model-connector/src/serve.rs index 9e3c9ff..6bac7c1 100644 --- a/crates/model-connector/src/serve.rs +++ b/crates/model-connector/src/serve.rs @@ -521,40 +521,39 @@ mod tests { } #[test] - fn serve_refuses_a_long_request_id_on_the_wire_without_dropping_the_connection() { + fn serve_refuses_a_long_request_id_on_the_wire_echoing_the_id_without_dropping_the_connection() + { let store = TestStore::new(); let backend = CountingBackend::new(true); let (_client, endpoint) = start(backend.clone(), &store); let security = crate::fixtures::security(KEY); + // Read the way ConnectorClient reads a reply: decrypt_result, which + // requires the result to name the request by its own id. let send = |id: &str| { + let invocation = fresh(id); let request = rmp_serde::to_vec( - &crate::envelope::encrypt_invocation(&fresh(id), &security).unwrap(), + &crate::envelope::encrypt_invocation(&invocation, &security).unwrap(), ) .unwrap(); let mut stream = TcpStream::connect(endpoint).unwrap(); write_transport_frame(&mut stream, &request, FRAME).unwrap(); let reply = read_transport_frame(&mut stream, FRAME).unwrap(); let envelope: TransportEnvelope = rmp_serde::from_slice(&reply).unwrap(); - let result: TransportResult = rmp_serde::from_slice( - &crate::envelope::decrypt_message(&envelope, &security).unwrap(), - ) - .unwrap(); - (reply.len(), result) + crate::envelope::decrypt_result(&envelope, &security, &invocation).unwrap() }; - let (_, served) = send(&"i".repeat(crate::envelope::MAX_IDENTITY_BYTES)); + let served = send(&"i".repeat(crate::envelope::MAX_IDENTITY_BYTES)); assert!(matches!( served.disposition, TransportDisposition::Transported { .. } )); assert_eq!(backend.executions.load(Ordering::SeqCst), 1); for length in [257, 258, 5_000, 60_000] { - let (size, result) = send(&"i".repeat(length)); + let result = send(&"i".repeat(length)); assert_eq!( result.disposition, TransportDisposition::Refused(Refusal::Malformed), "{length}" ); - assert!(size <= smallest_frame(), "{length}: {size}"); } assert_eq!(backend.executions.load(Ordering::SeqCst), 1); } diff --git a/crates/model-connector/src/service.rs b/crates/model-connector/src/service.rs index a2a7768..e377c19 100644 --- a/crates/model-connector/src/service.rs +++ b/crates/model-connector/src/service.rs @@ -199,8 +199,9 @@ impl TransportService { /// fit, the same bound `complete` is given: a recorded result that no /// longer fits it (the service reopened at a smaller frame) is answered /// with the typed `result_size` failure, which is recorded in its place. - /// A request id longer than `MAX_IDENTITY_BYTES` is refused as malformed; - /// the refusal names the id by its digest so it fits any legal frame. + /// A request id longer than `MAX_IDENTITY_BYTES` is refused as malformed, + /// the refusal echoing the id like every other. The connection is dropped + /// (`FrameBound`) only when that sealed refusal does not fit `max_frame_bytes`. pub fn begin( &mut self, envelope: &TransportEnvelope, @@ -216,10 +217,15 @@ impl TransportService { let max_concurrent_requests = caller.max_concurrent_requests; let invocation = decrypt_invocation(envelope, &security)?; if envelope.request_id.len() > MAX_IDENTITY_BYTES { - return Ok(TransportAdmission::Reply(encrypt_result( - &unnamed_refusal(&invocation, Refusal::Malformed), + let refusal = encrypt_result( + &TransportResult::refused(&invocation, Refusal::Malformed), &security, - )?)); + )?; + return if fits_frame(&refusal, max_frame_bytes)? { + Ok(TransportAdmission::Reply(refusal)) + } else { + Err(ServiceError::FrameBound) + }; } if envelope.ciphertext.len() > caller.max_payload_bytes { return self.reply_refusal(&invocation, Refusal::Policy, &security); @@ -477,16 +483,6 @@ fn hex(digest: [u8; 32]) -> String { text } -/// A refusal for an invocation whose request id is too long to echo: it names -/// the request by the hex SHA-256 of the id (64 bytes), so the reply fits the -/// smallest legal frame whatever the id's length. -fn unnamed_refusal(invocation: &TransportInvocation, reason: Refusal) -> TransportResult { - TransportResult { - request_id: hex(Sha256::digest(invocation.request_id().as_bytes()).into()), - ..TransportResult::refused(invocation, reason) - } -} - fn replay_store_error(error: impl std::fmt::Display) -> ServiceError { ServiceError::ReplayStore(error.to_string()) } @@ -989,42 +985,66 @@ mod tests { } #[test] - fn a_request_id_past_the_bound_is_refused_malformed_in_a_reply_that_fits_the_smallest_frame() { - let smallest = frame_bytes_for_payload(4096); + fn a_request_id_past_the_bound_is_refused_malformed_echoing_the_id() { for length in [ MAX_IDENTITY_BYTES + 1, MAX_IDENTITY_BYTES + 2, 5_000, 100_000, - 400_000, ] { let store = TestStore::new(); let mut service = open_service(&store, 1); let id = "i".repeat(length); let long = invocation_for(CALLER, &id); - let TransportAdmission::Reply(reply) = service - .begin(&raw_envelope(&long), 1_000, smallest) - .unwrap() + let TransportAdmission::Reply(reply) = + service.begin(&raw_envelope(&long), 1_000, FRAME).unwrap() else { panic!("an over-long id must be refused, not executed ({length})"); }; - assert!( - rmp_serde::to_vec(&reply).unwrap().len() <= smallest, - "{length}" - ); + assert_eq!(reply.request_id, id, "{length}"); let result: TransportResult = rmp_serde::from_slice(&decrypt_message(&reply, &security(KEY)).unwrap()).unwrap(); assert_eq!( result.disposition, TransportDisposition::Refused(Refusal::Malformed) ); - assert_eq!(result.request_id, reply.request_id); - assert_eq!(result.request_id.len(), 64, "the id is named, not echoed"); + assert_eq!(result.request_id, id); + result.validate_against(&long).unwrap(); // Nothing was recorded or held: the caller's one slot is free. let (_claim, _) = claim_for(&mut service, "after-the-long-one"); } } + #[test] + fn an_over_long_id_drops_the_connection_only_when_its_refusal_cannot_fit() { + let id = "i".repeat(5_000); + let long = invocation_for(CALLER, &id); + let sealed = encrypt_result( + &TransportResult::refused(&long, Refusal::Malformed), + &security(KEY), + ) + .unwrap(); + // The sealed size varies with the random nonce and ciphertext bytes + // (a byte of 128 or more takes two), so bracket it: the ciphertext + // plus the id is a floor, twice the ciphertext plus the id and the + // envelope's other fields is a ceiling. + let floor = sealed.ciphertext.len() + id.len(); + let ceiling = 2 * sealed.ciphertext.len() + id.len() + 512; + let store = TestStore::new(); + let mut service = open_service(&store, 1); + assert!(matches!( + service.begin(&raw_envelope(&long), 1_000, ceiling).unwrap(), + TransportAdmission::Reply(_) + )); + assert_eq!( + service + .begin(&raw_envelope(&long), 1_000, floor - 1) + .unwrap_err(), + ServiceError::FrameBound + ); + let (_claim, _) = claim_for(&mut service, "after-the-drop"); + } + #[test] fn a_frame_check_is_inclusive_of_the_serialized_envelope() { let envelope =