From 4c3adcfae90a3f67c1fa2b9001d7868ac209ebb3 Mon Sep 17 00:00:00 2001 From: Enrique Saurez Date: Thu, 5 Nov 2026 19:19:33 -0800 Subject: [PATCH 01/17] guest: retain layered mapping state and CLI identity Keep prepared mapping handles and the host table intact when handing them to the shared dispatcher. Read only the CLI-owned workload identity while retaining the resolved account defaults. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 4f77d672-f979-4d98-9305-63a5f57ae08c --- .../src/guest/managed_agent/dispatch/tests.rs | 97 ++++++++++++++++++- nvx-rs/src/guest/managed_agent/mappings.rs | 16 +++ nvx-rs/src/guest/managed_agent/mod.rs | 37 +++++++ nvx-rs/src/guest/managed_agent/tests.rs | 40 ++++++++ scripts/test-layout-allowlist.json | 2 + 5 files changed, 189 insertions(+), 3 deletions(-) diff --git a/nvx-rs/src/guest/managed_agent/dispatch/tests.rs b/nvx-rs/src/guest/managed_agent/dispatch/tests.rs index 4422aff6b..9b4190bce 100644 --- a/nvx-rs/src/guest/managed_agent/dispatch/tests.rs +++ b/nvx-rs/src/guest/managed_agent/dispatch/tests.rs @@ -67,13 +67,17 @@ impl Harness { /// Starts a direct agent, or a sandbox agent whose assembled root is `rootfs`, which /// attaches and waits for the broker's first RESET. - fn spawn_at(rootfs: Option<&Path>, mut system: TestSystem, expected_mappings: u32) -> Self { + fn spawn_at(rootfs: Option<&Path>, system: TestSystem, expected_mappings: u32) -> Self { + let config = AgentConfig::for_test(rootfs); + let mappings = HostMappings::announced_for(&config, expected_mappings); + Self::spawn_with(system, config, mappings) + } + + fn spawn_with(mut system: TestSystem, config: AgentConfig, mut mappings: HostMappings) -> Self { let (host, guest) = UnixStream::pair().unwrap(); let descriptor = FileDescriptor::from_owned(guest.into()); - let config = AgentConfig::for_test(rootfs); let thread = thread::spawn(move || { let mut session = Session::new(descriptor); - let mut mappings = HostMappings::announced_for(&config, expected_mappings); let outcome = run(&mut system, &mut session, &config, &mut mappings); (outcome, system) }); @@ -473,6 +477,93 @@ fn sandbox_root() -> (tempfile::TempDir, TestSystem, std::path::PathBuf) { (temporary, system, rootfs) } +#[test] +fn prepared_layered_mappings_and_cli_identity_use_the_existing_dispatcher() -> TestResult { + if super::super::tests::isolate_process_test( + "NVX_LAYERED_DISPATCH_CHILD", + "guest::managed_agent::dispatch::tests::prepared_layered_mappings_and_cli_identity_use_the_existing_dispatcher", + ) { + return Ok(()); + } + use crate::guest::mappings::{Mapper, MappingTable, Reserved, SourceRoot, TargetRoot}; + + let (_temporary, mut system, rootfs) = sandbox_root(); + system.direct_starter = Some(super::super::seam::tests::start_as_privileged); + let export = system.paths.hostfs_dir.join("root").canonicalize()?; + let source = SourceRoot::open(&export)?; + let target = TargetRoot::open_with(&mut system.kernel, &rootfs)?; + let mut mapper = Mapper::new(target, Reserved::container_root()); + let mut table = MappingTable::new(2).unwrap(); + table + .apply( + &maps_payload( + 0, + &[(b"0/src", b"/workspace", protocol::MAP_READ_ONLY)], + None, + ), + |batch| { + mapper + .apply_with(&mut system.kernel, &source, batch) + .map_err(|failure| (failure.entry(), failure.error().raw_os_error())) + }, + ) + .unwrap(); + let mappings = HostMappings::layered(table, mapper, source); + system.paths.hostfs_dir = rootfs.join("not-the-retained-export"); + let config = AgentConfig::layered( + b"nvx_identity=1000:1000", + c"workload".to_owned(), + c"/home/workload".to_owned(), + )?; + let mut harness = Harness::spawn_with(system, config, mappings); + harness.reset(1, 7); + assert_eq!(harness.request(APP_PING, &[]).kind, APP_READY); + assert_eq!( + harness.request(APP_EXEC, &exec_payload(None)), + response(APP_ERROR, 42, Errno::EAGAIN as i32, b"mappings-incomplete") + ); + harness.reset(2, 0); + assert_eq!( + harness.request(APP_MAPS, &maps_payload(0, &[(b"1", b"/out", 0)], None)), + response(APP_ERROR, 42, Errno::EINVAL as i32, b"invalid-request") + ); + assert_eq!( + harness + .request(APP_MAPS, &maps_payload(1, &[(b"1", b"/out", 0)], None)) + .kind, + APP_READY + ); + assert_eq!( + stdout_from( + harness.request_all( + 9, + APP_EXEC, + &workload_payload(&["/bin/sh", "-c", "printf layered; exit 7"], Some("/"))?, + ), + 9, + 7 + ), + b"layered" + ); + let system = harness.stop(); + let expected: Vec<_> = [ + ( + export.join("0/src"), + rootfs.join("workspace"), + Access::ReadOnly, + ), + (export.join("1"), rootfs.join("out"), Access::ReadWrite), + ] + .into_iter() + .flat_map(|(from, to, access)| attach_calls(&from, &to, access)) + .collect(); + assert_eq!(system.kernel.calls(), expected); + assert_eq!(system.spawn_calls, 1); + assert_eq!(system.entries_prepared, 0); + assert_eq!(system.runtime_binds, 0); + Ok(()) +} + #[test] fn sandbox_mappings_mount_in_order_beneath_its_root_and_unblock_workloads() { let (_temporary, system, rootfs) = sandbox_root(); diff --git a/nvx-rs/src/guest/managed_agent/mappings.rs b/nvx-rs/src/guest/managed_agent/mappings.rs index c63590da4..bd394312b 100644 --- a/nvx-rs/src/guest/managed_agent/mappings.rs +++ b/nvx-rs/src/guest/managed_agent/mappings.rs @@ -38,6 +38,8 @@ enum TableRoot { Direct, /// The sandbox agent's `ROOTFS`, the container's assembled root. Sandbox(PathBuf), + /// A layered root's retained mapper and export, never reopened by path. + Layered, } /// The mapper of the host mapping table beneath the table's root, and the root of the host's @@ -50,6 +52,19 @@ struct TableMapper { } impl HostMappings { + /// Adopts a layered root's table and prepared mapping handles without resetting either. + #[cfg_attr( + not(test), + expect(dead_code, reason = "SL7.7c hands over the layered root") + )] + pub(crate) fn layered(table: MappingTable, mapper: Mapper, export: SourceRoot) -> Self { + Self { + table, + root: TableRoot::Layered, + mapper: Some(TableMapper { mapper, export }), + } + } + /// Returns the state of the table of the agent that `config` configures, before /// [`prepare`] reads its announcement: the direct agent maps beneath `/`, and the sandbox /// agent beneath `ROOTFS`. @@ -172,6 +187,7 @@ fn open_table(system: &mut impl System, root: &TableRoot) -> Result return Err(Errno::EINVAL), }; let export = SourceRoot::open(&system.paths().hostfs_dir.join("root")).map_err(errno)?; Ok(TableMapper { diff --git a/nvx-rs/src/guest/managed_agent/mod.rs b/nvx-rs/src/guest/managed_agent/mod.rs index db4bd59ad..4c2538cdb 100644 --- a/nvx-rs/src/guest/managed_agent/mod.rs +++ b/nvx-rs/src/guest/managed_agent/mod.rs @@ -111,6 +111,43 @@ pub(crate) struct AgentConfig { entry: Option, } +impl AgentConfig { + /// Takes a layered cold boot's CLI identity and its already-resolved account defaults. + #[cfg_attr( + not(test), + expect(dead_code, reason = "SL7.7c selects layered cold boots") + )] + pub(crate) fn layered( + command_line: &[u8], + user: CString, + home: CString, + ) -> Result { + let mut identities = command_line + .split(u8::is_ascii_whitespace) + .filter_map(|token| token.strip_prefix(b"nvx_identity=")); + let identity = identities.next().ok_or(Errno::EINVAL)?; + let mut ids = identity.split(|&byte| byte == b':'); + let (Some(uid), Some(gid), None) = (ids.next(), ids.next(), ids.next()) else { + return Err(Errno::EINVAL); + }; + if identities.next().is_some() { + return Err(Errno::EINVAL); + } + launch::parse_id(uid)?; + launch::parse_id(gid)?; + Ok(Self { + rootfs: c"-".to_owned(), + hostname: c"nvx-managed".to_owned(), + uid: sys::c_string(uid)?, + gid: sys::c_string(gid)?, + user, + home, + direct: true, + entry: None, + }) + } +} + /// Runs `nvx-managed-agent` with the process's arguments. It exits with 0, or 125 when the agent /// fails. pub fn main() -> ExitCode { diff --git a/nvx-rs/src/guest/managed_agent/tests.rs b/nvx-rs/src/guest/managed_agent/tests.rs index 1b92c42f8..0edab9cee 100644 --- a/nvx-rs/src/guest/managed_agent/tests.rs +++ b/nvx-rs/src/guest/managed_agent/tests.rs @@ -50,6 +50,46 @@ fn sandbox_config(hostname: &[u8], uid: &[u8]) -> AgentConfig { } } +#[test] +fn a_layered_cold_boot_takes_only_the_clis_workload_identity() { + let config = AgentConfig::layered( + b"nvx_workload_uid=9 nvx_workload_gid=10 nvx_identity=1000:1001", + c"workload".to_owned(), + c"/home/workload".to_owned(), + ) + .unwrap(); + assert_eq!(config.uid.as_bytes(), b"1000"); + assert_eq!(config.gid.as_bytes(), b"1001"); + assert_eq!(config.user.as_c_str(), c"workload"); + assert_eq!(config.home.as_c_str(), c"/home/workload"); + assert!(config.direct); + assert!(config.entry.is_none()); +} + +#[test] +fn a_layered_cold_boot_refuses_missing_repeated_or_invalid_cli_identities() { + for command_line in [ + b"nvx_workload_uid=1000 nvx_workload_gid=1000".as_slice(), + b"nvx_identity=1000:1000 nvx_identity=1001:1001", + b"nvx_identity=", + b"nvx_identity=1000", + b"nvx_identity=:1000", + b"nvx_identity=1000:", + b"nvx_identity=1000:1000:1000", + b"nvx_identity=0:1000", + b"nvx_identity=1000:000", + b"nvx_identity=4294967295:1000", + b"nvx_identity=1000:4294967296", + b"nvx_identity=+1000:1000", + ] { + assert_eq!( + AgentConfig::layered(command_line, c"workload".to_owned(), c"/tmp".to_owned()).err(), + Some(Errno::EINVAL), + "{command_line:?}" + ); + } +} + #[test] fn the_sandbox_entry_takes_an_existing_account_and_one_machine_id_per_boot() { let (_temporary, system) = TestSystem::with_container_cgroup(); diff --git a/scripts/test-layout-allowlist.json b/scripts/test-layout-allowlist.json index ca8ea1991..94793cfd1 100644 --- a/scripts/test-layout-allowlist.json +++ b/scripts/test-layout-allowlist.json @@ -7,6 +7,8 @@ "nvx-rs/src/guest/exec.rs": 4, "nvx-rs/src/guest/init.rs": 8, "nvx-rs/src/guest/init/boot.rs": 1, + "nvx-rs/src/guest/managed_agent/mappings.rs": 1, + "nvx-rs/src/guest/managed_agent/mod.rs": 1, "nvx-rs/src/guest/reseed.rs": 1, "nvx-rs/src/guest/sys/entry.rs": 3, "nvx-rs/src/guest/sys/spawn.rs": 7, From 5bf5d0012181bd9a7321404d2db059c6f163cb3b Mon Sep 17 00:00:00 2001 From: Enrique Saurez Date: Thu, 5 Nov 2026 19:48:31 -0800 Subject: [PATCH 02/17] guest: parse workload identity with the init token reader Share the init token grammar so carriage returns and form feeds cannot create extra identities. Cover those cases and check that CLI IDs reach the native execution request. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 4f77d672-f979-4d98-9305-63a5f57ae08c --- .../src/guest/managed_agent/dispatch/tests.rs | 8 +++-- nvx-rs/src/guest/managed_agent/mod.rs | 6 ++-- nvx-rs/src/guest/managed_agent/tests.rs | 31 ++++++++++++------- 3 files changed, 27 insertions(+), 18 deletions(-) diff --git a/nvx-rs/src/guest/managed_agent/dispatch/tests.rs b/nvx-rs/src/guest/managed_agent/dispatch/tests.rs index 9b4190bce..0203ca931 100644 --- a/nvx-rs/src/guest/managed_agent/dispatch/tests.rs +++ b/nvx-rs/src/guest/managed_agent/dispatch/tests.rs @@ -488,7 +488,10 @@ fn prepared_layered_mappings_and_cli_identity_use_the_existing_dispatcher() -> T use crate::guest::mappings::{Mapper, MappingTable, Reserved, SourceRoot, TargetRoot}; let (_temporary, mut system, rootfs) = sandbox_root(); - system.direct_starter = Some(super::super::seam::tests::start_as_privileged); + system.direct_starter = Some(|request| { + assert_eq!((request.uid, request.gid), (1234, 2345)); + super::super::seam::tests::start_as_privileged(request) + }); let export = system.paths.hostfs_dir.join("root").canonicalize()?; let source = SourceRoot::open(&export)?; let target = TargetRoot::open_with(&mut system.kernel, &rootfs)?; @@ -511,7 +514,7 @@ fn prepared_layered_mappings_and_cli_identity_use_the_existing_dispatcher() -> T let mappings = HostMappings::layered(table, mapper, source); system.paths.hostfs_dir = rootfs.join("not-the-retained-export"); let config = AgentConfig::layered( - b"nvx_identity=1000:1000", + b"nvx_identity=1234:2345", c"workload".to_owned(), c"/home/workload".to_owned(), )?; @@ -560,7 +563,6 @@ fn prepared_layered_mappings_and_cli_identity_use_the_existing_dispatcher() -> T assert_eq!(system.kernel.calls(), expected); assert_eq!(system.spawn_calls, 1); assert_eq!(system.entries_prepared, 0); - assert_eq!(system.runtime_binds, 0); Ok(()) } diff --git a/nvx-rs/src/guest/managed_agent/mod.rs b/nvx-rs/src/guest/managed_agent/mod.rs index 4c2538cdb..e96f16f3d 100644 --- a/nvx-rs/src/guest/managed_agent/mod.rs +++ b/nvx-rs/src/guest/managed_agent/mod.rs @@ -122,9 +122,9 @@ impl AgentConfig { user: CString, home: CString, ) -> Result { - let mut identities = command_line - .split(u8::is_ascii_whitespace) - .filter_map(|token| token.strip_prefix(b"nvx_identity=")); + let mut identities = crate::protocol::cmdline::init_tokens(command_line) + .filter(|token| token.name == b"nvx_identity") + .filter_map(|token| token.value); let identity = identities.next().ok_or(Errno::EINVAL)?; let mut ids = identity.split(|&byte| byte == b':'); let (Some(uid), Some(gid), None) = (ids.next(), ids.next(), ids.next()) else { diff --git a/nvx-rs/src/guest/managed_agent/tests.rs b/nvx-rs/src/guest/managed_agent/tests.rs index 0edab9cee..52f07529c 100644 --- a/nvx-rs/src/guest/managed_agent/tests.rs +++ b/nvx-rs/src/guest/managed_agent/tests.rs @@ -52,18 +52,23 @@ fn sandbox_config(hostname: &[u8], uid: &[u8]) -> AgentConfig { #[test] fn a_layered_cold_boot_takes_only_the_clis_workload_identity() { - let config = AgentConfig::layered( - b"nvx_workload_uid=9 nvx_workload_gid=10 nvx_identity=1000:1001", - c"workload".to_owned(), - c"/home/workload".to_owned(), - ) - .unwrap(); - assert_eq!(config.uid.as_bytes(), b"1000"); - assert_eq!(config.gid.as_bytes(), b"1001"); - assert_eq!(config.user.as_c_str(), c"workload"); - assert_eq!(config.home.as_c_str(), c"/home/workload"); - assert!(config.direct); - assert!(config.entry.is_none()); + for command_line in [ + b"nvx_workload_uid=9 nvx_workload_gid=10 nvx_identity=1000:1001".as_slice(), + b"x\x0cnvx_identity=1001:1001 nvx_identity=1000:1001", + ] { + let config = AgentConfig::layered( + command_line, + c"workload".to_owned(), + c"/home/workload".to_owned(), + ) + .unwrap(); + assert_eq!(config.uid.as_bytes(), b"1000"); + assert_eq!(config.gid.as_bytes(), b"1001"); + assert_eq!(config.user.as_c_str(), c"workload"); + assert_eq!(config.home.as_c_str(), c"/home/workload"); + assert!(config.direct); + assert!(config.entry.is_none()); + } } #[test] @@ -81,6 +86,8 @@ fn a_layered_cold_boot_refuses_missing_repeated_or_invalid_cli_identities() { b"nvx_identity=4294967295:1000", b"nvx_identity=1000:4294967296", b"nvx_identity=+1000:1000", + b"x\rnvx_identity=1000:1000", + b"x\x0cnvx_identity=1000:1000", ] { assert_eq!( AgentConfig::layered(command_line, c"workload".to_owned(), c"/tmp".to_owned()).err(), From 6ee7f028b9c5dc2d2a34a0f101492d3046d6761b Mon Sep 17 00:00:00 2001 From: Enrique Saurez Date: Thu, 5 Nov 2026 21:48:57 -0800 Subject: [PATCH 03/17] guest: share the managed control loop with init Adopt prepared layered mappings and the init-owned collector through the existing control dispatcher. Keep framing and legacy sandbox policy in place. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 4f77d672-f979-4d98-9305-63a5f57ae08c --- nvx-rs/src/guest/managed_agent/dispatch.rs | 74 +++++++++- nvx-rs/src/guest/managed_agent/exec.rs | 64 +++++---- nvx-rs/src/guest/managed_agent/mappings.rs | 4 - nvx-rs/src/guest/managed_agent/mod.rs | 37 ----- nvx-rs/src/guest/managed_agent/session.rs | 150 ++++++++++++++++++++- scripts/test-layout-allowlist.json | 3 +- 6 files changed, 254 insertions(+), 78 deletions(-) diff --git a/nvx-rs/src/guest/managed_agent/dispatch.rs b/nvx-rs/src/guest/managed_agent/dispatch.rs index f46a6e641..1667082bb 100644 --- a/nvx-rs/src/guest/managed_agent/dispatch.rs +++ b/nvx-rs/src/guest/managed_agent/dispatch.rs @@ -1,4 +1,8 @@ -use crate::guest::sys::Errno; +use std::ffi::CString; + +use crate::guest::init::Runtime; +use crate::guest::mappings::{Mapper, MappingTable, SourceRoot}; +use crate::guest::sys::{self, Errno, FileDescriptor}; use crate::protocol::{ self, APP_CANCEL, APP_EXEC, APP_FEATURES, APP_MAPS, APP_PING, APP_READY, APP_STOP, APP_STOPPED, GuestFeatures, INVALID_REQUEST, MAPPINGS_INCOMPLETE, UNSUPPORTED_OPERATION, @@ -6,9 +10,68 @@ use crate::protocol::{ use super::diagnostics; use super::mappings::{HostMappings, MappingError}; -use super::seam::System; +use super::seam::{Paths, RealSystem, System}; use super::session::{Event, ReceiveFailure, Session}; -use super::{AgentConfig, exec, tty}; +use super::{AgentConfig, exec, launch, tty}; + +#[cfg_attr( + not(test), + expect( + dead_code, + reason = "the layered boot owner supplies the control handoff" + ) +)] +impl AgentConfig { + /// Takes a layered cold boot's CLI identity and its already-resolved account defaults. + pub(crate) fn layered( + command_line: &[u8], + user: CString, + home: CString, + ) -> Result { + let mut identities = crate::protocol::cmdline::init_tokens(command_line) + .filter(|token| token.name == b"nvx_identity") + .filter_map(|token| token.value); + let identity = identities.next().ok_or(Errno::EINVAL)?; + let mut ids = identity.split(|&byte| byte == b':'); + let (Some(uid), Some(gid), None) = (ids.next(), ids.next(), ids.next()) else { + return Err(Errno::EINVAL); + }; + if identities.next().is_some() { + return Err(Errno::EINVAL); + } + launch::parse_id(uid)?; + launch::parse_id(gid)?; + Ok(Self { + rootfs: c"-".to_owned(), + hostname: c"nvx-managed".to_owned(), + uid: sys::c_string(uid)?, + gid: sys::c_string(gid)?, + user, + home, + direct: true, + entry: None, + }) + } + + /// Serves an initialized layered root, returning shutdown to its init owner. + pub(crate) fn serve_layered( + &self, + runtime: &mut Runtime, + control: FileDescriptor, + (table, mapper, export): (MappingTable, Mapper, SourceRoot), + ) -> Result<(), String> { + let mut system = RealSystem::new(Paths::default()); + let mut session = Session::with_runtime(control, runtime)?; + let mut mappings = HostMappings::layered(table, mapper, export); + run(&mut system, &mut session, self, &mut mappings) + .map(|Outcome::Stopped| ()) + .map_err(|error| { + session + .runtime_failure() + .unwrap_or_else(|| format!("control session failed: {}", error.desc())) + }) + } +} /// Terminal outcome of the long-lived dispatch loop. #[derive(Debug)] @@ -24,6 +87,9 @@ pub(crate) fn run( config: &AgentConfig, mappings: &mut HostMappings, ) -> Result { + if session.stopping() { + return Ok(Outcome::Stopped); + } session.send_attach()?; loop { let payload = match session.next_event() { @@ -34,6 +100,7 @@ pub(crate) fn run( continue; } Ok(Event::Data(payload)) => payload, + Err(_) if session.stopping() => return Ok(Outcome::Stopped), Err(failure) => return Err(report_failure(system, failure)), }; // As in the C agent, a DATA record that fails also reports the `data` stage, after any @@ -43,6 +110,7 @@ pub(crate) fn run( match handle_data(system, session, config, mappings, &payload) { Ok(Some(outcome)) => return Ok(outcome), Ok(None) => {} + Err(_) if session.stopping() => return Ok(Outcome::Stopped), Err(error) => { diagnostics::error(system, "data", error as i32); return Err(error); diff --git a/nvx-rs/src/guest/managed_agent/exec.rs b/nvx-rs/src/guest/managed_agent/exec.rs index d0e18698e..1be19a52c 100644 --- a/nvx-rs/src/guest/managed_agent/exec.rs +++ b/nvx-rs/src/guest/managed_agent/exec.rs @@ -54,7 +54,7 @@ pub(crate) fn run( } } } - let mut workload = match launch(system, config, execution) { + let mut workload = match launch(system, session, config, execution) { Ok(launched) => launched, Err(LaunchError::Entry(error)) => { let diagnostic = @@ -63,7 +63,7 @@ pub(crate) fn run( return session.send_error(request_id, 125, LAUNCH_FAILED); } Err(LaunchError::Start(failure)) => { - settle_start(system, &failure)?; + settle_start(system, session, &failure)?; return session.send_error(request_id, 125, LAUNCH_FAILED); } Err(LaunchError::Other) => return session.send_error(request_id, 125, LAUNCH_FAILED), @@ -99,8 +99,11 @@ pub(crate) fn run( } Err(error) => { workload.terminate(); + if session.stopping() { + return Err(error); + } if !(config.direct && state.status_lost) - && let Err(failure) = workload.wait() + && let Err(failure) = workload.wait(session) { report_wait_failure(system, &failure); } @@ -114,8 +117,11 @@ pub(crate) fn run( } else { &mut buffer[..] }; + if let Some((id, status)) = session.take_status() { + workload.observe(id, status); + } state.forward(system, session, request_id, &mut workload, output)?; - workload.collect(); + workload.collect(session)?; if state.running(&workload) && execution.timeout_ms != 0 && started.elapsed() >= Duration::from_millis(u64::from(execution.timeout_ms)) @@ -128,7 +134,7 @@ pub(crate) fn run( if state.ending() && state.running(&workload) && !config.direct - && let Err(error) = workload.wait() + && let Err(error) = workload.wait(session) { state.lose_status(system, &error); } @@ -333,15 +339,15 @@ impl Workload { } } - fn collect(&mut self) { - collect(|id, status| match self { - Self::Direct(workload) => { - workload.observe(id, status); - } - Self::Sandbox(workload) => { - workload.observe(id, status); - } - }); + fn observe(&mut self, id: u32, status: std::process::ExitStatus) { + match self { + Self::Direct(workload) => workload.observe(id, status), + Self::Sandbox(workload) => workload.observe(id, status), + }; + } + + fn collect(&mut self, session: &Session) -> Result<(), Errno> { + session.collect(|id, status| self.observe(id, status)) } fn terminate(&mut self) { @@ -379,12 +385,12 @@ impl Workload { } } - fn wait(&mut self) -> std::io::Result { + fn wait(&mut self, session: &Session) -> std::io::Result { if let Self::Sandbox(workload) = self { return workload.wait(); } loop { - self.collect(); + self.collect(session)?; if let Some(status) = self.status() { return Ok(status); } @@ -417,7 +423,7 @@ impl Workload { } /// Feeds the agent's sole status collector to its active or failed-start owner. -fn collect(mut observe: impl FnMut(u32, std::process::ExitStatus)) { +pub(super) fn collect(mut observe: impl FnMut(u32, std::process::ExitStatus)) { while let Some(reaped) = sys::reap_any_nonblocking() { observe( reaped.process.raw().unsigned_abs(), @@ -427,18 +433,23 @@ fn collect(mut observe: impl FnMut(u32, std::process::ExitStatus)) { } /// Keeps a failed native launch owned until its child is collected or explicitly lost. -fn settle_start(system: &mut impl System, failure: &init::StartFailure) -> Result<(), Errno> { +fn settle_start( + system: &mut impl System, + session: &Session, + failure: &init::StartFailure, +) -> Result<(), Errno> { super::diagnostics::error(system, "exec-launch", sys::errno_of(&failure.error) as i32); if let Some(error) = failure.cleanup { super::diagnostics::error(system, "exec-kill", error as i32); } if let Some(process) = failure.process { + session.watch(process.raw().unsigned_abs()); loop { let mut settled = false; - collect(|id, status| { + session.collect(|id, status| { settled |= id == process.raw().unsigned_abs() && (status.code().is_some() || status.signal().is_some()); - }); + })?; if settled { break; } @@ -638,15 +649,17 @@ impl From for LaunchError { /// Starts a native direct workload, or the unchanged sandbox entry and placement sequence. fn launch( system: &mut impl System, + session: &Session, config: &AgentConfig, execution: &GuestExecPayload, ) -> Result { if config.direct { - collect(|_, _| {}); - return system + session.collect(|_, _| {})?; + let workload = system .start_direct(launch::direct_request(config, execution)?) - .map(Workload::Direct) - .map_err(LaunchError::Start); + .map_err(LaunchError::Start)?; + session.watch(workload.id()); + return Ok(Workload::Direct(workload)); } let (pipes, ends) = Pipes::new(false).map_err(|error| sys::errno_of(&error))?; let request = LaunchRequest { @@ -681,7 +694,7 @@ fn launch( { super::diagnostics::error(system, "exec-launch", error as i32); workload.terminate(); - if let Err(error) = workload.wait() { + if let Err(error) = workload.wait(session) { report_wait_failure(system, &error); } return Err(LaunchError::Other); @@ -810,6 +823,7 @@ fn handle_control( return Ok(ExecControl::SessionLost); } Ok(SessionEvent::Data(payload)) => payload, + Err(_) if session.stopping() => return Err(Errno::ECANCELED), Err(failure) => return Err(report_control_failure(system, failure)), }; let request = match session.accept_request(&payload) { diff --git a/nvx-rs/src/guest/managed_agent/mappings.rs b/nvx-rs/src/guest/managed_agent/mappings.rs index bd394312b..c451a88be 100644 --- a/nvx-rs/src/guest/managed_agent/mappings.rs +++ b/nvx-rs/src/guest/managed_agent/mappings.rs @@ -53,10 +53,6 @@ struct TableMapper { impl HostMappings { /// Adopts a layered root's table and prepared mapping handles without resetting either. - #[cfg_attr( - not(test), - expect(dead_code, reason = "SL7.7c hands over the layered root") - )] pub(crate) fn layered(table: MappingTable, mapper: Mapper, export: SourceRoot) -> Self { Self { table, diff --git a/nvx-rs/src/guest/managed_agent/mod.rs b/nvx-rs/src/guest/managed_agent/mod.rs index e96f16f3d..db4bd59ad 100644 --- a/nvx-rs/src/guest/managed_agent/mod.rs +++ b/nvx-rs/src/guest/managed_agent/mod.rs @@ -111,43 +111,6 @@ pub(crate) struct AgentConfig { entry: Option, } -impl AgentConfig { - /// Takes a layered cold boot's CLI identity and its already-resolved account defaults. - #[cfg_attr( - not(test), - expect(dead_code, reason = "SL7.7c selects layered cold boots") - )] - pub(crate) fn layered( - command_line: &[u8], - user: CString, - home: CString, - ) -> Result { - let mut identities = crate::protocol::cmdline::init_tokens(command_line) - .filter(|token| token.name == b"nvx_identity") - .filter_map(|token| token.value); - let identity = identities.next().ok_or(Errno::EINVAL)?; - let mut ids = identity.split(|&byte| byte == b':'); - let (Some(uid), Some(gid), None) = (ids.next(), ids.next(), ids.next()) else { - return Err(Errno::EINVAL); - }; - if identities.next().is_some() { - return Err(Errno::EINVAL); - } - launch::parse_id(uid)?; - launch::parse_id(gid)?; - Ok(Self { - rootfs: c"-".to_owned(), - hostname: c"nvx-managed".to_owned(), - uid: sys::c_string(uid)?, - gid: sys::c_string(gid)?, - user, - home, - direct: true, - entry: None, - }) - } -} - /// Runs `nvx-managed-agent` with the process's arguments. It exits with 0, or 125 when the agent /// fails. pub fn main() -> ExitCode { diff --git a/nvx-rs/src/guest/managed_agent/session.rs b/nvx-rs/src/guest/managed_agent/session.rs index bbf3f834c..13210620f 100644 --- a/nvx-rs/src/guest/managed_agent/session.rs +++ b/nvx-rs/src/guest/managed_agent/session.rs @@ -1,20 +1,92 @@ -use std::io; +use std::cell::{Cell, RefCell}; +use std::io::{self, Read, Write}; +use std::process::ExitStatus; +use crate::guest::init::Runtime; use crate::guest::nvxs::{GuestConnection, GuestConnectionError}; -use crate::guest::sys::{Errno, FileDescriptor}; +use crate::guest::sys::{self, Errno, FileDescriptor, Pollable}; use crate::protocol::nvxs::{GuestEvent, RecordError, RecordType, SessionError}; use crate::protocol::{self, APP_ERROR, OUTER_MAX_PAYLOAD, OUTER_RECEIVE_WINDOW}; /// The agent's control session: the guest's role of the shared NVXS session on the control /// console, and the NVXC frames that its DATA records carry. -pub(crate) struct Session { +pub(crate) struct Session<'r> { /// The guest's NVXS connection on the owned control-console descriptor. - connection: GuestConnection, + connection: GuestConnection>, /// Error of the connection's latest failure, which a connection that a failure ended reports /// again. last_error: Option, } +/// Byte transport only: the existing connection still owns all record framing. +struct Control<'r> { + descriptor: FileDescriptor, + runtime: Option>, + leader: Cell, + status: Cell>, + outcome: RefCell>, +} + +impl Control<'_> { + fn service(&self, observed: impl FnMut(u32, ExitStatus)) -> Result { + if self.outcome.borrow().is_err() { + return Err(Errno::ECHILD); + } + if let Some(runtime) = &self.runtime { + self.outcome.replace(runtime.borrow_mut().service(observed)); + } else { + super::exec::collect(observed); + } + self.outcome + .borrow() + .as_ref() + .copied() + .map_err(|_| Errno::ECHILD) + } +} + +impl Read for Control<'_> { + fn read(&mut self, bytes: &mut [u8]) -> io::Result { + if bytes.is_empty() { + return Read::read(&mut self.descriptor, bytes); + } + let Some(runtime) = &self.runtime else { + return Read::read(&mut self.descriptor, bytes); + }; + loop { + if self.service(|id, status| { + // Keep the first leader status across partial reads, not unrelated orphans. + if id == self.leader.get() && self.status.get().is_none() { + self.status.set(Some((id, status))); + } + })? { + return Err(Errno::ECANCELED.into()); + } + let signals = sys::poll( + [Pollable::File(&self.descriptor), runtime.borrow().wake()], + None, + )?[1]; + if signals.error { + return Err(Errno::EIO.into()); + } + if !signals.readable { + // A control-only wake includes HUP, which PollEvents does not expose. + return Read::read(&mut self.descriptor, bytes); + } + } + } +} + +impl Write for Control<'_> { + fn write(&mut self, bytes: &[u8]) -> io::Result { + Write::write(&mut self.descriptor, bytes) + } + + fn flush(&mut self) -> io::Result<()> { + Write::flush(&mut self.descriptor) + } +} + /// Record from the broker that the session accepted. pub(crate) enum Event { /// The broker reset the session. The agent abandons the earlier epoch's work, then @@ -82,20 +154,84 @@ impl DataFailure { } } -impl Session { +impl<'r> Session<'r> { /// Starts unattached session state on an owned control descriptor. pub(crate) fn new(descriptor: FileDescriptor) -> Self { + Self::open(descriptor, None) + } + + fn open(descriptor: FileDescriptor, runtime: Option<&'r mut Runtime>) -> Self { Self { - connection: GuestConnection::new(descriptor), + connection: GuestConnection::new(Control { + descriptor, + runtime: runtime.map(RefCell::new), + leader: Cell::new(0), + status: Cell::new(None), + outcome: RefCell::new(Ok(false)), + }), last_error: None, } } + /// Borrows init's sole collector for the lifetime of this control session. + pub(crate) fn with_runtime( + descriptor: FileDescriptor, + runtime: &'r mut Runtime, + ) -> Result { + let session = Self::open(descriptor, Some(runtime)); + session + .connection + .get_ref() + .service(|_, _| {}) + .map_err(|_| { + session + .runtime_failure() + .expect("the collector retained its failure") + })?; + Ok(session) + } + + /// Registers the newly launched leader before any control read can collect it. + pub(crate) fn watch(&self, id: u32) { + self.connection.get_ref().leader.set(id); + self.connection.get_ref().status.set(None); + } + + /// Takes a status collected while a partial control record was being read. + pub(crate) fn take_status(&self) -> Option<(u32, ExitStatus)> { + self.connection.get_ref().status.take() + } + + /// Collects in the caller's existing position, after handing over any earlier status. + pub(crate) fn collect(&self, mut observed: impl FnMut(u32, ExitStatus)) -> Result<(), Errno> { + if let Some((id, status)) = self.take_status() { + observed(id, status); + } + if self.connection.get_ref().service(observed)? { + return Err(Errno::ECANCELED); + } + Ok(()) + } + + /// A helper failure overrides even an already-latched normal termination. + pub(crate) fn stopping(&self) -> bool { + matches!(*self.connection.get_ref().outcome.borrow(), Ok(true)) + } + + /// Returns the original init diagnostic to its shutdown owner. + pub(crate) fn runtime_failure(&self) -> Option { + std::mem::replace( + &mut *self.connection.get_ref().outcome.borrow_mut(), + Ok(false), + ) + .err() + } + /// Borrows the control descriptor for polling and TTY operations. The connection never reads /// past the record in progress and holds no partial record between calls, so polling sees /// every record that the session has not returned. pub(crate) fn descriptor(&self) -> &FileDescriptor { - self.connection.get_ref() + &self.connection.get_ref().descriptor } /// Sends the initial guest-attach record. diff --git a/scripts/test-layout-allowlist.json b/scripts/test-layout-allowlist.json index 94793cfd1..d287998cc 100644 --- a/scripts/test-layout-allowlist.json +++ b/scripts/test-layout-allowlist.json @@ -7,8 +7,7 @@ "nvx-rs/src/guest/exec.rs": 4, "nvx-rs/src/guest/init.rs": 8, "nvx-rs/src/guest/init/boot.rs": 1, - "nvx-rs/src/guest/managed_agent/mappings.rs": 1, - "nvx-rs/src/guest/managed_agent/mod.rs": 1, + "nvx-rs/src/guest/managed_agent/dispatch.rs": 1, "nvx-rs/src/guest/reseed.rs": 1, "nvx-rs/src/guest/sys/entry.rs": 3, "nvx-rs/src/guest/sys/spawn.rs": 7, From 0485b3674c4b04c5aef12fb7209240ba1ee944f4 Mon Sep 17 00:00:00 2001 From: Enrique Saurez Date: Thu, 5 Nov 2026 22:02:28 -0800 Subject: [PATCH 04/17] guest: exercise init-owned managed control wakeups Cover fragmented control input, helper failure versus SIGTERM and terminal status delivery before native driver events. Reuse the existing outer-record reader and run explicitly selected ignored proofs in isolated subprocesses. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 4f77d672-f979-4d98-9305-63a5f57ae08c --- nvx-rs/src/guest/managed_agent/dispatch.rs | 2 +- .../src/guest/managed_agent/dispatch/tests.rs | 18 +- .../managed_agent/dispatch/tests/runtime.rs | 306 ++++++++++++++++++ nvx-rs/src/guest/sys/process/tests.rs | 2 +- 4 files changed, 320 insertions(+), 8 deletions(-) create mode 100644 nvx-rs/src/guest/managed_agent/dispatch/tests/runtime.rs diff --git a/nvx-rs/src/guest/managed_agent/dispatch.rs b/nvx-rs/src/guest/managed_agent/dispatch.rs index 1667082bb..3092ef420 100644 --- a/nvx-rs/src/guest/managed_agent/dispatch.rs +++ b/nvx-rs/src/guest/managed_agent/dispatch.rs @@ -76,7 +76,7 @@ impl AgentConfig { /// Terminal outcome of the long-lived dispatch loop. #[derive(Debug)] pub(crate) enum Outcome { - /// The host requested and acknowledged an orderly stop. + /// The host requested STOP, or init accepted an orderly termination. Stopped, } diff --git a/nvx-rs/src/guest/managed_agent/dispatch/tests.rs b/nvx-rs/src/guest/managed_agent/dispatch/tests.rs index 0203ca931..a7111b92b 100644 --- a/nvx-rs/src/guest/managed_agent/dispatch/tests.rs +++ b/nvx-rs/src/guest/managed_agent/dispatch/tests.rs @@ -212,15 +212,21 @@ impl Harness { } fn read_outer(&mut self) -> (OuterHeader, Vec) { - let mut header = [0u8; OUTER_HEADER_LEN]; - self.host.read_exact(&mut header).unwrap(); - let header = decode_outer_header(&header).unwrap(); - let mut payload = vec![0; header.length]; - self.host.read_exact(&mut payload).unwrap(); - (header, payload) + read_outer(&mut self.host) } } +fn read_outer(host: &mut UnixStream) -> (OuterHeader, Vec) { + let mut header = [0u8; OUTER_HEADER_LEN]; + host.read_exact(&mut header).unwrap(); + let header = decode_outer_header(&header).unwrap(); + let mut payload = vec![0; header.length]; + host.read_exact(&mut payload).unwrap(); + (header, payload) +} + +mod runtime; + fn response(kind: u8, request_id: u64, status: i32, payload: &[u8]) -> Response { Response { kind, diff --git a/nvx-rs/src/guest/managed_agent/dispatch/tests/runtime.rs b/nvx-rs/src/guest/managed_agent/dispatch/tests/runtime.rs new file mode 100644 index 000000000..b25fa0e68 --- /dev/null +++ b/nvx-rs/src/guest/managed_agent/dispatch/tests/runtime.rs @@ -0,0 +1,306 @@ +use std::time::{Duration, Instant}; + +use crate::guest::mappings::{Mapper, MappingTable, Reserved, SourceRoot, TargetRoot}; +use crate::guest::sys::tests::fork_isolated_test; +use crate::guest::sys::{self, ChildStatus, ForkResult, ProcessId, Signal}; + +use super::*; + +struct OwnedChild { + process: ProcessId, + descriptor: FileDescriptor, +} + +impl OwnedChild { + fn new(process: ProcessId) -> Self { + let descriptor = sys::open_pidfd(process).unwrap_or_else(|error| { + sys::kill_process(process).unwrap(); + let _ = sys::wait(process); + panic!("cannot retain the test child's identity: {error}"); + }); + Self { + process, + descriptor, + } + } +} + +impl Drop for OwnedChild { + fn drop(&mut self) { + assert!(matches!( + sys::kill_through_pidfd(self.descriptor.borrowed(), false), + Ok(()) | Err(Errno::ESRCH) + )); + // The Runtime may already have collected this owned child. + let _ = sys::wait(self.process); + } +} + +fn isolated(marker: &str, name: &str, check: impl FnOnce() + std::panic::UnwindSafe) { + if super::super::super::tests::isolate_process_test(marker, name) { + return; + } + match fork_isolated_test().unwrap() { + ForkResult::Child => { + sys::reset_child_signal(); + let result = std::panic::catch_unwind(check); + sys::exit_immediately(u8::from(result.is_err())); + } + ForkResult::Parent(child) => { + assert_eq!(sys::wait(child), Some(ChildStatus::Exited(0))); + } + } +} + +fn helper() -> OwnedChild { + match sys::fork().unwrap() { + ForkResult::Child => { + std::thread::sleep(Duration::from_secs(30)); + sys::exit_immediately(0); + } + ForkResult::Parent(process) => OwnedChild::new(process), + } +} + +fn peer( + check: impl FnOnce(UnixStream, UnixStream) + std::panic::UnwindSafe, +) -> (FileDescriptor, OwnedChild, UnixStream) { + let (host, guest) = UnixStream::pair().unwrap(); + let (proof, mut completed) = UnixStream::pair().unwrap(); + host.set_read_timeout(Some(Duration::from_secs(5))).unwrap(); + proof + .set_read_timeout(Some(Duration::from_secs(5))) + .unwrap(); + match sys::fork().unwrap() { + ForkResult::Child => { + drop(proof); + let probe = guest.try_clone().unwrap(); + drop(guest); + let result = std::panic::catch_unwind(|| check(host, probe)); + if result.is_ok() { + completed.write_all(&[1]).unwrap(); + } + sys::exit_immediately(u8::from(result.is_err())); + } + ForkResult::Parent(process) => { + drop(host); + drop(completed); + ( + FileDescriptor::from_owned(guest.into()), + OwnedChild::new(process), + proof, + ) + } + } +} + +fn completed(mut proof: UnixStream) { + let mut success = [0]; + proof.read_exact(&mut success).unwrap(); + assert_eq!(success, [1]); +} + +fn until(mut ready: impl FnMut() -> bool) { + let deadline = Instant::now() + Duration::from_secs(5); + while !ready() { + assert!(Instant::now() < deadline, "the owned peer made no progress"); + std::thread::sleep(Duration::from_millis(1)); + } +} + +fn fragment(host: &mut UnixStream, probe: UnixStream, record: &[u8]) { + use std::os::fd::AsFd; + + host.write_all(&record[..1]).unwrap(); + // Only the guest reads this endpoint: no readable byte proves it consumed the fragment. + until(|| !sys::readable_or_hung_up(&[probe.as_fd()], 0).unwrap()[0]); + drop(probe); +} + +fn prepared() -> (tempfile::TempDir, (MappingTable, Mapper, SourceRoot)) { + let (temporary, mut system, rootfs) = sandbox_root(); + let source = SourceRoot::open(&system.paths.hostfs_dir.join("root")).unwrap(); + let target = TargetRoot::open_with(&mut system.kernel, &rootfs).unwrap(); + ( + temporary, + ( + MappingTable::new(0).unwrap(), + Mapper::new(target, Reserved::container_root()), + source, + ), + ) +} + +fn termination(helper_ends: bool) { + let helper = helper(); + let parent = sys::open_pidfd(sys::current_process()).unwrap(); + let (descriptor, _peer, proof) = peer(|mut host, probe| { + assert_eq!(read_outer(&mut host).0.record_type, OUTER_GUEST_ATTACH); + let reset = encode_outer(OUTER_RESET, &INSTANCE, 1, 0, &[]).unwrap(); + fragment(&mut host, probe, &reset); + if helper_ends { + sys::kill_through_pidfd(helper.descriptor.borrowed(), false).unwrap(); + assert!(sys::readable_or_hung_up(&[helper.descriptor.borrowed()], 5000).unwrap()[0]); + } + sys::signal_through_pidfd(parent.borrowed(), Signal::SIGTERM as i32, false).unwrap(); + let mut traffic = Vec::new(); + host.read_to_end(&mut traffic).unwrap(); + assert!( + traffic.is_empty(), + "SIGTERM must not synthesize NVXC frames" + ); + }); + let mut runtime = + Runtime::new(&[CString::new(helper.process.raw().to_string()).unwrap()]).unwrap(); + let (_temporary, mappings) = prepared(); + let config = + AgentConfig::layered(b"nvx_identity=1000:1000", c"workload".into(), c"/".into()).unwrap(); + let result = config.serve_layered(&mut runtime, descriptor, mappings); + drop(runtime); + if helper_ends { + let failure = result.unwrap_err(); + assert!(failure.contains("the time daemon"), "{failure}"); + assert!(failure.contains("signal 9"), "{failure}"); + } else { + result.unwrap(); + } + completed(proof); +} + +#[test] +#[ignore = "requires root with CAP_SYS_NICE and CAP_SYS_RESOURCE"] +fn privileged_layered_runtime_stops_during_a_fragment_without_protocol_traffic() { + isolated( + "NVX_LAYERED_RUNTIME_STOP_CHILD", + "guest::managed_agent::dispatch::tests::runtime::privileged_layered_runtime_stops_during_a_fragment_without_protocol_traffic", + || termination(false), + ); +} + +#[test] +#[ignore = "requires root with CAP_SYS_NICE and CAP_SYS_RESOURCE"] +fn privileged_layered_runtime_helper_failure_wins_over_pending_sigterm() { + isolated( + "NVX_LAYERED_RUNTIME_HELPER_CHILD", + "guest::managed_agent::dispatch::tests::runtime::privileged_layered_runtime_helper_failure_wins_over_pending_sigterm", + || termination(true), + ); +} + +fn receive(host: &mut UnixStream) -> Response { + let (header, payload) = read_outer(host); + assert_eq!(header.record_type, OUTER_DATA); + let frame = decode_app(&payload).unwrap(); + response(frame.kind, frame.request_id, frame.status, frame.payload) +} + +#[test] +#[ignore = "requires root with CAP_SYS_NICE and CAP_SYS_RESOURCE"] +fn privileged_layered_runtime_delivers_a_fragment_collected_status_before_driver_events() { + isolated( + "NVX_LAYERED_RUNTIME_STATUS_CHILD", + "guest::managed_agent::dispatch::tests::runtime::privileged_layered_runtime_delivers_a_fragment_collected_status_before_driver_events", + || { + let temporary = tempfile::tempdir().unwrap(); + let pid_path = temporary.path().join("leader"); + let gate = temporary.path().join("gate"); + crate::guest::sys::tests::create_fifo( + &sys::path_c_string(&gate).unwrap(), + sys::Mode::from_bits_retain(0o600), + ) + .unwrap(); + let payload = workload_payload( + &[ + "/bin/sh", + "-c", + "printf '%s' \"$$\" > \"$1\"; read -r go < \"$2\"; printf observed; exit 7", + "sh", + pid_path.to_str().unwrap(), + gate.to_str().unwrap(), + ], + Some("/"), + ) + .unwrap(); + let helper = helper(); + let (descriptor, _peer, proof) = peer(|mut host, probe| { + assert_eq!(read_outer(&mut host).0.record_type, OUTER_GUEST_ATTACH); + host.write_all(&encode_outer(OUTER_RESET, &INSTANCE, 1, 0, &[]).unwrap()) + .unwrap(); + assert_eq!(read_outer(&mut host).0.record_type, OUTER_ACK); + let execution = encode_app(APP_EXEC, 9, 0, &payload); + host.write_all(&encode_outer(OUTER_DATA, &INSTANCE, 1, 1, &execution).unwrap()) + .unwrap(); + assert_eq!(read_outer(&mut host).0.record_type, OUTER_CREDIT); + until(|| pid_path.exists() && std::fs::metadata(&pid_path).unwrap().len() != 0); + let id: i64 = std::fs::read_to_string(&pid_path).unwrap().parse().unwrap(); + let leader = sys::open_pidfd(ProcessId::from_positive(id).unwrap()).unwrap(); + let ping = encode_outer( + OUTER_DATA, + &INSTANCE, + 1, + 2, + &encode_app(APP_PING, 98, 0, &[]), + ) + .unwrap(); + fragment(&mut host, probe, &ping); + std::fs::OpenOptions::new() + .write(true) + .open(&gate) + .unwrap() + .write_all(b"go\n") + .unwrap(); + assert!(sys::readable_or_hung_up(&[leader.borrowed()], 5000).unwrap()[0]); + // The driver never reaps: disappearance here proves Runtime collected while + // the same dispatcher remained inside the incomplete control record. + until(|| !Path::new(&format!("/proc/{id}")).exists()); + host.write_all(&ping[1..]).unwrap(); + assert_eq!(read_outer(&mut host).0.record_type, OUTER_CREDIT); + assert_eq!( + receive(&mut host), + response(APP_ERROR, 98, Errno::EBUSY as i32, b"busy") + ); + let mut output = Vec::new(); + loop { + let frame = receive(&mut host); + if frame.kind == APP_EXIT { + assert_eq!(frame, response(APP_EXIT, 9, 7, b"exit")); + break; + } + assert_eq!((frame.kind, frame.request_id), (APP_STDOUT, 9)); + output.extend(frame.payload); + } + assert_eq!(output, b"observed"); + host.write_all( + &encode_outer( + OUTER_DATA, + &INSTANCE, + 1, + 3, + &encode_app(APP_STOP, 99, 0, &[]), + ) + .unwrap(), + ) + .unwrap(); + assert_eq!(read_outer(&mut host).0.record_type, OUTER_CREDIT); + assert_eq!(receive(&mut host), response(APP_STOPPED, 99, 0, &[])); + }); + let mut runtime = + Runtime::new(&[CString::new(helper.process.raw().to_string()).unwrap()]).unwrap(); + let mut session = Session::with_runtime(descriptor, &mut runtime).unwrap(); + let mut system = TestSystem::new(Paths::default()); + system.direct_starter = Some(super::super::super::seam::tests::start_as_privileged); + let config = + AgentConfig::layered(b"nvx_identity=1000:1000", c"workload".into(), c"/".into()) + .unwrap(); + let mut mappings = HostMappings::announced_for(&config, 0); + assert!(matches!( + run(&mut system, &mut session, &config, &mut mappings), + Ok(Outcome::Stopped) + )); + assert!(system.diagnostics.is_empty(), "{:?}", system.diagnostics); + drop(session); + drop(runtime); + completed(proof); + }, + ); +} diff --git a/nvx-rs/src/guest/sys/process/tests.rs b/nvx-rs/src/guest/sys/process/tests.rs index b75477f61..ca1512d01 100644 --- a/nvx-rs/src/guest/sys/process/tests.rs +++ b/nvx-rs/src/guest/sys/process/tests.rs @@ -17,7 +17,7 @@ pub(crate) fn isolate_process_test(marker: &str, test: &str) -> bool { let status = std::process::Command::new( std::env::current_exe().expect("the test executable has a path"), ) - .args(["--exact", test, "--nocapture"]) + .args(["--exact", test, "--nocapture", "--include-ignored"]) .env(marker, "1") .status() .expect("the isolated test process starts"); From 516480033b3e3faefe38146fb37d5f11f7b40009 Mon Sep 17 00:00:00 2001 From: Enrique Saurez Date: Thu, 5 Nov 2026 22:10:12 -0800 Subject: [PATCH 05/17] guest: retain each runtime service result explicitly Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 4f77d672-f979-4d98-9305-63a5f57ae08c --- nvx-rs/src/guest/managed_agent/session.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/nvx-rs/src/guest/managed_agent/session.rs b/nvx-rs/src/guest/managed_agent/session.rs index 13210620f..ce6e0fc63 100644 --- a/nvx-rs/src/guest/managed_agent/session.rs +++ b/nvx-rs/src/guest/managed_agent/session.rs @@ -33,7 +33,7 @@ impl Control<'_> { return Err(Errno::ECHILD); } if let Some(runtime) = &self.runtime { - self.outcome.replace(runtime.borrow_mut().service(observed)); + *self.outcome.borrow_mut() = runtime.borrow_mut().service(observed); } else { super::exec::collect(observed); } From 11ed1fd4821b5a25a9df436d39e8e418b15cc60f Mon Sep 17 00:00:00 2001 From: Enrique Saurez Date: Thu, 5 Nov 2026 22:36:13 -0800 Subject: [PATCH 06/17] guest: exercise runtime ownership during failed starts Keep helper failure policy and the first launch and cleanup diagnostics while the existing dispatcher settles an owned failed child. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 4f77d672-f979-4d98-9305-63a5f57ae08c --- .../managed_agent/dispatch/tests/runtime.rs | 102 ++++++++++++++++++ 1 file changed, 102 insertions(+) diff --git a/nvx-rs/src/guest/managed_agent/dispatch/tests/runtime.rs b/nvx-rs/src/guest/managed_agent/dispatch/tests/runtime.rs index b25fa0e68..6ac939619 100644 --- a/nvx-rs/src/guest/managed_agent/dispatch/tests/runtime.rs +++ b/nvx-rs/src/guest/managed_agent/dispatch/tests/runtime.rs @@ -304,3 +304,105 @@ fn privileged_layered_runtime_delivers_a_fragment_collected_status_before_driver }, ); } + +#[test] +#[ignore = "requires root with CAP_SYS_NICE and CAP_SYS_RESOURCE"] +fn privileged_layered_runtime_keeps_helper_policy_and_first_error_during_failed_start() { + isolated( + "NVX_LAYERED_RUNTIME_FAILED_START_CHILD", + "guest::managed_agent::dispatch::tests::runtime::privileged_layered_runtime_keeps_helper_policy_and_first_error_during_failed_start", + || { + let temporary = tempfile::tempdir().unwrap(); + let marker = temporary.path().join("failed-child"); + let gate = temporary.path().join("gate"); + crate::guest::sys::tests::create_fifo( + &sys::path_c_string(&gate).unwrap(), + sys::Mode::from_bits_retain(0o600), + ) + .unwrap(); + let payload = workload_payload( + &[ + "/bin/sh", + "-c", + "read -r go < \"$1\"; exit 19", + "_", + gate.to_str().unwrap(), + marker.to_str().unwrap(), + ], + Some("/"), + ) + .unwrap(); + let helper = helper(); + let (descriptor, _peer, proof) = peer(|mut host, probe| { + drop(probe); + assert_eq!(read_outer(&mut host).0.record_type, OUTER_GUEST_ATTACH); + host.write_all(&encode_outer(OUTER_RESET, &INSTANCE, 1, 0, &[]).unwrap()) + .unwrap(); + assert_eq!(read_outer(&mut host).0.record_type, OUTER_ACK); + host.write_all( + &encode_outer( + OUTER_DATA, + &INSTANCE, + 1, + 1, + &encode_app(APP_EXEC, 9, 0, &payload), + ) + .unwrap(), + ) + .unwrap(); + assert_eq!(read_outer(&mut host).0.record_type, OUTER_CREDIT); + until(|| marker.exists() && std::fs::metadata(&marker).unwrap().len() != 0); + let id = std::fs::read_to_string(&marker) + .unwrap() + .trim() + .parse() + .unwrap(); + let failed = OwnedChild::new(ProcessId::from_positive(id).unwrap()); + assert!(!sys::readable_or_hung_up(&[failed.descriptor.borrowed()], 0).unwrap()[0]); + sys::kill_through_pidfd(helper.descriptor.borrowed(), false).unwrap(); + let mut traffic = Vec::new(); + host.read_to_end(&mut traffic).unwrap(); + assert!( + traffic.is_empty(), + "helper failure must end the control loop" + ); + drop(failed); + }); + let mut runtime = + Runtime::new(&[CString::new(helper.process.raw().to_string()).unwrap()]).unwrap(); + let mut session = Session::with_runtime(descriptor, &mut runtime).unwrap(); + let mut system = TestSystem::new(Paths::default()); + system.direct_starter = Some(start_with_failed_pidfd_and_cleanup); + let config = + AgentConfig::layered(b"nvx_identity=1000:1000", c"workload".into(), c"/".into()) + .unwrap(); + let mut mappings = HostMappings::announced_for(&config, 0); + assert_eq!( + run(&mut system, &mut session, &config, &mut mappings).err(), + Some(Errno::ECHILD) + ); + let failure = session.runtime_failure().unwrap(); + assert!(failure.contains("the time daemon"), "{failure}"); + let diagnostics = String::from_utf8(system.diagnostics).unwrap(); + assert!(diagnostics.starts_with(&format!( + "NVX-MANAGED-ERROR: stage=exec-launch status={}\n\ + NVX-MANAGED-ERROR: stage=exec-kill status={}\n", + Errno::EPERM as i32, + Errno::EPERM as i32 + ))); + assert!(!diagnostics.contains("stage=exec-wait"), "{diagnostics}"); + drop(session); + drop(runtime); + completed(proof); + let id = std::fs::read_to_string(&marker) + .unwrap() + .trim() + .parse() + .unwrap(); + assert_eq!( + sys::wait(ProcessId::from_positive(id).unwrap()), + Some(ChildStatus::Signaled(Signal::SIGKILL as i32)) + ); + }, + ); +} From ea057a942873b5f65af30b18d7d880764c8be52c Mon Sep 17 00:00:00 2001 From: Enrique Saurez Date: Fri, 6 Nov 2026 02:39:31 -0800 Subject: [PATCH 07/17] guest: retain optional layered exports Accept a zero-mapping root without an export, reject a positive table without its retained export, and keep unexpected MAPS on the existing accounting path. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 4f77d672-f979-4d98-9305-63a5f57ae08c --- nvx-rs/src/guest/managed_agent/dispatch.rs | 5 +-- .../src/guest/managed_agent/dispatch/tests.rs | 31 ++++++++++++++++++- .../managed_agent/dispatch/tests/runtime.rs | 8 +++-- nvx-rs/src/guest/managed_agent/mappings.rs | 20 ++++++++---- .../src/guest/managed_agent/mappings/tests.rs | 14 +++++++++ 5 files changed, 66 insertions(+), 12 deletions(-) diff --git a/nvx-rs/src/guest/managed_agent/dispatch.rs b/nvx-rs/src/guest/managed_agent/dispatch.rs index 3092ef420..3b29edc96 100644 --- a/nvx-rs/src/guest/managed_agent/dispatch.rs +++ b/nvx-rs/src/guest/managed_agent/dispatch.rs @@ -58,11 +58,12 @@ impl AgentConfig { &self, runtime: &mut Runtime, control: FileDescriptor, - (table, mapper, export): (MappingTable, Mapper, SourceRoot), + (table, mapper, export): (MappingTable, Mapper, Option), ) -> Result<(), String> { let mut system = RealSystem::new(Paths::default()); let mut session = Session::with_runtime(control, runtime)?; - let mut mappings = HostMappings::layered(table, mapper, export); + let mut mappings = HostMappings::layered(table, mapper, export) + .map_err(|error| format!("invalid layered mappings: {}", error.desc()))?; run(&mut system, &mut session, self, &mut mappings) .map(|Outcome::Stopped| ()) .map_err(|error| { diff --git a/nvx-rs/src/guest/managed_agent/dispatch/tests.rs b/nvx-rs/src/guest/managed_agent/dispatch/tests.rs index a7111b92b..b16ead8d7 100644 --- a/nvx-rs/src/guest/managed_agent/dispatch/tests.rs +++ b/nvx-rs/src/guest/managed_agent/dispatch/tests.rs @@ -517,7 +517,7 @@ fn prepared_layered_mappings_and_cli_identity_use_the_existing_dispatcher() -> T }, ) .unwrap(); - let mappings = HostMappings::layered(table, mapper, source); + let mappings = HostMappings::layered(table, mapper, Some(source))?; system.paths.hostfs_dir = rootfs.join("not-the-retained-export"); let config = AgentConfig::layered( b"nvx_identity=1234:2345", @@ -572,6 +572,35 @@ fn prepared_layered_mappings_and_cli_identity_use_the_existing_dispatcher() -> T Ok(()) } +#[test] +fn a_layered_root_without_an_export_refuses_maps_but_keeps_its_complete_table() { + use crate::guest::mappings::{Mapper, MappingTable, Reserved, TargetRoot}; + + let (_temporary, mut system, rootfs) = sandbox_root(); + let mapper = Mapper::new( + TargetRoot::open_with(&mut system.kernel, &rootfs).unwrap(), + Reserved::container_root(), + ); + let mappings = HostMappings::layered(MappingTable::new(0).unwrap(), mapper, None).unwrap(); + let config = + AgentConfig::layered(b"nvx_identity=1000:1000", c"workload".into(), c"/".into()).unwrap(); + let mut harness = Harness::spawn_with(system, config, mappings); + harness.reset(1, 0); + for payload in [ + maps_payload(0, &[], None), + maps_payload(0, &[(b"0", b"/workspace", 0)], None), + ] { + assert_eq!( + harness.request(APP_MAPS, &payload), + response(APP_ERROR, 42, Errno::EINVAL as i32, b"invalid-request") + ); + } + harness.reset(2, 0); + assert_eq!(harness.request(APP_PING, &[]).kind, APP_READY); + let system = harness.stop(); + assert!(system.kernel.calls().is_empty()); +} + #[test] fn sandbox_mappings_mount_in_order_beneath_its_root_and_unblock_workloads() { let (_temporary, system, rootfs) = sandbox_root(); diff --git a/nvx-rs/src/guest/managed_agent/dispatch/tests/runtime.rs b/nvx-rs/src/guest/managed_agent/dispatch/tests/runtime.rs index 6ac939619..47534b08c 100644 --- a/nvx-rs/src/guest/managed_agent/dispatch/tests/runtime.rs +++ b/nvx-rs/src/guest/managed_agent/dispatch/tests/runtime.rs @@ -117,16 +117,18 @@ fn fragment(host: &mut UnixStream, probe: UnixStream, record: &[u8]) { drop(probe); } -fn prepared() -> (tempfile::TempDir, (MappingTable, Mapper, SourceRoot)) { +fn prepared() -> ( + tempfile::TempDir, + (MappingTable, Mapper, Option), +) { let (temporary, mut system, rootfs) = sandbox_root(); - let source = SourceRoot::open(&system.paths.hostfs_dir.join("root")).unwrap(); let target = TargetRoot::open_with(&mut system.kernel, &rootfs).unwrap(); ( temporary, ( MappingTable::new(0).unwrap(), Mapper::new(target, Reserved::container_root()), - source, + None, ), ) } diff --git a/nvx-rs/src/guest/managed_agent/mappings.rs b/nvx-rs/src/guest/managed_agent/mappings.rs index c451a88be..0fae6cbd4 100644 --- a/nvx-rs/src/guest/managed_agent/mappings.rs +++ b/nvx-rs/src/guest/managed_agent/mappings.rs @@ -48,17 +48,24 @@ struct TableMapper { /// The mapper, which tracks the table's mappings, and latches at the first that fails. mapper: Mapper, /// The host export's mount, `hostfs_dir/root`. - export: SourceRoot, + export: Option, } impl HostMappings { /// Adopts a layered root's table and prepared mapping handles without resetting either. - pub(crate) fn layered(table: MappingTable, mapper: Mapper, export: SourceRoot) -> Self { - Self { + pub(crate) fn layered( + table: MappingTable, + mapper: Mapper, + export: Option, + ) -> Result { + if table.expected() != 0 && export.is_none() { + return Err(Errno::ENOENT); + } + Ok(Self { table, root: TableRoot::Layered, mapper: Some(TableMapper { mapper, export }), - } + }) } /// Returns the state of the table of the agent that `config` configures, before @@ -149,9 +156,10 @@ pub(crate) fn apply_request( None => open_table(system, root).map_err(|error| (0, error as i32))?, }; let opened = mapper.insert(opened); + let export = opened.export.as_ref().ok_or((0, Errno::ENOENT as i32))?; opened .mapper - .apply_with(system.kernel(), &opened.export, batch) + .apply_with(system.kernel(), export, batch) .map_err(|failure| (failure.entry(), failure.error().raw_os_error())) }) .map_err(|error| match error { @@ -188,7 +196,7 @@ fn open_table(system: &mut impl System, root: &TableRoot) -> Result) -> Vec { let mut payload = Vec::new(); payload.extend_from_slice(&first.to_le_bytes()); From 82a80285a1f3a90b18c9217389a7a3c59af73b9e Mon Sep 17 00:00:00 2001 From: Enrique Saurez Date: Fri, 6 Nov 2026 18:56:14 -0800 Subject: [PATCH 08/17] guest: propagate prelaunch runtime shutdowns Keep direct prelaunch collection outside the launch-error conversion so a pending SIGTERM or helper failure ends the session without a false launch-failed response. Exercise both after receiving and crediting a complete EXEC, and keep the helper-failure host test on the existing fake diagnostic boundary. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 4f77d672-f979-4d98-9305-63a5f57ae08c --- .../managed_agent/dispatch/tests/runtime.rs | 106 +++++++++++++++++- nvx-rs/src/guest/managed_agent/exec.rs | 4 +- 2 files changed, 108 insertions(+), 2 deletions(-) diff --git a/nvx-rs/src/guest/managed_agent/dispatch/tests/runtime.rs b/nvx-rs/src/guest/managed_agent/dispatch/tests/runtime.rs index 47534b08c..68cf31026 100644 --- a/nvx-rs/src/guest/managed_agent/dispatch/tests/runtime.rs +++ b/nvx-rs/src/guest/managed_agent/dispatch/tests/runtime.rs @@ -157,7 +157,19 @@ fn termination(helper_ends: bool) { let (_temporary, mappings) = prepared(); let config = AgentConfig::layered(b"nvx_identity=1000:1000", c"workload".into(), c"/".into()).unwrap(); - let result = config.serve_layered(&mut runtime, descriptor, mappings); + let result = if helper_ends { + let mut session = Session::with_runtime(descriptor, &mut runtime).unwrap(); + let mut system = TestSystem::new(Paths::default()); + let (table, mapper, export) = mappings; + let mut mappings = HostMappings::layered(table, mapper, export).unwrap(); + assert_eq!( + run(&mut system, &mut session, &config, &mut mappings).err(), + Some(Errno::ECHILD) + ); + Err(session.runtime_failure().unwrap()) + } else { + config.serve_layered(&mut runtime, descriptor, mappings) + }; drop(runtime); if helper_ends { let failure = result.unwrap_err(); @@ -196,6 +208,98 @@ fn receive(host: &mut UnixStream) -> Response { response(frame.kind, frame.request_id, frame.status, frame.payload) } +fn prelaunch_termination(helper_ends: bool) { + let helper = helper(); + let mut runtime = + Runtime::new(&[CString::new(helper.process.raw().to_string()).unwrap()]).unwrap(); + let (mut host, guest) = UnixStream::pair().unwrap(); + let mut session = + Session::with_runtime(FileDescriptor::from_owned(guest.into()), &mut runtime).unwrap(); + session.send_attach().unwrap(); + assert_eq!(read_outer(&mut host).0.record_type, OUTER_GUEST_ATTACH); + host.write_all(&encode_outer(OUTER_RESET, &INSTANCE, 1, 0, &[]).unwrap()) + .unwrap(); + assert!(matches!(session.next_event(), Ok(Event::Reset))); + session.acknowledge().unwrap(); + assert_eq!(read_outer(&mut host).0.record_type, OUTER_ACK); + let execution = workload_payload(&["/bin/true"], Some("/")).unwrap(); + host.write_all( + &encode_outer( + OUTER_DATA, + &INSTANCE, + 1, + 1, + &encode_app(APP_EXEC, 9, 0, &execution), + ) + .unwrap(), + ) + .unwrap(); + let Ok(Event::Data(payload)) = session.next_event() else { + panic!("the complete EXEC record was not received"); + }; + let Ok(request) = session.accept_request(&payload) else { + panic!("the complete EXEC request was not accepted"); + }; + assert_eq!(read_outer(&mut host).0.record_type, OUTER_CREDIT); + let execution = crate::protocol::decode_guest_exec_payload(request.payload).unwrap(); + if helper_ends { + sys::kill_through_pidfd(helper.descriptor.borrowed(), false).unwrap(); + assert!(sys::readable_or_hung_up(&[helper.descriptor.borrowed()], 5000).unwrap()[0]); + } else { + let process = sys::open_pidfd(sys::current_process()).unwrap(); + sys::signal_through_pidfd(process.borrowed(), Signal::SIGTERM as i32, false).unwrap(); + } + let mut system = TestSystem::new(Paths::default()); + let config = + AgentConfig::layered(b"nvx_identity=1000:1000", c"workload".into(), c"/".into()).unwrap(); + let error = super::super::super::exec::run( + &mut system, + &mut session, + &config, + request.request_id, + &execution, + ) + .err(); + drop(session); + drop(runtime); + let mut traffic = Vec::new(); + host.read_to_end(&mut traffic).unwrap(); + assert_eq!( + error, + Some(if helper_ends { + Errno::ECHILD + } else { + Errno::ECANCELED + }) + ); + assert!( + traffic.is_empty(), + "prelaunch shutdown must not send an application error" + ); + assert_eq!(system.spawn_calls, 0); + assert!(system.diagnostics.is_empty(), "{:?}", system.diagnostics); +} + +#[test] +#[ignore = "requires root with CAP_SYS_NICE and CAP_SYS_RESOURCE"] +fn privileged_layered_runtime_prelaunch_sigterm_sends_no_application_error() { + isolated( + "NVX_LAYERED_RUNTIME_PRELAUNCH_STOP_CHILD", + "guest::managed_agent::dispatch::tests::runtime::privileged_layered_runtime_prelaunch_sigterm_sends_no_application_error", + || prelaunch_termination(false), + ); +} + +#[test] +#[ignore = "requires root with CAP_SYS_NICE and CAP_SYS_RESOURCE"] +fn privileged_layered_runtime_prelaunch_helper_failure_sends_no_application_error() { + isolated( + "NVX_LAYERED_RUNTIME_PRELAUNCH_HELPER_CHILD", + "guest::managed_agent::dispatch::tests::runtime::privileged_layered_runtime_prelaunch_helper_failure_sends_no_application_error", + || prelaunch_termination(true), + ); +} + #[test] #[ignore = "requires root with CAP_SYS_NICE and CAP_SYS_RESOURCE"] fn privileged_layered_runtime_delivers_a_fragment_collected_status_before_driver_events() { diff --git a/nvx-rs/src/guest/managed_agent/exec.rs b/nvx-rs/src/guest/managed_agent/exec.rs index 1be19a52c..843ce747a 100644 --- a/nvx-rs/src/guest/managed_agent/exec.rs +++ b/nvx-rs/src/guest/managed_agent/exec.rs @@ -54,6 +54,9 @@ pub(crate) fn run( } } } + if config.direct { + session.collect(|_, _| {})?; + } let mut workload = match launch(system, session, config, execution) { Ok(launched) => launched, Err(LaunchError::Entry(error)) => { @@ -654,7 +657,6 @@ fn launch( execution: &GuestExecPayload, ) -> Result { if config.direct { - session.collect(|_, _| {})?; let workload = system .start_direct(launch::direct_request(config, execution)?) .map_err(LaunchError::Start)?; From 70dd71727cb34d62e5e1f5be9622d88ac5dd2f75 Mon Sep 17 00:00:00 2001 From: Enrique Saurez Date: Fri, 6 Nov 2026 02:44:38 -0800 Subject: [PATCH 09/17] guest: hand layered init control to the shared loop Keep the switched root and retained mapping handles alive while the existing managed dispatcher serves init control with the same Runtime. Remove the now-used wake and dispatcher hook allowances. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 4f77d672-f979-4d98-9305-63a5f57ae08c --- nvx-rs/src/guest/init.rs | 17 +++++++---------- nvx-rs/src/guest/managed_agent/dispatch.rs | 7 ------- scripts/test-layout-allowlist.json | 3 +-- 3 files changed, 8 insertions(+), 19 deletions(-) diff --git a/nvx-rs/src/guest/init.rs b/nvx-rs/src/guest/init.rs index 0956f7942..f54f629ac 100644 --- a/nvx-rs/src/guest/init.rs +++ b/nvx-rs/src/guest/init.rs @@ -853,10 +853,6 @@ impl Runtime { } /// Poll alongside control input so a blocked peer cannot hide a helper's end. - #[cfg_attr( - not(test), - expect(dead_code, reason = "SL7.6b polls the init runtime with control input") - )] pub(crate) fn wake(&self) -> Pollable<'_> { Pollable::Signals(self.system.signals.as_ref().expect("listening")) } @@ -907,12 +903,13 @@ fn serve_layered(runtime: &mut Runtime) -> Result<(), String> { ) .as_bytes(), ); - loop { - if runtime.service(|_, _| {})? { - return Ok(()); - } - runtime.terminating |= runtime.system.wait_signal() == Some(Signal::SIGTERM); - } + super::managed_agent::AgentConfig::layered(&boot.command_line, boot.user, boot.home) + .map_err(|error| format!("invalid layered identity: {}", error.desc()))? + .serve_layered( + runtime, + boot.control, + (boot.table, boot.mapper, boot.export), + ) } /// Keeps diagnostics, sync and direct power-off with the original PID-1 owner. diff --git a/nvx-rs/src/guest/managed_agent/dispatch.rs b/nvx-rs/src/guest/managed_agent/dispatch.rs index 3b29edc96..e08d2f496 100644 --- a/nvx-rs/src/guest/managed_agent/dispatch.rs +++ b/nvx-rs/src/guest/managed_agent/dispatch.rs @@ -14,13 +14,6 @@ use super::seam::{Paths, RealSystem, System}; use super::session::{Event, ReceiveFailure, Session}; use super::{AgentConfig, exec, launch, tty}; -#[cfg_attr( - not(test), - expect( - dead_code, - reason = "the layered boot owner supplies the control handoff" - ) -)] impl AgentConfig { /// Takes a layered cold boot's CLI identity and its already-resolved account defaults. pub(crate) fn layered( diff --git a/scripts/test-layout-allowlist.json b/scripts/test-layout-allowlist.json index d287998cc..5b7e6aa10 100644 --- a/scripts/test-layout-allowlist.json +++ b/scripts/test-layout-allowlist.json @@ -5,9 +5,8 @@ "nvx-rs/src/guest/cgroup.rs": 2, "nvx-rs/src/guest/entry.rs": 1, "nvx-rs/src/guest/exec.rs": 4, - "nvx-rs/src/guest/init.rs": 8, + "nvx-rs/src/guest/init.rs": 7, "nvx-rs/src/guest/init/boot.rs": 1, - "nvx-rs/src/guest/managed_agent/dispatch.rs": 1, "nvx-rs/src/guest/reseed.rs": 1, "nvx-rs/src/guest/sys/entry.rs": 3, "nvx-rs/src/guest/sys/spawn.rs": 7, From e29114a2390f4a80fe96ef52f21a5b610d097cad Mon Sep 17 00:00:00 2001 From: Enrique Saurez Date: Fri, 6 Nov 2026 02:47:10 -0800 Subject: [PATCH 10/17] guest: leave idle signal scripts in init tests The shared control loop now polls Runtime directly. Retain the old helper and signal scripts as test-only waiting fixtures instead of a second unused production wait path. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 4f77d672-f979-4d98-9305-63a5f57ae08c --- nvx-rs/src/guest/init.rs | 17 ------- nvx-rs/src/guest/init/tests.rs | 83 ++++++++++++++++++++++------------ 2 files changed, 54 insertions(+), 46 deletions(-) diff --git a/nvx-rs/src/guest/init.rs b/nvx-rs/src/guest/init.rs index f54f629ac..5a68a39eb 100644 --- a/nvx-rs/src/guest/init.rs +++ b/nvx-rs/src/guest/init.rs @@ -64,8 +64,6 @@ const NICE: i32 = -10; /// The signals that PID 1 reads from its signal descriptor: the end of a child, and the request /// to end the VM. const SIGNALS: [Signal; 2] = [Signal::SIGCHLD, Signal::SIGTERM]; -/// Pause before PID 1 polls its signal descriptor again after `poll` failed. -const POLL_RETRY: Duration = Duration::from_secs(1); /// The helpers that the shell `/init` hands over: the only processes whose end ends the VM. #[derive(Clone, Copy, Debug, PartialEq, Eq)] @@ -769,8 +767,6 @@ trait System { /// Collects the status of one child that ended, without waiting, or returns `None` when none /// did. fn reap(&mut self) -> Option<(ProcessId, ChildStatus, i32)>; - /// Waits for one of [`SIGNALS`], and returns it. - fn wait_signal(&mut self) -> Option; /// Writes the diagnostic `line` to the console. fn report(&mut self, line: &[u8]); /// Writes every file system's modified data to its storage, and then stops the VM with @@ -811,19 +807,6 @@ impl System for RealSystem { sys::reap_any_nonblocking().map(|child| (child.process, child.status, child.wait_status)) } - fn wait_signal(&mut self) -> Option { - let signals = self.signals.as_ref()?; - loop { - if let Some(event) = signals.read() { - return event.signal; - } - match sys::poll([Pollable::Signals(signals)], None) { - Ok(_) | Err(Errno::EINTR) => {} - Err(_) => std::thread::sleep(POLL_RETRY), - } - } - } - fn report(&mut self, line: &[u8]) { write_console(kernel::CONSOLE_DEVICE, line); } diff --git a/nvx-rs/src/guest/init/tests.rs b/nvx-rs/src/guest/init/tests.rs index 5632fcea1..3a017f74a 100644 --- a/nvx-rs/src/guest/init/tests.rs +++ b/nvx-rs/src/guest/init/tests.rs @@ -23,12 +23,12 @@ use crate::guest::sys::{ForkResult, wait}; /// Drives the unchanged helper/signal scripts through production preparation, collection and /// shutdown; layered boot itself is exercised by the boot tests. -fn init(system: &mut impl System, arguments: &[CString]) { +fn init(system: &mut impl Waiting, arguments: &[CString]) { let result = serve(system, arguments); finish(system, result); } -fn serve(system: &mut impl System, arguments: &[CString]) -> Result<(), String> { +fn serve(system: &mut impl Waiting, arguments: &[CString]) -> Result<(), String> { let helpers = prepare(system, arguments)?; let mut terminating = false; loop { @@ -40,6 +40,25 @@ fn serve(system: &mut impl System, arguments: &[CString]) -> Result<(), String> } } +trait Waiting: System { + fn wait_signal(&mut self) -> Option; +} + +impl Waiting for RealSystem { + fn wait_signal(&mut self) -> Option { + let signals = self.signals.as_ref()?; + loop { + if let Some(event) = signals.read() { + return event.signal; + } + match sys::poll([Pollable::Signals(signals)], None) { + Ok(_) | Err(Errno::EINTR) => {} + Err(_) => std::thread::sleep(Duration::from_secs(1)), + } + } + } +} + /// One operation that PID 1 asked its system for. #[derive(Clone, Debug, PartialEq, Eq)] enum Call { @@ -51,7 +70,7 @@ enum Call { Adopt(i32), /// [`System::reap`]. Reap, - /// [`System::wait_signal`]. + /// [`Waiting::wait_signal`]. Wait, /// [`System::report`] of this line. Report(String), @@ -73,7 +92,7 @@ struct Fake { /// The answers of [`System::reap`], in order: a child that ended, or `None` when none did. /// Once they run out, no child ended. reaped: VecDeque>, - /// The signals that [`System::wait_signal`] returns, in order. The check fails if PID 1 waits + /// The signals that [`Waiting::wait_signal`] returns, in order. The check fails if PID 1 waits /// for another. signals: VecDeque, } @@ -112,15 +131,6 @@ impl System for Fake { Some((pid(process), status, raw)) } - fn wait_signal(&mut self) -> Option { - self.calls.push(Call::Wait); - Some( - self.signals - .pop_front() - .expect("PID 1 waits for no signal that the check sends"), - ) - } - fn report(&mut self, line: &[u8]) { self.calls .push(Call::Report(String::from_utf8(line.to_vec()).unwrap())); @@ -131,6 +141,17 @@ impl System for Fake { } } +impl Waiting for Fake { + fn wait_signal(&mut self) -> Option { + self.calls.push(Call::Wait); + Some( + self.signals + .pop_front() + .expect("PID 1 waits for no signal that the check sends"), + ) + } +} + #[test] fn the_shared_collector_forwards_workload_status_and_keeps_helper_failure_fatal() { let helpers = Helpers { @@ -261,6 +282,16 @@ impl System for Subreaper { Some(child) } + fn report(&mut self, line: &[u8]) { + self.reports.push_str(std::str::from_utf8(line).unwrap()); + } + + fn shut_down(&mut self, status: u8) { + self.status = Some(status); + } +} + +impl Waiting for Subreaper { fn wait_signal(&mut self) -> Option { if let Some(before_wait) = self.before_wait.take() { before_wait(); @@ -269,14 +300,6 @@ impl System for Subreaper { self.signals.extend(signal); signal } - - fn report(&mut self, line: &[u8]) { - self.reports.push_str(std::str::from_utf8(line).unwrap()); - } - - fn shut_down(&mut self, status: u8) { - self.status = Some(status); - } } /// A system whose priorities, children, and statuses are the real ones, which records the @@ -312,6 +335,16 @@ impl System for Observer { self.real.reap() } + fn report(&mut self, line: &[u8]) { + self.reports.push_str(std::str::from_utf8(line).unwrap()); + } + + fn shut_down(&mut self, status: u8) { + self.status = Some(status); + } +} + +impl Waiting for Observer { fn wait_signal(&mut self) -> Option { let processes = std::iter::once(sys::current_process()).chain(self.helpers); self.observed = processes @@ -323,14 +356,6 @@ impl System for Observer { .collect(); Some(Signal::SIGTERM) } - - fn report(&mut self, line: &[u8]) { - self.reports.push_str(std::str::from_utf8(line).unwrap()); - } - - fn shut_down(&mut self, status: u8) { - self.status = Some(status); - } } /// Returns the process ID `raw`. From d2da5f9538815d65a9864cd1795b3d895162ae24 Mon Sep 17 00:00:00 2001 From: Enrique Saurez Date: Fri, 6 Nov 2026 02:58:18 -0800 Subject: [PATCH 11/17] guest: exercise real layered init through managed control Reuse the native host control driver for a dedicated PID-1 layered boot fixture. Check CLI identity and account defaults, retained zero or ordered mappings across reset, cancellation and clean shutdown without changing the existing default scenario. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 4f77d672-f979-4d98-9305-63a5f57ae08c --- nvx-rs/tests/managed_lifecycle.rs | 161 +++++++++++++++++++++++++++++- 1 file changed, 159 insertions(+), 2 deletions(-) diff --git a/nvx-rs/tests/managed_lifecycle.rs b/nvx-rs/tests/managed_lifecycle.rs index 57b9aa7dc..3b7a895f8 100644 --- a/nvx-rs/tests/managed_lifecycle.rs +++ b/nvx-rs/tests/managed_lifecycle.rs @@ -15,6 +15,9 @@ //! - `NVX_MANAGED_LIFECYCLE_TIMEOUT`: the longest wait for each operation, in seconds. //! - `NVX_MANAGED_LIFECYCLE_OUTPUT_DIR`: receives OpenVMM's output, `managed-lifecycle.log`, and //! the guest console, `managed-lifecycle-guest.log`, which the scenario then scans. +//! - `NVX_LAYERED_INIT_MAPS`, only for the dedicated real layered-init fixture: `0` for a live +//! share without an export, or `2` for the retained host export and an ordered mapping table. +//! That fixture boots the actual `nvx-init` as PID 1 with CLI identity `1000:1001`. //! //! The test also records OpenVMM's process ID in `NVX_ADVERSARIAL_OPENVMM_PID_JOURNAL`, if set, //! as the microVM scenarios do for the adversarial campaigns. @@ -36,7 +39,10 @@ use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use nvx::host::instance::generate_capability; use nvx::host::platform::process_start_time; use nvx::host::session::{self, ControlSession, ExecEvent}; -use nvx::protocol::{CAPABILITY_LEN, CWD_FAILED, ExitCategory, Workload, WorkloadEnvironment}; +use nvx::protocol::{ + self, APP_ERROR, APP_MAPS, APP_READY, CAPABILITY_LEN, CWD_FAILED, ExitCategory, HostMap, + Workload, WorkloadEnvironment, +}; /// An entry whose value has spaces, an equals sign, and a character beyond ASCII. const COMPLEX: &str = "COMPLEX=space = \u{2603}"; @@ -363,10 +369,136 @@ fn the_process_id_journal_has_the_entries_that_the_adversarial_executor_reads() assert_eq!(pids, [4321, u64::from(u32::MAX)]); } +fn layered_mapping_session( + guest: &Guest, + mut session: ControlSession, + count: u32, +) -> ControlSession { + assert!( + matches!(count, 0 | 2), + "the layered fixture has zero or two mappings" + ); + let maps = ["/workspace", "/run/task-state"].map(|target| HostMap { + source: "0".to_owned(), + target: target.to_owned(), + read_only: true, + }); + let parts = protocol::encode_maps_payloads(&maps, 40).unwrap(); + assert_eq!(parts.len(), 2); + let send = |session: &mut ControlSession, payload: &[u8], accepted| { + let deadline = Instant::now() + guest.timeout; + session + .send_request(APP_MAPS, 900, payload, deadline) + .unwrap(); + let expected = if accepted { + (APP_READY, 900, 0, Vec::new()) + } else { + ( + APP_ERROR, + 900, + 22, + protocol::INVALID_REQUEST.as_bytes().to_vec(), + ) + }; + assert_eq!(session.read_response(Some(deadline)).unwrap(), expected); + }; + if count == 0 { + send(&mut session, &parts[0], false); + } else { + let incomplete = |session: &mut ControlSession| { + let outcome = guest.exec( + session, + &["/bin/true"], + 5_000, + None, + WorkloadEnvironment::Default, + ); + assert_eq!( + outcome.refusal, + Some((11, protocol::MAPPINGS_INCOMPLETE.to_owned())) + ); + assert!(outcome.exit.is_none()); + }; + incomplete(&mut session); + send(&mut session, &parts[1], false); + send(&mut session, &parts[0], true); + drop(session); + session = guest.connect(); + incomplete(&mut session); + send(&mut session, &parts[1], true); + } + drop(session); + session = guest.connect(); + session.ping(Instant::now() + guest.timeout).unwrap(); + send(&mut session, &parts[0], false); + let defaults = guest.exec( + &mut session, + &[ + "/bin/sh", + "-c", + "test \"$(cat /proc/1/comm)\" = nvx-init && \ + test \"$(cat /proc/sys/kernel/hostname)\" = sl76b-real-root && \ + test -s /run/nvx/time/state && test \"$(wc -c () + .expect("NVX_LAYERED_INIT_MAPS is 0 or 2") + }); let command: Vec = serde_json::from_str(&variable("NVX_MANAGED_LIFECYCLE_COMMAND")) .expect("NVX_MANAGED_LIFECYCLE_COMMAND is a JSON array of strings"); let timeout = Duration::from_secs_f64( @@ -415,6 +547,9 @@ fn a_managed_guest_runs_request_scoped_workloads_until_it_stops() { session .ping(Instant::now() + timeout) .expect("the guest agent is ready"); + if let Some(count) = layered_maps { + session = layered_mapping_session(&guest, session, count); + } let exact = vec!["EMPTY=".to_owned(), COMPLEX.to_owned()]; let first = guest.exec( &mut session, @@ -476,7 +611,11 @@ fn a_managed_guest_runs_request_scoped_workloads_until_it_stops() { WorkloadEnvironment::Default, ); assert!( - identity.printed(WORKLOAD_IDENTITY), + identity.printed(if layered_maps.is_some() { + "1000:1001" + } else { + WORKLOAD_IDENTITY + }), "the workload identity is {identity:?}" ); // State persists across requests, while working directories and environments do not, and a @@ -603,4 +742,22 @@ fn a_managed_guest_runs_request_scoped_workloads_until_it_stops() { status.success(), "the managed OpenVMM process exited with {status}" ); + if let Some(count) = layered_maps { + let output = [ + fs::read(output_dir.join("managed-lifecycle-guest.log")).unwrap(), + fs::read(output_dir.join("managed-lifecycle.log")).unwrap(), + ] + .concat(); + let marker = format!("NVX-INIT-ROOT: switched mappings={count}"); + assert!( + output + .windows(marker.len()) + .any(|bytes| bytes == marker.as_bytes()) + ); + assert!( + !output + .windows(b"NVX-INIT-ERROR:".len()) + .any(|bytes| bytes == b"NVX-INIT-ERROR:") + ); + } } From 9698af55beadca99fd415dfbbc9f709df8f669d6 Mon Sep 17 00:00:00 2001 From: Enrique Saurez Date: Fri, 6 Nov 2026 03:19:56 -0800 Subject: [PATCH 12/17] guest: document layered init control shutdown Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 4f77d672-f979-4d98-9305-63a5f57ae08c --- nvx-rs/src/guest/init.rs | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/nvx-rs/src/guest/init.rs b/nvx-rs/src/guest/init.rs index 5a68a39eb..ec2e871c2 100644 --- a/nvx-rs/src/guest/init.rs +++ b/nvx-rs/src/guest/init.rs @@ -1,5 +1,5 @@ -//! `nvx-init TIME_DAEMON [ANNOUNCER]`: NVX's PID 1 for layered roots, which runs ahead of the -//! workloads. +//! `nvx-init TIME_DAEMON [ANNOUNCER]`: NVX's PID 1, which initializes a layered root and serves +//! its managed control session through the same dispatcher as blockless VMs. //! //! A layered root's early boot, which the shell `/init` runs, mounts the kernel's file systems, //! runs `nvx-time boot`, whose child becomes the time daemon, configures the network, and, when @@ -30,8 +30,8 @@ //! nothing from it; //! - ends the VM itself, without another guest program: it writes every file system's modified //! data to its storage, with `sync`, and then stops the VM through OpenVMM's control port with a -//! status: 0 after `SIGTERM`, or 125, as `nvx-init-agent`'s failures do, after a failure, which -//! an `NVX-INIT-ERROR:` line on the console names first. +//! status: 0 after a host `STOP` or `SIGTERM`, or 125, as `nvx-init-agent`'s failures do, after +//! a failure, which an `NVX-INIT-ERROR:` line on the console names first. //! //! Outside PID 1, `nvx-init` changes nothing: it fails with status 2 and a message. From 17ca29e3e256a132d01f2016e2b1de4e50e019d5 Mon Sep 17 00:00:00 2001 From: Enrique Saurez Date: Mon, 9 Nov 2026 11:55:56 -0800 Subject: [PATCH 13/17] guest: supervise init during backpressured control writes Make only runtime-backed control descriptors nonblocking and wait for writable readiness alongside the existing Runtime signal descriptor. Keep helper failure and SIGTERM handling live without polling on a timer or changing legacy sessions. Retain leader statuses collected during writes and hand them to the execution driver before each event. Preserve orderly termination during ATTACH and reset acknowledgement as well as workload output. Add focused readiness, backpressure, helper-precedence, pending-termination and write-collected-status regressions without relaxing existing checks or hooks. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 190278d5-cbae-4ec6-b47d-ddf469124598 --- nvx-rs/src/guest/managed_agent/dispatch.rs | 18 +- .../managed_agent/dispatch/tests/runtime.rs | 270 ++++++++++++++++++ nvx-rs/src/guest/managed_agent/exec.rs | 6 +- nvx-rs/src/guest/managed_agent/session.rs | 63 +++- nvx-rs/src/guest/sys/poll.rs | 25 +- nvx-rs/src/guest/sys/poll/tests.rs | 31 ++ 6 files changed, 386 insertions(+), 27 deletions(-) diff --git a/nvx-rs/src/guest/managed_agent/dispatch.rs b/nvx-rs/src/guest/managed_agent/dispatch.rs index e08d2f496..b94c61b97 100644 --- a/nvx-rs/src/guest/managed_agent/dispatch.rs +++ b/nvx-rs/src/guest/managed_agent/dispatch.rs @@ -84,13 +84,23 @@ pub(crate) fn run( if session.stopping() { return Ok(Outcome::Stopped); } - session.send_attach()?; + if let Err(error) = session.send_attach() { + return if session.stopping() { + Ok(Outcome::Stopped) + } else { + Err(error) + }; + } loop { let payload = match session.next_event() { Ok(Event::Reset) => { - session.acknowledge().inspect_err(|error| { - diagnostics::error(system, "reset-ack", *error as i32); - })?; + if let Err(error) = session.acknowledge() { + if session.stopping() { + return Ok(Outcome::Stopped); + } + diagnostics::error(system, "reset-ack", error as i32); + return Err(error); + } continue; } Ok(Event::Data(payload)) => payload, diff --git a/nvx-rs/src/guest/managed_agent/dispatch/tests/runtime.rs b/nvx-rs/src/guest/managed_agent/dispatch/tests/runtime.rs index 68cf31026..3418d55f3 100644 --- a/nvx-rs/src/guest/managed_agent/dispatch/tests/runtime.rs +++ b/nvx-rs/src/guest/managed_agent/dispatch/tests/runtime.rs @@ -512,3 +512,273 @@ fn privileged_layered_runtime_keeps_helper_policy_and_first_error_during_failed_ }, ); } + +struct KillParentOnPanic<'a>(&'a FileDescriptor); + +impl Drop for KillParentOnPanic<'_> { + fn drop(&mut self) { + if std::thread::panicking() { + assert!(matches!( + sys::kill_through_pidfd(self.0.borrowed(), false), + Ok(()) | Err(Errno::ESRCH) + )); + } + } +} + +fn narrow_output(probe: &UnixStream) { + use nix::sys::socket::{getsockopt, setsockopt, sockopt::SndBuf}; + + setsockopt(probe, SndBuf, &1024).unwrap(); + assert!(getsockopt(probe, SndBuf).unwrap() < 6000); +} + +fn output_is_blocked(probe: &UnixStream) { + use std::os::fd::AsFd; + + use nix::poll::{PollFd, PollFlags}; + + until(|| { + let mut descriptors = [PollFd::new(probe.as_fd(), PollFlags::POLLOUT)]; + nix::poll::poll(&mut descriptors, 0u16).unwrap(); + !descriptors[0] + .revents() + .unwrap() + .contains(PollFlags::POLLOUT) + }); +} + +fn reset_peer(host: &mut UnixStream) { + assert_eq!(read_outer(host).0.record_type, OUTER_GUEST_ATTACH); + host.write_all(&encode_outer(OUTER_RESET, &INSTANCE, 1, 0, &[]).unwrap()) + .unwrap(); + assert_eq!(read_outer(host).0.record_type, OUTER_ACK); +} + +fn only_stdout_prefix(mut traffic: &[u8]) { + assert!(!traffic.is_empty()); + while traffic.len() >= OUTER_HEADER_LEN { + let header = decode_outer_header(&traffic[..OUTER_HEADER_LEN]).unwrap(); + assert_eq!(header.record_type, OUTER_DATA); + let length = OUTER_HEADER_LEN + header.length; + if traffic.len() < length { + return; + } + let frame = decode_app(&traffic[OUTER_HEADER_LEN..length]).unwrap(); + assert_eq!((frame.kind, frame.request_id), (APP_STDOUT, 9)); + traffic = &traffic[length..]; + } +} + +fn backpressured_termination(helper_ends: bool) { + let helper = helper(); + let parent = sys::open_pidfd(sys::current_process()).unwrap(); + let (descriptor, _peer, proof) = peer(|mut host, probe| { + let _guard = KillParentOnPanic(&parent); + narrow_output(&probe); + reset_peer(&mut host); + output_is_blocked(&probe); + drop(probe); + if helper_ends { + sys::kill_through_pidfd(helper.descriptor.borrowed(), false).unwrap(); + assert!(sys::readable_or_hung_up(&[helper.descriptor.borrowed()], 5000).unwrap()[0]); + } + sys::signal_through_pidfd(parent.borrowed(), Signal::SIGTERM as i32, false).unwrap(); + let mut traffic = Vec::new(); + host.read_to_end(&mut traffic).unwrap(); + only_stdout_prefix(&traffic); + }); + let mut runtime = + Runtime::new(&[CString::new(helper.process.raw().to_string()).unwrap()]).unwrap(); + let mut session = Session::with_runtime(descriptor, &mut runtime).unwrap(); + session.send_attach().unwrap(); + assert!(matches!(session.next_event(), Ok(Event::Reset))); + session.acknowledge().unwrap(); + assert_eq!( + session.send_app(APP_STDOUT, 9, 0, &[0; 32768]), + Err(if helper_ends { + Errno::ECHILD + } else { + Errno::ECANCELED + }) + ); + if helper_ends { + let failure = session.runtime_failure().unwrap(); + assert!(failure.contains("the time daemon"), "{failure}"); + assert!(failure.contains("signal 9"), "{failure}"); + } else { + assert!(session.stopping()); + assert!(session.runtime_failure().is_none()); + } + drop(session); + drop(runtime); + completed(proof); +} + +#[test] +#[ignore = "requires root with CAP_SYS_NICE and CAP_SYS_RESOURCE"] +fn privileged_layered_runtime_stops_with_backpressured_control_output() { + isolated( + "NVX_LAYERED_RUNTIME_OUTPUT_STOP_CHILD", + "guest::managed_agent::dispatch::tests::runtime::privileged_layered_runtime_stops_with_backpressured_control_output", + || backpressured_termination(false), + ); +} + +#[test] +#[ignore = "requires root with CAP_SYS_NICE and CAP_SYS_RESOURCE"] +fn privileged_layered_runtime_helper_failure_wins_with_backpressured_control_output() { + isolated( + "NVX_LAYERED_RUNTIME_OUTPUT_HELPER_CHILD", + "guest::managed_agent::dispatch::tests::runtime::privileged_layered_runtime_helper_failure_wins_with_backpressured_control_output", + || backpressured_termination(true), + ); +} + +#[test] +#[ignore = "requires root with CAP_SYS_NICE and CAP_SYS_RESOURCE"] +fn privileged_layered_runtime_stops_before_attach_on_pending_sigterm() { + isolated( + "NVX_LAYERED_RUNTIME_ATTACH_STOP_CHILD", + "guest::managed_agent::dispatch::tests::runtime::privileged_layered_runtime_stops_before_attach_on_pending_sigterm", + || { + let helper = helper(); + let mut runtime = + Runtime::new(&[CString::new(helper.process.raw().to_string()).unwrap()]).unwrap(); + let (mut host, guest) = UnixStream::pair().unwrap(); + let mut session = + Session::with_runtime(FileDescriptor::from_owned(guest.into()), &mut runtime) + .unwrap(); + let process = sys::open_pidfd(sys::current_process()).unwrap(); + sys::signal_through_pidfd(process.borrowed(), Signal::SIGTERM as i32, false).unwrap(); + let mut system = TestSystem::new(Paths::default()); + let config = + AgentConfig::layered(b"nvx_identity=1000:1000", c"workload".into(), c"/".into()) + .unwrap(); + let mut mappings = HostMappings::announced_for(&config, 0); + assert!(matches!( + run(&mut system, &mut session, &config, &mut mappings), + Ok(Outcome::Stopped) + )); + assert!(system.diagnostics.is_empty(), "{:?}", system.diagnostics); + drop(session); + drop(runtime); + let mut traffic = Vec::new(); + host.read_to_end(&mut traffic).unwrap(); + assert!( + traffic.is_empty(), + "pending termination must not send ATTACH" + ); + }, + ); +} + +#[test] +#[ignore = "requires root with CAP_SYS_NICE and CAP_SYS_RESOURCE"] +fn privileged_layered_runtime_delivers_a_write_collected_status_before_driver_events() { + isolated( + "NVX_LAYERED_RUNTIME_OUTPUT_STATUS_CHILD", + "guest::managed_agent::dispatch::tests::runtime::privileged_layered_runtime_delivers_a_write_collected_status_before_driver_events", + || { + let temporary = tempfile::tempdir().unwrap(); + let pid_path = temporary.path().join("leader"); + let gate = sys::path_c_string(&temporary.path().join("gate")).unwrap(); + crate::guest::sys::tests::create_fifo(&gate, sys::Mode::from_bits_retain(0o600)) + .unwrap(); + let payload = workload_payload( + &[ + "/bin/sh", + "-c", + "printf '%s' \"$$\" > \"$1\"; head -c 6000 /dev/zero; read -r go < \"$2\"; exit 7", + "sh", + pid_path.to_str().unwrap(), + gate.to_str().unwrap(), + ], + Some("/"), + ) + .unwrap(); + let helper = helper(); + let parent = sys::open_pidfd(sys::current_process()).unwrap(); + let (descriptor, _peer, proof) = peer(|mut host, probe| { + let _guard = KillParentOnPanic(&parent); + narrow_output(&probe); + reset_peer(&mut host); + host.write_all( + &encode_outer( + OUTER_DATA, + &INSTANCE, + 1, + 1, + &encode_app(APP_EXEC, 9, 0, &payload), + ) + .unwrap(), + ) + .unwrap(); + assert_eq!(read_outer(&mut host).0.record_type, OUTER_CREDIT); + output_is_blocked(&probe); + drop(probe); + let id: i64 = std::fs::read_to_string(&pid_path).unwrap().parse().unwrap(); + let leader = sys::open_pidfd(ProcessId::from_positive(id).unwrap()).unwrap(); + let mut release = None; + until(|| { + match sys::open( + &gate, + sys::OpenFlags::O_WRONLY | sys::OpenFlags::O_NONBLOCK, + sys::Mode::empty(), + ) { + Ok(writer) => { + release = Some(writer); + true + } + Err(Errno::ENXIO) => false, + Err(error) => panic!("cannot release the owned leader: {error}"), + } + }); + release.unwrap().write_all(b"go\n").unwrap(); + assert!(sys::readable_or_hung_up(&[leader.borrowed()], 5000).unwrap()[0]); + until(|| !Path::new(&format!("/proc/{id}")).exists()); + let mut output = Vec::new(); + loop { + let frame = receive(&mut host); + if frame.kind == APP_EXIT { + assert_eq!(frame, response(APP_EXIT, 9, 7, b"exit")); + break; + } + assert_eq!((frame.kind, frame.request_id), (APP_STDOUT, 9)); + output.extend(frame.payload); + } + assert_eq!(output, vec![0; 6000]); + host.write_all( + &encode_outer( + OUTER_DATA, + &INSTANCE, + 1, + 2, + &encode_app(APP_STOP, 99, 0, &[]), + ) + .unwrap(), + ) + .unwrap(); + assert_eq!(read_outer(&mut host).0.record_type, OUTER_CREDIT); + assert_eq!(receive(&mut host), response(APP_STOPPED, 99, 0, &[])); + }); + let mut runtime = + Runtime::new(&[CString::new(helper.process.raw().to_string()).unwrap()]).unwrap(); + let mut session = Session::with_runtime(descriptor, &mut runtime).unwrap(); + let mut system = TestSystem::new(Paths::default()); + system.direct_starter = Some(super::super::super::seam::tests::start_as_privileged); + let config = + AgentConfig::layered(b"nvx_identity=1000:1000", c"workload".into(), c"/".into()) + .unwrap(); + let mut mappings = HostMappings::announced_for(&config, 0); + assert!(matches!( + run(&mut system, &mut session, &config, &mut mappings), + Ok(Outcome::Stopped) + )); + assert!(system.diagnostics.is_empty(), "{:?}", system.diagnostics); + drop(session); + drop(runtime); + completed(proof); + }, + ); +} diff --git a/nvx-rs/src/guest/managed_agent/exec.rs b/nvx-rs/src/guest/managed_agent/exec.rs index 843ce747a..69be440e1 100644 --- a/nvx-rs/src/guest/managed_agent/exec.rs +++ b/nvx-rs/src/guest/managed_agent/exec.rs @@ -120,9 +120,6 @@ pub(crate) fn run( } else { &mut buffer[..] }; - if let Some((id, status)) = session.take_status() { - workload.observe(id, status); - } state.forward(system, session, request_id, &mut workload, output)?; workload.collect(session)?; if state.running(&workload) @@ -519,6 +516,9 @@ impl RunState { buffer: &mut [u8], ) -> Result<(), Errno> { loop { + if let Some((id, status)) = session.take_status() { + workload.observe(id, status); + } let event = match workload.next_event(buffer) { Ok(Some(event)) => event, Ok(None) => return Ok(()), diff --git a/nvx-rs/src/guest/managed_agent/session.rs b/nvx-rs/src/guest/managed_agent/session.rs index ce6e0fc63..99111a654 100644 --- a/nvx-rs/src/guest/managed_agent/session.rs +++ b/nvx-rs/src/guest/managed_agent/session.rs @@ -28,6 +28,18 @@ struct Control<'r> { } impl Control<'_> { + fn service_leader(&self) -> Result<(), Errno> { + if self.service(|id, status| { + // Keep the first leader status across partial I/O, not unrelated orphans. + if id == self.leader.get() && self.status.get().is_none() { + self.status.set(Some((id, status))); + } + })? { + return Err(Errno::ECANCELED); + } + Ok(()) + } + fn service(&self, observed: impl FnMut(u32, ExitStatus)) -> Result { if self.outcome.borrow().is_err() { return Err(Errno::ECHILD); @@ -54,14 +66,7 @@ impl Read for Control<'_> { return Read::read(&mut self.descriptor, bytes); }; loop { - if self.service(|id, status| { - // Keep the first leader status across partial reads, not unrelated orphans. - if id == self.leader.get() && self.status.get().is_none() { - self.status.set(Some((id, status))); - } - })? { - return Err(Errno::ECANCELED.into()); - } + self.service_leader()?; let signals = sys::poll( [Pollable::File(&self.descriptor), runtime.borrow().wake()], None, @@ -71,7 +76,10 @@ impl Read for Control<'_> { } if !signals.readable { // A control-only wake includes HUP, which PollEvents does not expose. - return Read::read(&mut self.descriptor, bytes); + match self.descriptor.read(bytes) { + Err(Errno::EAGAIN) => {} + result => return result.map_err(io::Error::from), + } } } } @@ -79,7 +87,34 @@ impl Read for Control<'_> { impl Write for Control<'_> { fn write(&mut self, bytes: &[u8]) -> io::Result { - Write::write(&mut self.descriptor, bytes) + let Some(runtime) = &self.runtime else { + return Write::write(&mut self.descriptor, bytes); + }; + if bytes.is_empty() { + return Write::write(&mut self.descriptor, bytes); + } + loop { + self.service_leader()?; + match self.descriptor.write(bytes) { + Err(Errno::EAGAIN) => loop { + let events = sys::poll( + [ + Pollable::Writable(&self.descriptor), + runtime.borrow().wake(), + ], + None, + )?; + if events[1].error { + return Err(Errno::EIO.into()); + } + self.service_leader()?; + if events[0].writable || !events[1].readable { + break; + } + }, + result => return result.map_err(io::Error::from), + } + } } fn flush(&mut self) -> io::Result<()> { @@ -188,16 +223,20 @@ impl<'r> Session<'r> { .runtime_failure() .expect("the collector retained its failure") })?; + session + .descriptor() + .set_nonblocking() + .map_err(|error| format!("cannot configure layered control: {}", error.desc()))?; Ok(session) } - /// Registers the newly launched leader before any control read can collect it. + /// Registers the newly launched leader before any control I/O can collect it. pub(crate) fn watch(&self, id: u32) { self.connection.get_ref().leader.set(id); self.connection.get_ref().status.set(None); } - /// Takes a status collected while a partial control record was being read. + /// Takes a status collected while a control record was being read or written. pub(crate) fn take_status(&self) -> Option<(u32, ExitStatus)> { self.connection.get_ref().status.take() } diff --git a/nvx-rs/src/guest/sys/poll.rs b/nvx-rs/src/guest/sys/poll.rs index 65078733d..a6f9cfc0e 100644 --- a/nvx-rs/src/guest/sys/poll.rs +++ b/nvx-rs/src/guest/sys/poll.rs @@ -37,11 +37,13 @@ pub(crate) fn hung_up(descriptor: BorrowedFd<'_>) -> Result { .is_some_and(|events| events.contains(PollFlags::POLLHUP))) } -/// A descriptor owner that [`poll`] waits on for input. +/// A descriptor owner and the readiness that [`poll`] waits for. #[derive(Clone, Copy, Debug)] pub(crate) enum Pollable<'a> { /// A file, device, pipe, or socket. File(&'a FileDescriptor), + /// A file, device, pipe, or socket with space to write. + Writable(&'a FileDescriptor), /// A timer, which polls readable once it expires. Timer(&'a Timer), /// A signal descriptor, which polls readable while one of its signals is pending. @@ -54,7 +56,7 @@ impl Pollable<'_> { /// Borrows the owner's descriptor. fn borrowed(&self) -> BorrowedFd<'_> { match self { - Self::File(descriptor) => descriptor.borrowed(), + Self::File(descriptor) | Self::Writable(descriptor) => descriptor.borrowed(), Self::Timer(timer) => timer.borrowed(), Self::Signals(signals) => signals.borrowed(), Self::ChildSignal(wait) => wait.borrowed(), @@ -67,12 +69,14 @@ impl Pollable<'_> { pub(crate) struct PollEvents { /// The descriptor has input to read. pub(crate) readable: bool, - /// The descriptor has an error condition, which a read returns. + /// The descriptor has space to write. + pub(crate) writable: bool, + /// The descriptor has an error condition, which its next I/O returns. pub(crate) error: bool, } -/// Waits at most `timeout`, or without a limit for `None`, until one of `descriptors` has input -/// or an error, with one `poll` call and without allocating, and returns each descriptor's +/// Waits at most `timeout`, or without a limit for `None`, until one of `descriptors` has its +/// requested readiness or an error, with one `poll` call and without allocating, and returns their /// events in the caller's order. An interrupted call is `EINTR`, not a retry. /// /// A timeout of more than `i32::MAX` milliseconds waits for `i32::MAX` milliseconds, about 24 @@ -84,14 +88,19 @@ pub(crate) fn poll( let timeout = timeout.map_or(PollTimeout::NONE, |timeout| { PollTimeout::try_from(timeout.as_millis()).unwrap_or(PollTimeout::MAX) }); - let mut raw = descriptors - .each_ref() - .map(|descriptor| PollFd::new(descriptor.borrowed(), PollFlags::POLLIN)); + let mut raw = descriptors.each_ref().map(|descriptor| { + let interest = match descriptor { + Pollable::Writable(_) => PollFlags::POLLOUT, + _ => PollFlags::POLLIN, + }; + PollFd::new(descriptor.borrowed(), interest) + }); nix::poll::poll(&mut raw, timeout)?; Ok(raw.map(|descriptor| { let returned = descriptor.revents().unwrap_or(PollFlags::empty()); PollEvents { readable: returned.contains(PollFlags::POLLIN), + writable: returned.contains(PollFlags::POLLOUT), error: returned.contains(PollFlags::POLLERR), } })) diff --git a/nvx-rs/src/guest/sys/poll/tests.rs b/nvx-rs/src/guest/sys/poll/tests.rs index dfe8e631e..8a0c97cb9 100644 --- a/nvx-rs/src/guest/sys/poll/tests.rs +++ b/nvx-rs/src/guest/sys/poll/tests.rs @@ -19,8 +19,39 @@ fn poll_reports_input_in_the_callers_order() { PollEvents::default(), PollEvents { readable: true, + writable: false, error: false } + + #[test] + fn poll_distinguishes_backpressure_from_a_readable_wake() { + let (reader, writer) = create_pipe().unwrap(); + writer.set_nonblocking().unwrap(); + assert!( + poll([Pollable::Writable(&writer)], Some(Duration::ZERO)).unwrap()[0].writable + ); + let bytes = [0; 4096]; + loop { + match writer.write(&bytes) { + Ok(count) => assert!(count != 0), + Err(Errno::EAGAIN) => break, + Err(error) => panic!("cannot fill the owned pipe: {error}"), + } + } + let (wake, signal) = create_pipe().unwrap(); + signal.write_all(b"x").unwrap(); + let events = poll( + [Pollable::Writable(&writer), Pollable::File(&wake)], + Some(Duration::ZERO), + ) + .unwrap(); + assert_eq!(events[0], PollEvents::default()); + assert!(events[1].readable && !events[1].writable); + reader.read_exact(&mut [0; 4096]).unwrap(); + assert!( + poll([Pollable::Writable(&writer)], Some(Duration::ZERO)).unwrap()[0].writable + ); + } ] ); } From 9445f16ba83d62e48ca77e9be79aafe87071f587 Mon Sep 17 00:00:00 2001 From: Enrique Saurez Date: Mon, 9 Nov 2026 12:19:26 -0800 Subject: [PATCH 14/17] guest: keep writable readiness regression at module scope Move the new poll regression out of the existing assertion's token tree so Linux unit tests compile and the test inventory discovers it independently. Keep the original ordering assertions and the readiness regression unchanged. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 190278d5-cbae-4ec6-b47d-ddf469124598 --- nvx-rs/src/guest/sys/poll/tests.rs | 56 ++++++++++++++---------------- 1 file changed, 26 insertions(+), 30 deletions(-) diff --git a/nvx-rs/src/guest/sys/poll/tests.rs b/nvx-rs/src/guest/sys/poll/tests.rs index 8a0c97cb9..6054c63e0 100644 --- a/nvx-rs/src/guest/sys/poll/tests.rs +++ b/nvx-rs/src/guest/sys/poll/tests.rs @@ -22,36 +22,32 @@ fn poll_reports_input_in_the_callers_order() { writable: false, error: false } - - #[test] - fn poll_distinguishes_backpressure_from_a_readable_wake() { - let (reader, writer) = create_pipe().unwrap(); - writer.set_nonblocking().unwrap(); - assert!( - poll([Pollable::Writable(&writer)], Some(Duration::ZERO)).unwrap()[0].writable - ); - let bytes = [0; 4096]; - loop { - match writer.write(&bytes) { - Ok(count) => assert!(count != 0), - Err(Errno::EAGAIN) => break, - Err(error) => panic!("cannot fill the owned pipe: {error}"), - } - } - let (wake, signal) = create_pipe().unwrap(); - signal.write_all(b"x").unwrap(); - let events = poll( - [Pollable::Writable(&writer), Pollable::File(&wake)], - Some(Duration::ZERO), - ) - .unwrap(); - assert_eq!(events[0], PollEvents::default()); - assert!(events[1].readable && !events[1].writable); - reader.read_exact(&mut [0; 4096]).unwrap(); - assert!( - poll([Pollable::Writable(&writer)], Some(Duration::ZERO)).unwrap()[0].writable - ); - } ] ); } + +#[test] +fn poll_distinguishes_backpressure_from_a_readable_wake() { + let (reader, writer) = create_pipe().unwrap(); + writer.set_nonblocking().unwrap(); + assert!(poll([Pollable::Writable(&writer)], Some(Duration::ZERO)).unwrap()[0].writable); + let bytes = [0; 4096]; + loop { + match writer.write(&bytes) { + Ok(count) => assert!(count != 0), + Err(Errno::EAGAIN) => break, + Err(error) => panic!("cannot fill the owned pipe: {error}"), + } + } + let (wake, signal) = create_pipe().unwrap(); + signal.write_all(b"x").unwrap(); + let events = poll( + [Pollable::Writable(&writer), Pollable::File(&wake)], + Some(Duration::ZERO), + ) + .unwrap(); + assert_eq!(events[0], PollEvents::default()); + assert!(events[1].readable && !events[1].writable); + reader.read_exact(&mut [0; 4096]).unwrap(); + assert!(poll([Pollable::Writable(&writer)], Some(Duration::ZERO)).unwrap()[0].writable); +} From c7e971607a0ea3b638800b4d90a3fd549685c42b Mon Sep 17 00:00:00 2001 From: Enrique Saurez Date: Mon, 9 Nov 2026 12:50:29 -0800 Subject: [PATCH 15/17] guest: decode runtime test frames with a typed header Convert the bounded header slice using the existing session-test pattern so the Linux-only regression compiles against the fixed-size protocol decoder. Keep all framing assertions and production behavior unchanged. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 190278d5-cbae-4ec6-b47d-ddf469124598 --- nvx-rs/src/guest/managed_agent/dispatch/tests/runtime.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/nvx-rs/src/guest/managed_agent/dispatch/tests/runtime.rs b/nvx-rs/src/guest/managed_agent/dispatch/tests/runtime.rs index 3418d55f3..e9bfd863a 100644 --- a/nvx-rs/src/guest/managed_agent/dispatch/tests/runtime.rs +++ b/nvx-rs/src/guest/managed_agent/dispatch/tests/runtime.rs @@ -558,7 +558,7 @@ fn reset_peer(host: &mut UnixStream) { fn only_stdout_prefix(mut traffic: &[u8]) { assert!(!traffic.is_empty()); while traffic.len() >= OUTER_HEADER_LEN { - let header = decode_outer_header(&traffic[..OUTER_HEADER_LEN]).unwrap(); + let header = decode_outer_header(traffic[..OUTER_HEADER_LEN].try_into().unwrap()).unwrap(); assert_eq!(header.record_type, OUTER_DATA); let length = OUTER_HEADER_LEN + header.length; if traffic.len() < length { From aae549ad10505a5adc1c38ccb616229afc6b877c Mon Sep 17 00:00:00 2001 From: Enrique Saurez Date: Mon, 9 Nov 2026 17:18:06 -0800 Subject: [PATCH 16/17] tests: discard self-connected listener clients A Linux loopback connect may choose its destination as its own source port before the guest publishes its listener. A normal close leaves that port in TIME_WAIT and blocks the host's later publication. Share the existing listener client between cold-launch and clone scenarios and reset only that self-connection before retrying. Keep connect timing, read deadlines, and greeting assertions unchanged. Cover publication rebinding with a real self-connected TCP stream and a normal exchange. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 0c16e8c2-e1da-4c5f-9b3b-5535db03b05b --- tests/runtime-adapter/tests/common/mod.rs | 71 +++++++++++++++- .../runtime-adapter/tests/listener_client.rs | 85 +++++++++++++++++++ tests/runtime-adapter/tests/vm_clone.rs | 35 ++------ tests/runtime-adapter/tests/vm_layered.rs | 31 +------ 4 files changed, 162 insertions(+), 60 deletions(-) create mode 100644 tests/runtime-adapter/tests/listener_client.rs diff --git a/tests/runtime-adapter/tests/common/mod.rs b/tests/runtime-adapter/tests/common/mod.rs index f27089851..a43302879 100644 --- a/tests/runtime-adapter/tests/common/mod.rs +++ b/tests/runtime-adapter/tests/common/mod.rs @@ -20,9 +20,11 @@ use std::env; use std::error::Error; use std::ffi::{OsStr, OsString}; use std::fs::{self, File}; -use std::io; +use std::io::{self, Read, Write}; +use std::net::{Shutdown, SocketAddr, TcpStream}; use std::path::{Path, PathBuf}; use std::process::{Child, Command, Stdio}; +use std::sync::atomic::{AtomicBool, Ordering}; use std::thread; use std::time::{Duration, Instant}; @@ -468,6 +470,73 @@ impl Drop for Sandbox { } } +/// Exchanges a line each way with the guest until it succeeds or `stop` is set. +/// A connect runs to completion: cancelling a Windows loopback connect can leave a late +/// connection that consumes the guest's one-shot listener without exchanging its greeting. +pub(crate) fn reach_guest( + address: SocketAddr, + stop: &AtomicBool, + greeting: &str, + answer: &str, +) -> bool { + reach_guest_with(address, stop, greeting, answer, TcpStream::connect) +} + +/// Runs the listener client with a connector that can supply a real socket of a known kind. +pub(crate) fn reach_guest_with( + address: SocketAddr, + stop: &AtomicBool, + greeting: &str, + answer: &str, + mut connect: impl FnMut(SocketAddr) -> io::Result, +) -> bool { + while !stop.load(Ordering::SeqCst) { + let connected = connect(address); + #[cfg(target_os = "linux")] + let connected = connected.and_then(|stream| { + if stream.local_addr().unwrap() == stream.peer_addr().unwrap() { + // Linux can assign the unpublished destination as this client's source port. + // Reset that self-connection so TIME_WAIT cannot block the later publication. + nix::sys::socket::setsockopt( + &stream, + nix::sys::socket::sockopt::Linger, + &nix::libc::linger { + l_onoff: 1, + l_linger: 0, + }, + ) + .expect("reset the client's self-connection before publishing the guest's port"); + return Err(io::ErrorKind::ConnectionAborted.into()); + } + Ok(stream) + }); + if let Ok(mut stream) = connected { + stream + .set_read_timeout(Some(Duration::from_secs(5))) + .unwrap(); + let mut received = Vec::new(); + let mut byte = [0]; + while !received.ends_with(b"\n") && matches!(stream.read(&mut byte), Ok(1)) { + received.push(byte[0]); + } + if received == format!("{greeting}\n").as_bytes() { + stream.write_all(format!("{answer}\n").as_bytes()).unwrap(); + stream.shutdown(Shutdown::Write).unwrap(); + let mut rest = Vec::new(); + stream.read_to_end(&mut rest).unwrap(); + assert!(rest.is_empty(), "the guest's listener then sent {rest:?}"); + return true; + } + assert!( + received.is_empty(), + "the guest's listener greeted the host with {received:?}" + ); + } + thread::sleep(Duration::from_millis(100)); + } + false +} + /// Describes `error`, with its causes. pub(crate) fn describe(error: &dyn Error) -> String { let mut description = error.to_string(); diff --git a/tests/runtime-adapter/tests/listener_client.rs b/tests/runtime-adapter/tests/listener_client.rs new file mode 100644 index 000000000..fa57bc104 --- /dev/null +++ b/tests/runtime-adapter/tests/listener_client.rs @@ -0,0 +1,85 @@ +//! Real TCP regressions for the hardware scenarios' listener client. +#![cfg(feature = "host")] + +mod common; + +use std::io::{Read, Write}; +use std::net::TcpListener; +use std::sync::atomic::AtomicBool; +use std::thread; + +#[test] +fn a_published_listener_exchanges_its_greeting() { + let listener = TcpListener::bind(("127.0.0.1", 0)).unwrap(); + let address = listener.local_addr().unwrap(); + thread::scope(|scope| { + let server = scope.spawn(|| { + let (mut stream, _) = listener.accept().unwrap(); + stream.write_all(b"guest\n").unwrap(); + let mut answer = Vec::new(); + stream.read_to_end(&mut answer).unwrap(); + assert_eq!(answer, b"host\n"); + }); + assert!(common::reach_guest( + address, + &AtomicBool::new(false), + "guest", + "host" + )); + server.join().unwrap(); + }); +} + +#[test] +#[cfg(target_os = "linux")] +fn a_self_connection_does_not_hold_the_publication_port() { + use std::net::TcpStream; + use std::os::fd::AsRawFd; + use std::sync::atomic::Ordering; + + use nix::sys::socket::{ + AddressFamily, SockFlag, SockProtocol, SockType, SockaddrIn, bind, connect, getsockname, + setsockopt, socket, sockopt, + }; + + let client = socket( + AddressFamily::Inet, + SockType::Stream, + SockFlag::SOCK_CLOEXEC, + SockProtocol::Tcp, + ) + .unwrap(); + bind(client.as_raw_fd(), &SockaddrIn::new(127, 0, 0, 1, 0)).unwrap(); + let address: SockaddrIn = getsockname(client.as_raw_fd()).unwrap(); + connect(client.as_raw_fd(), &address).unwrap(); + let stream = TcpStream::from(client); + assert_eq!(stream.local_addr().unwrap(), stream.peer_addr().unwrap()); + let peer = stream.peer_addr().unwrap(); + let mut supplied = Some(stream); + let stop = AtomicBool::new(false); + assert!(!common::reach_guest_with( + peer, + &stop, + "guest", + "host", + |_| { + if let Some(stream) = supplied.take() { + Ok(stream) + } else { + stop.store(true, Ordering::SeqCst); + Err(std::io::ErrorKind::ConnectionRefused.into()) + } + } + )); + + let publisher = socket( + AddressFamily::Inet, + SockType::Stream, + SockFlag::SOCK_CLOEXEC, + SockProtocol::Tcp, + ) + .unwrap(); + setsockopt(&publisher, sockopt::ReuseAddr, &true).unwrap(); + bind(publisher.as_raw_fd(), &address) + .expect("the client must not leave the guest's publication port in TIME_WAIT"); +} diff --git a/tests/runtime-adapter/tests/vm_clone.rs b/tests/runtime-adapter/tests/vm_clone.rs index 21b4414ce..654fad486 100644 --- a/tests/runtime-adapter/tests/vm_clone.rs +++ b/tests/runtime-adapter/tests/vm_clone.rs @@ -208,11 +208,11 @@ use std::env; use std::ffi::{OsStr, OsString}; use std::fs::{self, File}; use std::hash::{BuildHasher, RandomState}; -use std::io::{self, Read, Write}; +use std::io; +#[cfg(target_os = "linux")] +use std::io::Read; use std::mem; -use std::net::{ - IpAddr, Ipv4Addr, Ipv6Addr, Shutdown, SocketAddr, TcpListener, TcpStream, UdpSocket, -}; +use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr, TcpListener, UdpSocket}; use std::num::NonZeroU64; use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicBool, Ordering}; @@ -4786,32 +4786,7 @@ fn a_clone_whose_mapping_fails_ends_before_its_sandbox_starts() { /// timeout gave up on could still open once OpenVMM publishes the listener, which would then take /// that connection, already closed, for its one client. fn reach_guest(address: SocketAddr, stop: &AtomicBool) -> bool { - while !stop.load(Ordering::SeqCst) { - if let Ok(mut stream) = TcpStream::connect(address) { - stream - .set_read_timeout(Some(Duration::from_secs(5))) - .unwrap(); - let mut greeting = Vec::new(); - let mut byte = [0]; - while !greeting.ends_with(b"\n") && matches!(stream.read(&mut byte), Ok(1)) { - greeting.push(byte[0]); - } - if greeting == format!("{GREETING}\n").as_bytes() { - stream.write_all(format!("{ANSWER}\n").as_bytes()).unwrap(); - stream.shutdown(Shutdown::Write).unwrap(); - let mut rest = Vec::new(); - stream.read_to_end(&mut rest).unwrap(); - assert!(rest.is_empty(), "the guest's listener then sent {rest:?}"); - return true; - } - assert!( - greeting.is_empty(), - "the guest's listener greeted the host with {greeting:?}" - ); - } - thread::sleep(Duration::from_millis(100)); - } - false + common::reach_guest(address, stop, GREETING, ANSWER) } /// Sets its flag once it drops, so that a thread that waits for the flag ends even when the check diff --git a/tests/runtime-adapter/tests/vm_layered.rs b/tests/runtime-adapter/tests/vm_layered.rs index 00307e614..5284504fa 100644 --- a/tests/runtime-adapter/tests/vm_layered.rs +++ b/tests/runtime-adapter/tests/vm_layered.rs @@ -43,9 +43,7 @@ use std::ffi::{OsStr, OsString}; use std::fs::{self, File}; use std::hash::{BuildHasher, RandomState}; use std::io::{self, Read, Seek, SeekFrom, Write}; -use std::net::{ - IpAddr, Ipv4Addr, Ipv6Addr, Shutdown, SocketAddr, TcpListener, TcpStream, UdpSocket, -}; +use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr, TcpListener, TcpStream, UdpSocket}; use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Mutex, PoisonError}; @@ -3017,32 +3015,7 @@ fn check_guest_listener(vm: &mut Vm<'_>, address: SocketAddr, reached: bool) { /// OpenVMM publishes the listener, which would then take that connection, already closed, for its /// one client. fn reach_guest(address: SocketAddr, stop: &AtomicBool) -> bool { - while !stop.load(Ordering::SeqCst) { - if let Ok(mut stream) = TcpStream::connect(address) { - stream - .set_read_timeout(Some(Duration::from_secs(5))) - .unwrap(); - let mut greeting = Vec::new(); - let mut byte = [0]; - while !greeting.ends_with(b"\n") && matches!(stream.read(&mut byte), Ok(1)) { - greeting.push(byte[0]); - } - if greeting == format!("{GREETING}\n").as_bytes() { - stream.write_all(format!("{ANSWER}\n").as_bytes()).unwrap(); - stream.shutdown(Shutdown::Write).unwrap(); - let mut rest = Vec::new(); - stream.read_to_end(&mut rest).unwrap(); - assert!(rest.is_empty(), "the guest's listener then sent {rest:?}"); - return true; - } - assert!( - greeting.is_empty(), - "the guest's listener greeted the host with {greeting:?}" - ); - } - thread::sleep(Duration::from_millis(100)); - } - false + common::reach_guest(address, stop, GREETING, ANSWER) } /// Serves the test proxy's `listener` until `stop`: answers each request with [`PROXY_RESPONSE`], From 061b99e153b52afa5554bcf6eec4e3408fa8bd63 Mon Sep 17 00:00:00 2001 From: Enrique Saurez Date: Mon, 9 Nov 2026 18:22:14 -0800 Subject: [PATCH 17/17] tests: retry listener connections reset before inspection Keep the original retry behavior when an ingress policy resets a connected client before peer inspection. Prove the race with a real reset socket, bound the positive listener control, and document the shared test fixture. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 0c16e8c2-e1da-4c5f-9b3b-5535db03b05b --- tests/runtime-adapter/README.md | 9 ++- tests/runtime-adapter/tests/common/mod.rs | 7 +- .../runtime-adapter/tests/listener_client.rs | 77 +++++++++++++++++-- 3 files changed, 82 insertions(+), 11 deletions(-) diff --git a/tests/runtime-adapter/README.md b/tests/runtime-adapter/README.md index 5c16c3525..d700bda5f 100644 --- a/tests/runtime-adapter/README.md +++ b/tests/runtime-adapter/README.md @@ -2559,6 +2559,11 @@ configuration, on Linux and Windows: in the initramfs too, and missing install metadata, which fails as NVX reports it. +`tests/listener_client.rs` checks, in the host-only configuration, the real TCP +greeting and answer of the VM tests' listener client on Linux and Windows. +On Linux it also checks that a self-connection leaves the guest's publication +port bindable and that a connection reset before inspection is retried. + `tests/staged_images.rs` checks, in the host-only configuration, on Linux and Windows: @@ -3217,8 +3222,8 @@ layered roots, captures of templates, and the starts and runs of their clones. The VM tests, `tests/vm_restart.rs`, `tests/vm_steps.rs`, `tests/vm_layered.rs`, `tests/vm_template.rs`, and `tests/vm_clone.rs`, share -their sandboxes and -the fixtures of their layered roots through `tests/common`. The sandboxes start +their sandboxes, the fixtures of their layered roots, and their host listener +client through `tests/common`. The sandboxes start the host program's commands with `nvx::host::supervisor::spawn_command`, so that on Windows a command cannot inherit the handles of a VM that a test launches in its own process meanwhile, such as its log and its capability's diff --git a/tests/runtime-adapter/tests/common/mod.rs b/tests/runtime-adapter/tests/common/mod.rs index a43302879..3c3540923 100644 --- a/tests/runtime-adapter/tests/common/mod.rs +++ b/tests/runtime-adapter/tests/common/mod.rs @@ -494,7 +494,12 @@ pub(crate) fn reach_guest_with( let connected = connect(address); #[cfg(target_os = "linux")] let connected = connected.and_then(|stream| { - if stream.local_addr().unwrap() == stream.peer_addr().unwrap() { + let peer = match stream.peer_addr() { + Ok(peer) => peer, + Err(error) if error.kind() == io::ErrorKind::NotConnected => return Err(error), + Err(error) => panic!("inspect the listener client's peer: {error}"), + }; + if stream.local_addr().unwrap() == peer { // Linux can assign the unpublished destination as this client's source port. // Reset that self-connection so TIME_WAIT cannot block the later publication. nix::sys::socket::setsockopt( diff --git a/tests/runtime-adapter/tests/listener_client.rs b/tests/runtime-adapter/tests/listener_client.rs index fa57bc104..f1a3c9f64 100644 --- a/tests/runtime-adapter/tests/listener_client.rs +++ b/tests/runtime-adapter/tests/listener_client.rs @@ -5,14 +5,24 @@ mod common; use std::io::{Read, Write}; use std::net::TcpListener; -use std::sync::atomic::AtomicBool; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::mpsc; use std::thread; +use std::time::Duration; #[test] fn a_published_listener_exchanges_its_greeting() { let listener = TcpListener::bind(("127.0.0.1", 0)).unwrap(); let address = listener.local_addr().unwrap(); + let stop = AtomicBool::new(false); + let (finished, completion) = mpsc::channel(); thread::scope(|scope| { + let stop_ref = &stop; + scope.spawn(move || { + if completion.recv_timeout(Duration::from_secs(5)).is_err() { + stop_ref.store(true, Ordering::SeqCst); + } + }); let server = scope.spawn(|| { let (mut stream, _) = listener.accept().unwrap(); stream.write_all(b"guest\n").unwrap(); @@ -20,22 +30,73 @@ fn a_published_listener_exchanges_its_greeting() { stream.read_to_end(&mut answer).unwrap(); assert_eq!(answer, b"host\n"); }); - assert!(common::reach_guest( - address, - &AtomicBool::new(false), - "guest", - "host" - )); + let reached = common::reach_guest(address, &stop, "guest", "host"); + finished + .send(()) + .expect("the listener exchange must finish before its deadline"); + assert!(reached); server.join().unwrap(); }); } +#[test] +#[cfg(target_os = "linux")] +fn a_connection_reset_before_inspection_is_retried() { + use std::net::TcpStream; + + use nix::sys::socket::{setsockopt, sockopt}; + + let listener = TcpListener::bind(("127.0.0.1", 0)).unwrap(); + let address = listener.local_addr().unwrap(); + let mut stream = TcpStream::connect(address).unwrap(); + let (server, _) = listener.accept().unwrap(); + setsockopt( + &server, + sockopt::Linger, + &nix::libc::linger { + l_onoff: 1, + l_linger: 0, + }, + ) + .unwrap(); + drop(server); + stream + .set_read_timeout(Some(Duration::from_secs(1))) + .unwrap(); + assert_eq!( + stream.read(&mut [0]).unwrap_err().kind(), + std::io::ErrorKind::ConnectionReset + ); + + let mut supplied = Some(stream); + let mut retried = false; + let stop = AtomicBool::new(false); + assert!(!common::reach_guest_with( + address, + &stop, + "guest", + "host", + |_| { + if let Some(stream) = supplied.take() { + Ok(stream) + } else { + retried = true; + stop.store(true, Ordering::SeqCst); + Err(std::io::ErrorKind::ConnectionRefused.into()) + } + } + )); + assert!( + retried, + "a reset connection must retry instead of panicking" + ); +} + #[test] #[cfg(target_os = "linux")] fn a_self_connection_does_not_hold_the_publication_port() { use std::net::TcpStream; use std::os::fd::AsRawFd; - use std::sync::atomic::Ordering; use nix::sys::socket::{ AddressFamily, SockFlag, SockProtocol, SockType, SockaddrIn, bind, connect, getsockname,