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 =