Persist encrypted peer approval delivery and bind discovery invite identities

core/archipelago/src/federation/handshake_delivery.rs

@@ -0,0 +1,199 @@
+//! Durable, node-encrypted approval replies. Relay acknowledgement is not peer
+//! acceptance: keep retrying the same invite until reciprocal membership exists.
+use anyhow::{Context, Result};
+use serde::{Deserialize, Serialize};
+use std::path::Path;
+use tokio::{fs, io::AsyncWriteExt};
+
+const FILE: &str = "federation/handshake-delivery.enc";
+const DOMAIN: &[u8] = b"archipelago-handshake-delivery-v1";
+static LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());
+
+#[derive(Clone, Serialize, Deserialize)]
+pub(crate) struct ApprovalReply {
+    pub request_id: String,
+    pub recipient: String,
+    pub expected_did: String,
+    pub invite_code: String,
+    pub attempts: u32,
+    pub next_attempt: i64,
+}
+
+async fn load(data_dir: &Path) -> Result<Vec<ApprovalReply>> {
+    let bytes = match fs::read(data_dir.join(FILE)).await {
+        Ok(bytes) => bytes,
+        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
+        Err(e) => return Err(e.into()),
+    };
+    let key = crate::storage_crypto::derive_key(data_dir, DOMAIN).await?;
+    let plaintext = crate::storage_crypto::open(&bytes, &key)?;
+    serde_json::from_slice(&plaintext)
+        .context("Invalid handshake delivery store; preserved for recovery")
+}
+
+async fn save(data_dir: &Path, entries: &[ApprovalReply]) -> Result<()> {
+    let key = crate::storage_crypto::derive_key(data_dir, DOMAIN).await?;
+    let bytes = crate::storage_crypto::seal(&serde_json::to_vec(entries)?, &key)?;
+    let path = data_dir.join(FILE);
+    let parent = path.parent().context("Delivery parent missing")?;
+    fs::create_dir_all(parent).await?;
+    let temporary = parent.join(format!(".delivery-{}.tmp", uuid::Uuid::new_v4()));
+    let result = async {
+        let mut file = fs::OpenOptions::new()
+            .create_new(true)
+            .write(true)
+            .mode(0o600)
+            .open(&temporary)
+            .await?;
+        file.write_all(&bytes).await?;
+        file.sync_all().await?;
+        drop(file);
+        fs::rename(&temporary, &path).await?;
+        fs::File::open(parent).await?.sync_all().await?;
+        Ok::<_, anyhow::Error>(())
+    }
+    .await;
+    if result.is_err() {
+        let _ = fs::remove_file(temporary).await;
+    }
+    result
+}
+
+pub(crate) async fn find(data_dir: &Path, request_id: &str) -> Result<Option<ApprovalReply>> {
+    let _guard = LOCK.lock().await;
+    Ok(load(data_dir)
+        .await?
+        .into_iter()
+        .find(|entry| entry.request_id == request_id))
+}
+
+pub(crate) async fn stage(data_dir: &Path, reply: ApprovalReply) -> Result<ApprovalReply> {
+    let _guard = LOCK.lock().await;
+    let mut entries = load(data_dir).await?;
+    if let Some(existing) = entries
+        .iter()
+        .find(|entry| entry.request_id == reply.request_id)
+    {
+        anyhow::ensure!(
+            existing.recipient == reply.recipient && existing.expected_did == reply.expected_did,
+            "Approval recipient changed; refusing delivery"
+        );
+        return Ok(existing.clone());
+    }
+    anyhow::ensure!(entries.len() < 1024, "Handshake delivery queue is full");
+    entries.push(reply.clone());
+    save(data_dir, &entries).await?;
+    Ok(reply)
+}
+
+/// Claim before sending, including failed sends. Concurrent polls cannot create
+/// retry storms; a crash after this write delays but never loses the reply.
+pub(crate) async fn claim(
+    data_dir: &Path,
+    request_id: &str,
+    now: i64,
+) -> Result<Option<ApprovalReply>> {
+    let _guard = LOCK.lock().await;
+    let mut entries = load(data_dir).await?;
+    let Some(entry) = entries
+        .iter_mut()
+        .find(|entry| entry.request_id == request_id)
+    else {
+        return Ok(None);
+    };
+    if entry.next_attempt > now {
+        return Ok(None);
+    }
+    entry.attempts = entry.attempts.saturating_add(1);
+    let delay = 30_i64
+        .saturating_mul(1_i64 << entry.attempts.min(7))
+        .min(3600);
+    entry.next_attempt = now.saturating_add(delay);
+    let claimed = entry.clone();
+    save(data_dir, &entries).await?;
+    Ok(Some(claimed))
+}
+
+pub(crate) async fn remove(data_dir: &Path, request_id: &str) -> Result<()> {
+    let _guard = LOCK.lock().await;
+    let mut entries = load(data_dir).await?;
+    let before = entries.len();
+    entries.retain(|entry| entry.request_id != request_id);
+    if entries.len() != before {
+        save(data_dir, &entries).await?;
+    }
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    async fn fixture() -> tempfile::TempDir {
+        let dir = tempfile::tempdir().unwrap();
+        fs::create_dir_all(dir.path().join("identity"))
+            .await
+            .unwrap();
+        fs::write(dir.path().join("identity/node_key"), [7; 32])
+            .await
+            .unwrap();
+        dir
+    }
+    fn reply() -> ApprovalReply {
+        ApprovalReply {
+            request_id: "request-1".into(),
+            recipient: "recipient".into(),
+            expected_did: "did:key:peer".into(),
+            invite_code: "secret-invite".into(),
+            attempts: 0,
+            next_attempt: 0,
+        }
+    }
+    #[tokio::test]
+    async fn encrypted_reply_survives_reload_and_retry_claim_is_exclusive() {
+        let dir = fixture().await;
+        stage(dir.path(), reply()).await.unwrap();
+        let raw = fs::read(dir.path().join(FILE)).await.unwrap();
+        assert!(!raw.windows(13).any(|bytes| bytes == b"secret-invite"));
+        assert_eq!(
+            find(dir.path(), "request-1")
+                .await
+                .unwrap()
+                .unwrap()
+                .invite_code,
+            "secret-invite"
+        );
+        let (a, b) = tokio::join!(
+            claim(dir.path(), "request-1", 100),
+            claim(dir.path(), "request-1", 100)
+        );
+        assert_eq!(
+            usize::from(a.unwrap().is_some()) + usize::from(b.unwrap().is_some()),
+            1
+        );
+        assert!(claim(dir.path(), "request-1", 159).await.unwrap().is_none());
+        assert_eq!(
+            claim(dir.path(), "request-1", 160)
+                .await
+                .unwrap()
+                .unwrap()
+                .attempts,
+            2
+        );
+        remove(dir.path(), "request-1").await.unwrap();
+        assert!(find(dir.path(), "request-1").await.unwrap().is_none());
+    }
+    #[tokio::test]
+    async fn corrupt_store_is_not_overwritten_and_recipient_cannot_change() {
+        let dir = fixture().await;
+        stage(dir.path(), reply()).await.unwrap();
+        let mut other = reply();
+        other.recipient = "different-recipient".into();
+        assert!(stage(dir.path(), other).await.is_err());
+        let path = dir.path().join(FILE);
+        let mut raw = fs::read(&path).await.unwrap();
+        raw[20] ^= 1;
+        fs::write(&path, &raw).await.unwrap();
+        assert!(stage(dir.path(), reply()).await.is_err());
+        assert_eq!(fs::read(path).await.unwrap(), raw);
+    }
+}

