running_process/broker/backend_lifecycle/
identity.rs

1//! Normalized daemon identity carried by `BackendHandle`.
2//!
3//! `DaemonProcess` is the typed form of `CacheManifest.current_daemon`. It is
4//! deliberately more specific than the generated protobuf message: paths are
5//! `PathBuf`s, executable hashes are fixed 32-byte arrays, and the IPC endpoint
6//! is required. That keeps malformed manifests out of the `BackendHandle` probe
7//! path.
8
9use std::convert::TryFrom;
10use std::fs;
11use std::io;
12use std::path::{Path, PathBuf};
13use std::time::{SystemTime, UNIX_EPOCH};
14
15use serde::{Deserialize, Deserializer, Serialize, Serializer};
16use sha2::{Digest, Sha256};
17
18use crate::broker::host_identity;
19use crate::broker::protocol::{self, CacheManifest, Endpoint};
20
21/// A backend daemon identity with fixed-width fields suitable for verification.
22///
23/// This mirrors `CacheManifest.current_daemon`, but normalizes protobuf strings
24/// and byte vectors into path and digest types that are harder to misuse.
25///
26/// Persist this value only after the daemon has selected its final IPC endpoint
27/// and executable. Later consumers can pass the same identity to
28/// [`crate::broker::backend_handle::BackendHandle::probe`] or store it as
29/// `CacheManifest.current_daemon`.
30///
31/// ```no_run
32/// use running_process::broker::backend_handle::DaemonProcess;
33/// use running_process::broker::protocol::{CacheManifest, Endpoint};
34///
35/// # fn example(mut manifest: CacheManifest)
36/// #     -> Result<CacheManifest, running_process::broker::backend_lifecycle::identity::IdentityError>
37/// # {
38/// let endpoint = Endpoint {
39///     namespace_id: "host-namespace".to_owned(),
40///     path: "running-process-backend.sock".to_owned(),
41/// };
42/// let daemon = DaemonProcess::current_process(endpoint, Some(600))?;
43///
44/// manifest.current_daemon = Some(daemon.to_proto());
45/// # Ok(manifest)
46/// # }
47/// ```
48#[derive(Debug, Clone, PartialEq, Eq)]
49pub struct DaemonProcess {
50    /// Operating-system process ID.
51    pub pid: u32,
52    /// Executable path recorded when the daemon identity was written.
53    pub exe_path: PathBuf,
54    /// BLAKE3 content hash of the daemon executable.
55    pub exe_hash: [u8; 32],
56    /// SHA-256 digest retained on the v1 wire for pre-BLAKE3 stable brokers.
57    pub legacy_exe_sha256: [u8; 32],
58    /// Host boot ID observed when the daemon started.
59    pub boot_id: String,
60    /// IPC endpoint used to connect to the daemon.
61    pub ipc_endpoint: Endpoint,
62    /// Daemon start timestamp in Unix milliseconds.
63    pub started_at_unix_ms: u64,
64    /// Optional idle timeout advertised by the daemon.
65    pub idle_timeout_secs: Option<u32>,
66}
67
68impl DaemonProcess {
69    /// Build a daemon identity for the current process.
70    ///
71    /// This is primarily useful for tests and direct-daemon consumers that have
72    /// just spawned a backend and need to persist a manifest entry.
73    ///
74    /// The executable digest is taken from `std::env::current_exe()` at the time
75    /// this method runs. If a daemon relocates or replaces its executable after
76    /// startup, record the final identity after relocation instead.
77    pub fn current_process(
78        ipc_endpoint: Endpoint,
79        idle_timeout_secs: Option<u32>,
80    ) -> Result<Self, IdentityError> {
81        let exe_path = std::env::current_exe().map_err(IdentityError::CurrentExe)?;
82        let exe_hash = executable_hash_file(&exe_path)?;
83        let legacy_exe_sha256 = sha256_file(&exe_path)?;
84        Ok(Self {
85            pid: std::process::id(),
86            exe_path,
87            exe_hash,
88            legacy_exe_sha256,
89            boot_id: host_identity::current().boot_id,
90            ipc_endpoint,
91            started_at_unix_ms: unix_now_ms(),
92            idle_timeout_secs,
93        })
94    }
95
96    /// Convert this identity into the protobuf form stored in `CacheManifest`.
97    ///
98    /// The conversion preserves the fixed-width BLAKE3 value as bytes and
99    /// names its algorithm explicitly on the wire.
100    pub fn to_proto(&self) -> protocol::DaemonProcess {
101        protocol::DaemonProcess {
102            pid: self.pid,
103            exe_path: self.exe_path.to_string_lossy().into_owned(),
104            exe_hash_algorithm: EXECUTABLE_HASH_ALGORITHM.to_owned(),
105            exe_hash: self.exe_hash.to_vec(),
106            ipc_endpoint: Some(self.ipc_endpoint.clone()),
107            started_at_unix_ms: self.started_at_unix_ms,
108            boot_id: self.boot_id.clone(),
109            idle_timeout_secs: self.idle_timeout_secs,
110        }
111    }
112
113    /// Encode a daemon identity for an endpoint probe.
114    ///
115    /// Tag 3 remains reserved in the current protobuf schema, but stable
116    /// pre-BLAKE3 brokers still decode it as the executable SHA-256.  Preserve
117    /// their read path by appending that historical unknown field after the
118    /// canonical BLAKE3 message. Current protobuf readers safely ignore it.
119    pub fn encode_probe_identity(&self, output: &mut Vec<u8>) -> Result<(), prost::EncodeError> {
120        use prost::Message;
121
122        self.to_proto().encode(output)?;
123        output.push(0x1a); // field 3, length-delimited
124        output.push(32); // fixed digest length, encoded as a one-byte varint
125        output.extend_from_slice(&self.legacy_exe_sha256);
126        Ok(())
127    }
128
129    /// Read and normalize `CacheManifest.current_daemon`.
130    ///
131    /// Returns `Ok(None)` when the manifest has no daemon entry. Malformed
132    /// entries, such as a missing endpoint or non-32-byte executable digest,
133    /// return an [`IdentityError`].
134    pub fn from_manifest_current_daemon(
135        manifest: &CacheManifest,
136    ) -> Result<Option<Self>, IdentityError> {
137        manifest
138            .current_daemon
139            .clone()
140            .map(Self::try_from)
141            .transpose()
142    }
143}
144
145impl TryFrom<protocol::DaemonProcess> for DaemonProcess {
146    type Error = IdentityError;
147
148    fn try_from(value: protocol::DaemonProcess) -> Result<Self, Self::Error> {
149        let ipc_endpoint = value.ipc_endpoint.ok_or(IdentityError::MissingEndpoint)?;
150        if value.exe_hash_algorithm != EXECUTABLE_HASH_ALGORITHM {
151            return Err(IdentityError::UnsupportedExecutableHashAlgorithm(
152                value.exe_hash_algorithm,
153            ));
154        }
155        let exe_hash =
156            vec_to_hash(value.exe_hash).map_err(IdentityError::InvalidExecutableHashLength)?;
157        Ok(Self {
158            pid: value.pid,
159            exe_path: PathBuf::from(value.exe_path),
160            exe_hash,
161            // Decoded identities are never served as a newly launched daemon.
162            // Keep the reserved compatibility payload local to identities we
163            // create from a verified executable path.
164            legacy_exe_sha256: [0; 32],
165            boot_id: value.boot_id,
166            ipc_endpoint,
167            started_at_unix_ms: value.started_at_unix_ms,
168            idle_timeout_secs: value.idle_timeout_secs,
169        })
170    }
171}
172
173impl From<&DaemonProcess> for protocol::DaemonProcess {
174    fn from(value: &DaemonProcess) -> Self {
175        value.to_proto()
176    }
177}
178
179impl Serialize for DaemonProcess {
180    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
181    where
182        S: Serializer,
183    {
184        DaemonProcessSerde::from(self).serialize(serializer)
185    }
186}
187
188impl<'de> Deserialize<'de> for DaemonProcess {
189    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
190    where
191        D: Deserializer<'de>,
192    {
193        let value = DaemonProcessSerde::deserialize(deserializer)?;
194        if value.exe_hash_algorithm != EXECUTABLE_HASH_ALGORITHM {
195            return Err(<D::Error as serde::de::Error>::custom(format!(
196                "unsupported daemon executable hash algorithm {:?}; expected blake3",
197                value.exe_hash_algorithm
198            )));
199        }
200        Ok(Self {
201            pid: value.pid,
202            exe_path: value.exe_path,
203            exe_hash: value.exe_hash,
204            legacy_exe_sha256: value.legacy_exe_sha256,
205            boot_id: value.boot_id,
206            ipc_endpoint: value.ipc_endpoint.into(),
207            started_at_unix_ms: value.started_at_unix_ms,
208            idle_timeout_secs: value.idle_timeout_secs,
209        })
210    }
211}
212
213/// Errors returned while normalizing daemon identity.
214#[derive(Debug, thiserror::Error)]
215pub enum IdentityError {
216    /// The protobuf daemon identity did not include an IPC endpoint.
217    #[error("daemon process is missing ipc_endpoint")]
218    MissingEndpoint,
219    /// The protobuf daemon identity used an unsupported or legacy hash contract.
220    #[error("unsupported daemon executable hash algorithm {0:?}; expected blake3")]
221    UnsupportedExecutableHashAlgorithm(String),
222    /// The protobuf daemon identity had an executable digest with the wrong size.
223    #[error("daemon process exe_hash must be 32 bytes, got {0}")]
224    InvalidExecutableHashLength(usize),
225    /// The current executable path could not be read.
226    #[error("failed to resolve current executable: {0}")]
227    CurrentExe(io::Error),
228    /// A filesystem operation failed while hashing the executable.
229    #[error("failed to hash executable: {0}")]
230    Io(#[from] io::Error),
231}
232
233#[derive(Debug, Clone, Serialize, Deserialize)]
234struct DaemonProcessSerde {
235    pid: u32,
236    exe_path: PathBuf,
237    exe_hash_algorithm: String,
238    exe_hash: [u8; 32],
239    legacy_exe_sha256: [u8; 32],
240    boot_id: String,
241    ipc_endpoint: EndpointSerde,
242    started_at_unix_ms: u64,
243    idle_timeout_secs: Option<u32>,
244}
245
246impl From<&DaemonProcess> for DaemonProcessSerde {
247    fn from(value: &DaemonProcess) -> Self {
248        Self {
249            pid: value.pid,
250            exe_path: value.exe_path.clone(),
251            exe_hash_algorithm: EXECUTABLE_HASH_ALGORITHM.to_owned(),
252            exe_hash: value.exe_hash,
253            legacy_exe_sha256: value.legacy_exe_sha256,
254            boot_id: value.boot_id.clone(),
255            ipc_endpoint: EndpointSerde::from(&value.ipc_endpoint),
256            started_at_unix_ms: value.started_at_unix_ms,
257            idle_timeout_secs: value.idle_timeout_secs,
258        }
259    }
260}
261
262#[derive(Debug, Clone, Serialize, Deserialize)]
263struct EndpointSerde {
264    namespace_id: String,
265    path: String,
266}
267
268impl From<&Endpoint> for EndpointSerde {
269    fn from(value: &Endpoint) -> Self {
270        Self {
271            namespace_id: value.namespace_id.clone(),
272            path: value.path.clone(),
273        }
274    }
275}
276
277impl From<EndpointSerde> for Endpoint {
278    fn from(value: EndpointSerde) -> Self {
279        Endpoint {
280            namespace_id: value.namespace_id,
281            path: value.path,
282        }
283    }
284}
285
286/// BLAKE3 content hash used by the daemon identity wire contract.
287pub fn executable_hash_file(path: &Path) -> Result<[u8; 32], io::Error> {
288    crate::content_hash::blake3_file(path).map(|hash| *hash.as_bytes())
289}
290
291/// SHA-256 helper retained for the independent process-probe wire contract.
292pub fn sha256_file(path: &Path) -> Result<[u8; 32], io::Error> {
293    let bytes = fs::read(path)?;
294    let digest = Sha256::digest(&bytes);
295    let mut out = [0_u8; 32];
296    out.copy_from_slice(&digest);
297    Ok(out)
298}
299
300fn vec_to_hash(bytes: Vec<u8>) -> Result<[u8; 32], usize> {
301    let len = bytes.len();
302    let Ok(out) = bytes.try_into() else {
303        return Err(len);
304    };
305    Ok(out)
306}
307
308const EXECUTABLE_HASH_ALGORITHM: &str = "blake3";
309
310fn unix_now_ms() -> u64 {
311    SystemTime::now()
312        .duration_since(UNIX_EPOCH)
313        .map(|duration| duration.as_millis() as u64)
314        .unwrap_or(0)
315}
316
317#[cfg(test)]
318mod broker_dance_identity_tests {
319    //! The broker owns the daemon "dance" (relocation / identity / lifecycle),
320    //! and it keys a backend on its `DaemonProcess` identity — whose distinctive
321    //! field is `exe_hash`, the content hash of the (relocated) executable.
322    //!
323    //! These pin the invariant that lets the broker keep two *builds* apart
324    //! instead of letting them displace each other as "stale-version" — the
325    //! exact collision that spawn-stormed soldr's per-process self-managed
326    //! daemon when the identity did NOT carry the hash (zackees/soldr#2352).
327    use super::*;
328    use prost::Message;
329
330    /// Schema used by brokers released before the BLAKE3 identity migration.
331    #[derive(Clone, PartialEq, Message)]
332    struct LegacyDaemonProcess {
333        #[prost(uint32, tag = "1")]
334        pid: u32,
335        #[prost(string, tag = "2")]
336        exe_path: String,
337        #[prost(bytes = "vec", tag = "3")]
338        exe_sha256: Vec<u8>,
339    }
340
341    #[test]
342    fn executable_identity_hash_uses_blake3() {
343        let path =
344            std::env::temp_dir().join(format!("running-process-946-hash-{}", std::process::id()));
345        std::fs::write(&path, b"daemon image bytes").expect("write fixture");
346        let actual = executable_hash_file(&path).expect("hash fixture");
347        std::fs::remove_file(&path).ok();
348
349        assert_eq!(actual, *blake3::hash(b"daemon image bytes").as_bytes());
350    }
351
352    #[test]
353    fn blake3_identity_dual_writes_a_legacy_sha256_for_stable_brokers() {
354        let identity = DaemonProcess::current_process(endpoint("compat.sock"), None)
355            .expect("current daemon identity");
356        let current = identity.to_proto();
357        let mut encoded = Vec::new();
358        identity
359            .encode_probe_identity(&mut encoded)
360            .expect("encode compatibility identity");
361        let legacy = LegacyDaemonProcess::decode(encoded.as_slice())
362            .expect("pre-blake3 broker decodes current identity");
363
364        assert_eq!(current.exe_hash_algorithm, EXECUTABLE_HASH_ALGORITHM);
365        assert_eq!(legacy.exe_sha256.len(), 32);
366        assert_eq!(
367            legacy.exe_sha256,
368            sha256_file(&identity.exe_path)
369                .expect("sha256 executable")
370                .to_vec(),
371            "the legacy wire field remains verifiable by a stable broker"
372        );
373    }
374
375    fn endpoint(path: &str) -> Endpoint {
376        Endpoint {
377            namespace_id: "ns".to_owned(),
378            path: path.to_owned(),
379        }
380    }
381
382    fn identity(exe_hash: [u8; 32]) -> DaemonProcess {
383        // Everything else is held constant so `exe_hash` is the only variable:
384        // a distinct *build* of the same daemon differs only in its bytes.
385        DaemonProcess {
386            pid: 1234,
387            exe_path: PathBuf::from("runtime/soldr-self/v0.8.44-deadbeef/soldr.exe"),
388            exe_hash,
389            legacy_exe_sha256: [0x24; 32],
390            boot_id: "boot-1".to_owned(),
391            ipc_endpoint: endpoint("rpb-v2-soldr-daemon-0123456789abcdef-0"),
392            started_at_unix_ms: 1,
393            idle_timeout_secs: Some(600),
394        }
395    }
396
397    #[test]
398    fn distinct_builds_get_distinct_identities() {
399        let a = identity([0xAA; 32]);
400
401        // One byte of the binary changed => a rebuild.
402        let mut rebuilt = [0xAA; 32];
403        rebuilt[0] = 0xBB;
404        let b = identity(rebuilt);
405
406        assert_ne!(
407            a, b,
408            "a different executable hash must produce a distinct daemon identity, \
409             so the broker can never conflate two builds (no stale-version war)"
410        );
411
412        // Same bytes => same identity: a client and the same-build daemon it
413        // spawns rendezvous on ONE identity, with no shared file or negotiation.
414        assert_eq!(
415            a,
416            identity([0xAA; 32]),
417            "identical executable bytes must yield the same identity"
418        );
419    }
420
421    #[test]
422    fn exe_sha256_survives_the_manifest_wire_round_trip() {
423        // The broker distinguishes backends off the `CacheManifest` on the wire,
424        // so the 32-byte content hash must round-trip intact — otherwise two
425        // builds could collapse to one identity in transit.
426        let original = identity([0x42; 32]);
427        let proto = original.to_proto();
428        assert_eq!(proto.exe_hash_algorithm, "blake3");
429        assert_eq!(
430            proto.exe_hash.len(),
431            32,
432            "the wire form must carry the full 32-byte BLAKE3 hash"
433        );
434        let restored = DaemonProcess::try_from(proto).expect("identity round-trips");
435        let mut expected = original;
436        expected.legacy_exe_sha256 = [0; 32];
437        assert_eq!(
438            restored, expected,
439            "the canonical BLAKE3 identity must survive the manifest round-trip; the legacy probe field is not persisted"
440        );
441    }
442
443    #[test]
444    fn legacy_sha256_wire_identity_is_rejected_actionably() {
445        // Pre-#946 peers populated reserved field 3 and know nothing about the
446        // new algorithm/hash fields. Prost drops the unknown legacy field, so
447        // the explicit empty algorithm marker is what turns version skew into
448        // a contract error instead of a false executable mismatch.
449        let legacy = protocol::DaemonProcess {
450            pid: 1234,
451            exe_path: "legacy-daemon".to_owned(),
452            exe_hash_algorithm: String::new(),
453            exe_hash: Vec::new(),
454            ipc_endpoint: Some(endpoint("legacy.sock")),
455            started_at_unix_ms: 1,
456            boot_id: "boot-1".to_owned(),
457            idle_timeout_secs: None,
458        };
459
460        let error = DaemonProcess::try_from(legacy).expect_err("legacy SHA-256 must be fenced");
461        assert!(matches!(
462            error,
463            IdentityError::UnsupportedExecutableHashAlgorithm(ref algorithm)
464                if algorithm.is_empty()
465        ));
466        assert!(error.to_string().contains("expected blake3"));
467    }
468}
running_process/broker/backend_lifecycle/identity.rs

running_process/broker/backend_lifecycle/
identity.rs
