GameCult/ModelConnector
Watch
9
0
Fork
You've already forked ModelConnector
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Long request ids: the Malformed refusal echoes the id; the client refuses locally (long-id-refusal-unreadable)

Browse source 
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 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-10-28 06:40:18 +01:00
commit ec9722a2d8
3 changed files with 87 additions and 39 deletions
Download patch file Download diff file Expand all files Collapse all files

33
crates/model-connector/src/client.rs
View file

@ -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<TransportResult, ConnectorClientError> {
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"
);
}
}

19
crates/model-connector/src/serve.rs
View file

@ -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);
}

74
crates/model-connector/src/service.rs
View file

@ -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 =