core/archipelago/src/federation/invites.rs

@@ -134,6 +134,32 @@ pub fn parse_invite(code: &str) -> Result<ParsedInvite> {
     })
 }
 
+/// Bind a Nostr-discovery reply to the node the operator requested, and cap
+/// its grant before any local node entry or callback is written. Legacy invites
+/// default to Trusted, which must never transiently authorize discovery peers.
+pub(crate) fn restrict_discovery_invite(code: &str, expected_did: &str) -> Result<String> {
+    use base64::Engine;
+    let parsed = parse_invite(code)?;
+    anyhow::ensure!(
+        !expected_did.is_empty() && parsed.did == expected_did,
+        "Peer invite does not match the requested node"
+    );
+    anyhow::ensure!(
+        crate::identity::did_key_from_pubkey_hex(&parsed.pubkey)? == parsed.did,
+        "Peer invite DID does not match its identity key"
+    );
+    let bytes = base64::engine::general_purpose::URL_SAFE_NO_PAD.decode(
+        code.strip_prefix("fed1:")
+            .context("Invalid invite prefix")?,
+    )?;
+    let mut payload: serde_json::Value = serde_json::from_slice(&bytes)?;
+    payload["trust"] = serde_json::json!("observer");
+    Ok(format!(
+        "fed1:{}",
+        base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(serde_json::to_vec(&payload)?)
+    ))
+}
+
 /// Accept an invite: parse code, verify the remote node, add to federation.
 pub async fn accept_invite(
     data_dir: &Path,
@@ -621,3 +647,35 @@ mod tests {
         assert_eq!(nodes.len(), 1, "re-accept should not duplicate");
     }
 }
