osdk_core/container/
apply.rs

1//! Stale-checked, atomic application of one ready native mirror candidate.
2
3use std::fs::{OpenOptions, Permissions};
4use std::io::{self, Write};
5use std::path::{Path, PathBuf};
6
7use serde::Serialize;
8
9use super::plan::{
10    Fingerprint, MirrorPlanBundle, NativeConfigCandidate, NativeConfigFormat,
11    NativeInputFingerprint, NativeInputState, PlanApplicability, PlanError,
12    MAX_NATIVE_CONFIG_BYTES,
13};
14
15pub const MIRROR_APPLY_SCHEMA_VERSION: u32 = 1;
16
17#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
18pub struct MirrorApplyReport {
19    pub schema_version: u32,
20    pub plan_id: Fingerprint,
21    pub path: String,
22    pub content_sha256: Fingerprint,
23    pub backup_path: Option<String>,
24}
25
26#[derive(Clone, Copy, Debug, PartialEq, Eq, thiserror::Error)]
27pub enum MirrorApplyError {
28    #[error(transparent)]
29    Plan(#[from] PlanError),
30    #[error("only a ready mirror plan can be applied")]
31    PlanNotReady,
32    #[error("a ready mirror plan must contain exactly one native candidate")]
33    CandidateCount,
34    #[error("native candidate identity does not match the mirror plan")]
35    CandidateMismatch,
36    #[error("native configuration changed after the mirror plan was created")]
37    StaleInput,
38    #[error("native candidate is not valid for its declared format")]
39    InvalidCandidate,
40    #[error("could not write native configuration ({0:?})")]
41    Write(io::ErrorKind),
42}
43
44/// Apply exactly one candidate while holding an osdk-owned cross-process lock.
45///
46/// The plan and candidate hashes are verified first. Once locked, the native
47/// input is captured again with the same no-follow and size constraints used
48/// by planning. Candidate bytes are parsed before a sibling file is flushed
49/// and atomically installed. Native services are deliberately not restarted.
50pub fn apply_mirror_plan(
51    bundle: &MirrorPlanBundle,
52    lock_path: &Path,
53) -> Result<MirrorApplyReport, MirrorApplyError> {
54    bundle.plan.validate()?;
55    if bundle.plan.applicability != PlanApplicability::Ready {
56        return Err(MirrorApplyError::PlanNotReady);
57    }
58    let [candidate] = bundle.candidates.as_slice() else {
59        return Err(MirrorApplyError::CandidateCount);
60    };
61    if bundle.plan.inputs.len() != 1
62        || bundle.plan.inputs.first() != Some(candidate.input())
63        || bundle.plan.candidates.first() != Some(candidate.fingerprint())
64        || candidate.fingerprint().size != candidate.bytes().len() as u64
65        || candidate.fingerprint().content_sha256 != Fingerprint::for_bytes(candidate.bytes())
66    {
67        return Err(MirrorApplyError::CandidateMismatch);
68    }
69    validate_candidate(candidate)?;
70
71    let _lock = crate::lock::FileLock::acquire(lock_path)
72        .map_err(|_| MirrorApplyError::Write(io::ErrorKind::Other))?;
73    let target = Path::new(&candidate.fingerprint().path);
74    let current = match super::plan::NativeConfigSnapshot::capture(target, MAX_NATIVE_CONFIG_BYTES)
75    {
76        Ok(current) => current,
77        Err(PlanError::InputChanged) => return Err(MirrorApplyError::StaleInput),
78        Err(error) => return Err(MirrorApplyError::Plan(error)),
79    };
80    if current.fingerprint() != candidate.input() {
81        return Err(MirrorApplyError::StaleInput);
82    }
83
84    let permissions = target
85        .metadata()
86        .ok()
87        .map(|metadata| metadata.permissions());
88    let backup_path = current
89        .bytes()
90        .map(|bytes| write_backup(target, bytes, permissions.as_ref()))
91        .transpose()?;
92    atomic_write(
93        target,
94        candidate.bytes(),
95        permissions.as_ref(),
96        candidate.input(),
97    )?;
98    Ok(MirrorApplyReport {
99        schema_version: MIRROR_APPLY_SCHEMA_VERSION,
100        plan_id: bundle.plan.plan_id.clone(),
101        path: candidate.fingerprint().path.clone(),
102        content_sha256: candidate.fingerprint().content_sha256.clone(),
103        backup_path: backup_path
104            .map(|path| {
105                path.to_str()
106                    .map(str::to_owned)
107                    .ok_or(MirrorApplyError::Plan(PlanError::NonUtf8Path))
108            })
109            .transpose()?,
110    })
111}
112
113fn validate_candidate(candidate: &NativeConfigCandidate) -> Result<(), MirrorApplyError> {
114    match candidate.fingerprint().format {
115        NativeConfigFormat::Json => serde_json::from_slice::<serde_json::Value>(candidate.bytes())
116            .map(|_| ())
117            .map_err(|_| MirrorApplyError::InvalidCandidate),
118        NativeConfigFormat::Toml => std::str::from_utf8(candidate.bytes())
119            .ok()
120            .and_then(|value| value.parse::<toml::Value>().ok())
121            .map(|_| ())
122            .ok_or(MirrorApplyError::InvalidCandidate),
123    }
124}
125
126fn atomic_write(
127    target: &Path,
128    bytes: &[u8],
129    permissions: Option<&Permissions>,
130    expected_input: &NativeInputFingerprint,
131) -> Result<(), MirrorApplyError> {
132    let parent = target
133        .parent()
134        .ok_or(MirrorApplyError::Write(io::ErrorKind::InvalidInput))?;
135    let mut temporary = temporary_path(parent, target);
136    let mut file = None;
137    for attempt in 0..1024u32 {
138        temporary.set_extension(format!("osdk-tmp-{}-{attempt}", std::process::id()));
139        match OpenOptions::new()
140            .create_new(true)
141            .write(true)
142            .open(&temporary)
143        {
144            Ok(opened) => {
145                file = Some(opened);
146                break;
147            }
148            Err(error) if error.kind() == io::ErrorKind::AlreadyExists => continue,
149            Err(error) => return Err(MirrorApplyError::Write(error.kind())),
150        }
151    }
152    let mut file = file.ok_or(MirrorApplyError::Write(io::ErrorKind::AlreadyExists))?;
153    if let Err(error) = file.write_all(bytes) {
154        let _ = std::fs::remove_file(&temporary);
155        return Err(MirrorApplyError::Write(error.kind()));
156    }
157    if let Some(permissions) = permissions {
158        if let Err(error) = file.set_permissions(permissions.clone()) {
159            let _ = std::fs::remove_file(&temporary);
160            return Err(MirrorApplyError::Write(error.kind()));
161        }
162    }
163    if let Err(error) = file.sync_all() {
164        let _ = std::fs::remove_file(&temporary);
165        return Err(MirrorApplyError::Write(error.kind()));
166    }
167    drop(file);
168
169    let unchanged = match expected_input.state {
170        NativeInputState::RegularFile => {
171            super::plan::NativeConfigSnapshot::capture(target, MAX_NATIVE_CONFIG_BYTES)
172                .is_ok_and(|snapshot| snapshot.fingerprint() == expected_input)
173        }
174        NativeInputState::Missing => std::fs::symlink_metadata(target)
175            .is_err_and(|error| error.kind() == io::ErrorKind::NotFound),
176    };
177    if !unchanged {
178        let _ = std::fs::remove_file(&temporary);
179        return Err(MirrorApplyError::StaleInput);
180    }
181    if let Err(error) = atomic_replace(&temporary, target).and_then(|_| sync_parent(parent)) {
182        let _ = std::fs::remove_file(&temporary);
183        return Err(MirrorApplyError::Write(error.kind()));
184    }
185    Ok(())
186}
187
188fn write_backup(
189    target: &Path,
190    bytes: &[u8],
191    source_permissions: Option<&Permissions>,
192) -> Result<PathBuf, MirrorApplyError> {
193    let parent = target
194        .parent()
195        .ok_or(MirrorApplyError::Write(io::ErrorKind::InvalidInput))?;
196    let name = target
197        .file_name()
198        .and_then(|name| name.to_str())
199        .ok_or(MirrorApplyError::Plan(PlanError::NonUtf8Path))?;
200    for attempt in 0..1024u32 {
201        let backup = parent.join(format!(
202            ".{name}.osdk-backup-{}-{attempt}",
203            std::process::id()
204        ));
205        let mut options = OpenOptions::new();
206        options.create_new(true).write(true);
207        #[cfg(unix)]
208        {
209            use std::os::unix::fs::OpenOptionsExt;
210            options.mode(0o600);
211        }
212        match options.open(&backup) {
213            Ok(mut file) => {
214                let result = file
215                    .write_all(bytes)
216                    .and_then(|_| {
217                        #[cfg(unix)]
218                        {
219                            use std::os::unix::fs::PermissionsExt;
220                            let mut permissions = source_permissions
221                                .cloned()
222                                .unwrap_or_else(|| Permissions::from_mode(0o600));
223                            permissions.set_mode(permissions.mode() & 0o700);
224                            file.set_permissions(permissions)?;
225                        }
226                        #[cfg(not(unix))]
227                        if let Some(permissions) = source_permissions {
228                            file.set_permissions(permissions.clone())?;
229                        }
230                        Ok(())
231                    })
232                    .and_then(|_| file.sync_all());
233                if let Err(error) = result {
234                    let _ = std::fs::remove_file(&backup);
235                    return Err(MirrorApplyError::Write(error.kind()));
236                }
237                return Ok(backup);
238            }
239            Err(error) if error.kind() == io::ErrorKind::AlreadyExists => continue,
240            Err(error) => return Err(MirrorApplyError::Write(error.kind())),
241        }
242    }
243    Err(MirrorApplyError::Write(io::ErrorKind::AlreadyExists))
244}
245
246fn temporary_path(parent: &Path, target: &Path) -> PathBuf {
247    let name = target
248        .file_name()
249        .and_then(|name| name.to_str())
250        .unwrap_or("native-config");
251    parent.join(format!(".{name}"))
252}
253
254#[cfg(not(windows))]
255fn atomic_replace(source: &Path, destination: &Path) -> io::Result<()> {
256    std::fs::rename(source, destination)
257}
258
259#[cfg(windows)]
260fn atomic_replace(source: &Path, destination: &Path) -> io::Result<()> {
261    use std::os::windows::ffi::OsStrExt;
262    use windows_sys::Win32::Storage::FileSystem::{
263        MoveFileExW, MOVEFILE_REPLACE_EXISTING, MOVEFILE_WRITE_THROUGH,
264    };
265
266    let source: Vec<u16> = source.as_os_str().encode_wide().chain(Some(0)).collect();
267    let destination: Vec<u16> = destination
268        .as_os_str()
269        .encode_wide()
270        .chain(Some(0))
271        .collect();
272    let result = unsafe {
273        MoveFileExW(
274            source.as_ptr(),
275            destination.as_ptr(),
276            MOVEFILE_REPLACE_EXISTING | MOVEFILE_WRITE_THROUGH,
277        )
278    };
279    if result == 0 {
280        Err(io::Error::last_os_error())
281    } else {
282        Ok(())
283    }
284}
285
286#[cfg(unix)]
287fn sync_parent(parent: &Path) -> io::Result<()> {
288    std::fs::File::open(parent)?.sync_all()
289}
290
291#[cfg(not(unix))]
292fn sync_parent(_parent: &Path) -> io::Result<()> {
293    Ok(())
294}
295
296#[cfg(test)]
297mod tests {
298    use std::collections::BTreeSet;
299
300    use super::*;
301    use crate::container::plan::{
302        ActivationRequirement, DockerTargetKind, MirrorPlanDraft, MirrorPlanTarget,
303        NativeConfigSnapshot, RequiredPrivilege, ValidationStep,
304    };
305
306    fn ready_bundle(path: &Path, bytes: &[u8]) -> MirrorPlanBundle {
307        let snapshot = NativeConfigSnapshot::capture(path, MAX_NATIVE_CONFIG_BYTES).unwrap();
308        let candidate =
309            NativeConfigCandidate::new(&snapshot, NativeConfigFormat::Json, bytes.to_vec())
310                .unwrap();
311        let plan = MirrorPlanDraft {
312            target: MirrorPlanTarget::Docker {
313                context: Fingerprint::for_bytes(b"default"),
314                kind: DockerTargetKind::Local,
315                endpoint: None,
316                version: Some("29.0.0".into()),
317            },
318            applicability: PlanApplicability::Ready,
319            policy_fingerprint: Fingerprint::for_bytes(b"policy"),
320            inputs: vec![snapshot.fingerprint().clone()],
321            candidates: vec![candidate.fingerprint().clone()],
322            changes: Vec::new(),
323            privilege: RequiredPrivilege::CurrentUser,
324            activation: ActivationRequirement::RestartDaemon,
325            validation: BTreeSet::from([ValidationStep::CompareInputFingerprints]),
326            warnings: BTreeSet::new(),
327        }
328        .finalize()
329        .unwrap();
330        MirrorPlanBundle {
331            plan,
332            candidates: vec![candidate],
333        }
334    }
335
336    #[test]
337    fn applies_one_ready_candidate_atomically() {
338        let temporary = tempfile::tempdir().unwrap();
339        let target = temporary.path().join("daemon.json");
340        std::fs::write(&target, b"{\"debug\":true}").unwrap();
341        let bundle = ready_bundle(
342            &target,
343            b"{\"registry-mirrors\":[\"https://mirror.example/\"]}",
344        );
345        let report = apply_mirror_plan(&bundle, &temporary.path().join("apply.lock")).unwrap();
346        assert_eq!(report.plan_id, bundle.plan.plan_id);
347        assert!(report
348            .backup_path
349            .as_deref()
350            .is_some_and(|path| Path::new(path).is_file()));
351        assert_eq!(
352            std::fs::read(report.backup_path.as_deref().unwrap()).unwrap(),
353            b"{\"debug\":true}"
354        );
355        #[cfg(unix)]
356        {
357            use std::os::unix::fs::PermissionsExt;
358            let backup = report.backup_path.as_deref().unwrap();
359            assert_eq!(
360                std::fs::metadata(backup).unwrap().permissions().mode() & 0o777,
361                0o600
362            );
363        }
364        assert_eq!(
365            std::fs::read(&target).unwrap(),
366            bundle.candidates[0].bytes()
367        );
368    }
369
370    #[test]
371    fn creates_a_missing_target_without_inventing_a_backup() {
372        let temporary = tempfile::tempdir().unwrap();
373        let lock_dir = temporary.path().join("locks");
374        std::fs::create_dir(&lock_dir).unwrap();
375        let target = temporary.path().join("daemon.json");
376        let bundle = ready_bundle(&target, b"{\"registry-mirrors\":[]}");
377        let report = apply_mirror_plan(&bundle, &lock_dir.join("apply.lock")).unwrap();
378        assert_eq!(report.backup_path, None);
379        assert_eq!(
380            std::fs::read(&target).unwrap(),
381            bundle.candidates[0].bytes()
382        );
383    }
384
385    #[test]
386    fn rejects_stale_inputs_without_overwriting_them() {
387        let temporary = tempfile::tempdir().unwrap();
388        let target = temporary.path().join("daemon.json");
389        std::fs::write(&target, b"{}").unwrap();
390        let bundle = ready_bundle(&target, b"{\"debug\":true}");
391        std::fs::write(&target, b"{\"changed\":true}").unwrap();
392        assert_eq!(
393            apply_mirror_plan(&bundle, &temporary.path().join("apply.lock")).unwrap_err(),
394            MirrorApplyError::StaleInput
395        );
396        assert_eq!(std::fs::read(&target).unwrap(), b"{\"changed\":true}");
397        assert_eq!(
398            std::fs::read_dir(temporary.path())
399                .unwrap()
400                .filter_map(Result::ok)
401                .filter(|entry| entry.file_name().to_string_lossy().contains("osdk-backup"))
402                .count(),
403            0
404        );
405    }
406
407    #[test]
408    fn rejects_tampered_non_ready_and_mismatched_candidates() {
409        let temporary = tempfile::tempdir().unwrap();
410        let target = temporary.path().join("daemon.json");
411        std::fs::write(&target, b"{}").unwrap();
412        let mut bundle = ready_bundle(&target, b"{}");
413        bundle.plan.applicability = PlanApplicability::ManualOnly;
414        assert_eq!(
415            apply_mirror_plan(&bundle, &temporary.path().join("apply.lock")).unwrap_err(),
416            MirrorApplyError::Plan(PlanError::InvalidPlan)
417        );
418
419        let snapshot = NativeConfigSnapshot::capture(&target, MAX_NATIVE_CONFIG_BYTES).unwrap();
420        let candidate = NativeConfigCandidate::new(
421            &snapshot,
422            NativeConfigFormat::Json,
423            b"{\"other\":true}".to_vec(),
424        )
425        .unwrap();
426        let mut manual = ready_bundle(&target, b"{}");
427        manual.plan = MirrorPlanDraft {
428            target: manual.plan.target.clone(),
429            applicability: PlanApplicability::ManualOnly,
430            policy_fingerprint: manual.plan.policy_fingerprint.clone(),
431            inputs: manual.plan.inputs.clone(),
432            candidates: manual.plan.candidates.clone(),
433            changes: manual.plan.changes.clone(),
434            privilege: manual.plan.privilege,
435            activation: manual.plan.activation,
436            validation: manual.plan.validation.clone(),
437            warnings: manual.plan.warnings.clone(),
438        }
439        .finalize()
440        .unwrap();
441        assert_eq!(
442            apply_mirror_plan(&manual, &temporary.path().join("apply.lock")).unwrap_err(),
443            MirrorApplyError::PlanNotReady
444        );
445
446        let mut mismatch = ready_bundle(&target, b"{}");
447        mismatch.candidates = vec![candidate];
448        assert_eq!(
449            apply_mirror_plan(&mismatch, &temporary.path().join("apply.lock")).unwrap_err(),
450            MirrorApplyError::CandidateMismatch
451        );
452    }
453}
osdk_core/container/apply.rs

osdk_core/container/
apply.rs
