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.