//! Node-to-node messaging over Tor. //! Sends messages to peer .onion addresses via SOCKS5 proxy. //! Messages are persisted to disk and survive restarts. use anyhow::{Context, Result}; use serde::{Deserialize, Serialize}; use std::path::{Path, PathBuf}; use std::sync::{Mutex, OnceLock}; const MAX_STORED: usize = 200; #[derive(Debug, Clone, Serialize, Deserialize)] pub struct IncomingMessage { pub from_pubkey: String, #[serde(default)] pub from_onion: Option, /// Sender's node name (for display in group chat). #[serde(default)] pub from_name: Option, pub message: String, pub timestamp: String, /// "sent" or "received" #[serde(default = "default_received")] pub direction: String, } fn default_received() -> String { "received".to_string() } #[derive(Debug, Default, Serialize, Deserialize)] struct MessageStore { messages: Vec, } fn store() -> &'static Mutex { static STORE: OnceLock> = OnceLock::new(); STORE.get_or_init(|| Mutex::new(MessageStore::default())) } fn data_path() -> &'static Mutex> { static PATH: OnceLock>> = OnceLock::new(); PATH.get_or_init(|| Mutex::new(None)) } /// Initialize message store — load from disk. Call once at startup. pub async fn init(data_dir: &Path) { let path = data_dir.join("messages.json"); *data_path().lock().unwrap_or_else(|e| e.into_inner()) = Some(path.clone()); if let Ok(content) = tokio::fs::read_to_string(&path).await { if let Ok(loaded) = serde_json::from_str::(&content) { let mut guard = store().lock().unwrap_or_else(|e| e.into_inner()); *guard = loaded; tracing::info!("Loaded {} messages from disk", guard.messages.len()); } } } /// Persist current messages to disk. /// Serializes under the lock, then writes asynchronously via spawn_blocking /// to avoid blocking the tokio runtime. fn persist() { let guard = store().lock().unwrap_or_else(|e| e.into_inner()); let path_guard = data_path().lock().unwrap_or_else(|e| e.into_inner()); if let Some(ref path) = *path_guard { if let Ok(content) = serde_json::to_string(&*guard) { let path = path.clone(); drop(path_guard); drop(guard); tokio::task::spawn(async move { let _ = tokio::fs::write(&path, content).await; }); } } } /// Store a received message (called from HTTP handler). pub fn store_received_sync(from_pubkey: &str, message: &str, from_name: Option<&str>) { let ts = chrono::Utc::now().to_rfc3339(); let mut guard = store().lock().unwrap_or_else(|e| e.into_inner()); // Deduplication: skip if same pubkey + message within last 30 seconds let dominated = guard.messages.iter().rev().take(20).any(|m| { m.from_pubkey == from_pubkey && m.message == message && m.direction == "received" && within_seconds(&m.timestamp, &ts, 30) }); if dominated { return; } guard.messages.push(IncomingMessage { from_pubkey: from_pubkey.to_string(), from_onion: None, from_name: from_name.map(|s| s.to_string()), message: message.to_string(), timestamp: ts, direction: "received".to_string(), }); trim_messages(&mut guard); drop(guard); persist(); } pub async fn store_received(from_pubkey: &str, message: &str, from_name: Option<&str>) { store_received_sync(from_pubkey, message, from_name); } /// Store a sent message (for display in Archipelago channel). pub fn store_sent(message: &str) { let mut guard = store().lock().unwrap_or_else(|e| e.into_inner()); guard.messages.push(IncomingMessage { from_pubkey: "me".to_string(), from_onion: None, from_name: None, message: message.to_string(), timestamp: chrono::Utc::now().to_rfc3339(), direction: "sent".to_string(), }); trim_messages(&mut guard); drop(guard); persist(); } /// Get all messages (sent + received) for UI display. pub fn get_received() -> Vec { store() .lock() .unwrap_or_else(|e| e.into_inner()) .messages .clone() } fn trim_messages(store: &mut MessageStore) { if store.messages.len() > MAX_STORED { let drain = store.messages.len() - MAX_STORED; store.messages.drain(0..drain); } } /// Check if two ISO8601 timestamps are within N seconds of each other. fn within_seconds(ts1: &str, ts2: &str, secs: i64) -> bool { use chrono::DateTime; let a = DateTime::parse_from_rfc3339(ts1).ok(); let b = DateTime::parse_from_rfc3339(ts2).ok(); match (a, b) { (Some(a), Some(b)) => (a - b).num_seconds().unsigned_abs() < secs as u64, _ => false, } } // ─── E2E Encryption ───────────────────────────────────────────── use crate::mesh::crypto; use base64::Engine; /// Encrypt a message for a recipient using X25519 ECDH + ChaCha20-Poly1305. /// Returns base64-encoded ciphertext (nonce + encrypted data). fn encrypt_for_peer( our_signing_key: &ed25519_dalek::SigningKey, their_pubkey_hex: &str, plaintext: &str, ) -> Result { let their_pubkey_bytes: [u8; 32] = hex::decode(their_pubkey_hex) .context("Invalid peer pubkey hex")? .try_into() .map_err(|_| anyhow::anyhow!("Invalid peer pubkey length"))?; let their_x25519 = crypto::ed25519_pubkey_to_x25519(&their_pubkey_bytes)?; let our_x25519 = crypto::ed25519_secret_to_x25519(our_signing_key); let shared = crypto::x25519_shared_secret(&our_x25519, &their_x25519); // HKDF to derive message key (domain separation for Tor messages) let msg_key_bytes = crypto::hkdf_sha256( b"archipelago-tor-msg-v1", &shared, b"message-encryption", 32, )?; let msg_key: [u8; 32] = msg_key_bytes .try_into() .map_err(|_| anyhow::anyhow!("HKDF key length mismatch"))?; let encrypted = crypto::encrypt(&msg_key, plaintext.as_bytes())?; Ok(base64::engine::general_purpose::STANDARD.encode(&encrypted)) } /// Decrypt a message from a sender using X25519 ECDH + ChaCha20-Poly1305. pub fn decrypt_from_peer( our_signing_key: &ed25519_dalek::SigningKey, sender_pubkey_hex: &str, encrypted_b64: &str, ) -> Result { let sender_pubkey_bytes: [u8; 32] = hex::decode(sender_pubkey_hex) .context("Invalid sender pubkey hex")? .try_into() .map_err(|_| anyhow::anyhow!("Invalid sender pubkey length"))?; let sender_x25519 = crypto::ed25519_pubkey_to_x25519(&sender_pubkey_bytes)?; let our_x25519 = crypto::ed25519_secret_to_x25519(our_signing_key); let shared = crypto::x25519_shared_secret(&our_x25519, &sender_x25519); let msg_key_bytes = crypto::hkdf_sha256( b"archipelago-tor-msg-v1", &shared, b"message-encryption", 32, )?; let msg_key: [u8; 32] = msg_key_bytes .try_into() .map_err(|_| anyhow::anyhow!("HKDF key length mismatch"))?; let encrypted = base64::engine::general_purpose::STANDARD .decode(encrypted_b64) .context("Invalid base64 ciphertext")?; let plaintext_bytes = crypto::decrypt(&msg_key, &encrypted)?; String::from_utf8(plaintext_bytes).context("Decrypted message is not valid UTF-8") } // ─── Tor Messaging ────────────────────────────────────────────── /// Tor v3 onion hostname is 56 base32 chars (a-z, 2-7). Reject invalid formats. fn validate_onion(onion: &str) -> Result<()> { let host = onion.trim_end_matches(".onion"); if host.len() != 56 { anyhow::bail!( "Invalid onion address (expected 56 chars, got {}). The peer may have wrong data - try removing and re-adding via Discover.", host.len() ); } let valid = host .chars() .all(|c| c.is_ascii_lowercase() || ('2'..='7').contains(&c)); if !valid { anyhow::bail!("Invalid onion address: must be 56 base32 chars (a-z, 2-7)"); } Ok(()) } /// Send an encrypted message to a peer over Tor. /// The message is encrypted with ChaCha20-Poly1305 using an X25519 shared secret /// derived from both nodes' ed25519 keys. pub async fn send_to_peer( onion: &str, fips_npub: Option<&str>, from_pubkey: &str, message: &str, signing_key: Option<&ed25519_dalek::SigningKey>, recipient_pubkey: Option<&str>, from_name: Option<&str>, ) -> Result<()> { validate_onion(onion)?; // Encrypt message if we have both keys let (payload_message, encrypted) = match (signing_key, recipient_pubkey) { (Some(sk), Some(rpk)) => match encrypt_for_peer(sk, rpk, message) { Ok(enc) => (enc, true), Err(e) => { tracing::warn!("Encryption failed, sending plaintext: {}", e); (message.to_string(), false) } }, _ => (message.to_string(), false), }; let mut body = serde_json::json!({ "from_pubkey": from_pubkey, "message": payload_message, "timestamp": chrono::Utc::now().to_rfc3339(), "encrypted": encrypted, }); if let Some(name) = from_name { body["from_name"] = serde_json::Value::String(name.to_string()); } let (resp, transport) = crate::fips::dial::PeerRequest::new( fips_npub, onion, "/archipelago/node-message", ) .timeout(std::time::Duration::from_secs(60)) .send_json(&body) .await .map_err(|e| { let msg = e.to_string(); if msg.contains("connection refused") || msg.contains("Connection refused") { anyhow::anyhow!("Peer unreachable. Check Tor (127.0.0.1:9050) and FIPS daemon status.") } else if msg.contains("timeout") || msg.contains("timed out") { anyhow::anyhow!("Connection timed out. The peer may be offline.") } else { anyhow::anyhow!("Failed to send: {}", msg) } })?; if !resp.status().is_success() { anyhow::bail!( "Peer returned {} {} (via {}). The peer may need /archipelago/ in its nginx config.", resp.status().as_u16(), resp.status().canonical_reason().unwrap_or(""), transport, ); } Ok(()) } /// Check if a peer is reachable (ping). FIPS is preferred when an npub /// is known, Tor is the fallback. pub async fn check_peer_reachable(onion: &str, fips_npub: Option<&str>) -> Result { validate_onion(onion)?; match crate::fips::dial::PeerRequest::new(fips_npub, onion, "/health") .timeout(std::time::Duration::from_secs(30)) .send_get() .await { Ok((resp, _)) => Ok(resp.status().is_success()), Err(_) => Ok(false), } }