lfg2025/archy
1
0
Fork 0
Code Issues 27 Pull Requests 5 Actions Packages Projects Releases 3 Wiki Activity

Archipelago — open-source initial import

Browse Source
This commit is contained in:
Archipelago
2026-08-12 10:55:49 +00:00
commit dc80b552a1
1639 changed files with 352647 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/archipelago/src/federation/invites.rs
+595
View File
@@ -0,0 +1,595 @@
//! Federation invite creation, parsing, and acceptance.
use anyhow::{Context, Result};
use std::path::Path;
use super::storage::{add_node, load_invites, load_nodes, save_invites, save_nodes};
use super::types::{FederatedNode, FederationInvite, TrustLevel};
/// Parsed contents of a federation invite code.
#[derive(Debug, Clone)]
pub struct ParsedInvite {
pub did: String,
pub onion: String,
pub pubkey: String,
/// Per-invite randomness; retained by parsers but not consumed
/// end-to-end — the outer signature binds the relationship.
#[allow(dead_code)]
pub token: String,
/// Inviter's FIPS npub if advertised in the code.
pub fips_npub: Option<String>,
/// Trust level the invite grants both sides. Absent in legacy codes,
/// which default to Trusted.
pub trust_level: TrustLevel,
}
/// Generate an invite code. Format: `fed1:<base64(json{did, onion, pubkey, token, fips_npub?, trust})>`.
/// `fips_npub` is only included when the local node has a materialised FIPS key.
/// `trust_level` is the level BOTH sides assign to each other for this invite
/// ("Invite a Peer" = Observer, "Link Your Nodes" = Trusted).
pub async fn create_invite(
data_dir: &Path,
did: &str,
onion: &str,
pubkey: &str,
fips_npub: Option<&str>,
trust_level: TrustLevel,
) -> Result<String> {
use base64::Engine;
use rand::Rng;
let mut token_bytes = [0u8; 16];
rand::thread_rng().fill(&mut token_bytes);
let token = hex::encode(token_bytes);
let mut payload = serde_json::json!({
"did": did,
"onion": onion,
"pubkey": pubkey,
"token": token,
"trust": trust_level.to_string(),
});
if let Some(npub) = fips_npub {
payload["fips_npub"] = serde_json::Value::String(npub.to_string());
}
let json = serde_json::to_string(&payload).context("Failed to serialize invite")?;
let code = format!(
"fed1:{}",
base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(json.as_bytes())
);
let invite = FederationInvite {
code: code.clone(),
did: did.to_string(),
onion: onion.to_string(),
pubkey: pubkey.to_string(),
created_at: chrono::Utc::now().to_rfc3339(),
accepted: false,
fips_npub: fips_npub.map(|s| s.to_string()),
trust_level,
};
let mut invites = load_invites(data_dir).await?;
invites.outgoing.push(invite);
save_invites(data_dir, &invites).await?;
Ok(code)
}
/// Parse an invite code into its components.
pub fn parse_invite(code: &str) -> Result<ParsedInvite> {
use base64::Engine;
let encoded = code
.strip_prefix("fed1:")
.ok_or_else(|| anyhow::anyhow!("Invalid invite format: must start with fed1:"))?;
let bytes = base64::engine::general_purpose::URL_SAFE_NO_PAD
.decode(encoded)
.context("Invalid base64 in invite code")?;
let payload: serde_json::Value =
serde_json::from_slice(&bytes).context("Invalid JSON in invite")?;
let did = payload["did"]
.as_str()
.ok_or_else(|| anyhow::anyhow!("Missing did in invite"))?
.to_string();
let onion = payload["onion"]
.as_str()
.ok_or_else(|| anyhow::anyhow!("Missing onion in invite"))?
.to_string();
let pubkey = payload["pubkey"]
.as_str()
.ok_or_else(|| anyhow::anyhow!("Missing pubkey in invite"))?
.to_string();
let token = payload["token"]
.as_str()
.ok_or_else(|| anyhow::anyhow!("Missing token in invite"))?
.to_string();
let fips_npub = payload
.get("fips_npub")
.and_then(|v| v.as_str())
.map(|s| s.to_string());
// Legacy codes (pre trust-threading) carry no "trust" field → Trusted,
// which matches what both sides did before the field existed.
let trust_level = payload
.get("trust")
.and_then(|v| v.as_str())
.and_then(TrustLevel::parse)
.unwrap_or(TrustLevel::Trusted);
Ok(ParsedInvite {
did,
onion,
pubkey,
token,
fips_npub,
trust_level,
})
}
/// Accept an invite: parse code, verify the remote node, add to federation.
pub async fn accept_invite(
data_dir: &Path,
code: &str,
local_did: &str,
local_onion: &str,
local_pubkey: &str,
local_fips_npub: Option<&str>,
local_name: Option<&str>,
sign_fn: impl FnOnce(&[u8]) -> String,
) -> Result<FederatedNode> {
let ParsedInvite {
did,
onion,
pubkey,
token,
fips_npub,
trust_level,
} = parse_invite(code)?;
// Refuse self-peering. If the invite's did / onion / pubkey matches
// our own, adding it pollutes the federation list with a node that
// sees itself as its own peer and causes sync loops. The user
// almost certainly pasted the wrong invite.
if did == local_did || pubkey == local_pubkey || {
let a = onion.trim_end_matches(".onion");
let b = local_onion.trim_end_matches(".onion");
!a.is_empty() && a == b
} {
anyhow::bail!(
"Refusing to federate with self — invite points at this node's own did / onion / pubkey"
);
}
// Make accept idempotent: drop any existing entry that conflicts with
// this invite — same DID (same node, refreshing the link), same onion
// (node rotated identity but kept its hidden service), or same pubkey
// (DID and onion reformatted but the underlying key is the same).
// Whatever is there gets replaced so re-accepting an invite is always
// safe and the user never has to manually remove an entry first.
let mut nodes = load_nodes(data_dir).await?;
let onion_norm = onion.trim_end_matches(".onion");
let before = nodes.len();
nodes.retain(|n| {
n.did != did && n.onion.trim_end_matches(".onion") != onion_norm && n.pubkey != pubkey
});
if nodes.len() != before {
save_nodes(data_dir, &nodes).await?;
tracing::info!(
removed = before - nodes.len(),
new_did = %did,
onion = %onion,
"Replaced stale federation entry on re-accept"
);
}
let node = FederatedNode {
did: did.clone(),
pubkey,
onion,
name: None,
// The invite code itself says what this relationship is —
// Observer for "Invite a Peer", Trusted for "Link Your Nodes".
trust_level,
added_at: chrono::Utc::now().to_rfc3339(),
last_seen: None,
last_state: None,
fips_npub: fips_npub.clone(),
last_transport: None,
last_transport_at: None,
};
add_node(data_dir, node.clone()).await?;
// Record as incoming accepted invite
let mut invites = load_invites(data_dir).await?;
invites.incoming.push(FederationInvite {
code: code.to_string(),
did: did.clone(),
onion: node.onion.clone(),
pubkey: node.pubkey.clone(),
created_at: chrono::Utc::now().to_rfc3339(),
accepted: true,
fips_npub,
trust_level,
});
save_invites(data_dir, &invites).await?;
// Notify remote node (best-effort, FIPS-first → Tor fallback)
let _ = notify_join(
&node.onion,
node.fips_npub.as_deref(),
local_did,
local_onion,
local_pubkey,
local_fips_npub,
local_name,
Some(&token),
trust_level,
sign_fn,
)
.await;
Ok(node)
}
/// Best-effort notification to the remote node that we joined their federation.
/// Prefers FIPS (if the remote advertised an npub in their invite) and
/// falls back to Tor. Signs the message with our ed25519 key so the
/// remote peer can verify authenticity regardless of transport.
#[allow(clippy::too_many_arguments)]
pub(crate) async fn notify_join(
remote_onion: &str,
remote_fips_npub: Option<&str>,
local_did: &str,
local_onion: &str,
local_pubkey: &str,
local_fips_npub: Option<&str>,
local_name: Option<&str>,
invite_token: Option<&str>,
trust_level: TrustLevel,
sign_fn: impl FnOnce(&[u8]) -> String,
) -> Result<()> {
// Sign the canonical message: "peer-joined:{did}:{onion}:{pubkey}"
// Signature domain intentionally unchanged — fips_npub + name are
// carried as unsigned informational fields. Name is display-only
// (any identity claim is anchored on the signed did/pubkey); the
// FIPS daemon's own Noise handshake authenticates the transport
// session regardless of the advertised npub.
//
// invite_token + trust are also unsigned: the inviter treats the token
// as a lookup key into its own stored invites (authoritative for the
// granted level) and only ever accepts the trust claim as a DOWNGRADE,
// so neither field can escalate privileges.
let sign_data = format!("peer-joined:{}:{}:{}", local_did, local_onion, local_pubkey);
let signature = sign_fn(sign_data.as_bytes());
let mut params = serde_json::json!({
"did": local_did,
"onion": local_onion,
"pubkey": local_pubkey,
"signature": signature,
"trust": trust_level.to_string(),
});
if let Some(npub) = local_fips_npub {
params["fips_npub"] = serde_json::Value::String(npub.to_string());
}
if let Some(name) = local_name {
params["name"] = serde_json::Value::String(name.to_string());
}
if let Some(token) = invite_token {
params["invite_token"] = serde_json::Value::String(token.to_string());
}
let body = serde_json::json!({
"method": "federation.peer-joined",
"params": params,
});
// Deliver the notification in the BACKGROUND with retries, and return
// immediately. Two reasons:
// 1. The join RPC must not block on this. Awaiting a cold FIPS overlay
// (no shared FIPS path between LAN and remote/Tailscale peers) stalled
// the whole join until FIPS timed out, surfacing as "Request timeout"
// in the UI even though the local membership was already saved.
// 2. If this single best-effort POST failed, the inviter never learned
// about us → asymmetric federation (they couldn't see us). Retrying in
// the background until it lands makes federation converge to symmetric.
// `fips_timeout` fast-fails a dead FIPS path so the Tor fallback (which
// answers an onion in ~3-5s) is reached quickly on each attempt.
let remote_onion = remote_onion.to_string();
let remote_fips_npub = remote_fips_npub.map(|s| s.to_string());
tokio::spawn(async move {
// ~5 attempts with linear backoff: 0s, 10s, 20s, 30s, 40s — covers a
// peer that is briefly unreachable (restarting, publishing its onion)
// without hammering it.
for attempt in 1..=5u32 {
let res = crate::fips::dial::PeerRequest::new(
remote_fips_npub.as_deref(),
&remote_onion,
"/rpc/v1",
)
.service(crate::settings::transport::PeerService::Federation)
.timeout(std::time::Duration::from_secs(30))
.fips_timeout(std::time::Duration::from_secs(6))
.send_json(&body)
.await;
match res {
Ok((resp, transport)) if resp.status().is_success() => {
tracing::info!(
attempt,
transport = %transport,
"peer-joined notification delivered to inviter"
);
return;
}
Ok((resp, _)) => tracing::warn!(
attempt,
status = %resp.status(),
"peer-joined notification rejected; will retry"
),
Err(e) => {
tracing::warn!(attempt, error = %e, "peer-joined notification failed; will retry")
}
}
tokio::time::sleep(std::time::Duration::from_secs(10 * attempt as u64)).await;
}
tracing::warn!(
onion = %remote_onion,
"peer-joined notification gave up after retries — peer may not see us until next sync"
);
});
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
use crate::federation::storage::load_nodes;
#[tokio::test]
async fn test_create_and_parse_invite() {
let dir = tempfile::tempdir().unwrap();
let code = create_invite(
dir.path(),
"did:key:z1",
"test.onion",
"aabbcc",
None,
TrustLevel::Trusted,
)
.await
.unwrap();
assert!(code.starts_with("fed1:"));
let parsed = parse_invite(&code).unwrap();
assert_eq!(parsed.did, "did:key:z1");
assert_eq!(parsed.onion, "test.onion");
assert_eq!(parsed.pubkey, "aabbcc");
assert_eq!(parsed.token.len(), 32); // 16 bytes = 32 hex chars
assert!(parsed.fips_npub.is_none());
}
#[tokio::test]
async fn test_invite_roundtrips_fips_npub() {
let dir = tempfile::tempdir().unwrap();
let fips = "npub1fipstest0000000000000000000000000000000000";
let code = create_invite(
dir.path(),
"did:key:z1",
"test.onion",
"aabbcc",
Some(fips),
TrustLevel::Trusted,
)
.await
.unwrap();
let parsed = parse_invite(&code).unwrap();
assert_eq!(parsed.fips_npub.as_deref(), Some(fips));
}
#[tokio::test]
async fn test_parse_invite_tolerates_missing_fips() {
// Older invites minted before fips_npub existed must still parse.
use base64::Engine;
let legacy = serde_json::json!({
"did": "did:key:zOld",
"onion": "old.onion",
"pubkey": "00",
"token": "aa",
});
let code = format!(
"fed1:{}",
base64::engine::general_purpose::URL_SAFE_NO_PAD
.encode(serde_json::to_string(&legacy).unwrap())
);
let parsed = parse_invite(&code).unwrap();
assert_eq!(parsed.did, "did:key:zOld");
assert!(parsed.fips_npub.is_none());
}
#[tokio::test]
async fn test_observer_invite_threads_trust_level() {
// "Invite a Peer" mints an observer invite; the acceptor must land
// on Observer, not Trusted (the original bug).
let dir = tempfile::tempdir().unwrap();
let code = create_invite(
dir.path(),
"did:key:zRemote",
"remote.onion",
"remotepub",
None,
TrustLevel::Observer,
)
.await
.unwrap();
let parsed = parse_invite(&code).unwrap();
assert_eq!(parsed.trust_level, TrustLevel::Observer);
let dir2 = tempfile::tempdir().unwrap();
let node = accept_invite(
dir2.path(),
&code,
"did:key:zLocal",
"local.onion",
"localpub",
None,
None,
|_| "test-sig".to_string(),
)
.await
.unwrap();
assert_eq!(node.trust_level, TrustLevel::Observer);
// The inviter's stored outgoing invite carries the level too — this
// is what peer-joined uses as the authoritative grant on the far side.
let invites = load_invites(dir.path()).await.unwrap();
assert_eq!(invites.outgoing.len(), 1);
assert_eq!(invites.outgoing[0].trust_level, TrustLevel::Observer);
}
#[test]
fn test_parse_legacy_invite_defaults_to_trusted() {
// Codes minted before the trust field existed must stay Trusted.
use base64::Engine;
let legacy = serde_json::json!({
"did": "did:key:zOld",
"onion": "old.onion",
"pubkey": "00",
"token": "aa",
});
let code = format!(
"fed1:{}",
base64::engine::general_purpose::URL_SAFE_NO_PAD
.encode(serde_json::to_string(&legacy).unwrap())
);
let parsed = parse_invite(&code).unwrap();
assert_eq!(parsed.trust_level, TrustLevel::Trusted);
}
#[test]
fn test_parse_invalid_invite() {
assert!(parse_invite("invalid").is_err());
assert!(parse_invite("fed1:not-valid-base64!!!").is_err());
}
#[tokio::test]
async fn test_accept_invite_creates_node() {
let dir = tempfile::tempdir().unwrap();
let code = create_invite(
dir.path(),
"did:key:zRemote",
"remote.onion",
"remotepub",
None,
TrustLevel::Trusted,
)
.await
.unwrap();
// Accept from a different "local" perspective
let dir2 = tempfile::tempdir().unwrap();
let node = accept_invite(
dir2.path(),
&code,
"did:key:zLocal",
"local.onion",
"localpub",
None,
None,
|_| "test-sig".to_string(),
)
.await
.unwrap();
assert_eq!(node.did, "did:key:zRemote");
assert_eq!(node.trust_level, TrustLevel::Trusted);
let nodes = load_nodes(dir2.path()).await.unwrap();
assert_eq!(nodes.len(), 1);
}
#[tokio::test]
async fn test_accept_invite_persists_fips_npub() {
let dir = tempfile::tempdir().unwrap();
let fips = "npub1remotefipsaddrxxxxxxxxxxxxxxxxxxxxxxxxxx";
let code = create_invite(
dir.path(),
"did:key:zRemote",
"remote.onion",
"remotepub",
Some(fips),
TrustLevel::Trusted,
)
.await
.unwrap();
let dir2 = tempfile::tempdir().unwrap();
let node = accept_invite(
dir2.path(),
&code,
"did:key:zLocal",
"local.onion",
"localpub",
None,
None,
|_| "test-sig".to_string(),
)
.await
.unwrap();
assert_eq!(node.fips_npub.as_deref(), Some(fips));
}
#[tokio::test]
async fn test_accept_invite_is_idempotent() {
// Re-accepting the same invite is a no-op refresh — it must not
// duplicate the entry and must not error. This is the contract the
// UI relies on: clicking "Join" twice or refreshing after an
// identity rotation always converges to one entry.
let dir = tempfile::tempdir().unwrap();
let code = create_invite(
dir.path(),
"did:key:zRemote",
"remote.onion",
"remotepub",
None,
TrustLevel::Trusted,
)
.await
.unwrap();
let dir2 = tempfile::tempdir().unwrap();
accept_invite(
dir2.path(),
&code,
"did:key:zLocal",
"local.onion",
"localpub",
None,
None,
|_| "test-sig".to_string(),
)
.await
.unwrap();
accept_invite(
dir2.path(),
&code,
"did:key:zLocal",
"local.onion",
"localpub",
None,
None,
|_| "test-sig".to_string(),
)
.await
.unwrap();
let nodes = load_nodes(dir2.path()).await.unwrap();
assert_eq!(nodes.len(), 1, "re-accept should not duplicate");
}
}
core/archipelago/src/federation/mod.rs
+27
View File
@@ -0,0 +1,27 @@
//! Node federation: trusted multi-node clusters with state sync.
//!
//! Nodes federate by exchanging invite codes containing DID + onion address.
//! Trust is bilateral — both sides must agree. Federated nodes periodically
//! sync container status, health metrics, and availability.
mod invites;
pub mod pending;
mod storage;
mod sync;
mod types;
// Re-export all public items so `crate::federation::*` continues to work.
pub use invites::{accept_invite, create_invite, parse_invite};
// Crate-internal: used by the periodic federation auto-sync to re-assert
// membership to peers that don't list us back (asymmetry self-heal).
pub(crate) use invites::notify_join;
// Crate-internal: peer-joined resolves the granted trust level by matching
// the acceptor's invite_token against our stored outgoing invites.
pub(crate) use storage::load_invites;
#[allow(unused_imports)]
pub use storage::{
add_node, fips_npub_for_onion, load_nodes, load_removed_dids, record_peer_transport,
remove_node, save_nodes, set_trust_level, update_node,
};
pub use sync::{build_local_state, deploy_to_peer, sync_with_peer, sync_with_peer_by_did};
pub use types::{AppStatus, FederatedNode, NodeStateSnapshot, TrustLevel};
core/archipelago/src/federation/pending.rs
+374
View File
@@ -0,0 +1,374 @@
//! Pending peer-discovery requests received over Nostr.
//!
//! When another node discovers us via Nostr presence and sends an encrypted
//! `PeerRequest` (NIP-44 DM), we store the request here instead of acting
//! on it. The user explicitly approves or rejects each request via the
//! Federation UI; only on approval do we generate a federation invite code
//! and ship it back over the same encrypted channel.
//!
//! Nothing in this module ever exposes the local onion address. The onion
//! is only added to the wire later, by the approval handler, and only
//! inside a NIP-44 ciphertext addressed to the requester's nostr pubkey.
use anyhow::{Context, Result};
use serde::{Deserialize, Serialize};
use std::path::Path;
use tokio::fs;
const PENDING_FILE: &str = "federation/pending_requests.json";
const MAX_PENDING_PER_PUBKEY: usize = 5;
const PENDING_EXPIRY_DAYS: i64 = 30;
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum PendingState {
/// Inbound: a remote node sent us a peer request, awaiting local approval.
Pending,
/// Outbound: we sent a peer request, awaiting their approval (and the
/// invite code they will send back via NIP-44 if they accept).
Sent,
/// Approved locally — the inbound request has been turned into a federation
/// invite that has been shipped back to the requester. Kept as history.
Approved,
/// Rejected locally. Kept as history so the same npub can't immediately
/// re-request without the user noticing.
Rejected,
/// Auto-expired after `PENDING_EXPIRY_DAYS` with no action.
Expired,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PendingPeerRequest {
/// UUID — stable identifier the FE refers to when approving/rejecting.
pub id: String,
/// Sender's Nostr secp256k1 pubkey (hex). Authoritative for routing
/// the encrypted NIP-44 reply on approval.
pub from_nostr_pubkey: String,
/// Sender's Nostr pubkey in bech32 npub format (display only).
pub from_nostr_npub: String,
/// Sender's claimed archipelago DID. Verified at *approval* time
/// (when their onion arrives via federation.peer-joined), not now —
/// the requester could lie here, but the worst case is a wasted
/// approval slot.
pub from_did: String,
/// Optional friendly name the requester typed.
pub from_name: Option<String>,
/// Optional one-line message the requester attached.
pub message: Option<String>,
pub received_at: String,
pub state: PendingState,
/// True if this row represents an outbound request we sent (`Sent`)
/// rather than an inbound one we received (`Pending`).
#[serde(default)]
pub outbound: bool,
}
#[derive(Debug, Default, Serialize, Deserialize)]
pub struct PendingRequestsFile {
pub requests: Vec<PendingPeerRequest>,
}
pub async fn load_pending(data_dir: &Path) -> Result<Vec<PendingPeerRequest>> {
let path = data_dir.join(PENDING_FILE);
if !path.exists() {
return Ok(Vec::new());
}
let content = fs::read_to_string(&path)
.await
.context("Failed to read pending requests file")?;
let file: PendingRequestsFile = serde_json::from_str(&content).unwrap_or_default();
Ok(file.requests)
}
pub async fn save_pending(data_dir: &Path, requests: &[PendingPeerRequest]) -> Result<()> {
let path = data_dir.join(PENDING_FILE);
if let Some(parent) = path.parent() {
fs::create_dir_all(parent)
.await
.context("Failed to create federation dir")?;
}
let file = PendingRequestsFile {
requests: requests.to_vec(),
};
let content =
serde_json::to_string_pretty(&file).context("Failed to serialize pending requests")?;
fs::write(&path, content)
.await
.context("Failed to write pending requests file")?;
Ok(())
}
/// Sweep auto-expired entries. Returns the cleaned list, mutated in place.
fn expire_stale(requests: &mut Vec<PendingPeerRequest>) {
let cutoff = chrono::Utc::now() - chrono::Duration::days(PENDING_EXPIRY_DAYS);
for r in requests.iter_mut() {
if !matches!(r.state, PendingState::Pending | PendingState::Sent) {
continue;
}
if let Ok(ts) = chrono::DateTime::parse_from_rfc3339(&r.received_at) {
if ts.with_timezone(&chrono::Utc) < cutoff {
r.state = PendingState::Expired;
}
}
}
}
/// Insert a new inbound peer request. Returns the stored row (with id),
/// or `None` if the request was deduplicated or rate-limited.
///
/// Dedup rule: if the same (from_nostr_pubkey, from_did) already has a
/// `Pending` OR `Approved` entry, do not insert a second one. Including
/// `Approved` is what stops an already-approved peer from re-spawning a fresh
/// pending row every time their request re-syncs (the reported "approve, Poll
/// Now, see approved + a new pending" loop). `Rejected` is intentionally NOT
/// matched so a previously-rejected peer can still ask again later. Otherwise
/// count `Pending` entries per pubkey and reject beyond `MAX_PENDING_PER_PUBKEY`.
pub async fn insert_inbound(
data_dir: &Path,
from_nostr_pubkey: String,
from_nostr_npub: String,
from_did: String,
from_name: Option<String>,
message: Option<String>,
) -> Result<Option<PendingPeerRequest>> {
let mut requests = load_pending(data_dir).await?;
expire_stale(&mut requests);
let already_handled = requests.iter().any(|r| {
r.from_nostr_pubkey == from_nostr_pubkey
&& r.from_did == from_did
&& matches!(r.state, PendingState::Pending | PendingState::Approved)
&& !r.outbound
});
if already_handled {
save_pending(data_dir, &requests).await?;
return Ok(None);
}
let live_count = requests
.iter()
.filter(|r| {
r.from_nostr_pubkey == from_nostr_pubkey
&& matches!(r.state, PendingState::Pending)
&& !r.outbound
})
.count();
if live_count >= MAX_PENDING_PER_PUBKEY {
save_pending(data_dir, &requests).await?;
anyhow::bail!(
"rate-limited: {} already has {} pending requests",
from_nostr_pubkey,
live_count
);
}
let row = PendingPeerRequest {
id: uuid::Uuid::new_v4().to_string(),
from_nostr_pubkey,
from_nostr_npub,
from_did,
from_name,
message,
received_at: chrono::Utc::now().to_rfc3339(),
state: PendingState::Pending,
outbound: false,
};
requests.push(row.clone());
save_pending(data_dir, &requests).await?;
Ok(Some(row))
}
/// Record an outbound peer request we just sent, so the user can see it
/// in the "sent" tab and so the eventual NIP-44 invite reply can be
/// matched against it.
pub async fn insert_outbound(
data_dir: &Path,
to_nostr_pubkey: String,
to_nostr_npub: String,
to_did: String,
to_name: Option<String>,
message: Option<String>,
) -> Result<PendingPeerRequest> {
let mut requests = load_pending(data_dir).await?;
expire_stale(&mut requests);
requests.retain(|r| {
!(r.outbound
&& r.from_nostr_pubkey == to_nostr_pubkey
&& matches!(r.state, PendingState::Sent))
});
let row = PendingPeerRequest {
id: uuid::Uuid::new_v4().to_string(),
from_nostr_pubkey: to_nostr_pubkey,
from_nostr_npub: to_nostr_npub,
from_did: to_did,
from_name: to_name,
message,
received_at: chrono::Utc::now().to_rfc3339(),
state: PendingState::Sent,
outbound: true,
};
requests.push(row.clone());
save_pending(data_dir, &requests).await?;
Ok(row)
}
pub async fn find_by_id(data_dir: &Path, id: &str) -> Result<Option<PendingPeerRequest>> {
let requests = load_pending(data_dir).await?;
Ok(requests.into_iter().find(|r| r.id == id))
}
pub async fn set_state(data_dir: &Path, id: &str, state: PendingState) -> Result<()> {
let mut requests = load_pending(data_dir).await?;
if let Some(r) = requests.iter_mut().find(|r| r.id == id) {
r.state = state;
} else {
anyhow::bail!("Pending request not found: {}", id);
}
save_pending(data_dir, &requests).await?;
Ok(())
}
/// Remove a pending request entirely. Used when the sender cancels an
/// outbound request they initiated and we want it gone (not just marked
/// Rejected/Cancelled — those states fill up the UI audit trail).
pub async fn delete(data_dir: &Path, id: &str) -> Result<()> {
let mut requests = load_pending(data_dir).await?;
let before = requests.len();
requests.retain(|r| r.id != id);
if requests.len() == before {
anyhow::bail!("Pending request not found: {}", id);
}
save_pending(data_dir, &requests).await?;
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test]
async fn test_insert_inbound_then_dedupes() {
let dir = tempfile::tempdir().unwrap();
let r1 = insert_inbound(
dir.path(),
"npk1".into(),
"npub1".into(),
"did:key:zABC".into(),
None,
None,
)
.await
.unwrap();
assert!(r1.is_some());
let r2 = insert_inbound(
dir.path(),
"npk1".into(),
"npub1".into(),
"did:key:zABC".into(),
None,
None,
)
.await
.unwrap();
assert!(r2.is_none(), "duplicate Pending request should be ignored");
}
#[tokio::test]
async fn test_approved_request_does_not_respawn_pending() {
// Regression for the "approve → Poll Now → approved + a fresh pending"
// loop: once a request is Approved, a re-synced inbound for the same
// peer must NOT create a new Pending row.
let dir = tempfile::tempdir().unwrap();
let r1 = insert_inbound(
dir.path(),
"npk1".into(),
"npub1".into(),
"did:key:zABC".into(),
None,
None,
)
.await
.unwrap()
.expect("first insert stored");
set_state(dir.path(), &r1.id, PendingState::Approved)
.await
.unwrap();
let r2 = insert_inbound(
dir.path(),
"npk1".into(),
"npub1".into(),
"did:key:zABC".into(),
None,
None,
)
.await
.unwrap();
assert!(
r2.is_none(),
"an already-approved peer must not re-spawn a pending request"
);
let pending = load_pending(dir.path()).await.unwrap();
assert_eq!(
pending
.iter()
.filter(|r| matches!(r.state, PendingState::Pending))
.count(),
0,
"no Pending rows should remain after approval + re-sync"
);
}
#[tokio::test]
async fn test_rate_limit() {
let dir = tempfile::tempdir().unwrap();
for i in 0..MAX_PENDING_PER_PUBKEY {
let res = insert_inbound(
dir.path(),
"npk-spammer".into(),
"npub-spammer".into(),
format!("did:key:zVar{}", i),
None,
None,
)
.await
.unwrap();
assert!(res.is_some());
}
let result = insert_inbound(
dir.path(),
"npk-spammer".into(),
"npub-spammer".into(),
"did:key:zOverflow".into(),
None,
None,
)
.await;
assert!(result.is_err(), "should rate-limit beyond MAX");
}
#[tokio::test]
async fn test_set_state_round_trip() {
let dir = tempfile::tempdir().unwrap();
let row = insert_inbound(
dir.path(),
"npk2".into(),
"npub2".into(),
"did:key:zXYZ".into(),
Some("Bob".into()),
Some("hi".into()),
)
.await
.unwrap()
.unwrap();
set_state(dir.path(), &row.id, PendingState::Approved)
.await
.unwrap();
let reloaded = find_by_id(dir.path(), &row.id).await.unwrap().unwrap();
assert_eq!(reloaded.state, PendingState::Approved);
}
}
core/archipelago/src/federation/storage.rs
+532
View File
@@ -0,0 +1,532 @@
//! Federation persistent storage: node list and invite management on disk.
use anyhow::{Context, Result};
use serde::{Deserialize, Serialize};
use std::path::Path;
use tokio::fs;
use super::types::{FederatedNode, FederationInvite, NodeStateSnapshot, TrustLevel};
pub(crate) const FEDERATION_DIR: &str = "federation";
pub(crate) const NODES_FILE: &str = "nodes.json";
pub(crate) const INVITES_FILE: &str = "invites.json";
/// Tombstones: DIDs the operator explicitly removed. Kept so transitive
/// federation discovery can't silently re-add a peer they deleted.
pub(crate) const REMOVED_FILE: &str = "removed-nodes.json";
/// Top-level file structures.
#[derive(Debug, Default, Serialize, Deserialize)]
pub(crate) struct NodesFile {
pub(crate) nodes: Vec<FederatedNode>,
}
#[derive(Debug, Default, Serialize, Deserialize)]
pub(crate) struct RemovedFile {
pub(crate) removed: Vec<RemovedNode>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub(crate) struct RemovedNode {
pub(crate) did: String,
pub(crate) removed_at: String,
}
#[derive(Debug, Default, Serialize, Deserialize)]
pub(crate) struct InvitesFile {
pub(crate) outgoing: Vec<FederationInvite>,
pub(crate) incoming: Vec<FederationInvite>,
}
/// Ensure federation directory exists.
pub(crate) async fn ensure_dir(data_dir: &Path) -> Result<std::path::PathBuf> {
let dir = data_dir.join(FEDERATION_DIR);
fs::create_dir_all(&dir)
.await
.context("Failed to create federation directory")?;
Ok(dir)
}
// ──────────────────────────── Node Management ────────────────────────────
pub async fn load_nodes(data_dir: &Path) -> Result<Vec<FederatedNode>> {
let dir = data_dir.join(FEDERATION_DIR);
let path = dir.join(NODES_FILE);
if !path.exists() {
return Ok(Vec::new());
}
let content = fs::read_to_string(&path)
.await
.context("Failed to read federation nodes")?;
let file: NodesFile = serde_json::from_str(&content).unwrap_or_default();
Ok(dedup_nodes_by_onion(file.nodes))
}
/// Collapse entries that share an onion. An onion is a node's stable, unique
/// network identity, so two entries with the same onion are the SAME physical
/// node lingering under two dids (e.g. after a did/key change). Returning both
/// duplicates the node in the trusted-node list (B1) and the chat list (B2).
/// Keep the first occurrence and merge any missing fips_npub/name/last_state
/// from the duplicates into it, then drop them. Non-destructive to disk; the
/// deduped list persists the next time nodes are saved (add/sync).
fn dedup_nodes_by_onion(nodes: Vec<FederatedNode>) -> Vec<FederatedNode> {
use std::collections::HashMap;
let mut by_onion: HashMap<String, usize> = HashMap::new();
let mut out: Vec<FederatedNode> = Vec::with_capacity(nodes.len());
for node in nodes {
let key = node.onion.trim_end_matches(".onion").to_string();
if key.is_empty() {
out.push(node);
continue;
}
if let Some(&idx) = by_onion.get(&key) {
let kept = &mut out[idx];
if kept.fips_npub.is_none() {
kept.fips_npub = node.fips_npub;
}
if kept.name.is_none() {
kept.name = node.name;
}
if kept.last_state.is_none() {
kept.last_state = node.last_state;
}
continue;
}
by_onion.insert(key, out.len());
out.push(node);
}
out
}
/// Look up a federated peer's FIPS npub given their onion address.
/// Returns `None` when the onion isn't in our federation list or the
/// peer hasn't advertised a FIPS key. Matching is suffix-tolerant so
/// callers can pass `abc` or `abc.onion` interchangeably.
pub async fn fips_npub_for_onion(data_dir: &Path, onion: &str) -> Option<String> {
let target = onion.trim_end_matches(".onion");
let nodes = load_nodes(data_dir).await.ok()?;
nodes
.iter()
.find(|n| n.onion.trim_end_matches(".onion") == target)
.and_then(|n| n.fips_npub.clone())
}
/// Record the transport used on the most recent successful peer reach.
/// Used for the "FIPS"/"Tor" badge on each node card in the UI — we write
/// what we actually used, not what was predicted.
///
/// Matches by DID first (precise) and falls back to onion (when the
/// caller didn't carry the DID through). No-op if the peer isn't in
/// our federation list.
pub async fn record_peer_transport(
data_dir: &Path,
did: Option<&str>,
onion: Option<&str>,
transport: &str,
) -> Result<()> {
let mut nodes = load_nodes(data_dir).await?;
let now = chrono::Utc::now().to_rfc3339();
let onion_target = onion.map(|o| o.trim_end_matches(".onion"));
let mut modified = false;
for node in nodes.iter_mut() {
let did_match = did.is_some_and(|d| d == node.did);
let onion_match = onion_target.is_some_and(|t| node.onion.trim_end_matches(".onion") == t);
if did_match || onion_match {
node.last_transport = Some(transport.to_string());
node.last_transport_at = Some(now.clone());
node.last_seen = Some(now.clone());
modified = true;
break;
}
}
if modified {
save_nodes(data_dir, &nodes).await?;
}
Ok(())
}
pub async fn save_nodes(data_dir: &Path, nodes: &[FederatedNode]) -> Result<()> {
let dir = ensure_dir(data_dir).await?;
let file = NodesFile {
nodes: nodes.to_vec(),
};
let content = serde_json::to_string_pretty(&file).context("Failed to serialize nodes")?;
fs::write(dir.join(NODES_FILE), content)
.await
.context("Failed to write federation nodes")?;
Ok(())
}
pub async fn add_node(data_dir: &Path, node: FederatedNode) -> Result<Vec<FederatedNode>> {
let mut nodes = load_nodes(data_dir).await?;
let exists = nodes.iter().any(|n| n.did == node.did);
if exists {
anyhow::bail!("Node with DID {} is already federated", node.did);
}
// Explicitly (re-)adding a node clears any prior tombstone so the
// operator can intentionally bring back a previously removed peer.
// Propagate failure BEFORE mutating the node list: with the tombstone
// still in place, sync reconciliation would silently re-remove the
// node the operator just added.
untombstone_did(data_dir, &node.did)
.await
.context("clear removal tombstone")?;
nodes.push(node);
save_nodes(data_dir, &nodes).await?;
Ok(nodes)
}
pub async fn remove_node(data_dir: &Path, did: &str) -> Result<Vec<FederatedNode>> {
let mut nodes = load_nodes(data_dir).await?;
let before = nodes.len();
nodes.retain(|n| n.did != did);
if nodes.len() == before {
anyhow::bail!("No federated node with DID {}", did);
}
// Tombstone the DID so transitive federation discovery (a still-federated
// peer advertising this DID as one of *its* trusted peers) can't silently
// re-add it. Tombstone FIRST and propagate failure: a remove whose
// tombstone never landed isn't a remove — the peer would quietly
// reappear after the next sync. Tombstoning is idempotent, so if the
// node-list save below fails the operator's retry works cleanly.
tombstone_did(data_dir, did)
.await
.context("persist removal tombstone")?;
save_nodes(data_dir, &nodes).await?;
Ok(nodes)
}
/// Load the set of tombstoned (operator-removed) DIDs.
pub async fn load_removed_dids(data_dir: &Path) -> Result<std::collections::HashSet<String>> {
let path = data_dir.join(FEDERATION_DIR).join(REMOVED_FILE);
if !path.exists() {
return Ok(std::collections::HashSet::new());
}
let content = fs::read_to_string(&path)
.await
.context("Failed to read removed nodes")?;
let file: RemovedFile = serde_json::from_str(&content).unwrap_or_default();
Ok(file.removed.into_iter().map(|r| r.did).collect())
}
/// Record a DID as removed. Idempotent.
pub async fn tombstone_did(data_dir: &Path, did: &str) -> Result<()> {
let dir = ensure_dir(data_dir).await?;
let path = dir.join(REMOVED_FILE);
let mut file: RemovedFile = if path.exists() {
serde_json::from_str(&fs::read_to_string(&path).await.unwrap_or_default())
.unwrap_or_default()
} else {
RemovedFile::default()
};
if !file.removed.iter().any(|r| r.did == did) {
file.removed.push(RemovedNode {
did: did.to_string(),
removed_at: chrono::Utc::now().to_rfc3339(),
});
let content = serde_json::to_string_pretty(&file).context("serialize removed nodes")?;
fs::write(&path, content)
.await
.context("Failed to write removed nodes")?;
}
Ok(())
}
/// Clear a DID's tombstone (operator explicitly re-added it).
pub async fn untombstone_did(data_dir: &Path, did: &str) -> Result<()> {
let path = data_dir.join(FEDERATION_DIR).join(REMOVED_FILE);
if !path.exists() {
return Ok(());
}
let mut file: RemovedFile =
serde_json::from_str(&fs::read_to_string(&path).await.unwrap_or_default())
.unwrap_or_default();
let before = file.removed.len();
file.removed.retain(|r| r.did != did);
if file.removed.len() != before {
let content = serde_json::to_string_pretty(&file).context("serialize removed nodes")?;
fs::write(&path, content)
.await
.context("Failed to write removed nodes")?;
}
Ok(())
}
pub async fn set_trust_level(
data_dir: &Path,
did: &str,
trust: TrustLevel,
) -> Result<Vec<FederatedNode>> {
let mut nodes = load_nodes(data_dir).await?;
let node = nodes
.iter_mut()
.find(|n| n.did == did)
.ok_or_else(|| anyhow::anyhow!("No federated node with DID {}", did))?;
node.trust_level = trust;
save_nodes(data_dir, &nodes).await?;
Ok(nodes)
}
/// Update a federated node's metadata (onion, pubkey, name, last_seen).
pub async fn update_node(data_dir: &Path, updated: &FederatedNode) -> Result<()> {
let mut nodes = load_nodes(data_dir).await?;
if let Some(node) = nodes.iter_mut().find(|n| n.did == updated.did) {
if !updated.onion.is_empty() {
node.onion = updated.onion.clone();
}
if !updated.pubkey.is_empty() {
node.pubkey = updated.pubkey.clone();
}
if updated.name.is_some() {
node.name = updated.name.clone();
}
if updated.last_seen.is_some() {
node.last_seen = updated.last_seen.clone();
}
save_nodes(data_dir, &nodes).await?;
}
Ok(())
}
pub async fn update_node_state(data_dir: &Path, did: &str, state: NodeStateSnapshot) -> Result<()> {
let mut nodes = load_nodes(data_dir).await?;
if let Some(node) = nodes.iter_mut().find(|n| n.did == did) {
node.last_seen = Some(state.timestamp.clone());
// Update node name from sync if provided (peer announced their name)
if let Some(ref name) = state.node_name {
if !name.is_empty() {
node.name = Some(name.clone());
}
}
// Learn the peer's FIPS npub from their state snapshot so
// federations established before v1.4 (pre-fips_npub) start
// routing over FIPS on the very next sync. Refresh if the peer
// rotated their FIPS key, too.
if let Some(ref npub) = state.own_fips_npub {
if !npub.is_empty() && node.fips_npub.as_deref().map(str::trim) != Some(npub.trim()) {
node.fips_npub = Some(npub.clone());
}
}
node.last_state = Some(state);
save_nodes(data_dir, &nodes).await?;
}
Ok(())
}
// ──────────────────────────── Invite Storage ────────────────────────────
pub(crate) async fn load_invites(data_dir: &Path) -> Result<InvitesFile> {
let dir = data_dir.join(FEDERATION_DIR);
let path = dir.join(INVITES_FILE);
if !path.exists() {
return Ok(InvitesFile::default());
}
let content = fs::read_to_string(&path)
.await
.context("Failed to read invites")?;
let file: InvitesFile = serde_json::from_str(&content).unwrap_or_default();
Ok(file)
}
pub(crate) async fn save_invites(data_dir: &Path, invites: &InvitesFile) -> Result<()> {
let dir = ensure_dir(data_dir).await?;
let content = serde_json::to_string_pretty(invites).context("Failed to serialize invites")?;
fs::write(dir.join(INVITES_FILE), content)
.await
.context("Failed to write invites")?;
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
use crate::federation::types::AppStatus;
fn make_node(did: &str, onion: &str) -> FederatedNode {
FederatedNode {
did: did.to_string(),
pubkey: "aabbccdd".to_string(),
onion: onion.to_string(),
name: None,
trust_level: TrustLevel::Trusted,
added_at: "2026-01-01T00:00:00Z".to_string(),
last_seen: None,
last_state: None,
fips_npub: None,
last_transport: None,
last_transport_at: None,
}
}
#[test]
fn test_dedup_nodes_by_onion_collapses_same_onion() {
// Two entries share an onion (same physical node under two dids) — must
// collapse to one, keeping the first did and merging fips_npub/name (B1/B2).
let mut dup = make_node("did:key:zDUP", "shared.onion");
dup.fips_npub = Some("npub1merged".to_string());
dup.name = Some("Sapien".to_string());
let nodes = vec![
make_node("did:key:zKEEP", "shared.onion"),
dup,
make_node("did:key:zOTHER", "other.onion"),
];
let out = dedup_nodes_by_onion(nodes);
assert_eq!(out.len(), 2, "two distinct onions remain");
let kept = out.iter().find(|n| n.onion == "shared.onion").unwrap();
assert_eq!(kept.did, "did:key:zKEEP", "keeps first did for the onion");
assert_eq!(
kept.fips_npub.as_deref(),
Some("npub1merged"),
"merges fips_npub from the dropped duplicate"
);
assert_eq!(
kept.name.as_deref(),
Some("Sapien"),
"merges name from the dup"
);
}
#[test]
fn test_dedup_onion_suffix_insensitive() {
// The ".onion" suffix must not affect the match.
let nodes = vec![
make_node("did:key:z1", "abc"),
make_node("did:key:z2", "abc.onion"),
];
assert_eq!(dedup_nodes_by_onion(nodes).len(), 1);
}
#[tokio::test]
async fn test_load_nodes_empty_when_no_file() {
let dir = tempfile::tempdir().unwrap();
let nodes = load_nodes(dir.path()).await.unwrap();
assert!(nodes.is_empty());
}
#[tokio::test]
async fn test_save_and_load_nodes_roundtrip() {
let dir = tempfile::tempdir().unwrap();
let nodes = vec![
make_node("did:key:z1", "a.onion"),
make_node("did:key:z2", "b.onion"),
];
save_nodes(dir.path(), &nodes).await.unwrap();
let loaded = load_nodes(dir.path()).await.unwrap();
assert_eq!(loaded.len(), 2);
assert_eq!(loaded[0].did, "did:key:z1");
}
#[tokio::test]
async fn test_add_node_deduplicates_by_did() {
let dir = tempfile::tempdir().unwrap();
add_node(dir.path(), make_node("did:key:z1", "a.onion"))
.await
.unwrap();
let result = add_node(dir.path(), make_node("did:key:z1", "b.onion")).await;
assert!(result.is_err());
}
#[tokio::test]
async fn test_remove_node_by_did() {
let dir = tempfile::tempdir().unwrap();
add_node(dir.path(), make_node("did:key:z1", "a.onion"))
.await
.unwrap();
add_node(dir.path(), make_node("did:key:z2", "b.onion"))
.await
.unwrap();
let result = remove_node(dir.path(), "did:key:z1").await.unwrap();
assert_eq!(result.len(), 1);
assert_eq!(result[0].did, "did:key:z2");
}
#[tokio::test]
async fn test_remove_nonexistent_node_errors() {
let dir = tempfile::tempdir().unwrap();
let result = remove_node(dir.path(), "did:key:nonexistent").await;
assert!(result.is_err());
}
#[tokio::test]
async fn test_remove_tombstones_and_readd_clears_it() {
let dir = tempfile::tempdir().unwrap();
add_node(dir.path(), make_node("did:key:z1", "a.onion"))
.await
.unwrap();
// No tombstones yet.
assert!(load_removed_dids(dir.path()).await.unwrap().is_empty());
// Removing tombstones the DID so transitive discovery won't re-add it.
remove_node(dir.path(), "did:key:z1").await.unwrap();
let removed = load_removed_dids(dir.path()).await.unwrap();
assert!(
removed.contains("did:key:z1"),
"removed DID must be tombstoned"
);
// Explicitly re-adding clears the tombstone (intentional re-federate).
add_node(dir.path(), make_node("did:key:z1", "a.onion"))
.await
.unwrap();
assert!(
!load_removed_dids(dir.path())
.await
.unwrap()
.contains("did:key:z1"),
"explicit re-add must clear the tombstone"
);
}
#[tokio::test]
async fn test_set_trust_level() {
let dir = tempfile::tempdir().unwrap();
add_node(dir.path(), make_node("did:key:z1", "a.onion"))
.await
.unwrap();
let nodes = set_trust_level(dir.path(), "did:key:z1", TrustLevel::Observer)
.await
.unwrap();
assert_eq!(nodes[0].trust_level, TrustLevel::Observer);
}
#[tokio::test]
async fn test_update_node_state() {
let dir = tempfile::tempdir().unwrap();
add_node(dir.path(), make_node("did:key:z1", "a.onion"))
.await
.unwrap();
let state = NodeStateSnapshot {
timestamp: "2026-03-10T12:00:00Z".to_string(),
node_name: None,
apps: vec![AppStatus {
id: "bitcoin".to_string(),
status: "running".to_string(),
version: Some("27.0".to_string()),
}],
cpu_usage_percent: Some(45.2),
mem_used_bytes: Some(4_000_000_000),
mem_total_bytes: Some(8_000_000_000),
disk_used_bytes: None,
disk_total_bytes: None,
uptime_secs: Some(86400),
tor_active: Some(true),
nostr_npub: None,
own_fips_npub: None,
federated_peers: Vec::new(),
lat: None,
lon: None,
};
update_node_state(dir.path(), "did:key:z1", state)
.await
.unwrap();
let nodes = load_nodes(dir.path()).await.unwrap();
assert!(nodes[0].last_seen.is_some());
let ls = nodes[0].last_state.as_ref().unwrap();
assert_eq!(ls.apps.len(), 1);
assert_eq!(ls.cpu_usage_percent, Some(45.2));
}
}
core/archipelago/src/federation/sync.rs
+493
View File
@@ -0,0 +1,493 @@
//! Federation state sync and remote deployment.
//!
//! Requests prefer FIPS (direct ULA dial, ~LAN latency) and fall back to
//! Tor on any network failure. See `crate::fips::dial::PeerRequest` for
//! the fallback mechanics.
use anyhow::{Context, Result};
use std::path::Path;
use super::storage::update_node_state;
use super::types::{AppStatus, FederatedNode, FederationPeerHint, NodeStateSnapshot, TrustLevel};
use crate::fips::dial::PeerRequest;
/// Sync state with a single federated peer. Tries FIPS first; falls back
/// to Tor on any transport-level failure.
pub async fn sync_with_peer(
data_dir: &Path,
peer: &FederatedNode,
local_did: &str,
sign_fn: impl FnOnce(&[u8]) -> String,
) -> Result<NodeStateSnapshot> {
let timestamp = chrono::Utc::now().to_rfc3339();
let signature = sign_fn(timestamp.as_bytes());
let body = serde_json::json!({
"method": "federation.get-state",
"params": {}
});
let (resp, transport) = PeerRequest::new(peer.fips_npub.as_deref(), &peer.onion, "/rpc/v1")
.service(crate::settings::transport::PeerService::Federation)
.header("X-Federation-DID", local_did)
.header("X-Federation-Sig", signature)
.header("X-Federation-Timestamp", timestamp)
.timeout(std::time::Duration::from_secs(30))
// Fast-fail a cold/unreachable FIPS overlay (common between LAN and
// remote/Tailscale peers that share no FIPS path) so the Tor fallback —
// which answers an onion in ~3-5s — isn't stuck behind the full 30s FIPS
// budget. Without this, a state sync to a FIPS-unreachable peer "took
// ages" and join/sync appeared to time out even though Tor was healthy.
.fips_timeout(std::time::Duration::from_secs(6))
.send_json(&body)
.await
.context("Failed to reach federated peer")?;
if !resp.status().is_success() {
anyhow::bail!("Peer returned {} (via {})", resp.status(), transport);
}
// Record transport used so the UI badge on this peer's card reflects
// the transport that actually carried the call, not a prediction.
if let Err(e) = super::storage::record_peer_transport(
data_dir,
Some(&peer.did),
Some(&peer.onion),
&transport.to_string(),
)
.await
{
tracing::warn!("Failed to persist peer transport badge: {e:#}");
}
let result: serde_json::Value = resp.json().await.context("Invalid response from peer")?;
let state_val = result
.get("result")
.ok_or_else(|| anyhow::anyhow!("No result in peer response"))?;
let state: NodeStateSnapshot =
serde_json::from_value(state_val.clone()).context("Failed to parse peer state")?;
update_node_state(data_dir, &peer.did, state.clone()).await?;
// Transitive federation: merge in peers our (Trusted) source advertised
// so we can route directly to them over FIPS without a second invite
// hop. Only runs when the source is Trusted — Observer-level peers
// don't get to expand our federation on their own authority.
if peer.trust_level == TrustLevel::Trusted {
if let Err(e) =
merge_transitive_peers(data_dir, &peer.did, local_did, &state.federated_peers).await
{
tracing::warn!(
peer_did = %peer.did,
error = %e,
"Transitive federation merge failed (non-fatal)"
);
}
}
Ok(state)
}
/// Convenience wrapper: look up a federated peer by DID, derive our
/// own local_did / signing context from the node identity on disk, and
/// call sync_with_peer. Used by transitive-discovery code paths where
/// the caller only knows the peer's DID (e.g. the peer-joined RPC's
/// follow-up task).
pub async fn sync_with_peer_by_did(data_dir: &Path, peer_did: &str) -> Result<NodeStateSnapshot> {
let nodes = super::storage::load_nodes(data_dir).await?;
let peer = nodes
.into_iter()
.find(|n| n.did == peer_did)
.ok_or_else(|| anyhow::anyhow!("Unknown federation peer: {}", peer_did))?;
let identity_dir = data_dir.join("identity");
let node_identity = crate::identity::NodeIdentity::load_or_create(&identity_dir).await?;
let local_pubkey_hex = node_identity.pubkey_hex();
let local_did = crate::identity::did_key_from_pubkey_hex(&local_pubkey_hex)?;
sync_with_peer(data_dir, &peer, &local_did, |data| node_identity.sign(data)).await
}
/// Merge peers advertised by a Trusted federated node into our own
/// federation list. New peers are added at `Trusted` — hints only
/// arrive from peers we already trust, and `build_local_state` only
/// re-exports our Trusted list, so transitive membership carries the
/// same trust the direct-invite path gives. Existing peers get their
/// `fips_npub` refreshed if we hadn't learned it yet.
///
/// Peers we are (us) or that we already track by DID are skipped.
async fn merge_transitive_peers(
data_dir: &std::path::Path,
source_did: &str,
local_did: &str,
hints: &[FederationPeerHint],
) -> Result<()> {
if hints.is_empty() {
return Ok(());
}
let mut nodes = super::storage::load_nodes(data_dir).await?;
// Tombstoned DIDs: peers the operator explicitly removed. Never re-add
// them via transitive discovery, or deleted (e.g. stale test) nodes
// reappear on the next sync with any peer that still lists them.
let removed = super::storage::load_removed_dids(data_dir)
.await
.unwrap_or_default();
let mut added = 0u32;
let mut refreshed = 0u32;
for hint in hints {
// Don't import the source peer advertising itself, or our own DID
// when the source advertises us back as one of its trusted peers.
if hint.did == source_did || hint.did == local_did {
continue;
}
// Skip anything the operator deliberately removed.
if removed.contains(&hint.did) {
continue;
}
if let Some(existing) = nodes.iter_mut().find(|n| n.did == hint.did) {
// Already known — just refresh fips_npub if we didn't have one.
if existing.fips_npub.is_none() && hint.fips_npub.is_some() {
existing.fips_npub = hint.fips_npub.clone();
refreshed += 1;
}
continue;
}
// Same physical node advertised under a DIFFERENT did? Match on the
// onion (its stable network identity). Without this, a node that
// appears under two dids (e.g. after a key/did change) gets added
// twice — showing up duplicated in the trusted-node list (B1) and as
// two separate mesh chat contacts (B2). Merge into the existing entry.
let hint_onion = hint.onion.trim_end_matches(".onion");
if !hint_onion.is_empty() {
if let Some(existing) = nodes
.iter_mut()
.find(|n| n.onion.trim_end_matches(".onion") == hint_onion)
{
if existing.fips_npub.is_none() && hint.fips_npub.is_some() {
existing.fips_npub = hint.fips_npub.clone();
}
if existing.name.is_none() && hint.name.is_some() {
existing.name = hint.name.clone();
}
refreshed += 1;
continue;
}
}
nodes.push(FederatedNode {
did: hint.did.clone(),
pubkey: hint.pubkey.clone(),
onion: hint.onion.clone(),
name: hint.name.clone(),
trust_level: TrustLevel::Trusted,
added_at: chrono::Utc::now().to_rfc3339(),
last_seen: None,
last_state: None,
fips_npub: hint.fips_npub.clone(),
last_transport: None,
last_transport_at: None,
});
added += 1;
}
if added > 0 || refreshed > 0 {
super::storage::save_nodes(data_dir, &nodes).await?;
tracing::info!(
source_did = %source_did,
added,
refreshed,
"Transitive federation merge complete"
);
}
Ok(())
}
/// Build the local node's state snapshot for sharing with peers.
///
/// `federated_peers` should be the caller's full list of federated
/// nodes; `build_local_state` filters them down to a `FederationPeerHint`
/// so receivers can perform transitive pairing (learn peers-of-peers
/// and route directly over FIPS from now on). Only peers we trust are
/// shared — an Untrusted/Observer node should not be re-exported
/// through us to the network.
#[allow(clippy::too_many_arguments)]
pub fn build_local_state(
apps: Vec<AppStatus>,
cpu: f64,
mem_used: u64,
mem_total: u64,
disk_used: u64,
disk_total: u64,
uptime: u64,
tor_active: bool,
server_name: Option<String>,
nostr_npub: Option<String>,
own_fips_npub: Option<String>,
federated_peers: &[FederatedNode],
// Only Some when the node has opted in via server.set-location's
// `share` flag — see NodeStateSnapshot::lat/lon's doc comment.
shared_location: Option<(f64, f64)>,
) -> NodeStateSnapshot {
let hints = federated_peers
.iter()
.filter(|n| n.trust_level == TrustLevel::Trusted)
.map(|n| FederationPeerHint {
did: n.did.clone(),
pubkey: n.pubkey.clone(),
onion: n.onion.clone(),
name: n.name.clone(),
fips_npub: n.fips_npub.clone(),
})
.collect();
NodeStateSnapshot {
timestamp: chrono::Utc::now().to_rfc3339(),
node_name: server_name,
apps,
cpu_usage_percent: Some(cpu),
mem_used_bytes: Some(mem_used),
mem_total_bytes: Some(mem_total),
disk_used_bytes: Some(disk_used),
disk_total_bytes: Some(disk_total),
uptime_secs: Some(uptime),
tor_active: Some(tor_active),
nostr_npub,
own_fips_npub,
federated_peers: hints,
lat: shared_location.map(|(lat, _)| lat),
lon: shared_location.map(|(_, lon)| lon),
}
}
/// Deploy an app to a remote federated peer over Tor.
/// Only works if the peer is trusted and the app exists in our marketplace.
pub async fn deploy_to_peer(
peer: &FederatedNode,
app_id: &str,
version: &str,
marketplace_url: &str,
local_did: &str,
sign_fn: impl FnOnce(&[u8]) -> String,
) -> Result<serde_json::Value> {
if peer.trust_level != TrustLevel::Trusted {
anyhow::bail!(
"Can only deploy to trusted peers (current: {})",
peer.trust_level
);
}
let timestamp = chrono::Utc::now().to_rfc3339();
let signature = sign_fn(timestamp.as_bytes());
let body = serde_json::json!({
"method": "package.install",
"params": {
"id": app_id,
"version": version,
"marketplace-url": marketplace_url,
}
});
let (resp, transport) = PeerRequest::new(peer.fips_npub.as_deref(), &peer.onion, "/rpc/v1")
.service(crate::settings::transport::PeerService::Federation)
.header("X-Federation-DID", local_did)
.header("X-Federation-Sig", signature)
.header("X-Federation-Timestamp", timestamp)
.timeout(std::time::Duration::from_secs(120))
.send_json(&body)
.await
.context("Failed to reach federated peer for deploy")?;
if !resp.status().is_success() {
anyhow::bail!(
"Remote node returned HTTP {} (via {})",
resp.status(),
transport
);
}
let result: serde_json::Value = resp.json().await.context("Invalid response from peer")?;
if let Some(err) = result.get("error") {
if !err.is_null() {
let msg = err
.get("message")
.and_then(|m| m.as_str())
.unwrap_or("Unknown remote error");
anyhow::bail!("Remote node refused deploy: {}", msg);
}
}
Ok(serde_json::json!({
"deployed": true,
"app_id": app_id,
"peer_did": peer.did,
"peer_onion": peer.onion,
}))
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_build_local_state() {
let state = build_local_state(
vec![AppStatus {
id: "lnd".to_string(),
status: "running".to_string(),
version: Some("0.18".to_string()),
}],
25.5,
2_000_000_000,
8_000_000_000,
100_000_000_000,
500_000_000_000,
3600,
true,
Some("Test Node".to_string()),
None,
None,
&[],
None,
);
assert_eq!(state.apps.len(), 1);
assert_eq!(state.cpu_usage_percent, Some(25.5));
assert_eq!(state.tor_active, Some(true));
assert_eq!(state.node_name, Some("Test Node".to_string()));
assert!(state.federated_peers.is_empty());
assert_eq!(state.lat, None);
}
#[test]
fn build_local_state_filters_non_trusted_peers() {
let peers = vec![
FederatedNode {
did: "did:key:zTrusted".into(),
pubkey: "aa".into(),
onion: "t.onion".into(),
name: None,
trust_level: TrustLevel::Trusted,
added_at: "now".into(),
last_seen: None,
last_state: None,
fips_npub: Some("npub1a".into()),
last_transport: None,
last_transport_at: None,
},
FederatedNode {
did: "did:key:zObserver".into(),
pubkey: "bb".into(),
onion: "o.onion".into(),
name: None,
trust_level: TrustLevel::Observer,
added_at: "now".into(),
last_seen: None,
last_state: None,
fips_npub: Some("npub1b".into()),
last_transport: None,
last_transport_at: None,
},
FederatedNode {
did: "did:key:zUntrusted".into(),
pubkey: "cc".into(),
onion: "u.onion".into(),
name: None,
trust_level: TrustLevel::Untrusted,
added_at: "now".into(),
last_seen: None,
last_state: None,
fips_npub: None,
last_transport: None,
last_transport_at: None,
},
];
let state = build_local_state(
vec![],
0.0,
0,
0,
0,
0,
0,
true,
None,
None,
None,
&peers,
None,
);
assert_eq!(state.federated_peers.len(), 1);
assert_eq!(state.federated_peers[0].did, "did:key:zTrusted");
assert_eq!(
state.federated_peers[0].fips_npub.as_deref(),
Some("npub1a")
);
}
#[tokio::test]
async fn merge_transitive_peers_skips_source_and_local_node() {
let dir = tempfile::tempdir().unwrap();
super::super::storage::save_nodes(
dir.path(),
&[FederatedNode {
did: "did:key:zSource".into(),
pubkey: "aa".into(),
onion: "source.onion".into(),
name: Some("Source".into()),
trust_level: TrustLevel::Trusted,
added_at: "now".into(),
last_seen: None,
last_state: None,
fips_npub: None,
last_transport: None,
last_transport_at: None,
}],
)
.await
.unwrap();
merge_transitive_peers(
dir.path(),
"did:key:zSource",
"did:key:zLocal",
&[
FederationPeerHint {
did: "did:key:zSource".into(),
pubkey: "aa".into(),
onion: "source.onion".into(),
name: Some("Source".into()),
fips_npub: None,
},
FederationPeerHint {
did: "did:key:zLocal".into(),
pubkey: "bb".into(),
onion: "local.onion".into(),
name: Some("Local".into()),
fips_npub: None,
},
FederationPeerHint {
did: "did:key:zPeer".into(),
pubkey: "cc".into(),
onion: "peer.onion".into(),
name: Some("Kitchen".into()),
fips_npub: Some("npub1peer".into()),
},
],
)
.await
.unwrap();
let nodes = super::super::storage::load_nodes(dir.path()).await.unwrap();
assert_eq!(nodes.len(), 2);
assert!(nodes.iter().all(|n| n.did != "did:key:zLocal"));
let peer = nodes
.iter()
.find(|n| n.did == "did:key:zPeer")
.expect("trusted transitive peer should be added");
assert_eq!(peer.name.as_deref(), Some("Kitchen"));
assert_eq!(peer.trust_level, TrustLevel::Trusted);
assert_eq!(peer.fips_npub.as_deref(), Some("npub1peer"));
}
}
core/archipelago/src/federation/types.rs
+237
View File
@@ -0,0 +1,237 @@
//! Federation type definitions.
use serde::{Deserialize, Serialize};
/// Trust level for a federated node.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum TrustLevel {
Trusted,
Observer,
Untrusted,
}
impl std::fmt::Display for TrustLevel {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
TrustLevel::Trusted => write!(f, "trusted"),
TrustLevel::Observer => write!(f, "observer"),
TrustLevel::Untrusted => write!(f, "untrusted"),
}
}
}
impl TrustLevel {
/// Parse the lowercase wire form ("trusted" | "observer" | "untrusted").
pub fn parse(s: &str) -> Option<Self> {
match s {
"trusted" => Some(TrustLevel::Trusted),
"observer" => Some(TrustLevel::Observer),
"untrusted" => Some(TrustLevel::Untrusted),
_ => None,
}
}
/// The lower (less privileged) of two levels: Untrusted < Observer < Trusted.
pub fn min(self, other: Self) -> Self {
fn rank(l: TrustLevel) -> u8 {
match l {
TrustLevel::Untrusted => 0,
TrustLevel::Observer => 1,
TrustLevel::Trusted => 2,
}
}
if rank(self) <= rank(other) {
self
} else {
other
}
}
}
/// A federated node in our cluster.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FederatedNode {
pub did: String,
pub pubkey: String,
pub onion: String,
#[serde(default)]
pub name: Option<String>,
pub trust_level: TrustLevel,
pub added_at: String,
#[serde(default)]
pub last_seen: Option<String>,
#[serde(default)]
pub last_state: Option<NodeStateSnapshot>,
/// FIPS mesh npub (bech32) for this peer, when they advertise one.
/// Lets the transport router prefer FIPS over Tor for peer traffic.
#[serde(default)]
pub fips_npub: Option<String>,
/// Transport kind used on the most recent successful reach
/// ("fips" | "tor" | "mesh" | "lan"). Written after each successful
/// PeerRequest so the UI can show a ground-truth badge ("this peer
/// is currently being reached over FIPS") instead of a prediction
/// based on available addresses.
#[serde(default)]
pub last_transport: Option<String>,
/// RFC 3339 timestamp of the last_transport value.
#[serde(default)]
pub last_transport_at: Option<String>,
}
/// State snapshot received from a federated peer during sync.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct NodeStateSnapshot {
pub timestamp: String,
#[serde(default)]
pub node_name: Option<String>,
#[serde(default)]
pub apps: Vec<AppStatus>,
#[serde(default)]
pub cpu_usage_percent: Option<f64>,
#[serde(default)]
pub mem_used_bytes: Option<u64>,
#[serde(default)]
pub mem_total_bytes: Option<u64>,
#[serde(default)]
pub disk_used_bytes: Option<u64>,
#[serde(default)]
pub disk_total_bytes: Option<u64>,
#[serde(default)]
pub uptime_secs: Option<u64>,
#[serde(default)]
pub tor_active: Option<bool>,
/// bech32-encoded Nostr identity pubkey (npub1…) for cross-transport
/// peer identification in the mesh UI. Optional: older nodes that
/// haven't synced after this field was added will report None.
#[serde(default)]
pub nostr_npub: Option<String>,
/// The sender's own FIPS npub (bech32). Lets pre-FIPS federation
/// pairs — who federated before v1.4 added fips_npub to the invite
/// code — discover each other's FIPS identity on the next state
/// sync and route over FIPS from then on. Optional for back-compat
/// with older peers.
#[serde(default)]
pub own_fips_npub: Option<String>,
/// Minimal summary of peers this node trusts, used for transitive
/// federation: when Alice syncs with Bob, she learns Bob's trusted
/// peers and adds them as Observers on her side so `fips_npub` is
/// known and future state-syncs can route directly. Bounded to one
/// hop (Alice doesn't auto-promote Observer-via-Bob to Trusted nor
/// re-export them in her own state snapshots).
#[serde(default)]
pub federated_peers: Vec<FederationPeerHint>,
/// This node's own location, for the Mesh Map — only present when the
/// sender has opted in via `server.set-location`'s `share` flag. Absent
/// (not just null) for nodes that haven't opted in, so older receivers
/// and the map's "no location shared" state both fall out naturally.
#[serde(default)]
pub lat: Option<f64>,
#[serde(default)]
pub lon: Option<f64>,
}
/// Minimal peer summary shared via `NodeStateSnapshot.federated_peers`.
/// Excludes sensitive/per-receiver fields like trust_level and added_at.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FederationPeerHint {
pub did: String,
pub pubkey: String,
pub onion: String,
#[serde(default)]
pub name: Option<String>,
#[serde(default)]
pub fips_npub: Option<String>,
}
/// Status of a single app/container on a remote node.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AppStatus {
pub id: String,
pub status: String, // "running", "stopped", "installed"
#[serde(default)]
pub version: Option<String>,
}
/// A pending invite (outgoing or incoming).
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FederationInvite {
pub code: String,
pub did: String,
pub onion: String,
pub pubkey: String,
pub created_at: String,
#[serde(default)]
pub accepted: bool,
/// Inviter's FIPS mesh npub if advertised in the code.
#[serde(default)]
pub fips_npub: Option<String>,
/// Trust level this invite grants both sides ("Invite a Peer" mints
/// Observer invites, "Link Your Nodes" mints Trusted ones). Defaults
/// to Trusted for invites minted before this field existed.
#[serde(default = "default_invite_trust")]
pub trust_level: TrustLevel,
}
fn default_invite_trust() -> TrustLevel {
TrustLevel::Trusted
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_trust_level_serialization() {
let json = serde_json::to_string(&TrustLevel::Trusted).unwrap();
assert_eq!(json, "\"trusted\"");
let parsed: TrustLevel = serde_json::from_str("\"observer\"").unwrap();
assert_eq!(parsed, TrustLevel::Observer);
}
#[test]
fn test_federated_node_serialization_roundtrip() {
let node = FederatedNode {
did: "did:key:zABC".to_string(),
pubkey: "aabbccdd".to_string(),
onion: "test.onion".to_string(),
name: None,
trust_level: TrustLevel::Trusted,
added_at: "2026-01-01T00:00:00Z".to_string(),
last_seen: None,
last_state: None,
fips_npub: None,
last_transport: None,
last_transport_at: None,
};
let json = serde_json::to_string(&node).unwrap();
let parsed: FederatedNode = serde_json::from_str(&json).unwrap();
assert_eq!(parsed.did, "did:key:zABC");
assert_eq!(parsed.trust_level, TrustLevel::Trusted);
assert!(parsed.last_state.is_none());
assert!(parsed.fips_npub.is_none());
}
#[test]
fn test_federated_node_deserializes_without_fips_field() {
// Backward compat: nodes on older versions omit fips_npub entirely.
let json = r#"{
"did": "did:key:zOld",
"pubkey": "0011",
"onion": "old.onion",
"trust_level": "trusted",
"added_at": "2026-01-01T00:00:00Z"
}"#;
let parsed: FederatedNode = serde_json::from_str(json).unwrap();
assert!(parsed.fips_npub.is_none());
}
#[test]
fn test_node_state_snapshot_defaults() {
let json = r#"{"timestamp": "2026-01-01T00:00:00Z"}"#;
let state: NodeStateSnapshot = serde_json::from_str(json).unwrap();
assert!(state.apps.is_empty());
assert!(state.cpu_usage_percent.is_none());
}
}