archy/core/archipelago/src/node_message.rs

//! 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<String>,
    /// Sender's node name (for display in group chat).
    #[serde(default)]
    pub from_name: Option<String>,
    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<IncomingMessage>,
}

fn store() -> &'static Mutex<MessageStore> {
    static STORE: OnceLock<Mutex<MessageStore>> = OnceLock::new();
    STORE.get_or_init(|| Mutex::new(MessageStore::default()))
}

fn data_path() -> &'static Mutex<Option<PathBuf>> {
    static PATH: OnceLock<Mutex<Option<PathBuf>>> = 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::<MessageStore>(&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<IncomingMessage> {
    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<String> {
    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<String> {
    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,
    from_pubkey: &str,
    message: &str,
    signing_key: Option<&ed25519_dalek::SigningKey>,
    recipient_pubkey: Option<&str>,
    from_name: Option<&str>,
) -> Result<()> {
    validate_onion(onion)?;

    let host = if onion.ends_with(".onion") {
        onion.to_string()
    } else {
        format!("{}.onion", onion)
    };
    let url = format!("http://{}/archipelago/node-message", host);

    // 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 proxy =
        reqwest::Proxy::all(crate::constants::TOR_SOCKS_PROXY).context("Invalid Tor proxy")?;
    let client = reqwest::Client::builder()
        .proxy(proxy)
        .timeout(std::time::Duration::from_secs(60))
        .build()
        .context("Failed to build HTTP client")?;

    let resp = client.post(&url).json(&body).send().await.map_err(|e| {
        let msg = e.to_string();
        if msg.contains("connection refused") || msg.contains("Connection refused") {
            anyhow::anyhow!("Tor not reachable at 127.0.0.1:9050. Is Tor running?")
        } else if msg.contains("timeout") || msg.contains("timed out") {
            anyhow::anyhow!(
                "Connection timed out. The peer may be offline or unreachable over Tor."
            )
        } else {
            anyhow::anyhow!("Failed to send over Tor: {}", msg)
        }
    })?;

    if !resp.status().is_success() {
        anyhow::bail!(
            "Peer returned {} {}. The peer may need /archipelago/ in its nginx config.",
            resp.status().as_u16(),
            resp.status().canonical_reason().unwrap_or("")
        );
    }
    Ok(())
}

/// Check if a peer is reachable (ping over Tor).
pub async fn check_peer_reachable(onion: &str) -> Result<bool> {
    validate_onion(onion)?;

    let host = if onion.ends_with(".onion") {
        onion.to_string()
    } else {
        format!("{}.onion", onion)
    };
    let url = format!("http://{}/health", host);
    let proxy =
        reqwest::Proxy::all(crate::constants::TOR_SOCKS_PROXY).context("Invalid Tor proxy")?;
    let client = reqwest::Client::builder()
        .proxy(proxy)
        .timeout(std::time::Duration::from_secs(30))
        .build()
        .context("Failed to build HTTP client")?;

    match client.get(&url).send().await {
        Ok(resp) => Ok(resp.status().is_success()),
        Err(_) => Ok(false),
    }
}