archy/core/archipelago/src/node_message.rs

//! Node-to-node messaging over Tor.
//! Sends messages to peer .onion addresses via SOCKS5 proxy.

use anyhow::{Context, Result};
use serde::{Deserialize, Serialize};
use std::sync::{Mutex, OnceLock};

const TOR_SOCKS: &str = "socks5h://127.0.0.1:9050";

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct IncomingMessage {
    pub from_pubkey: String,
    pub from_onion: Option<String>,
    pub message: String,
    pub timestamp: String,
}

fn received_messages() -> &'static Mutex<Vec<IncomingMessage>> {
    static RECEIVED: OnceLock<Mutex<Vec<IncomingMessage>>> = OnceLock::new();
    RECEIVED.get_or_init(|| Mutex::new(Vec::new()))
}

const MAX_STORED: usize = 100;

/// Store a received message (called from HTTP handler).
pub fn store_received_sync(from_pubkey: &str, message: &str) {
    let mut guard = received_messages().lock().unwrap_or_else(|e| e.into_inner());
    guard.push(IncomingMessage {
        from_pubkey: from_pubkey.to_string(),
        from_onion: None,
        message: message.to_string(),
        timestamp: chrono::Utc::now().to_rfc3339(),
    });
    let len = guard.len();
    if len > MAX_STORED {
        guard.drain(0..len - MAX_STORED);
    }
}

pub async fn store_received(from_pubkey: &str, message: &str) {
    store_received_sync(from_pubkey, message);
}

/// Get received messages for UI display.
pub fn get_received() -> Vec<IncomingMessage> {
    received_messages().lock().unwrap_or_else(|e| e.into_inner()).clone()
}

/// 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() || (c >= '2' && c <= '7'));
    if !valid {
        anyhow::bail!("Invalid onion address: must be 56 base32 chars (a-z, 2-7)");
    }
    Ok(())
}

/// Send a message to a peer over Tor.
pub async fn send_to_peer(onion: &str, from_pubkey: &str, message: &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);
    let body = serde_json::json!({
        "from_pubkey": from_pubkey,
        "message": message,
        "timestamp": chrono::Utc::now().to_rfc3339(),
    });

    let proxy = reqwest::Proxy::all(TOR_SOCKS).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 the Tor container 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> {
    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(TOR_SOCKS).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),
    }
}