soma_self_update/
validation.rs

1use std::process::Stdio;
2
3#[cfg(windows)]
4use process_wrap::tokio::JobObject;
5#[cfg(unix)]
6use process_wrap::tokio::ProcessGroup;
7use process_wrap::tokio::{ChildWrapper, CommandWrap, KillOnDrop};
8use tokio::io::{AsyncRead, AsyncReadExt};
9
10#[cfg(unix)]
11use crate::staging::VALIDATION_MODE;
12use crate::{Result, StagedArtifact, UpdateError, Updater};
13
14const OUTPUT_LIMIT: usize = 16 * 1024;
15
16/// An artifact that executed successfully and reported its exact target version.
17#[derive(Debug)]
18pub struct ValidatedArtifact {
19    pub(crate) staged: StagedArtifact,
20    #[cfg(unix)]
21    pub(crate) identity: ArtifactIdentity,
22}
23
24#[cfg(unix)]
25#[derive(Clone, Copy, Debug, Eq, PartialEq)]
26pub(crate) struct ArtifactIdentity {
27    device: u64,
28    inode: u64,
29}
30
31#[cfg(unix)]
32impl ArtifactIdentity {
33    pub(crate) fn from_metadata(metadata: &std::fs::Metadata) -> Self {
34        use std::os::unix::fs::MetadataExt;
35        Self {
36            device: metadata.dev(),
37            inode: metadata.ino(),
38        }
39    }
40}
41
42impl ValidatedArtifact {
43    pub fn path(&self) -> &std::path::Path {
44        self.staged.path()
45    }
46    pub fn target_version(&self) -> &str {
47        self.staged.target_version()
48    }
49    pub fn sha256(&self) -> &str {
50        self.staged.sha256()
51    }
52
53    #[cfg(unix)]
54    pub(crate) fn intended_mode(&self) -> u32 {
55        self.staged.intended_mode
56    }
57
58    #[cfg(unix)]
59    pub(crate) fn revalidate_source_executable(&self, path: &std::path::Path) -> Result<()> {
60        self.staged.revalidate_source_executable(path)
61    }
62}
63
64impl Updater {
65    /// Executes `--version` and consumes the staged artifact on success.
66    pub async fn validate(&self, staged: StagedArtifact) -> Result<ValidatedArtifact> {
67        let path = staged.path().to_path_buf();
68        #[cfg(unix)]
69        let identity = validated_path_identity(&path)?;
70        let timeout = self.policy().validation_timeout();
71        let deadline = tokio::time::Instant::now() + timeout;
72        let child = match tokio::time::timeout_at(deadline, spawn_validator(&path)).await {
73            Ok(result) => result?,
74            Err(_) => return Err(UpdateError::ValidationTimedOut { timeout }),
75        };
76        let mut child = ValidationProcessGuard::new(child);
77        let stdout = child
78            .child_mut()
79            .stdout()
80            .take()
81            .expect("piped stdout is configured");
82        let stderr = child
83            .child_mut()
84            .stderr()
85            .take()
86            .expect("piped stderr is configured");
87        let completed = tokio::time::timeout_at(deadline, async {
88            // Descendants may inherit the validator's output handles. Tear down
89            // the whole group as soon as the leader exits so the readers can
90            // observe EOF without turning a successful validation into a timeout.
91            let status = async {
92                let status = child.leader_mut().wait().await;
93                let terminated = child.terminate_and_drain(&path).await;
94                match (status, terminated) {
95                    (Ok(status), Ok(())) => Ok(status),
96                    (Err(error), _) => Err(UpdateError::io(&path, error)),
97                    (Ok(_), Err(error)) => Err(error),
98                }
99            };
100            let (status, stdout, stderr) =
101                tokio::join!(status, read_bounded(stdout), read_bounded(stderr));
102            (status, stdout, stderr)
103        })
104        .await;
105        let (status, stdout, stderr) = match completed {
106            Ok((status, stdout, stderr)) => (
107                status?,
108                stdout.map_err(|error| UpdateError::io(&path, error))?,
109                stderr.map_err(|error| UpdateError::io(&path, error))?,
110            ),
111            Err(_) => {
112                let _ = Box::into_pin(child.child_mut().kill()).await;
113                child.disarm();
114                return Err(UpdateError::ValidationTimedOut { timeout });
115            }
116        };
117        if stdout.overflowed {
118            return Err(UpdateError::ValidationOutputTooLarge {
119                stream: "stdout",
120                limit: OUTPUT_LIMIT,
121            });
122        }
123        if stderr.overflowed {
124            return Err(UpdateError::ValidationOutputTooLarge {
125                stream: "stderr",
126                limit: OUTPUT_LIMIT,
127            });
128        }
129        let stderr_text = String::from_utf8_lossy(&stderr.bytes).into_owned();
130        if !status.success() {
131            return Err(UpdateError::ValidationFailed {
132                code: status.code(),
133                stderr: stderr_text,
134            });
135        }
136        let output =
137            String::from_utf8(stdout.bytes).map_err(|_| UpdateError::InvalidVersionOutput)?;
138        let expected = staged.target_version();
139        let matches = output.split_ascii_whitespace().any(|token| {
140            token.trim_matches(|character: char| character.is_ascii_punctuation()) == expected
141        });
142        if !matches {
143            return Err(UpdateError::VersionMismatch {
144                expected: expected.to_owned(),
145                output: output.trim().to_owned(),
146            });
147        }
148        #[cfg(unix)]
149        if validated_path_identity(&path)? != identity {
150            return Err(UpdateError::ArtifactIdentityChanged { path });
151        }
152        Ok(ValidatedArtifact {
153            staged,
154            #[cfg(unix)]
155            identity,
156        })
157    }
158}
159
160struct ValidationProcessGuard {
161    child: Box<dyn ChildWrapper>,
162    armed: bool,
163}
164
165impl ValidationProcessGuard {
166    fn new(child: Box<dyn ChildWrapper>) -> Self {
167        Self { child, armed: true }
168    }
169
170    fn child_mut(&mut self) -> &mut dyn ChildWrapper {
171        self.child.as_mut()
172    }
173
174    fn leader_mut(&mut self) -> &mut dyn ChildWrapper {
175        self.child.inner_mut()
176    }
177
178    async fn terminate_and_drain(&mut self, path: &std::path::Path) -> Result<()> {
179        match self.child.start_kill() {
180            Ok(()) => {}
181            #[cfg(unix)]
182            Err(error) if error.raw_os_error() == Some(nix::libc::ESRCH) => {}
183            Err(error) => return Err(UpdateError::io(path, error)),
184        }
185        self.child
186            .wait()
187            .await
188            .map_err(|error| UpdateError::io(path, error))?;
189        self.disarm();
190        Ok(())
191    }
192
193    fn disarm(&mut self) {
194        self.armed = false;
195    }
196}
197
198impl Drop for ValidationProcessGuard {
199    fn drop(&mut self) {
200        if self.armed {
201            let _ = self.child.start_kill();
202        }
203    }
204}
205
206#[cfg(unix)]
207fn validated_path_identity(path: &std::path::Path) -> Result<ArtifactIdentity> {
208    use std::os::unix::fs::PermissionsExt;
209
210    let metadata = std::fs::symlink_metadata(path).map_err(|error| UpdateError::io(path, error))?;
211    if !metadata.file_type().is_file() || metadata.permissions().mode() & 0o7777 != VALIDATION_MODE
212    {
213        return Err(UpdateError::InvalidStagedArtifact {
214            path: path.to_path_buf(),
215        });
216    }
217    Ok(ArtifactIdentity::from_metadata(&metadata))
218}
219
220async fn spawn_validator(path: &std::path::Path) -> Result<Box<dyn ChildWrapper>> {
221    // Tokio's asynchronous file close can briefly race exec on Linux and
222    // surface ETXTBSY even after the staged writer has been flushed and
223    // converted back to a closed std file. Retry only that transient kernel
224    // condition; every other spawn error remains immediate and typed.
225    for _ in 0..10 {
226        let result = validator_command(path).spawn();
227        match result {
228            Ok(child) => return Ok(child),
229            Err(error) if error.raw_os_error() == Some(26) => {
230                tokio::time::sleep(std::time::Duration::from_millis(5)).await;
231            }
232            Err(error) => return Err(UpdateError::io(path, error)),
233        }
234    }
235    validator_command(path)
236        .spawn()
237        .map_err(|error| UpdateError::io(path, error))
238}
239
240fn validator_command(path: &std::path::Path) -> CommandWrap {
241    let mut command = CommandWrap::with_new(path, |command| {
242        command
243            .arg("--version")
244            .stdin(Stdio::null())
245            .stdout(Stdio::piped())
246            .stderr(Stdio::piped());
247    });
248    command.wrap(KillOnDrop);
249    #[cfg(unix)]
250    command.wrap(ProcessGroup::leader());
251    #[cfg(windows)]
252    command.wrap(JobObject);
253    command
254}
255
256struct BoundedOutput {
257    bytes: Vec<u8>,
258    overflowed: bool,
259}
260
261async fn read_bounded(mut reader: impl AsyncRead + Unpin) -> std::io::Result<BoundedOutput> {
262    let mut bytes = Vec::with_capacity(OUTPUT_LIMIT.min(4096));
263    let mut buffer = [0_u8; 4096];
264    let mut overflowed = false;
265    loop {
266        let read = reader.read(&mut buffer).await?;
267        if read == 0 {
268            break;
269        }
270        let remaining = OUTPUT_LIMIT.saturating_sub(bytes.len());
271        bytes.extend_from_slice(&buffer[..read.min(remaining)]);
272        overflowed |= read > remaining;
273    }
274    Ok(BoundedOutput { bytes, overflowed })
275}
soma_self_update/validation.rs

soma_self_update/
validation.rs
