Implement supervised rootless network readiness

crates/lazaret-linux/Cargo.toml

@@ -8,6 +8,7 @@ license.workspace = true
 [dependencies]
 lazaret-core = { version = "=0.0.0", path = "../lazaret-core" }
 rustix.workspace = true
+serde.workspace = true
 
 [[bin]]
 name = "lazaret-launch-wrapper"

crates/lazaret-linux/src/ipc.rs

@@ -1,6 +1,12 @@
-use std::os::fd::OwnedFd;
+use std::io;
+use std::os::fd::{AsFd, OwnedFd};
 
-use rustix::net::{AddressFamily, SocketFlags, SocketType, socketpair};
+use lazaret_core::codec::{decode, encode};
+use rustix::net::{
+    AddressFamily, RecvFlags, SendFlags, SocketFlags, SocketType, recv, send, socketpair,
+};
+
+const MAX_PRIVATE_PACKET: usize = 64 * 1024;
 
 /// # Errors
 /// Returns an error when the private sequenced-packet pair cannot be created.
@@ -26,6 +32,35 @@ pub fn private_stream_pair() -> std::io::Result<(OwnedFd, OwnedFd)> {
     .map_err(Into::into)
 }
 
+/// # Errors
+/// Returns an error when a typed private packet is oversized or cannot be sent atomically.
+pub fn send_packet<T: serde::Serialize>(descriptor: impl AsFd, message: &T) -> io::Result<()> {
+    let encoded = encode(message).map_err(|error| io::Error::other(error.to_string()))?;
+    if encoded.is_empty() || encoded.len() > MAX_PRIVATE_PACKET {
+        return Err(io::Error::other("private packet has invalid size"));
+    }
+    let sent = send(descriptor.as_fd(), &encoded, SendFlags::empty())?;
+    if sent == encoded.len() {
+        Ok(())
+    } else {
+        Err(io::Error::other("private packet send was incomplete"))
+    }
+}
+
+/// # Errors
+/// Returns an error when a private packet is empty, oversized, malformed, or unknown-version.
+pub fn receive_packet<T>(descriptor: impl AsFd) -> io::Result<T>
+where
+    T: serde::de::DeserializeOwned + serde::Serialize,
+{
+    let mut encoded = vec![0_u8; MAX_PRIVATE_PACKET + 1];
+    let (_, received) = recv(descriptor.as_fd(), &mut encoded, RecvFlags::empty())?;
+    if received == 0 || received > MAX_PRIVATE_PACKET {
+        return Err(io::Error::other("private packet has invalid size"));
+    }
+    decode(&encoded[..received]).map_err(|error| io::Error::other(error.to_string()))
+}
+
 #[cfg(test)]
 mod tests {
     use std::os::fd::AsFd;

crates/lazaret-linux/src/process.rs

@@ -272,6 +272,55 @@ fn require_captured_parent_edge(parent_edge: ParentEdge) -> io::Result<()> {
 }
 
 impl ChildBroker {
+    /// Runs a non-durable direct child under the same spawn/reap exclusion as owned children.
+    ///
+    /// # Errors
+    /// Returns an error when causal launch, exact waiting, or the command exit status fails.
+    #[allow(clippy::disallowed_methods, clippy::disallowed_types)]
+    pub fn run_transient(&self, target: &TargetCommand) -> io::Result<ProcessExit> {
+        let _guard = self
+            .exclusion
+            .lock()
+            .map_err(|_| io::Error::other("child broker lock poisoned"))?;
+        require_parent_edge_standard_io(target.parent_edge(), ChildStandardIoPolicy::Silenced)?;
+        require_default_sigchld()?;
+        require_reapable_sigchld()?;
+        let parent = current_process_identity()?;
+        let (release, wait) = launch_barrier()?;
+        let mut command = private_role_command(WRAPPER_ROLE);
+        append_parent_identity(&mut command, parent);
+        target.append_to(&mut command);
+        command
+            .stdin(wait.into_stdio())
+            .stdout(Stdio::null())
+            .stderr(Stdio::null());
+        let mut child = command.spawn()?;
+        let process = match acquire_spawned_child(child.id()) {
+            Ok(process) => process,
+            Err(error) => return fail_before_pidfd(release, &mut child, error),
+        };
+        if let Err(error) = release.release() {
+            return fail_after_pidfd_closed(process, error);
+        }
+        drop(child);
+        let status =
+            rustix::process::waitid(WaitId::PidFd(process.pidfd.as_fd()), WaitIdOptions::EXITED)?
+                .ok_or_else(|| io::Error::other("transient child exited without wait status"))?;
+        let exit = ProcessExit {
+            exit_code: status.exit_status(),
+            terminating_signal: status.terminating_signal(),
+        };
+        drop(process);
+        if exit.exit_code == Some(0) && exit.terminating_signal.is_none() {
+            Ok(exit)
+        } else {
+            Err(io::Error::other(format!(
+                "transient child failed with exit {:?} and signal {:?}",
+                exit.exit_code, exit.terminating_signal
+            )))
+        }
+    }
+
     /// Spawns the repository wrapper, stages exact ownership, accepts its acknowledgment,
     /// and only then releases target execution.
     ///
@@ -308,13 +357,46 @@ impl ChildBroker {
         expected_parent: PidfsProcessIdentityV1,
         ownership: ProcessOwnershipContext,
         target: &TargetCommand,
-        mut standard_io: ChildStandardIo,
+        standard_io: ChildStandardIo,
         stage: Stage,
         acknowledge: Acknowledge,
     ) -> io::Result<RetainedProcess>
     where
         Stage: FnOnce(StagedProcessOwnership) -> io::Result<()>,
         Acknowledge: FnOnce(StagedProcessOwnership) -> io::Result<ProcessOwnershipCommitAck>,
+    {
+        self.acquire_with_stdio_after_ack(
+            expected_parent,
+            ownership,
+            target,
+            standard_io,
+            stage,
+            acknowledge,
+            || Ok(()),
+        )
+    }
+
+    /// Acquires an exact child while retaining its blocked wrapper across one acknowledged
+    /// parent transition.
+    ///
+    /// # Errors
+    /// Refuses and reaps the blocked wrapper when staging, acknowledgment, transition, or release
+    /// fails.
+    #[allow(clippy::disallowed_methods, clippy::disallowed_types)]
+    pub fn acquire_with_stdio_after_ack<Stage, Acknowledge, Transition>(
+        &self,
+        expected_parent: PidfsProcessIdentityV1,
+        ownership: ProcessOwnershipContext,
+        target: &TargetCommand,
+        mut standard_io: ChildStandardIo,
+        stage: Stage,
+        acknowledge: Acknowledge,
+        transition: Transition,
+    ) -> io::Result<RetainedProcess>
+    where
+        Stage: FnOnce(StagedProcessOwnership) -> io::Result<()>,
+        Acknowledge: FnOnce(StagedProcessOwnership) -> io::Result<ProcessOwnershipCommitAck>,
+        Transition: FnOnce() -> io::Result<()>,
     {
         let _guard = self
             .exclusion
@@ -364,6 +446,9 @@ impl ChildBroker {
                 ),
             );
         }
+        if let Err(error) = transition() {
+            return fail_after_pidfd(release, process, error);
+        }
         if let Err(error) = release.release() {
             return fail_after_pidfd_closed(process, error);
         }

crates/lazaret-linux/tests/acquisition.rs

@@ -44,6 +44,7 @@ fn exit(result: std::io::Result<()>) -> ExitCode {
 
 fn run_all() -> std::io::Result<()> {
     successful_acquisition()?;
+    acknowledged_transition_precedes_exact_wrapper_release()?;
     lifetime_pipe_descriptor_contract()?;
     failed_stages_and_acknowledgments()?;
     production_wrapper_refuses_former_harness_path()?;
@@ -53,6 +54,47 @@ fn run_all() -> std::io::Result<()> {
     concurrent_acquisitions_serialize()
 }
 
+#[allow(clippy::disallowed_methods, clippy::disallowed_types)]
+fn acknowledged_transition_precedes_exact_wrapper_release() -> std::io::Result<()> {
+    let target_marker = marker("transition-target");
+    let parent = current_process_identity()?;
+    let staged = std::cell::Cell::new(None);
+    let acknowledged = std::cell::Cell::new(false);
+    let target = TargetCommand::new("/usr/bin/touch").argument(&target_marker);
+    let process = child_broker().acquire_with_stdio_after_ack(
+        parent,
+        context(),
+        &target,
+        ChildStandardIo::inherit(),
+        |ownership| {
+            staged.set(Some(ownership));
+            Ok(())
+        },
+        |ownership| {
+            if staged.get() != Some(ownership) {
+                return Err(std::io::Error::other("transition staged ownership changed"));
+            }
+            acknowledged.set(true);
+            Ok(ProcessOwnershipCommitAck::v1(ownership))
+        },
+        || {
+            if !acknowledged.get() || target_marker.exists() {
+                return Err(std::io::Error::other(
+                    "wrapper was released outside the acknowledged transition boundary",
+                ));
+            }
+            Ok(())
+        },
+    )?;
+    child_broker().reap(process)?;
+    if !target_marker.is_file() {
+        return Err(std::io::Error::other(
+            "acknowledged transition did not release the exact wrapper",
+        ));
+    }
+    std::fs::remove_file(target_marker)
+}
+
 #[allow(clippy::disallowed_methods, clippy::disallowed_types)]
 fn lifetime_pipe_descriptor_contract() -> std::io::Result<()> {
     let held_pipe_marker = marker("lifetime-pipe");