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
68/// Hash material recorded when constructing a current daemon identity.
69///
70/// [`Self::LegacyCompatible`] preserves the historical default for consumers
71/// that must authenticate to brokers released before the BLAKE3 migration.
72/// [`Self::Blake3Only`] avoids a second full executable read when the
73/// application has established that the legacy slot must remain all zeroes.
74#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
75pub enum DaemonIdentityHashPolicy {
76    /// Compute BLAKE3 and the legacy SHA-256 compatibility digest.
77    #[default]
78    LegacyCompatible,
79    /// Compute BLAKE3 only and encode the fixed-width legacy slot as zeroes.
80    Blake3Only,
81}
82
83impl DaemonProcess {
84    /// Build a daemon identity for the current process.
85    ///
86    /// This is primarily useful for tests and direct-daemon consumers that have
87    /// just spawned a backend and need to persist a manifest entry.
88    ///
89    /// The executable digest is taken from `std::env::current_exe()` at the time
90    /// this method runs. If a daemon relocates or replaces its executable after
91    /// startup, record the final identity after relocation instead.
92    pub fn current_process(
93        ipc_endpoint: Endpoint,
94        idle_timeout_secs: Option<u32>,
95    ) -> Result<Self, IdentityError> {
96        Self::current_process_with_hash_policy(
97            ipc_endpoint,
98            idle_timeout_secs,
99            DaemonIdentityHashPolicy::LegacyCompatible,
100        )
101    }
102
103    /// Build a current-process identity with an explicit legacy-digest policy.
104    ///
105    /// The default [`Self::current_process`] remains legacy compatible. This
106    /// variant exists for a direct daemon whose stable contract fixes the
107    /// historical SHA-256 probe field to zero and must avoid the extra file
108    /// read that computing that digest would require.
109    pub fn current_process_with_hash_policy(
110        ipc_endpoint: Endpoint,
111        idle_timeout_secs: Option<u32>,
112        hash_policy: DaemonIdentityHashPolicy,
113    ) -> Result<Self, IdentityError> {
114        let exe_path = std::env::current_exe().map_err(IdentityError::CurrentExe)?;
115        let exe_hash = executable_hash_file(&exe_path)?;
116        let legacy_exe_sha256 = match hash_policy {
117            DaemonIdentityHashPolicy::LegacyCompatible => sha256_file(&exe_path)?,
118            DaemonIdentityHashPolicy::Blake3Only => [0; 32],
119        };
120        Ok(Self {
121            pid: std::process::id(),
122            exe_path,
123            exe_hash,
124            legacy_exe_sha256,
125            boot_id: host_identity::current().boot_id,
126            ipc_endpoint,
127            started_at_unix_ms: unix_now_ms(),
128            idle_timeout_secs,
129        })
130    }
131
132    /// Convert this identity into the protobuf form stored in `CacheManifest`.
133    ///
134    /// The conversion preserves the fixed-width BLAKE3 value as bytes and
135    /// names its algorithm explicitly on the wire.
136    pub fn to_proto(&self) -> protocol::DaemonProcess {
137        protocol::DaemonProcess {
138            pid: self.pid,
139            exe_path: self.exe_path.to_string_lossy().into_owned(),
140            exe_hash_algorithm: EXECUTABLE_HASH_ALGORITHM.to_owned(),
141            exe_hash: self.exe_hash.to_vec(),
142            ipc_endpoint: Some(self.ipc_endpoint.clone()),
143            started_at_unix_ms: self.started_at_unix_ms,
144            boot_id: self.boot_id.clone(),
145            idle_timeout_secs: self.idle_timeout_secs,
146        }
147    }
148
149    /// Encode a daemon identity for an endpoint probe.
150    ///
151    /// Tag 3 remains reserved in the current protobuf schema, but stable
152    /// pre-BLAKE3 brokers still decode it as the executable SHA-256.  Preserve
153    /// their read path by appending that historical unknown field after the
154    /// canonical BLAKE3 message. Current protobuf readers safely ignore it.
155    pub fn encode_probe_identity(&self, output: &mut Vec<u8>) -> Result<(), prost::EncodeError> {
156        use prost::Message;
157
158        self.to_proto().encode(output)?;
159        output.push(0x1a); // field 3, length-delimited
160        output.push(32); // fixed digest length, encoded as a one-byte varint
161        output.extend_from_slice(&self.legacy_exe_sha256);
162        Ok(())
163    }
164
165    /// Read and normalize `CacheManifest.current_daemon`.
166    ///
167    /// Returns `Ok(None)` when the manifest has no daemon entry. Malformed
168    /// entries, such as a missing endpoint or non-32-byte executable digest,
169    /// return an [`IdentityError`].
170    pub fn from_manifest_current_daemon(
171        manifest: &CacheManifest,
172    ) -> Result<Option<Self>, IdentityError> {
173        manifest
174            .current_daemon
175            .clone()
176            .map(Self::try_from)
177            .transpose()
178    }
179}
180
181impl TryFrom<protocol::DaemonProcess> for DaemonProcess {
182    type Error = IdentityError;
183
184    fn try_from(value: protocol::DaemonProcess) -> Result<Self, Self::Error> {
185        let ipc_endpoint = value.ipc_endpoint.ok_or(IdentityError::MissingEndpoint)?;
186        if value.exe_hash_algorithm != EXECUTABLE_HASH_ALGORITHM {
187            return Err(IdentityError::UnsupportedExecutableHashAlgorithm(
188                value.exe_hash_algorithm,
189            ));
190        }
191        let exe_hash =
192            vec_to_hash(value.exe_hash).map_err(IdentityError::InvalidExecutableHashLength)?;
193        Ok(Self {
194            pid: value.pid,
195            exe_path: PathBuf::from(value.exe_path),
196            exe_hash,
197            // Decoded identities are never served as a newly launched daemon.
198            // Keep the reserved compatibility payload local to identities we
199            // create from a verified executable path.
200            legacy_exe_sha256: [0; 32],
201            boot_id: value.boot_id,
202            ipc_endpoint,
203            started_at_unix_ms: value.started_at_unix_ms,
204            idle_timeout_secs: value.idle_timeout_secs,
205        })
206    }
207}
208
209impl From<&DaemonProcess> for protocol::DaemonProcess {
210    fn from(value: &DaemonProcess) -> Self {
211        value.to_proto()
212    }
213}
214
215impl Serialize for DaemonProcess {
216    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
217    where
218        S: Serializer,
219    {
220        DaemonProcessSerde::from(self).serialize(serializer)
221    }
222}
223
224impl<'de> Deserialize<'de> for DaemonProcess {
225    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
226    where
227        D: Deserializer<'de>,
228    {
229        let value = DaemonProcessSerde::deserialize(deserializer)?;
230        if value.exe_hash_algorithm != EXECUTABLE_HASH_ALGORITHM {
231            return Err(<D::Error as serde::de::Error>::custom(format!(
232                "unsupported daemon executable hash algorithm {:?}; expected blake3",
233                value.exe_hash_algorithm
234            )));
235        }
236        Ok(Self {
237            pid: value.pid,
238            exe_path: value.exe_path,
239            exe_hash: value.exe_hash,
240            legacy_exe_sha256: value.legacy_exe_sha256,
241            boot_id: value.boot_id,
242            ipc_endpoint: value.ipc_endpoint.into(),
243            started_at_unix_ms: value.started_at_unix_ms,
244            idle_timeout_secs: value.idle_timeout_secs,
245        })
246    }
247}
248
249/// Errors returned while normalizing daemon identity.
250#[derive(Debug, thiserror::Error)]
251pub enum IdentityError {
252    /// The protobuf daemon identity did not include an IPC endpoint.
253    #[error("daemon process is missing ipc_endpoint")]
254    MissingEndpoint,
255    /// The protobuf daemon identity used an unsupported or legacy hash contract.
256    #[error("unsupported daemon executable hash algorithm {0:?}; expected blake3")]
257    UnsupportedExecutableHashAlgorithm(String),
258    /// The protobuf daemon identity had an executable digest with the wrong size.
259    #[error("daemon process exe_hash must be 32 bytes, got {0}")]
260    InvalidExecutableHashLength(usize),
261    /// The current executable path could not be read.
262    #[error("failed to resolve current executable: {0}")]
263    CurrentExe(io::Error),
264    /// A filesystem operation failed while hashing the executable.
265    #[error("failed to hash executable: {0}")]
266    Io(#[from] io::Error),
267}
268
269#[derive(Debug, Clone, Serialize, Deserialize)]
270struct DaemonProcessSerde {
271    pid: u32,
272    exe_path: PathBuf,
273    exe_hash_algorithm: String,
274    exe_hash: [u8; 32],
275    #[serde(default)]
276    legacy_exe_sha256: [u8; 32],
277    boot_id: String,
278    ipc_endpoint: EndpointSerde,
279    started_at_unix_ms: u64,
280    idle_timeout_secs: Option<u32>,
281}
282
283impl From<&DaemonProcess> for DaemonProcessSerde {
284    fn from(value: &DaemonProcess) -> Self {
285        Self {
286            pid: value.pid,
287            exe_path: value.exe_path.clone(),
288            exe_hash_algorithm: EXECUTABLE_HASH_ALGORITHM.to_owned(),
289            exe_hash: value.exe_hash,
290            legacy_exe_sha256: value.legacy_exe_sha256,
291            boot_id: value.boot_id.clone(),
292            ipc_endpoint: EndpointSerde::from(&value.ipc_endpoint),
293            started_at_unix_ms: value.started_at_unix_ms,
294            idle_timeout_secs: value.idle_timeout_secs,
295        }
296    }
297}
298
299#[derive(Debug, Clone, Serialize, Deserialize)]
300struct EndpointSerde {
301    namespace_id: String,
302    path: String,
303}
304
305impl From<&Endpoint> for EndpointSerde {
306    fn from(value: &Endpoint) -> Self {
307        Self {
308            namespace_id: value.namespace_id.clone(),
309            path: value.path.clone(),
310        }
311    }
312}
313
314impl From<EndpointSerde> for Endpoint {
315    fn from(value: EndpointSerde) -> Self {
316        Endpoint {
317            namespace_id: value.namespace_id,
318            path: value.path,
319        }
320    }
321}
322
323/// BLAKE3 content hash used by the daemon identity wire contract.
324pub fn executable_hash_file(path: &Path) -> Result<[u8; 32], io::Error> {
325    crate::content_hash::blake3_file(path).map(|hash| *hash.as_bytes())
326}
327
328/// SHA-256 helper retained for the independent process-probe wire contract.
329pub fn sha256_file(path: &Path) -> Result<[u8; 32], io::Error> {
330    let bytes = fs::read(path)?;
331    let digest = Sha256::digest(&bytes);
332    let mut out = [0_u8; 32];
333    out.copy_from_slice(&digest);
334    Ok(out)
335}
336
337fn vec_to_hash(bytes: Vec<u8>) -> Result<[u8; 32], usize> {
338    let len = bytes.len();
339    let Ok(out) = bytes.try_into() else {
340        return Err(len);
341    };
342    Ok(out)
343}
344
345const EXECUTABLE_HASH_ALGORITHM: &str = "blake3";
346
347fn unix_now_ms() -> u64 {
348    SystemTime::now()
349        .duration_since(UNIX_EPOCH)
350        .map(|duration| duration.as_millis() as u64)
351        .unwrap_or(0)
352}
353
354#[cfg(test)]
355mod broker_dance_identity_tests {
356    //! The broker owns the daemon "dance" (relocation / identity / lifecycle),
357    //! and it keys a backend on its `DaemonProcess` identity — whose distinctive
358    //! field is `exe_hash`, the content hash of the (relocated) executable.
359    //!
360    //! These pin the invariant that lets the broker keep two *builds* apart
361    //! instead of letting them displace each other as "stale-version" — the
362    //! exact collision that spawn-stormed soldr's per-process self-managed
363    //! daemon when the identity did NOT carry the hash (zackees/soldr#2352).
364    use super::*;
365    use prost::Message;
366
367    /// Schema used by brokers released before the BLAKE3 identity migration.
368    #[derive(Clone, PartialEq, Message)]
369    struct LegacyDaemonProcess {
370        #[prost(uint32, tag = "1")]
371        pid: u32,
372        #[prost(string, tag = "2")]
373        exe_path: String,
374        #[prost(bytes = "vec", tag = "3")]
375        exe_sha256: Vec<u8>,
376    }
377
378    #[test]
379    fn executable_identity_hash_uses_blake3() {
380        let path =
381            std::env::temp_dir().join(format!("running-process-946-hash-{}", std::process::id()));
382        std::fs::write(&path, b"daemon image bytes").expect("write fixture");
383        let actual = executable_hash_file(&path).expect("hash fixture");
384        std::fs::remove_file(&path).ok();
385
386        assert_eq!(actual, *blake3::hash(b"daemon image bytes").as_bytes());
387    }
388
389    #[test]
390    fn blake3_identity_dual_writes_a_legacy_sha256_for_stable_brokers() {
391        let identity = DaemonProcess::current_process(endpoint("compat.sock"), None)
392            .expect("current daemon identity");
393        let current = identity.to_proto();
394        let mut encoded = Vec::new();
395        identity
396            .encode_probe_identity(&mut encoded)
397            .expect("encode compatibility identity");
398        let legacy = LegacyDaemonProcess::decode(encoded.as_slice())
399            .expect("pre-blake3 broker decodes current identity");
400
401        assert_eq!(current.exe_hash_algorithm, EXECUTABLE_HASH_ALGORITHM);
402        assert_eq!(legacy.exe_sha256.len(), 32);
403        assert_eq!(
404            legacy.exe_sha256,
405            sha256_file(&identity.exe_path)
406                .expect("sha256 executable")
407                .to_vec(),
408            "the legacy wire field remains verifiable by a stable broker"
409        );
410    }
411
412    #[test]
413    fn blake3_only_identity_skips_the_legacy_sha256_pass_and_keeps_tag_three_zeroed() {
414        let identity = DaemonProcess::current_process_with_hash_policy(
415            endpoint("blake3-only.sock"),
416            None,
417            DaemonIdentityHashPolicy::Blake3Only,
418        )
419        .expect("current daemon identity");
420        let mut encoded = Vec::new();
421        identity
422            .encode_probe_identity(&mut encoded)
423            .expect("encode compatibility identity");
424        let legacy = LegacyDaemonProcess::decode(encoded.as_slice())
425            .expect("pre-blake3 broker decodes current identity");
426
427        assert_eq!(identity.legacy_exe_sha256, [0; 32]);
428        assert_eq!(legacy.exe_sha256, [0; 32]);
429        assert_eq!(
430            identity.exe_hash,
431            executable_hash_file(&identity.exe_path).expect("blake3 executable")
432        );
433    }
434
435    #[test]
436    fn blake3_only_identity_sidecar_records_a_zero_legacy_digest() {
437        let dir = tempfile::tempdir().expect("tempdir");
438        let path = dir.path().join("identity.json");
439        let identity = DaemonProcess::current_process_with_hash_policy(
440            endpoint("blake3-only-sidecar.sock"),
441            None,
442            DaemonIdentityHashPolicy::Blake3Only,
443        )
444        .expect("current daemon identity");
445
446        crate::broker::backend_sdk::write_daemon_identity_file(&path, &identity)
447            .expect("write sidecar");
448        let json: serde_json::Value =
449            serde_json::from_slice(&std::fs::read(&path).expect("read sidecar"))
450                .expect("parse sidecar");
451
452        let legacy = json
453            .get("legacy_exe_sha256")
454            .and_then(serde_json::Value::as_array)
455            .expect("legacy digest array");
456        assert_eq!(legacy.len(), 32);
457        assert!(legacy.iter().all(|byte| byte == &serde_json::json!(0)));
458    }
459
460    fn endpoint(path: &str) -> Endpoint {
461        Endpoint {
462            namespace_id: "ns".to_owned(),
463            path: path.to_owned(),
464        }
465    }
466
467    fn identity(exe_hash: [u8; 32]) -> DaemonProcess {
468        // Everything else is held constant so `exe_hash` is the only variable:
469        // a distinct *build* of the same daemon differs only in its bytes.
470        DaemonProcess {
471            pid: 1234,
472            exe_path: PathBuf::from("runtime/soldr-self/v0.8.44-deadbeef/soldr.exe"),
473            exe_hash,
474            legacy_exe_sha256: [0x24; 32],
475            boot_id: "boot-1".to_owned(),
476            ipc_endpoint: endpoint("rpb-v2-soldr-daemon-0123456789abcdef-0"),
477            started_at_unix_ms: 1,
478            idle_timeout_secs: Some(600),
479        }
480    }
481
482    #[test]
483    fn distinct_builds_get_distinct_identities() {
484        let a = identity([0xAA; 32]);
485
486        // One byte of the binary changed => a rebuild.
487        let mut rebuilt = [0xAA; 32];
488        rebuilt[0] = 0xBB;
489        let b = identity(rebuilt);
490
491        assert_ne!(
492            a, b,
493            "a different executable hash must produce a distinct daemon identity, \
494             so the broker can never conflate two builds (no stale-version war)"
495        );
496
497        // Same bytes => same identity: a client and the same-build daemon it
498        // spawns rendezvous on ONE identity, with no shared file or negotiation.
499        assert_eq!(
500            a,
501            identity([0xAA; 32]),
502            "identical executable bytes must yield the same identity"
503        );
504    }
505
506    #[test]
507    fn pre_4_10_4_json_defaults_the_missing_legacy_sha256() {
508        let original = identity([0x42; 32]);
509        let mut legacy_json = serde_json::to_value(&original).expect("serialize identity");
510        let object = legacy_json.as_object_mut().expect("identity JSON object");
511        assert!(object.remove("legacy_exe_sha256").is_some());
512
513        let restored: DaemonProcess =
514            serde_json::from_value(legacy_json).expect("read pre-4.10.4 identity JSON");
515        let mut expected = original;
516        expected.legacy_exe_sha256 = [0; 32];
517        assert_eq!(restored, expected);
518    }
519
520    #[test]
521    fn exe_sha256_survives_the_manifest_wire_round_trip() {
522        // The broker distinguishes backends off the `CacheManifest` on the wire,
523        // so the 32-byte content hash must round-trip intact — otherwise two
524        // builds could collapse to one identity in transit.
525        let original = identity([0x42; 32]);
526        let proto = original.to_proto();
527        assert_eq!(proto.exe_hash_algorithm, "blake3");
528        assert_eq!(
529            proto.exe_hash.len(),
530            32,
531            "the wire form must carry the full 32-byte BLAKE3 hash"
532        );
533        let restored = DaemonProcess::try_from(proto).expect("identity round-trips");
534        let mut expected = original;
535        expected.legacy_exe_sha256 = [0; 32];
536        assert_eq!(
537            restored, expected,
538            "the canonical BLAKE3 identity must survive the manifest round-trip; the legacy probe field is not persisted"
539        );
540    }
541
542    #[test]
543    fn legacy_sha256_wire_identity_is_rejected_actionably() {
544        // Pre-#946 peers populated reserved field 3 and know nothing about the
545        // new algorithm/hash fields. Prost drops the unknown legacy field, so
546        // the explicit empty algorithm marker is what turns version skew into
547        // a contract error instead of a false executable mismatch.
548        let legacy = protocol::DaemonProcess {
549            pid: 1234,
550            exe_path: "legacy-daemon".to_owned(),
551            exe_hash_algorithm: String::new(),
552            exe_hash: Vec::new(),
553            ipc_endpoint: Some(endpoint("legacy.sock")),
554            started_at_unix_ms: 1,
555            boot_id: "boot-1".to_owned(),
556            idle_timeout_secs: None,
557        };
558
559        let error = DaemonProcess::try_from(legacy).expect_err("legacy SHA-256 must be fenced");
560        assert!(matches!(
561            error,
562            IdentityError::UnsupportedExecutableHashAlgorithm(ref algorithm)
563                if algorithm.is_empty()
564        ));
565        assert!(error.to_string().contains("expected blake3"));
566    }
567}
running_process/broker/backend_lifecycle/identity.rs

running_process/broker/backend_lifecycle/
identity.rs
