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

running_process/broker/
manifest.rs