+
+#[cfg(test)]
+mod discovery_invite_scope_tests {
+    use super::*;
+    use base64::Engine;
+    #[test]
+    fn discovery_reply_binds_identity_and_caps_legacy_trust_before_acceptance() {
+        let key = "33".repeat(32);
+        let did = crate::identity::did_key_from_pubkey_hex(&key).unwrap();
+        let payload =
+            serde_json::json!({"did":did,"pubkey":key,"onion":"test.onion","token":"test-token"});
+        let code = format!(
+            "fed1:{}",
+            base64::engine::general_purpose::URL_SAFE_NO_PAD
+                .encode(serde_json::to_vec(&payload).unwrap())
+        );
+        let restricted = restrict_discovery_invite(&code, &did).unwrap();
+        let parsed = parse_invite(&restricted).unwrap();
+        assert_eq!(parsed.trust_level, TrustLevel::Observer);
+        assert_eq!(parsed.token, "test-token");
+        assert!(restrict_discovery_invite(&code, "did:key:someone-else").is_err());
+        assert!(restrict_discovery_invite(&code, "").is_err());
+        let mut forged = payload;
+        forged["pubkey"] = serde_json::json!("44".repeat(32));
+        let forged = format!(
+            "fed1:{}",
+            base64::engine::general_purpose::URL_SAFE_NO_PAD
+                .encode(serde_json::to_vec(&forged).unwrap())
+        );
+        assert!(restrict_discovery_invite(&forged, &did).is_err());
+    }
+}

core/archipelago/src/federation/mod.rs

@@ -6,12 +6,14 @@
 
 mod invites;
 pub mod pending;
+pub(crate) mod handshake_delivery;
 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};
+pub(crate) use invites::restrict_discovery_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;

core/archipelago/src/federation/pending.rs

@@ -14,6 +14,9 @@ use anyhow::{Context, Result};
 use serde::{Deserialize, Serialize};
 use std::path::Path;
 use tokio::fs;
+use tokio::io::AsyncWriteExt;
+
+static PENDING_STORE_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());
 
 const PENDING_FILE: &str = "federation/pending_requests.json";
 const MAX_PENDING_PER_PUBKEY: usize = 5;
@@ -76,11 +79,12 @@ pub async fn load_pending(data_dir: &Path) -> Result<Vec<PendingPeerRequest>> {
     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();
+    let file: PendingRequestsFile = serde_json::from_str(&content)
+        .context("Invalid pending requests file; preserving existing data")?;
     Ok(file.requests)
 }
 
