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

soma_self_update/
validation.rs
