fips_core/discovery/
local.rs

1//! Same-host instance discovery via a small registry under `~/.fips`.
2//!
3//! This is deliberately lower-tech than mDNS: processes owned by the same
4//! user publish a JSON record with loopback-reachable transport contacts.
5//! Consumers treat records as routing hints only. The Noise handshake still
6//! authenticates the advertised `npub`, so a stale or spoofed file cannot
7//! impersonate a peer.
8
9use std::fs;
10use std::io;
11use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr};
12use std::path::{Path, PathBuf};
13
14use serde::{Deserialize, Serialize};
15use sha2::{Digest, Sha256};
16use thiserror::Error;
17use tracing::debug;
18
19pub const LOCAL_INSTANCE_RECORD_VERSION: u16 = 1;
20const ENV_DIR: &str = "FIPS_LOCAL_INSTANCE_DIR";
21const ENV_DISABLE: &str = "FIPS_LOCAL_INSTANCE_DISCOVERY";
22
23/// Runtime configuration for the same-host JSON registry.
24#[derive(Debug, Clone, Serialize, Deserialize)]
25#[serde(deny_unknown_fields)]
26pub struct LocalInstanceDiscoveryConfig {
27    /// Master switch. Disabled in plain `Config::default()` so generic FIPS
28    /// nodes don't cross-feed through the user's home directory by accident.
29    /// Embedded endpoints with a discovery scope enable it explicitly.
30    #[serde(default)]
31    pub enabled: bool,
32    /// Optional registry directory. Defaults to `$FIPS_LOCAL_INSTANCE_DIR`,
33    /// then `~/.fips/instances`.
34    #[serde(default, skip_serializing_if = "Option::is_none")]
35    pub dir: Option<String>,
36    /// How often to refresh our own JSON record.
37    #[serde(default = "LocalInstanceDiscoveryConfig::default_publish_interval_secs")]
38    pub publish_interval_secs: u64,
39    /// Steady-state scan cadence after the startup sweep window.
40    #[serde(default = "LocalInstanceDiscoveryConfig::default_scan_interval_secs")]
41    pub scan_interval_secs: u64,
42    /// Scan cadence during the short startup sweep window.
43    #[serde(default = "LocalInstanceDiscoveryConfig::default_startup_scan_interval_secs")]
44    pub startup_scan_interval_secs: u64,
45    /// Duration of the startup sweep window.
46    #[serde(default = "LocalInstanceDiscoveryConfig::default_startup_scan_duration_secs")]
47    pub startup_scan_duration_secs: u64,
48    /// Records older than this are ignored and best-effort removed.
49    #[serde(default = "LocalInstanceDiscoveryConfig::default_stale_after_secs")]
50    pub stale_after_secs: u64,
51}
52
53impl Default for LocalInstanceDiscoveryConfig {
54    fn default() -> Self {
55        Self {
56            enabled: false,
57            dir: None,
58            publish_interval_secs: Self::default_publish_interval_secs(),
59            scan_interval_secs: Self::default_scan_interval_secs(),
60            startup_scan_interval_secs: Self::default_startup_scan_interval_secs(),
61            startup_scan_duration_secs: Self::default_startup_scan_duration_secs(),
62            stale_after_secs: Self::default_stale_after_secs(),
63        }
64    }
65}
66
67impl LocalInstanceDiscoveryConfig {
68    fn default_publish_interval_secs() -> u64 {
69        30
70    }
71    fn default_scan_interval_secs() -> u64 {
72        60
73    }
74    fn default_startup_scan_interval_secs() -> u64 {
75        5
76    }
77    fn default_startup_scan_duration_secs() -> u64 {
78        20
79    }
80    fn default_stale_after_secs() -> u64 {
81        180
82    }
83
84    pub(crate) fn publish_interval_ms(&self) -> u64 {
85        secs_to_ms_floor(self.publish_interval_secs, 1)
86    }
87
88    pub(crate) fn scan_interval_ms(&self) -> u64 {
89        secs_to_ms_floor(self.scan_interval_secs, 1)
90    }
91
92    pub(crate) fn startup_scan_interval_ms(&self) -> u64 {
93        secs_to_ms_floor(self.startup_scan_interval_secs, 1)
94    }
95
96    pub(crate) fn startup_scan_duration_ms(&self) -> u64 {
97        self.startup_scan_duration_secs.saturating_mul(1000)
98    }
99
100    pub(crate) fn stale_after_ms(&self) -> u64 {
101        secs_to_ms_floor(self.stale_after_secs, 1)
102    }
103}
104
105fn secs_to_ms_floor(secs: u64, min_secs: u64) -> u64 {
106    secs.max(min_secs).saturating_mul(1000)
107}
108
109#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
110#[serde(deny_unknown_fields)]
111pub struct LocalInstanceContact {
112    pub transport: String,
113    pub addr: String,
114}
115
116#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
117#[serde(deny_unknown_fields)]
118pub struct LocalInstanceRecord {
119    pub version: u16,
120    pub npub: String,
121    pub discovery_scope: String,
122    pub pid: u32,
123    pub started_at_ms: u64,
124    pub updated_at_ms: u64,
125    #[serde(default)]
126    pub contacts: Vec<LocalInstanceContact>,
127}
128
129#[derive(Debug, Error)]
130pub enum LocalInstanceRegistryError {
131    #[error("same-host FIPS discovery disabled")]
132    Disabled,
133    #[error("could not resolve FIPS local instance registry directory")]
134    NoRegistryDir,
135    #[error("local instance registry IO failed at {path}: {source}")]
136    Io {
137        path: PathBuf,
138        #[source]
139        source: io::Error,
140    },
141    #[error("local instance registry serialization failed: {0}")]
142    Json(#[from] serde_json::Error),
143}
144
145#[derive(Debug, Clone)]
146pub struct LocalInstanceRegistry {
147    dir: PathBuf,
148    record_path: PathBuf,
149    npub: String,
150    discovery_scope: String,
151    pid: u32,
152    started_at_ms: u64,
153}
154
155impl LocalInstanceRegistry {
156    pub fn new(
157        npub: impl Into<String>,
158        discovery_scope: impl Into<String>,
159        config: &LocalInstanceDiscoveryConfig,
160        started_at_ms: u64,
161    ) -> Result<Self, LocalInstanceRegistryError> {
162        if !config.enabled || env_disables_discovery() {
163            return Err(LocalInstanceRegistryError::Disabled);
164        }
165
166        let npub = npub.into();
167        let discovery_scope = discovery_scope.into();
168        let dir = registry_dir(config.dir.as_deref())?;
169        let pid = std::process::id();
170        let record_path = dir.join(record_filename(&npub, &discovery_scope, pid));
171
172        Ok(Self {
173            dir,
174            record_path,
175            npub,
176            discovery_scope,
177            pid,
178            started_at_ms,
179        })
180    }
181
182    pub fn publish(
183        &self,
184        contacts: Vec<LocalInstanceContact>,
185        now_ms: u64,
186    ) -> Result<(), LocalInstanceRegistryError> {
187        if contacts.is_empty() {
188            self.remove()?;
189            return Ok(());
190        }
191
192        ensure_private_dir(&self.dir)?;
193        let record = LocalInstanceRecord {
194            version: LOCAL_INSTANCE_RECORD_VERSION,
195            npub: self.npub.clone(),
196            discovery_scope: self.discovery_scope.clone(),
197            pid: self.pid,
198            started_at_ms: self.started_at_ms,
199            updated_at_ms: now_ms,
200            contacts,
201        };
202        let data = serde_json::to_vec_pretty(&record)?;
203        let tmp_path = self
204            .record_path
205            .with_extension(format!("json.tmp-{}", self.pid));
206        fs::write(&tmp_path, data).map_err(|source| LocalInstanceRegistryError::Io {
207            path: tmp_path.clone(),
208            source,
209        })?;
210        set_private_file_permissions(&tmp_path)?;
211        fs::rename(&tmp_path, &self.record_path).map_err(|source| {
212            let _ = fs::remove_file(&tmp_path);
213            LocalInstanceRegistryError::Io {
214                path: self.record_path.clone(),
215                source,
216            }
217        })?;
218        Ok(())
219    }
220
221    pub fn remove(&self) -> Result<(), LocalInstanceRegistryError> {
222        match fs::remove_file(&self.record_path) {
223            Ok(()) => Ok(()),
224            Err(err) if err.kind() == io::ErrorKind::NotFound => Ok(()),
225            Err(source) => Err(LocalInstanceRegistryError::Io {
226                path: self.record_path.clone(),
227                source,
228            }),
229        }
230    }
231
232    pub fn scan(
233        &self,
234        now_ms: u64,
235        stale_after_ms: u64,
236    ) -> Result<Vec<LocalInstanceRecord>, LocalInstanceRegistryError> {
237        let entries = match fs::read_dir(&self.dir) {
238            Ok(entries) => entries,
239            Err(err) if err.kind() == io::ErrorKind::NotFound => return Ok(Vec::new()),
240            Err(source) => {
241                return Err(LocalInstanceRegistryError::Io {
242                    path: self.dir.clone(),
243                    source,
244                });
245            }
246        };
247
248        let mut records = Vec::new();
249        for entry in entries {
250            let entry = match entry {
251                Ok(entry) => entry,
252                Err(err) => {
253                    debug!(error = %err, "local instance registry: skipping unreadable entry");
254                    continue;
255                }
256            };
257            let path = entry.path();
258            if path.extension().and_then(|s| s.to_str()) != Some("json") {
259                continue;
260            }
261
262            let text = match fs::read_to_string(&path) {
263                Ok(text) => text,
264                Err(err) => {
265                    debug!(path = %path.display(), error = %err, "local instance registry: skipping unreadable record");
266                    continue;
267                }
268            };
269            let record: LocalInstanceRecord = match serde_json::from_str(&text) {
270                Ok(record) => record,
271                Err(err) => {
272                    debug!(path = %path.display(), error = %err, "local instance registry: skipping malformed record");
273                    continue;
274                }
275            };
276            if record.version != LOCAL_INSTANCE_RECORD_VERSION {
277                continue;
278            }
279            if record.discovery_scope != self.discovery_scope {
280                continue;
281            }
282            if record.npub == self.npub && record.pid == self.pid {
283                continue;
284            }
285            if now_ms.saturating_sub(record.updated_at_ms) > stale_after_ms {
286                let _ = fs::remove_file(&path);
287                continue;
288            }
289            if record.contacts.is_empty() {
290                continue;
291            }
292            records.push(record);
293        }
294
295        records.sort_by(|a, b| b.updated_at_ms.cmp(&a.updated_at_ms));
296        Ok(records)
297    }
298}
299
300pub fn contact_for_transport_addr(
301    transport: impl Into<String>,
302    local_addr: SocketAddr,
303) -> Option<LocalInstanceContact> {
304    if local_addr.port() == 0 {
305        return None;
306    }
307
308    let addr = if local_addr.ip().is_unspecified() {
309        match local_addr.ip() {
310            IpAddr::V4(_) => SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), local_addr.port()),
311            IpAddr::V6(_) => SocketAddr::new(IpAddr::V6(Ipv6Addr::LOCALHOST), local_addr.port()),
312        }
313    } else {
314        local_addr
315    };
316
317    Some(LocalInstanceContact {
318        transport: transport.into(),
319        addr: addr.to_string(),
320    })
321}
322
323fn env_disables_discovery() -> bool {
324    std::env::var(ENV_DISABLE)
325        .ok()
326        .map(|value| {
327            let value = value.trim().to_ascii_lowercase();
328            matches!(value.as_str(), "0" | "false" | "off" | "no" | "disabled")
329        })
330        .unwrap_or(false)
331}
332
333fn registry_dir(configured: Option<&str>) -> Result<PathBuf, LocalInstanceRegistryError> {
334    if let Some(path) = configured
335        && !path.trim().is_empty()
336    {
337        return Ok(PathBuf::from(path));
338    }
339    if let Ok(path) = std::env::var(ENV_DIR)
340        && !path.trim().is_empty()
341    {
342        return Ok(PathBuf::from(path));
343    }
344    dirs::home_dir()
345        .map(|home| home.join(".fips").join("instances"))
346        .ok_or(LocalInstanceRegistryError::NoRegistryDir)
347}
348
349fn record_filename(npub: &str, discovery_scope: &str, pid: u32) -> String {
350    let mut hasher = Sha256::new();
351    hasher.update(discovery_scope.as_bytes());
352    hasher.update([0]);
353    hasher.update(npub.as_bytes());
354    hasher.update([0]);
355    hasher.update(pid.to_le_bytes());
356    format!("{}.json", hex::encode(hasher.finalize()))
357}
358
359fn ensure_private_dir(path: &Path) -> Result<(), LocalInstanceRegistryError> {
360    fs::create_dir_all(path).map_err(|source| LocalInstanceRegistryError::Io {
361        path: path.to_path_buf(),
362        source,
363    })?;
364    set_private_dir_permissions(path)
365}
366
367#[cfg(unix)]
368fn set_private_dir_permissions(path: &Path) -> Result<(), LocalInstanceRegistryError> {
369    use std::os::unix::fs::PermissionsExt;
370    fs::set_permissions(path, fs::Permissions::from_mode(0o700)).map_err(|source| {
371        LocalInstanceRegistryError::Io {
372            path: path.to_path_buf(),
373            source,
374        }
375    })
376}
377
378#[cfg(not(unix))]
379fn set_private_dir_permissions(_path: &Path) -> Result<(), LocalInstanceRegistryError> {
380    Ok(())
381}
382
383#[cfg(unix)]
384fn set_private_file_permissions(path: &Path) -> Result<(), LocalInstanceRegistryError> {
385    use std::os::unix::fs::PermissionsExt;
386    fs::set_permissions(path, fs::Permissions::from_mode(0o600)).map_err(|source| {
387        LocalInstanceRegistryError::Io {
388            path: path.to_path_buf(),
389            source,
390        }
391    })
392}
393
394#[cfg(not(unix))]
395fn set_private_file_permissions(_path: &Path) -> Result<(), LocalInstanceRegistryError> {
396    Ok(())
397}
398
399#[cfg(test)]
400mod tests {
401    use super::*;
402
403    fn config_for(dir: &Path) -> LocalInstanceDiscoveryConfig {
404        LocalInstanceDiscoveryConfig {
405            enabled: true,
406            dir: Some(dir.to_string_lossy().to_string()),
407            ..LocalInstanceDiscoveryConfig::default()
408        }
409    }
410
411    fn record(npub: &str, scope: &str, pid: u32, updated_at_ms: u64) -> LocalInstanceRecord {
412        LocalInstanceRecord {
413            version: LOCAL_INSTANCE_RECORD_VERSION,
414            npub: npub.to_string(),
415            discovery_scope: scope.to_string(),
416            pid,
417            started_at_ms: 1,
418            updated_at_ms,
419            contacts: vec![LocalInstanceContact {
420                transport: "udp".to_string(),
421                addr: "127.0.0.1:22121".to_string(),
422            }],
423        }
424    }
425
426    #[test]
427    fn wildcard_ipv4_contact_uses_loopback() {
428        let contact =
429            contact_for_transport_addr("udp", "0.0.0.0:22121".parse().unwrap()).expect("contact");
430
431        assert_eq!(contact.transport, "udp");
432        assert_eq!(contact.addr, "127.0.0.1:22121");
433    }
434
435    #[test]
436    fn wildcard_ipv6_contact_uses_loopback() {
437        let contact =
438            contact_for_transport_addr("udp", "[::]:22121".parse().unwrap()).expect("contact");
439
440        assert_eq!(contact.addr, "[::1]:22121");
441    }
442
443    #[test]
444    fn publish_and_remove_record() {
445        let temp = tempfile::tempdir().unwrap();
446        let registry =
447            LocalInstanceRegistry::new("npub-self", "scope-a", &config_for(temp.path()), 100)
448                .unwrap();
449
450        registry
451            .publish(
452                vec![LocalInstanceContact {
453                    transport: "udp".to_string(),
454                    addr: "127.0.0.1:22121".to_string(),
455                }],
456                200,
457            )
458            .unwrap();
459
460        let text = fs::read_to_string(&registry.record_path).unwrap();
461        let parsed: LocalInstanceRecord = serde_json::from_str(&text).unwrap();
462        assert_eq!(parsed.npub, "npub-self");
463        assert_eq!(parsed.discovery_scope, "scope-a");
464        assert_eq!(parsed.updated_at_ms, 200);
465
466        registry.remove().unwrap();
467        assert!(!registry.record_path.exists());
468    }
469
470    #[test]
471    fn scan_filters_self_scope_and_stale_records() {
472        let temp = tempfile::tempdir().unwrap();
473        let registry =
474            LocalInstanceRegistry::new("npub-self", "scope-a", &config_for(temp.path()), 100)
475                .unwrap();
476        ensure_private_dir(temp.path()).unwrap();
477
478        let cases = [
479            record("npub-peer", "scope-a", 2, 900),
480            record("npub-self", "scope-a", registry.pid, 900),
481            record("npub-other-scope", "scope-b", 3, 900),
482            record("npub-stale", "scope-a", 4, 100),
483        ];
484        for (index, record) in cases.iter().enumerate() {
485            let path = temp.path().join(format!("{index}.json"));
486            fs::write(path, serde_json::to_vec(record).unwrap()).unwrap();
487        }
488
489        let records = registry.scan(1000, 500).unwrap();
490
491        assert_eq!(records.len(), 1);
492        assert_eq!(records[0].npub, "npub-peer");
493    }
494}
fips_core/discovery/local.rs

fips_core/discovery/
local.rs