-pub async fn save_pending(data_dir: &Path, requests: &[PendingPeerRequest]) -> Result<()> {
+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)
@@ -92,9 +96,27 @@ pub async fn save_pending(data_dir: &Path, requests: &[PendingPeerRequest]) -> R
     };
     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")?;
+    let parent = path.parent().context("Pending requests parent missing")?;
+    let temporary = parent.join(format!(".pending-{}.tmp", uuid::Uuid::new_v4()));
+    let result = async {
+        let mut file = fs::OpenOptions::new()
+            .write(true)
+            .create_new(true)
+            .mode(0o600)
+            .open(&temporary)
+            .await?;
+        file.write_all(content.as_bytes()).await?;
+        file.sync_all().await?;
+        drop(file);
+        fs::rename(&temporary, &path).await?;
+        fs::File::open(parent).await?.sync_all().await?;
+        Ok::<_, anyhow::Error>(())
+    }
+    .await;
+    if result.is_err() {
+        let _ = fs::remove_file(&temporary).await;
+    }
+    result.context("Failed to atomically save pending requests")?;
     Ok(())
 }
 
@@ -102,7 +124,10 @@ pub async fn save_pending(data_dir: &Path, requests: &[PendingPeerRequest]) -> R
 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) {
+        if !matches!(
+            r.state,
+            PendingState::Pending | PendingState::Sent | PendingState::Approved
+        ) {
             continue;
         }
         if let Ok(ts) = chrono::DateTime::parse_from_rfc3339(&r.received_at) {
@@ -131,6 +156,7 @@ pub async fn insert_inbound(
     from_name: Option<String>,
     message: Option<String>,
 ) -> Result<Option<PendingPeerRequest>> {
+    let _guard = PENDING_STORE_LOCK.lock().await;
     let mut requests = load_pending(data_dir).await?;
     expire_stale(&mut requests);
 
@@ -189,6 +215,7 @@ pub async fn insert_outbound(
     to_name: Option<String>,
     message: Option<String>,
 ) -> Result<PendingPeerRequest> {
+    let _guard = PENDING_STORE_LOCK.lock().await;
     let mut requests = load_pending(data_dir).await?;
     expire_stale(&mut requests);
     requests.retain(|r| {
@@ -218,6 +245,7 @@ pub async fn find_by_id(data_dir: &Path, id: &str) -> Result<Option<PendingPeerR
 }
 
 pub async fn set_state(data_dir: &Path, id: &str, state: PendingState) -> Result<()> {
+    let _guard = PENDING_STORE_LOCK.lock().await;
     let mut requests = load_pending(data_dir).await?;
     if let Some(r) = requests.iter_mut().find(|r| r.id == id) {
         r.state = state;
@@ -228,10 +256,32 @@ pub async fn set_state(data_dir: &Path, id: &str, state: PendingState) -> Result
     Ok(())
 }
 
+/// Resolve a pending decision once; concurrent approval/rejection cannot
+/// overwrite each other after a slow network request.
+pub async fn decide(data_dir: &Path, id: &str, decision: PendingState) -> Result<()> {
+    anyhow::ensure!(
+        matches!(decision, PendingState::Approved | PendingState::Rejected),
+        "Invalid pending decision"
+    );
+    let _guard = PENDING_STORE_LOCK.lock().await;
+    let mut requests = load_pending(data_dir).await?;
+    let row = requests
+        .iter_mut()
+        .find(|row| row.id == id)
+        .context("Pending request not found")?;
+    anyhow::ensure!(
+        !row.outbound && row.state == PendingState::Pending,
+        "Request has already been decided"
+    );
+    row.state = decision;
+    save_pending(data_dir, &requests).await
+}
+
 /// 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 _guard = PENDING_STORE_LOCK.lock().await;
     let mut requests = load_pending(data_dir).await?;
     let before = requests.len();
     requests.retain(|r| r.id != id);
@@ -372,3 +422,140 @@ mod tests {
         assert_eq!(reloaded.state, PendingState::Approved);
     }
 }
+
+#[cfg(test)]
+mod persistence_regressions {
+    use super::*;
+
+    #[tokio::test]
+    async fn concurrent_requests_are_not_lost() {
+        let dir = tempfile::tempdir().unwrap();
+        let mut tasks = Vec::new();
+        for i in 0..24 {
+            let path = dir.path().to_path_buf();
+            tasks.push(tokio::spawn(async move {
+                insert_inbound(
+                    &path,
+                    format!("key-{i}"),
+                    format!("npub-{i}"),
+                    format!("did:key:{i}"),
+                    None,
+                    None,
+                )
+                .await
+                .unwrap()
+            }));
+        }
+        for task in tasks {
+            assert!(task.await.unwrap().is_some());
+        }
+        let rows = load_pending(dir.path()).await.unwrap();
+        assert_eq!(rows.len(), 24);
+        assert_eq!(
+            rows.iter()
+                .map(|r| &r.id)
+                .collect::<std::collections::HashSet<_>>()
+                .len(),
+            24
+        );
+    }
+
+    #[tokio::test]
+    async fn malformed_store_is_preserved_instead_of_replaced_with_one_request() {
+        let dir = tempfile::tempdir().unwrap();
+        let path = dir.path().join(PENDING_FILE);
+        fs::create_dir_all(path.parent().unwrap()).await.unwrap();
+        let damaged = b"{incomplete existing requests";
+        fs::write(&path, damaged).await.unwrap();
+        assert!(insert_inbound(
+            dir.path(),
+            "key".into(),
+            "npub".into(),
+            "did:key:test".into(),
+            None,
+            None
+        )
+        .await
+        .is_err());
+        assert_eq!(fs::read(path).await.unwrap(), damaged);
+    }
+}
+
+#[cfg(test)]
+mod decision_regressions {
+    use super::*;
+    #[tokio::test]
+    async fn only_one_concurrent_operator_decision_wins() {
+        let dir = tempfile::tempdir().unwrap();
+        let row = insert_inbound(
+            dir.path(),
+            "key".into(),
+            "npub".into(),
+            "did:key:peer".into(),
+            None,
+            None,
+        )
+        .await
+        .unwrap()
+        .unwrap();
+        let (approve, reject) = tokio::join!(
+            decide(dir.path(), &row.id, PendingState::Approved),
+            decide(dir.path(), &row.id, PendingState::Rejected)
+        );
+        assert_eq!(
+            usize::from(approve.is_ok()) + usize::from(reject.is_ok()),
+            1
+        );
+        let saved = find_by_id(dir.path(), &row.id).await.unwrap().unwrap();
+        assert_eq!(
+            saved.state,
+            if approve.is_ok() {
+                PendingState::Approved
+            } else {
+                PendingState::Rejected
+            }
+        );
+    }
+    #[tokio::test]
+    async fn an_expired_approval_does_not_block_a_new_request_forever() {
+        let dir = tempfile::tempdir().unwrap();
+        let mut row = insert_inbound(
+            dir.path(),
+            "key".into(),
+            "npub".into(),
+            "did:key:peer".into(),
+            None,
+            None,
+        )
+        .await
+        .unwrap()
+        .unwrap();
+        row.state = PendingState::Approved;
+        row.received_at = (chrono::Utc::now() - chrono::Duration::days(31)).to_rfc3339();
+        save_pending(dir.path(), &[row]).await.unwrap();
+        let renewed = insert_inbound(
+            dir.path(),
+            "key".into(),
+            "npub".into(),
+            "did:key:peer".into(),
+            None,
+            None,
+        )
+        .await
+        .unwrap();
+        assert!(renewed.is_some());
+        let rows = load_pending(dir.path()).await.unwrap();
+        assert_eq!(
+            rows.iter()
+                .filter(|r| r.state == PendingState::Expired)
+                .count(),
+            1
+        );
+        assert_eq!(
+            rows.iter()
+                .filter(|r| r.state == PendingState::Pending)
+                .count(),
+            1
+        );
+    }
+}