diff --git a/crates/claude-connector/src/backend/tests.rs b/crates/claude-connector/src/backend/tests.rs index 1ee69dc..009775a 100644 --- a/crates/claude-connector/src/backend/tests.rs +++ b/crates/claude-connector/src/backend/tests.rs @@ -1956,7 +1956,7 @@ fn rate_limit_event_updates_quota_cell() { #[test] fn a_rejected_turn_still_reports_its_windows_and_a_bad_window_costs_only_itself() { - let event = r#"{"type":"rate_limit_event","rate_limit_info":{"status":"rejected","unifiedWindows":{"five_hour":{"utilization":0.9996,"resetsAt":1793307000,"status":"rejected"},"seven_day":{"resetsAt":1793656800,"status":"allowed"},"odd":{"utilization":"high","resetsAt":1,"status":"allowed"},"negative":{"utilization":-0.5,"resetsAt":1,"status":"allowed"}}}}"#; + let event = r#"{"type":"rate_limit_event","rate_limit_info":{"status":"rejected","unifiedWindows":{"five_hour":{"utilization":0.9996,"resetsAt":1793307000,"status":"rejected"},"seven_day":{"resetsAt":1793656800,"status":"allowed"},"idle":{"utilization":0,"resetsAt":1793307000,"status":"allowed"},"odd":{"utilization":"high","resetsAt":1,"status":"allowed"},"negative":{"utilization":-0.5,"resetsAt":1,"status":"allowed"}}}}"#; let rig = rig(vec![Script::lines(vec![ init_line(MODEL, SESSION), event.to_string(), @@ -1968,7 +1968,11 @@ fn a_rejected_turn_still_reports_its_windows_and_a_bad_window_costs_only_itself( rejected.status = "rejected".to_string(); // Unknown is never zero: a window missing a field or of another type is // skipped, not read as empty. - assert_eq!(quota_of(&rig).windows, vec![rejected]); + // A window at exactly zero use is a reading, and is kept. + assert_eq!( + quota_of(&rig).windows, + vec![rejected, window("idle", 0, 1_793_307_000_000)] + ); } #[test] @@ -2261,7 +2265,7 @@ mod real { } #[test] - fn a_token_too_large_for_a_pipe_is_refused_not_hung() { + fn a_token_too_large_for_a_pipe_is_refused_at_its_bound_not_hung() { let fake = fake("exit 0", limits()); let config = config(fake.dir.path(), INSTANCE, limits()); let env = child_env(&config, &request(vec![user("x")])); @@ -2273,9 +2277,13 @@ mod real { .to_string_lossy() .into_owned(), ]; - let huge = "t".repeat(1 << 20); - let refused = port.spawn(&argv, &env, fake.dir.path(), Vec::new(), &huge); - assert!(refused.is_err()); + let spawn = |token: &str| port.spawn(&argv, &env, fake.dir.path(), Vec::new(), token); + let at_bound = spawn(&"t".repeat(4096)).expect("4096 bytes fit the bound"); + drop(at_bound); + assert_eq!( + spawn(&"t".repeat(4097)).err().map(|e| e.kind()), + Some(io::ErrorKind::InvalidInput) + ); } #[test] diff --git a/crates/claude-connector/src/process.rs b/crates/claude-connector/src/process.rs index 1b35fc4..af9bd13 100644 --- a/crates/claude-connector/src/process.rs +++ b/crates/claude-connector/src/process.rs @@ -67,21 +67,30 @@ mod unix { /// hundred bytes, so anything near the pipe's capacity is not one. const MAX_TOKEN_BYTES: usize = 4096; - /// Puts the pipe's read end at [`TOKEN_FD`] in the child. Runs between - /// fork and exec, so it calls nothing but `dup2` and `fcntl`. - fn place_token_fd(source: RawFd) -> io::Result<()> { + fn cvt(result: libc::c_int) -> io::Result { + if result < 0 { + Err(io::Error::last_os_error()) + } else { + Ok(result) + } + } + + /// Makes `source` also descriptor `target`, open across exec. Runs between + /// fork and exec in the child, so it calls nothing but `dup2` and `fcntl`. + fn place_fd(source: RawFd, target: RawFd) -> io::Result<()> { // SAFETY: dup2 and fcntl take integers only and are async-signal-safe. unsafe { - if source == TOKEN_FD { + if source == target { // dup2 onto itself changes nothing, so the close-on-exec flag // the pipe was created with must be cleared by hand. - let flags = libc::fcntl(TOKEN_FD, libc::F_GETFD); - if flags < 0 || libc::fcntl(TOKEN_FD, libc::F_SETFD, flags & !libc::FD_CLOEXEC) < 0 - { - return Err(io::Error::last_os_error()); - } - } else if libc::dup2(source, TOKEN_FD) < 0 { - return Err(io::Error::last_os_error()); + let flags = cvt(libc::fcntl(target, libc::F_GETFD))?; + cvt(libc::fcntl( + target, + libc::F_SETFD, + flags & !libc::FD_CLOEXEC, + ))?; + } else { + cvt(libc::dup2(source, target))?; } } Ok(()) @@ -126,13 +135,13 @@ mod unix { .process_group(0); // The write end closes here, so the child sees the token and then // the end of the pipe. Both ends are close-on-exec; only the copy - // `place_token_fd` makes reaches the child. + // `place_fd` makes reaches the child. let (reader, mut writer) = std::io::pipe()?; writer.write_all(token.as_bytes())?; drop(writer); let source = reader.as_raw_fd(); // SAFETY: the closure calls only async-signal-safe functions. - unsafe { command.pre_exec(move || place_token_fd(source)) }; + unsafe { command.pre_exec(move || place_fd(source, TOKEN_FD)) }; let mut child = command.spawn()?; drop(reader); let (stdin_pipe, stdout, stderr) = @@ -271,4 +280,70 @@ mod unix { self.wait(); } } + + #[cfg(test)] + mod tests { + use super::*; + + fn cloexec(fd: RawFd) -> bool { + // SAFETY: fcntl on an integer descriptor. + let flags = unsafe { libc::fcntl(fd, libc::F_GETFD) }; + assert!(flags >= 0, "descriptor {fd} is not open"); + flags & libc::FD_CLOEXEC != 0 + } + + /// A descriptor at or above 300 on the same pipe, so a test never + /// touches a low number another thread may be using. + fn high_dup(source: RawFd, close_on_exec: bool) -> RawFd { + let command = if close_on_exec { + libc::F_DUPFD_CLOEXEC + } else { + libc::F_DUPFD + }; + // SAFETY: fcntl on an integer descriptor. + let fd = unsafe { libc::fcntl(source, command, 300) }; + assert!(fd >= 300); + fd + } + + fn close(fd: RawFd) { + // SAFETY: closing a descriptor the test owns. + unsafe { libc::close(fd) }; + } + + #[test] + fn placing_a_descriptor_on_itself_clears_close_on_exec() { + let (reader, _writer) = std::io::pipe().unwrap(); + let fd = high_dup(reader.as_raw_fd(), true); + assert!(cloexec(fd)); + place_fd(fd, fd).unwrap(); + assert!(!cloexec(fd), "it must survive exec"); + // One that already survives is left alone, not refused. + place_fd(fd, fd).unwrap(); + assert!(!cloexec(fd)); + close(fd); + } + + #[test] + fn placing_a_descriptor_elsewhere_copies_it_open_across_exec() { + let (reader, _writer) = std::io::pipe().unwrap(); + let source = high_dup(reader.as_raw_fd(), true); + let target = high_dup(reader.as_raw_fd(), true); + assert!(cloexec(target)); + place_fd(source, target).unwrap(); + assert!(!cloexec(target), "the copy survives exec"); + assert!(cloexec(source), "the source does not"); + close(source); + close(target); + } + + #[test] + fn a_descriptor_that_is_not_open_is_refused_either_way() { + let (reader, _writer) = std::io::pipe().unwrap(); + let closed = high_dup(reader.as_raw_fd(), true); + close(closed); + assert!(place_fd(closed, closed).is_err()); + assert!(place_fd(closed, closed + 1).is_err()); + } + } } diff --git a/crates/model-connector/src/client.rs b/crates/model-connector/src/client.rs index e390258..47b6c94 100644 --- a/crates/model-connector/src/client.rs +++ b/crates/model-connector/src/client.rs @@ -256,4 +256,56 @@ mod fixture_tests { "the client must not even connect" ); } + + #[cfg(feature = "daemon")] + #[test] + fn a_readout_is_accepted_only_for_the_identity_that_asked() { + use crate::readout::PlanQuotaSnapshot; + // (caller the answer is sealed for, request id override, accepted) + let cases = [ + ("caller-1", None, true), + ("caller-2", None, false), + ("caller-1", Some("some-other-request"), false), + ]; + for (answer_caller, answer_request, accepted) in cases { + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let endpoint = listener.local_addr().unwrap(); + let key = security("readout-client-key"); + let server = std::thread::spawn(move || { + let (mut stream, _) = listener.accept().unwrap(); + let request = read_transport_frame(&mut stream, 64 * 1024).unwrap(); + let RequestFrame::SpendReadout(asked) = decode_request_frame(&request).unwrap() + else { + panic!("the client sent something other than a readout request"); + }; + let request_id = answer_request.unwrap_or(&asked.request_id); + let readout = SpendReadout::new(1, Vec::new(), PlanQuotaSnapshot::default()); + let sealed = + encrypt_spend_readout(answer_caller, request_id, &readout, &key).unwrap(); + write_transport_frame( + &mut stream, + &encode_spend_readout_response_frame(&sealed).unwrap(), + 64 * 1024, + ) + .unwrap(); + }); + let client = ConnectorClient::new( + endpoint, + "readout-client-key", + 64 * 1024, + Some(Duration::from_secs(1)), + ) + .unwrap(); + let answer = client.spend_readout("caller-1"); + if accepted { + assert_eq!(answer.unwrap().at_unix_ms, 1); + } else { + assert!(matches!( + answer, + Err(ConnectorClientError::Transport(ServiceError::OuterIdentity)) + )); + } + server.join().unwrap(); + } + } } diff --git a/crates/model-connector/src/envelope.rs b/crates/model-connector/src/envelope.rs index 90800c8..70d8ed2 100644 --- a/crates/model-connector/src/envelope.rs +++ b/crates/model-connector/src/envelope.rs @@ -1307,4 +1307,29 @@ mod tests { ServiceError::Encryption ); } + + #[cfg(feature = "daemon")] + #[test] + fn a_response_is_read_as_its_own_operation_only() { + use crate::readout::PlanQuotaSnapshot; + let key = TransportKey::from_connection_secret("readout-test-key").unwrap(); + let readout = SpendReadout::new(1, Vec::new(), PlanQuotaSnapshot::default()); + let sealed = encrypt_spend_readout("caller-1", "readout-1", &readout, &key).unwrap(); + let readout_frame = encode_spend_readout_response_frame(&sealed).unwrap(); + let invoke_frame = encode_result_frame(&sealed).unwrap(); + assert_eq!( + decode_spend_readout_response_frame(&readout_frame).unwrap(), + sealed + ); + assert_eq!(decode_result_frame(&invoke_frame).unwrap(), sealed); + assert_eq!( + decode_result_frame(&readout_frame), + Err(ServiceError::Carriage) + ); + assert_eq!( + decode_spend_readout_response_frame(&invoke_frame), + Err(ServiceError::Carriage) + ); + assert_eq!(decrypt_spend_readout(&sealed, &key).unwrap(), readout); + } } diff --git a/crates/model-connector/src/service.rs b/crates/model-connector/src/service.rs index 695f05b..b60daf0 100644 --- a/crates/model-connector/src/service.rs +++ b/crates/model-connector/src/service.rs @@ -3837,10 +3837,19 @@ mod tests { fn a_readout_that_cannot_fit_the_frame_is_a_frame_bound_failure() { let store = TestStore::new(); let service = open_service(&store, 4); + let asked = encrypt_spend_readout_request(CALLER, "readout-1", &security(KEY)).unwrap(); + let answer = service + .spend_readout(&asked, 1_000, PlanQuotaSnapshot::default(), FRAME) + .unwrap(); + let exact = encode_spend_readout_response_frame(&answer).unwrap().len(); + assert!(readout_as(&service, CALLER, KEY, 1_000, exact).is_ok()); + assert_eq!( + readout_as(&service, CALLER, KEY, 1_000, exact - 1).unwrap_err(), + ServiceError::FrameBound + ); assert_eq!( readout_as(&service, CALLER, KEY, 1_000, 64).unwrap_err(), ServiceError::FrameBound ); - assert!(readout_as(&service, CALLER, KEY, 1_000, FRAME).is_ok()); } }