From 3f0c3e58346e230d98b7ec2ba2bb631254bfe460 Mon Sep 17 00:00:00 2001 From: Marcin Zieba Date: Tue, 15 Sep 2026 12:42:28 +0200 Subject: [PATCH] fix: bound a PDU frame with one deadline The session set a five second timeout on each read call, and read_exact loops over read. Every partial read restarted the clock, so a peer that sent one byte every four seconds held the session open forever and the master never got its responses. The receive step now computes one deadline when the first byte of a frame arrives, and gives each following read only the time that remains. The wait for the first byte stays unbounded, so an idle session does not reconnect on its own. Exhausted time fails the frame before the socket call, because a zero timeout means block forever and the standard library rejects a zero duration. The receive step takes the frame budget as an argument. The run loop passes the existing constant. The tests pass a short budget, which pins the boundary in under a second. This follows the netlink acquisition step, which computes one deadline for a whole inventory and passes the remaining time into each wait. A fake master drips one frame byte every two seconds against the actual binary, then asserts that the session ends near five seconds, names the deadline in its log, and accepts the reconnection. A socket pair test drives the receive step with a 100 ms budget and asserts expiry, success inside the budget, and an unbounded idle wait. Both tests fail against the unfixed code. An adversarial review noted that a final read can return just after the deadline and still complete the frame. The overrun is bounded by one read, because the deadline is fixed and the peer cannot extend it. Rejecting a complete PDU at that boundary would discard a valid request, and the sub-millisecond window cannot be tested without flake. Closes #15 --- src/session.rs | 103 +++++++++++++++++++++++++++++++++++++--- tests/real_namespace.rs | 84 +++++++++++++++++++++++++++++++- 2 files changed, 179 insertions(+), 8 deletions(-) diff --git a/src/session.rs b/src/session.rs index a86d3e0..8443693 100644 --- a/src/session.rs +++ b/src/session.rs @@ -1,6 +1,6 @@ use std::io::{Error, ErrorKind, Read, Result, Write}; use std::os::unix::net::UnixStream; -use std::time::Duration; +use std::time::{Duration, Instant}; use agentx::ByteOrder; use agentx::encodings::{ID, Value, VarBindList}; @@ -42,7 +42,7 @@ pub fn run(config: &Config, tables: &TableReader) -> Result<()> { ); loop { - let (header, bytes) = receive(&mut stream)?; + let (header, bytes) = receive(&mut stream, IO_TIMEOUT)?; log::debug!( "AgentX request {:?}, packet {}", header.ty, @@ -190,7 +190,7 @@ fn reply(header: &Header) -> Response { } fn acknowledge(stream: &mut UnixStream, request: &Header) -> Result
{ - let (header, bytes) = receive(stream)?; + let (header, bytes) = receive(stream, IO_TIMEOUT)?; // A Response PDU carries sysUpTime, error and index, then a VarBindList the // master is free to populate. net-snmp does: 40 payload bytes for Open and 32 // for Register. Only the fixed 8 byte block is read below, so a longer payload @@ -221,29 +221,56 @@ fn acknowledge(stream: &mut UnixStream, request: &Header) -> Result
{ Ok(header) } -fn receive(stream: &mut UnixStream) -> Result<(Header, Vec)> { +fn receive(stream: &mut UnixStream, frame_budget: Duration) -> Result<(Header, Vec)> { let mut bytes = vec![0; 20]; stream.read_exact(&mut bytes[..1])?; let idle_timeout = stream.read_timeout()?; - stream.set_read_timeout(Some(IO_TIMEOUT))?; - stream.read_exact(&mut bytes[1..])?; + let deadline = Instant::now() + frame_budget; + read_before(stream, &mut bytes[1..], deadline)?; let header = Header::from_bytes(&bytes)?; if header.version != 1 || header.payload_length > MAX_PAYLOAD || header.payload_length % 4 != 0 { return Err(invalid("invalid AgentX version or payload length")); } bytes.resize(20 + header.payload_length as usize, 0); - stream.read_exact(&mut bytes[20..])?; + read_before(stream, &mut bytes[20..], deadline)?; stream.set_read_timeout(idle_timeout)?; Ok((header, bytes)) } +fn read_before(stream: &mut UnixStream, mut bytes: &mut [u8], deadline: Instant) -> Result<()> { + while !bytes.is_empty() { + let remaining = deadline.saturating_duration_since(Instant::now()); + if remaining.is_zero() { + return Err(frame_deadline()); + } + stream.set_read_timeout(Some(remaining))?; + match stream.read(bytes) { + Ok(0) => return Err(ErrorKind::UnexpectedEof.into()), + Ok(count) => bytes = &mut bytes[count..], + Err(error) if error.kind() == ErrorKind::Interrupted => {} + Err(error) if matches!(error.kind(), ErrorKind::TimedOut | ErrorKind::WouldBlock) => { + return Err(frame_deadline()); + } + Err(error) => return Err(error), + } + } + Ok(()) +} + +fn frame_deadline() -> Error { + Error::new(ErrorKind::TimedOut, "AgentX PDU frame deadline exceeded") +} + fn invalid(message: &str) -> Error { Error::new(ErrorKind::InvalidData, message) } #[cfg(test)] mod tests { + use std::thread; + use std::time::Instant; + use agentx::encodings::SearchRangeList; use super::*; @@ -313,4 +340,66 @@ mod tests { assert_eq!(acknowledged.session_id, session_id); } } + + #[test] + fn a_started_frame_must_complete_before_its_budget() { + let budget = Duration::from_millis(100); + let frame = Header::new(Type::Ping).to_bytes(); + let (mut receiver, sender) = UnixStream::pair().expect("create slow socket pair"); + let drip = drip_frame( + sender, + frame.clone(), + Duration::ZERO, + Duration::from_millis(25), + ); + let started = Instant::now(); + let result = receive(&mut receiver, budget); + let elapsed = started.elapsed(); + drop(receiver); + drip.join().expect("join slow frame sender"); + + let error = match result { + Err(error) => error, + Ok(_) => panic!("frame completed after {elapsed:?}, past its {budget:?} budget"), + }; + assert_eq!(error.kind(), ErrorKind::TimedOut); + assert!(error.to_string().contains("frame deadline"), "{error}"); + assert!( + elapsed >= budget, + "frame failed before its budget: {elapsed:?}" + ); + assert!( + elapsed < Duration::from_millis(500), + "frame deadline was not bounded: {elapsed:?}" + ); + + let (mut receiver, sender) = UnixStream::pair().expect("create timely socket pair"); + let idle = budget + Duration::from_millis(50); + let drip = drip_frame(sender, frame, idle, Duration::from_millis(3)); + let started = Instant::now(); + let (header, _) = receive(&mut receiver, budget).expect("receive frame within its budget"); + let elapsed = started.elapsed(); + drip.join().expect("join timely frame sender"); + assert_eq!(header.ty, Type::Ping); + assert!(elapsed >= idle, "idle wait ended after {elapsed:?}"); + } + + fn drip_frame( + mut stream: UnixStream, + frame: Vec, + initial_delay: Duration, + gap: Duration, + ) -> thread::JoinHandle<()> { + thread::spawn(move || { + thread::sleep(initial_delay); + for (index, byte) in frame.into_iter().enumerate() { + if index != 0 { + thread::sleep(gap); + } + if stream.write_all(&[byte]).is_err() { + return; + } + } + }) + } } diff --git a/tests/real_namespace.rs b/tests/real_namespace.rs index 00604d9..c8a0244 100644 --- a/tests/real_namespace.rs +++ b/tests/real_namespace.rs @@ -3,7 +3,7 @@ use std::fs; use std::io::{Read, Write}; use std::os::unix::{fs::PermissionsExt, net::UnixStream, process::CommandExt}; use std::path::Path; -use std::process::{Child, Command, Output}; +use std::process::{Child, Command, Output, Stdio}; use std::str::FromStr; use std::time::{Duration, Instant}; @@ -29,6 +29,27 @@ struct Master { child: Child, } +struct LoggedSubagent { + child: Option, +} + +impl LoggedSubagent { + fn stop(mut self) -> Output { + let mut child = self.child.take().expect("subagent is running"); + child.kill().expect("stop test subagent"); + child.wait_with_output().expect("reap test subagent") + } +} + +impl Drop for LoggedSubagent { + fn drop(&mut self) { + if let Some(mut child) = self.child.take() { + let _ = child.kill(); + let _ = child.wait(); + } + } +} + enum ProcessIdentity { Unprivileged, TestProcess, @@ -124,6 +145,67 @@ fn write_subagent_config(path: &Path, config: String) { .expect("make AgentX test config readable by the subagent user"); } +#[test] +fn a_dripped_frame_ends_the_actual_binary_session_at_five_seconds() { + let directory = tempfile::tempdir().expect("create test directory"); + let socket = directory.path().join("master"); + let master = AgentxMaster::bind(&socket); + let config = directory.path().join("config.toml"); + fs::write(&config, format!("socket = {socket:?}\n")).expect("write test configuration"); + let child = Command::new(env!("CARGO_BIN_EXE_agentx-ifstack")) + .arg("--config") + .arg(config) + .env_remove("RUST_LOG") + .stdout(Stdio::null()) + .stderr(Stdio::piped()) + .spawn() + .expect("start agentx-ifstack test process"); + let subagent = LoggedSubagent { child: Some(child) }; + let mut stream = master.connect(NETWORK_ORDER, 100, 127); + stream + .set_read_timeout(Some(Duration::from_secs(7))) + .expect("bound session-end wait"); + let mut writer = stream.try_clone().expect("clone fake master socket"); + let frame = header(Type::Ping, NETWORK_ORDER, 100, 3).to_bytes(); + let (sent, received) = std::sync::mpsc::sync_channel(0); + let drip = std::thread::spawn(move || { + for (index, byte) in frame.into_iter().take(4).enumerate() { + if index != 0 { + std::thread::sleep(Duration::from_secs(2)); + } + if writer.write_all(&[byte]).is_err() { + return; + } + if index == 0 { + sent.send(Instant::now()).expect("record first frame byte"); + } + } + }); + + let started = received + .recv_timeout(Duration::from_secs(1)) + .expect("first frame byte was sent"); + let ended = stream.read(&mut [0]); + let elapsed = started.elapsed(); + let _reconnected = matches!(ended, Ok(0)).then(|| master.connect(NETWORK_ORDER, 101, 127)); + let output = subagent.stop(); + drip.join().expect("join frame sender"); + let stderr = String::from_utf8(output.stderr).expect("subagent logs are UTF-8"); + + assert!( + matches!(ended, Ok(0)), + "session remained open after {elapsed:?}: {ended:?}\n{stderr}" + ); + assert!( + elapsed >= Duration::from_millis(4_500) && elapsed < Duration::from_millis(6_500), + "session ended after {elapsed:?}, not about five seconds\n{stderr}" + ); + assert!( + stderr.contains("AgentX session ended: AgentX PDU frame deadline"), + "session-end log did not name the frame deadline:\n{stderr}" + ); +} + impl Drop for Master { fn drop(&mut self) { // Drop runs during assertion unwinding. Do not hide the first failure.