running_process/broker/
manifest.rs

1//! CacheManifest persistence and central-registry helpers.
2//!
3//! Phase 2 of #228 (#231). The broker and standalone cleanup tool both
4//! use this module. Manifests are prost-encoded protobuf and carry a
5//! `self_sha256` digest over the encoded manifest with that field
6//! cleared.
7
8use std::fs::{self, OpenOptions};
9use std::io::{self, Write};
10use std::path::{Path, PathBuf};
11use std::time::{SystemTime, UNIX_EPOCH};
12
13#[cfg(not(windows))]
14use std::fs::File;
15
16use prost::Message;
17use sha2::{Digest, Sha256};
18
19use crate::broker::host_identity;
20use crate::broker::lifecycle::names::{validate_service_name, validate_version, PipePathError};
21use crate::broker::protocol::{CacheManifest, HostIdentity};
22use crate::broker::secure_dir;
23
24/// Filename written inside each daemon cache root.
25pub const ROOT_MANIFEST_FILE: &str = ".running-process-manifest.pb";
26
27/// Stable v1 manifest media type.
28pub const CACHE_MANIFEST_MEDIA_TYPE: &str = "application/vnd.running-process.cache-manifest.v1";
29
30/// Highest manifest schema this crate understands.
31pub const SUPPORTED_MANIFEST_SCHEMA_VERSION: u32 = 1;
32
33/// Test/development override for the central manifest registry directory.
34pub const RUNNING_PROCESS_MANIFEST_DIR_ENV: &str = "RUNNING_PROCESS_MANIFEST_DIR";
35
36/// Errors returned by manifest persistence and validation.
37#[derive(Debug, thiserror::Error)]
38pub enum ManifestError {
39    /// Filesystem operation failed.
40    #[error("manifest I/O failed: {0}")]
41    Io(#[from] io::Error),
42    /// Protobuf decode failed.
43    #[error("manifest protobuf decode failed: {0}")]
44    Decode(#[from] prost::DecodeError),
45    /// Protobuf encode failed.
46    #[error("manifest protobuf encode failed: {0}")]
47    Encode(#[from] prost::EncodeError),
48    /// The manifest's self_sha256 digest did not match its content.
49    #[error("manifest self_sha256 mismatch")]
50    Corruption,
51    /// The manifest uses a newer schema version than this crate supports.
52    #[error("manifest schema too new: got {got}, supported {supported}")]
53    SchemaTooNew {
54        /// Manifest schema version read from disk.
55        got: u32,
56        /// Maximum schema version this crate can read.
57        supported: u32,
58    },
59    /// Service/version validation failed while deriving a registry path.
60    #[error(transparent)]
61    InvalidName(#[from] PipePathError),
62    /// A path had no parent directory.
63    #[error("manifest path has no parent: {0}")]
64    MissingParent(PathBuf),
65    /// Central-registry permissions are too broad.
66    #[error("central manifest registry has insecure permissions: {0}")]
67    InsecureRegistry(PathBuf),
68}
69
70/// Result of scanning one central-registry entry.
71#[derive(Debug)]
72pub struct ManifestScanEntry {
73    /// Full path to the manifest file.
74    pub path: PathBuf,
75    /// Read result for that path.
76    pub result: Result<CacheManifest, ManifestError>,
77}
78
79/// Write `<cache_root>/.running-process-manifest.pb` atomically.
80pub fn write_to_root(cache_root: &Path, manifest: &CacheManifest) -> Result<(), ManifestError> {
81    fs::create_dir_all(cache_root)?;
82    secure_dir::ensure_private_dir(cache_root)?;
83    let target = cache_root.join(ROOT_MANIFEST_FILE);
84    write_manifest_file(&target, manifest)
85}
86
87/// Write `<central_registry>/{service}-{version}.pb` atomically.
88pub fn write_to_central(
89    service_name: &str,
90    version: &str,
91    manifest: &CacheManifest,
92) -> Result<PathBuf, ManifestError> {
93    let dir = central_registry_dir();
94    write_to_central_in_dir(&dir, service_name, version, manifest)
95}
96
97/// Testable variant of [`write_to_central`] with an explicit registry dir.
98pub fn write_to_central_in_dir(
99    registry_dir: &Path,
100    service_name: &str,
101    version: &str,
102    manifest: &CacheManifest,
103) -> Result<PathBuf, ManifestError> {
104    ensure_central_registry_dir(registry_dir)?;
105    let target = central_manifest_path(registry_dir, service_name, version)?;
106    write_manifest_file(&target, manifest)?;
107    Ok(target)
108}
109
110/// Read and integrity-verify a CacheManifest.
111pub fn read_manifest(path: &Path) -> Result<CacheManifest, ManifestError> {
112    let bytes = fs::read(path)?;
113    let manifest = CacheManifest::decode(bytes.as_slice())?;
114    verify_schema(&manifest)?;
115    verify_self_sha256(&manifest)?;
116    Ok(manifest)
117}
118
119/// Enumerate parseable manifests for this host and boot.
120///
121/// Corrupt or stale manifests are skipped. Use [`scan_central`] when
122/// callers need error details.
123pub fn enumerate_central(registry_dir: &Path) -> Vec<CacheManifest> {
124    let current_host = host_identity::current();
125    enumerate_central_for_host(registry_dir, &current_host)
126}
127
128/// Testable variant of [`enumerate_central`] with an explicit current host.
129pub fn enumerate_central_for_host(
130    registry_dir: &Path,
131    current_host: &HostIdentity,
132) -> Vec<CacheManifest> {
133    scan_central(registry_dir)
134        .into_iter()
135        .filter_map(|entry| match entry.result {
136            Ok(manifest) if manifest_matches_host(&manifest, current_host) => Some(manifest),
137            _ => None,
138        })
139        .collect()
140}
141
142/// Scan every `.pb` file in a registry and keep per-file errors.
143pub fn scan_central(registry_dir: &Path) -> Vec<ManifestScanEntry> {
144    match secure_dir::private_dir_permissions_are_private(registry_dir) {
145        Ok(true) => {}
146        Ok(false) => {
147            return vec![ManifestScanEntry {
148                path: registry_dir.to_path_buf(),
149                result: Err(ManifestError::InsecureRegistry(registry_dir.to_path_buf())),
150            }];
151        }
152        Err(_) if !registry_dir.exists() => return Vec::new(),
153        Err(err) => {
154            return vec![ManifestScanEntry {
155                path: registry_dir.to_path_buf(),
156                result: Err(ManifestError::Io(err)),
157            }];
158        }
159    }
160
161    let read_dir = match fs::read_dir(registry_dir) {
162        Ok(read_dir) => read_dir,
163        Err(_) => return Vec::new(),
164    };
165
166    let mut out = Vec::new();
167    for entry in read_dir.flatten() {
168        let path = entry.path();
169        if path.extension().and_then(|s| s.to_str()) != Some("pb") {
170            continue;
171        }
172        let result = read_manifest(&path);
173        out.push(ManifestScanEntry { path, result });
174    }
175    out.sort_by(|a, b| a.path.cmp(&b.path));
176    out
177}
178
179/// Return the platform central-registry directory.
180///
181/// `RUNNING_PROCESS_MANIFEST_DIR` is honored as a test/development
182/// override. Production callers should leave it unset.
183pub fn central_registry_dir() -> PathBuf {
184    if let Some(path) = std::env::var_os(RUNNING_PROCESS_MANIFEST_DIR_ENV) {
185        return PathBuf::from(path);
186    }
187
188    #[cfg(windows)]
189    {
190        dirs::data_dir()
191            .unwrap_or_else(|| PathBuf::from(r"C:\ProgramData"))
192            .join("running-process")
193            .join("manifests")
194    }
195    #[cfg(target_os = "macos")]
196    {
197        dirs::home_dir()
198            .unwrap_or_else(std::env::temp_dir)
199            .join("Library")
200            .join("Application Support")
201            .join("running-process")
202            .join("manifests")
203    }
204    #[cfg(all(unix, not(target_os = "macos")))]
205    {
206        if let Some(data_home) = std::env::var_os("XDG_DATA_HOME") {
207            PathBuf::from(data_home)
208                .join("running-process")
209                .join("manifests")
210        } else {
211            dirs::home_dir()
212                .unwrap_or_else(std::env::temp_dir)
213                .join(".local")
214                .join("share")
215                .join("running-process")
216                .join("manifests")
217        }
218    }
219}
220
221/// Ensure the central-registry directory exists with private permissions.
222pub fn ensure_central_registry_dir(path: &Path) -> Result<(), ManifestError> {
223    secure_dir::ensure_private_dir(path)?;
224    if !secure_dir::private_dir_permissions_are_private(path)? {
225        return Err(ManifestError::InsecureRegistry(path.to_path_buf()));
226    }
227    Ok(())
228}
229
230/// Compute the central-registry path for one service/version manifest.
231pub fn central_manifest_path(
232    registry_dir: &Path,
233    service_name: &str,
234    version: &str,
235) -> Result<PathBuf, ManifestError> {
236    validate_service_name(service_name)?;
237    validate_version(version)?;
238    Ok(registry_dir.join(format!("{service_name}-{version}.pb")))
239}
240
241/// Clone `manifest`, fill schema/media/hash fields, and return the copy.
242pub fn manifest_with_self_sha256(manifest: &CacheManifest) -> Result<CacheManifest, ManifestError> {
243    let mut out = manifest.clone();
244    out.manifest_schema_version = SUPPORTED_MANIFEST_SCHEMA_VERSION;
245    if out.media_type.is_empty() {
246        out.media_type = CACHE_MANIFEST_MEDIA_TYPE.to_string();
247    }
248    out.self_sha256.clear();
249    let digest = sha256_for_manifest(&out)?;
250    out.self_sha256 = digest.to_vec();
251    Ok(out)
252}
253
254/// Compute the SHA-256 digest with `self_sha256` cleared.
255pub fn sha256_for_manifest(manifest: &CacheManifest) -> Result<[u8; 32], ManifestError> {
256    let mut clone = manifest.clone();
257    clone.self_sha256.clear();
258    let mut bytes = Vec::new();
259    clone.encode(&mut bytes)?;
260    let digest = Sha256::digest(&bytes);
261    let mut out = [0_u8; 32];
262    out.copy_from_slice(&digest);
263    Ok(out)
264}
265
266fn write_manifest_file(path: &Path, manifest: &CacheManifest) -> Result<(), ManifestError> {
267    let manifest = manifest_with_self_sha256(manifest)?;
268    let mut bytes = Vec::new();
269    manifest.encode(&mut bytes)?;
270    atomic_write(path, &bytes)
271}
272
273fn verify_schema(manifest: &CacheManifest) -> Result<(), ManifestError> {
274    if manifest.manifest_schema_version > SUPPORTED_MANIFEST_SCHEMA_VERSION {
275        return Err(ManifestError::SchemaTooNew {
276            got: manifest.manifest_schema_version,
277            supported: SUPPORTED_MANIFEST_SCHEMA_VERSION,
278        });
279    }
280    Ok(())
281}
282
283fn verify_self_sha256(manifest: &CacheManifest) -> Result<(), ManifestError> {
284    if manifest.self_sha256.len() != 32 {
285        return Err(ManifestError::Corruption);
286    }
287    let expected = sha256_for_manifest(manifest)?;
288    if manifest.self_sha256.as_slice() != expected {
289        return Err(ManifestError::Corruption);
290    }
291    Ok(())
292}
293
294fn manifest_matches_host(manifest: &CacheManifest, current_host: &HostIdentity) -> bool {
295    let Some(host) = manifest.host.as_ref() else {
296        return true;
297    };
298    (host.machine_id.is_empty() || host.machine_id == current_host.machine_id)
299        && (host.boot_id.is_empty() || host.boot_id == current_host.boot_id)
300}
301
302/// Atomically replace the contents of `path` with `bytes`.
303///
304/// `pub(super)` so [`super::protocol_v2::manifest_io`] can reuse the
305/// exact same write semantics (tempfile + sync + rename + parent
306/// fsync) for v2 manifest files.
307pub(super) fn write_atomic(path: &Path, bytes: &[u8]) -> Result<(), ManifestError> {
308    atomic_write(path, bytes)
309}
310
311fn atomic_write(path: &Path, bytes: &[u8]) -> Result<(), ManifestError> {
312    let parent = path
313        .parent()
314        .ok_or_else(|| ManifestError::MissingParent(path.to_path_buf()))?;
315    fs::create_dir_all(parent)?;
316    let tmp = temp_path_for(path);
317
318    let write_result = (|| -> Result<(), ManifestError> {
319        let mut file = OpenOptions::new().write(true).create_new(true).open(&tmp)?;
320        file.write_all(bytes)?;
321        file.sync_all()?;
322        drop(file);
323        replace_file(&tmp, path)?;
324        sync_parent(parent)?;
325        Ok(())
326    })();
327
328    if write_result.is_err() {
329        let _ = fs::remove_file(&tmp);
330    }
331    write_result
332}
333
334fn temp_path_for(path: &Path) -> PathBuf {
335    let file_name = path
336        .file_name()
337        .and_then(|s| s.to_str())
338        .unwrap_or("manifest.pb");
339    let nanos = SystemTime::now()
340        .duration_since(UNIX_EPOCH)
341        .map(|d| d.as_nanos())
342        .unwrap_or(0);
343    path.with_file_name(format!(".{file_name}.tmp-{}-{nanos}", std::process::id()))
344}
345
346#[cfg(not(windows))]
347fn replace_file(tmp: &Path, target: &Path) -> io::Result<()> {
348    fs::rename(tmp, target)
349}
350
351#[cfg(windows)]
352fn replace_file(tmp: &Path, target: &Path) -> io::Result<()> {
353    use std::os::windows::ffi::OsStrExt;
354    use windows_sys::Win32::Storage::FileSystem::{ReplaceFileW, REPLACEFILE_WRITE_THROUGH};
355
356    if !target.exists() {
357        return fs::rename(tmp, target);
358    }
359
360    fn wide(path: &Path) -> Vec<u16> {
361        path.as_os_str()
362            .encode_wide()
363            .chain(std::iter::once(0))
364            .collect()
365    }
366
367    let target_w = wide(target);
368    let tmp_w = wide(tmp);
369    let ok = unsafe {
370        ReplaceFileW(
371            target_w.as_ptr(),
372            tmp_w.as_ptr(),
373            std::ptr::null(),
374            REPLACEFILE_WRITE_THROUGH,
375            std::ptr::null_mut(),
376            std::ptr::null_mut(),
377        )
378    };
379    if ok == 0 {
380        Err(io::Error::last_os_error())
381    } else {
382        Ok(())
383    }
384}
385
386#[cfg(not(windows))]
387fn sync_parent(parent: &Path) -> io::Result<()> {
388    File::open(parent)?.sync_all()
389}
390
391#[cfg(windows)]
392fn sync_parent(_parent: &Path) -> io::Result<()> {
393    Ok(())
394}
395
396#[cfg(test)]
397mod tests {
398    use super::*;
399    use crate::broker::protocol::Operation;
400
401    fn sample_manifest() -> CacheManifest {
402        let host = host_identity::current();
403        CacheManifest {
404            manifest_schema_version: 1,
405            media_type: CACHE_MANIFEST_MEDIA_TYPE.to_string(),
406            self_sha256: Vec::new(),
407            host: Some(host),
408            current_operation: Some(Operation {
409                kind: 0,
410                started_at_unix_ms: 1,
411                expected_done_unix_ms: 0,
412            }),
413            valid_until_unix_ms: 0,
414            service_name: "zccache".to_string(),
415            service_version: "1.2.3".to_string(),
416            broker_envelope_version: "v1".to_string(),
417            created_at_unix_ms: 1,
418            last_active_unix_ms: 2,
419            roots: Vec::new(),
420            current_daemon: None,
421            cleanup_policy: None,
422            broker_instance: "shared".to_string(),
423            depends_on: Vec::new(),
424            provides: Vec::new(),
425            observability: None,
426            bundle_id: "bundle".to_string(),
427        }
428    }
429
430    #[test]
431    fn self_hash_roundtrip() {
432        let manifest = manifest_with_self_sha256(&sample_manifest()).unwrap();
433        assert_eq!(manifest.self_sha256.len(), 32);
434        verify_self_sha256(&manifest).unwrap();
435    }
436
437    #[test]
438    fn central_path_validates_inputs() {
439        let dir = Path::new("/tmp/registry");
440        assert!(central_manifest_path(dir, "zccache", "1.2.3").is_ok());
441        assert!(central_manifest_path(dir, "Zccache", "1.2.3").is_err());
442        assert!(central_manifest_path(dir, "zccache", "../../../evil").is_err());
443    }
444
445    #[test]
446    fn central_registry_permissions_are_private_after_ensure() {
447        let tmp = tempfile::tempdir().unwrap();
448        let registry = tmp.path().join("registry");
449        ensure_central_registry_dir(&registry).unwrap();
450        assert!(secure_dir::private_dir_permissions_are_private(&registry).unwrap());
451    }
452}
running_process/broker/manifest.rs

running_process/broker/
manifest.rs
