archy/core/archipelago/src/api/handler.rs

use crate::api::rpc::RpcHandler;
use crate::content_server;
use crate::electrs_status;
use crate::node_message as node_msg;
use crate::config::Config;
use crate::session::{self, SessionStore};
use crate::state::StateManager;
use anyhow::Result;
use futures_util::{SinkExt, StreamExt};
use hyper::{Method, Request, Response, StatusCode};
use hyper_ws_listener::WsStream;
use std::sync::Arc;
use tokio::sync::broadcast;
use tokio_tungstenite::tungstenite::Message;
use tracing::{debug, info};

pub struct ApiHandler {
    config: Config,
    rpc_handler: Arc<RpcHandler>,
    state_manager: Arc<StateManager>,
    session_store: SessionStore,
}

impl ApiHandler {
    pub async fn new(config: Config, state_manager: Arc<StateManager>) -> Result<Self> {
        let session_store = SessionStore::new();
        let rpc_handler = Arc::new(
            RpcHandler::new(config.clone(), state_manager.clone(), session_store.clone()).await?,
        );

        Ok(Self {
            config,
            rpc_handler,
            state_manager,
            session_store,
        })
    }

    /// Check if the request has a valid session cookie.
    async fn is_authenticated(&self, headers: &hyper::HeaderMap) -> bool {
        match session::extract_session_cookie(headers) {
            Some(token) => self.session_store.validate(&token).await,
            None => false,
        }
    }

    /// Build a 401 Unauthorized JSON response.
    fn unauthorized() -> Response<hyper::Body> {
        let body = serde_json::json!({ "error": "Unauthorized" });
        let body_bytes = serde_json::to_vec(&body).unwrap_or_default();
        Response::builder()
            .status(StatusCode::UNAUTHORIZED)
            .header("Content-Type", "application/json")
            .body(hyper::Body::from(body_bytes))
            .unwrap()
    }

    /// Derive the allowed CORS origin from the config host IP.
    fn cors_origin(&self) -> String {
        format!("http://{}", self.config.host_ip)
    }

    pub async fn handle_request(
        &self,
        req: Request<hyper::Body>,
    ) -> Result<Response<hyper::Body>> {
        let path = req.uri().path().to_string();
        let method = req.method().clone();

        // Handle CORS preflight for all routes
        if method == Method::OPTIONS {
            let origin = self.cors_origin();
            return Ok(Response::builder()
                .status(StatusCode::NO_CONTENT)
                .header("Access-Control-Allow-Origin", &origin)
                .header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
                .header("Access-Control-Allow-Headers", "Content-Type")
                .header("Access-Control-Allow-Credentials", "true")
                .header("Vary", "Origin")
                .body(hyper::Body::empty())
                .unwrap());
        }

        // WebSocket upgrade — validate session before upgrading
        if method == Method::GET && path == "/ws/db" {
            if !self.is_authenticated(req.headers()).await {
                return Ok(Self::unauthorized());
            }
            return Self::handle_websocket(req, self.state_manager.clone()).await;
        }

        // Convert body to bytes for non-WS routes
        let headers = req.headers().clone();
        let (parts, body) = req.into_parts();
        let body_bytes = hyper::body::to_bytes(body).await
            .map_err(|e| anyhow::anyhow!("Failed to read body: {}", e))?;
        let req_with_bytes = Request::from_parts(parts, hyper::Body::from(body_bytes.clone()));

        debug!("{} {}", method, path);

        match (method, path.as_str()) {
            // RPC — auth is handled inside rpc handler per-method
            (Method::POST, "/rpc/v1") => self.rpc_handler.handle(req_with_bytes).await,

            // Health — unauthenticated
            (Method::GET, "/health") => Ok(Response::builder()
                .status(StatusCode::OK)
                .body(hyper::Body::from("OK"))
                .unwrap()),

            // Node message — P2P endpoint (authenticated by source validation, not cookie)
            (Method::POST, "/archipelago/node-message") => {
                Self::handle_node_message(body_bytes).await
            }

            // Content serving — peers access shared content over Tor (no session auth)
            (Method::GET, p) if p.starts_with("/content/") => {
                Self::handle_content_request(p, &headers, &self.config).await
            }

            // Content catalog — list available content (no session auth, for peers)
            (Method::GET, "/content") => {
                Self::handle_content_catalog(&self.config).await
            }

            // Electrs status — unauthenticated (read-only sync status)
            (Method::GET, "/electrs-status") => Self::handle_electrs_status().await,

            // Container logs — requires session
            (Method::GET, path) if path.starts_with("/api/container/logs") => {
                if !self.is_authenticated(&headers).await {
                    return Ok(Self::unauthorized());
                }
                Self::handle_container_logs_http(self.rpc_handler.clone(), path, &self.cors_origin()).await
            }

            // LND proxy — requires session
            (Method::GET, path) if path.starts_with("/proxy/lnd/") => {
                if !self.is_authenticated(&headers).await {
                    return Ok(Self::unauthorized());
                }
                Self::handle_lnd_proxy(path, &self.cors_origin()).await
            }

            _ => Ok(Response::builder()
                .status(StatusCode::NOT_FOUND)
                .body(hyper::Body::from("Not Found"))
                .unwrap()),
        }
    }

    async fn handle_container_logs_http(
        rpc: Arc<RpcHandler>,
        path: &str,
        cors_origin: &str,
    ) -> Result<Response<hyper::Body>> {
        let query = path
            .strip_prefix("/api/container/logs")
            .and_then(|s| s.strip_prefix('?'))
            .unwrap_or("");
        let params: std::collections::HashMap<String, String> =
            query
                .split('&')
                .filter_map(|p| {
                    let mut it = p.splitn(2, '=');
                    let k = it.next()?.to_string();
                    let v = it.next()?.to_string();
                    Some((k, v))
                })
                .collect();

        let app_id = params.get("app_id").map(|s| s.as_str()).unwrap_or("lnd");

        // Validate app_id format
        if !is_valid_app_id(app_id) {
            let body = serde_json::json!({ "error": "Invalid app_id" });
            let body_bytes = serde_json::to_vec(&body).unwrap_or_default();
            return Ok(Response::builder()
                .status(StatusCode::BAD_REQUEST)
                .header("Content-Type", "application/json")
                .body(hyper::Body::from(body_bytes))
                .unwrap());
        }

        let lines = params
            .get("lines")
            .and_then(|s| s.parse::<u32>().ok())
            .unwrap_or(200);

        match rpc.get_container_logs_value(app_id, lines).await {
            Ok(value) => {
                let body = serde_json::json!({ "result": value });
                let body_bytes = serde_json::to_vec(&body).unwrap_or_default();
                Ok(Response::builder()
                    .status(StatusCode::OK)
                    .header("Content-Type", "application/json")
                    .header("Access-Control-Allow-Origin", cors_origin)
                    .header("Access-Control-Allow-Credentials", "true")
                    .header("Vary", "Origin")
                    .body(hyper::Body::from(body_bytes))
                    .unwrap())
            }
            Err(e) => {
                let body = serde_json::json!({ "error": e.to_string() });
                let body_bytes = serde_json::to_vec(&body).unwrap_or_default();
                Ok(Response::builder()
                    .status(StatusCode::INTERNAL_SERVER_ERROR)
                    .header("Content-Type", "application/json")
                    .header("Access-Control-Allow-Origin", cors_origin)
                    .header("Access-Control-Allow-Credentials", "true")
                    .header("Vary", "Origin")
                    .body(hyper::Body::from(body_bytes))
                    .unwrap())
            }
        }
    }

    async fn handle_node_message(body: hyper::body::Bytes) -> Result<Response<hyper::Body>> {
        #[derive(serde::Deserialize)]
        struct Incoming {
            from_pubkey: Option<String>,
            message: Option<String>,
        }
        let incoming: Incoming = serde_json::from_slice(&body).unwrap_or(Incoming {
            from_pubkey: None,
            message: None,
        });
        if let (Some(from), Some(msg)) = (incoming.from_pubkey, incoming.message) {
            // Validate from_pubkey is a valid hex ed25519 pubkey
            if !is_valid_pubkey_hex(&from) {
                return Ok(Response::builder()
                    .status(StatusCode::BAD_REQUEST)
                    .header("Content-Type", "application/json")
                    .body(hyper::Body::from(r#"{"error":"Invalid pubkey format"}"#))
                    .unwrap());
            }
            // Sanitize log output to prevent log injection
            let safe_from = sanitize_log_string(&from);
            let safe_msg = sanitize_log_string(&msg);
            tracing::info!("Received message from {}: {}", safe_from, safe_msg);
            // Sanitize stored message content (strip HTML entities)
            let clean_from = sanitize_html(&from);
            let clean_msg = sanitize_html(&msg);
            node_msg::store_received(&clean_from, &clean_msg).await;
        }
        Ok(Response::builder()
            .status(StatusCode::OK)
            .header("Content-Type", "application/json")
            .body(hyper::Body::from(r#"{"ok":true}"#))
            .unwrap())
    }

    async fn handle_electrs_status() -> Result<Response<hyper::Body>> {
        let status = electrs_status::get_electrs_sync_status().await;
        let body = serde_json::to_vec(&status).unwrap_or_default();
        Ok(Response::builder()
            .status(StatusCode::OK)
            .header("Content-Type", "application/json")
            .body(hyper::Body::from(body))
            .unwrap())
    }

    async fn handle_lnd_proxy(path: &str, cors_origin: &str) -> Result<Response<hyper::Body>> {
        let suffix = path.strip_prefix("/proxy/lnd").unwrap_or("/");
        let url = format!("http://127.0.0.1:8080{}", suffix);
        match reqwest::get(&url).await {
            Ok(resp) => {
                let status = resp.status().as_u16();
                let headers = resp.headers().clone();
                let body = resp.bytes().await.unwrap_or_default();
                let mut builder = Response::builder().status(status);
                if let Some(ct) = headers.get("content-type") {
                    if let Ok(s) = ct.to_str() {
                        builder = builder.header("Content-Type", s);
                    }
                }
                builder
                    .header("Access-Control-Allow-Origin", cors_origin)
                    .header("Access-Control-Allow-Credentials", "true")
                    .header("Vary", "Origin")
                    .body(hyper::Body::from(body))
                    .map_err(|e| anyhow::anyhow!("response build: {}", e))
            }
            Err(e) => {
                let body = serde_json::json!({ "error": e.to_string() });
                let body_bytes = serde_json::to_vec(&body).unwrap_or_default();
                Ok(Response::builder()
                    .status(StatusCode::BAD_GATEWAY)
                    .header("Content-Type", "application/json")
                    .header("Access-Control-Allow-Origin", cors_origin)
                    .header("Access-Control-Allow-Credentials", "true")
                    .header("Vary", "Origin")
                    .body(hyper::Body::from(body_bytes))
                    .unwrap())
            }
        }
    }

    async fn handle_content_catalog(config: &Config) -> Result<Response<hyper::Body>> {
        match content_server::load_catalog(&config.data_dir).await {
            Ok(catalog) => {
                // Only expose public metadata, not file paths
                let items: Vec<serde_json::Value> = catalog
                    .items
                    .iter()
                    .map(|i| {
                        serde_json::json!({
                            "id": i.id,
                            "filename": i.filename,
                            "mime_type": i.mime_type,
                            "size_bytes": i.size_bytes,
                            "description": i.description,
                            "access": i.access,
                        })
                    })
                    .collect();
                let body = serde_json::to_vec(&serde_json::json!({ "items": items }))
                    .unwrap_or_default();
                Ok(Response::builder()
                    .status(StatusCode::OK)
                    .header("Content-Type", "application/json")
                    .body(hyper::Body::from(body))
                    .unwrap())
            }
            Err(e) => {
                let body = serde_json::json!({ "error": e.to_string() });
                let body_bytes = serde_json::to_vec(&body).unwrap_or_default();
                Ok(Response::builder()
                    .status(StatusCode::INTERNAL_SERVER_ERROR)
                    .header("Content-Type", "application/json")
                    .body(hyper::Body::from(body_bytes))
                    .unwrap())
            }
        }
    }

    async fn handle_content_request(
        path: &str,
        headers: &hyper::HeaderMap,
        config: &Config,
    ) -> Result<Response<hyper::Body>> {
        let content_id = path.strip_prefix("/content/").unwrap_or("");
        if content_id.is_empty() || !is_valid_app_id(content_id) {
            return Ok(Response::builder()
                .status(StatusCode::BAD_REQUEST)
                .body(hyper::Body::from("Invalid content ID"))
                .unwrap());
        }

        // Extract payment token from X-Payment-Token header
        let payment_token = headers
            .get("x-payment-token")
            .and_then(|v| v.to_str().ok())
            .map(|s| s.to_string());

        // Parse Range header for streaming support
        let range = headers
            .get("range")
            .and_then(|v| v.to_str().ok())
            .and_then(content_server::parse_range_header);

        match content_server::serve_content(
            &config.data_dir,
            content_id,
            payment_token.as_deref(),
            range,
        )
        .await
        {
            Ok(content_server::ServeResult::Ok(bytes, mime_type)) => {
                let len = bytes.len();
                Ok(Response::builder()
                    .status(StatusCode::OK)
                    .header("Content-Type", mime_type)
                    .header("Content-Length", len.to_string())
                    .header("Accept-Ranges", "bytes")
                    .body(hyper::Body::from(bytes))
                    .unwrap())
            }
            Ok(content_server::ServeResult::Partial {
                bytes,
                mime_type,
                start,
                end,
                total,
            }) => {
                Ok(Response::builder()
                    .status(StatusCode::PARTIAL_CONTENT)
                    .header("Content-Type", mime_type)
                    .header("Content-Length", bytes.len().to_string())
                    .header("Content-Range", format!("bytes {}-{}/{}", start, end, total))
                    .header("Accept-Ranges", "bytes")
                    .body(hyper::Body::from(bytes))
                    .unwrap())
            }
            Ok(content_server::ServeResult::PaymentRequired(price_sats)) => {
                let body = serde_json::json!({
                    "error": "Payment required",
                    "price_sats": price_sats,
                    "payment_header": "X-Payment-Token",
                });
                let body_bytes = serde_json::to_vec(&body).unwrap_or_default();
                Ok(Response::builder()
                    .status(StatusCode::PAYMENT_REQUIRED)
                    .header("Content-Type", "application/json")
                    .body(hyper::Body::from(body_bytes))
                    .unwrap())
            }
            Ok(content_server::ServeResult::NotFound) | Err(_) => {
                Ok(Response::builder()
                    .status(StatusCode::NOT_FOUND)
                    .body(hyper::Body::from("Content not found"))
                    .unwrap())
            }
        }
    }

    async fn handle_websocket(
        req: Request<hyper::Body>,
        state_manager: Arc<StateManager>,
    ) -> Result<Response<hyper::Body>> {
        let (response, ws_fut_opt) = hyper_ws_listener::create_ws(req)
            .map_err(|e| anyhow::anyhow!("WebSocket upgrade failed: {}", e))?;

        if let Some(ws_fut) = ws_fut_opt {
            tokio::spawn(async move {
                let ws_stream: WsStream = match ws_fut.await {
                    Ok(Ok(s)) => s,
                    Ok(Err(e)) => {
                        debug!("WebSocket handshake failed (hyper): {}", e);
                        return;
                    }
                    Err(e) => {
                        debug!("WebSocket task join failed: {}", e);
                        return;
                    }
                };
                info!("WebSocket /ws/db connected");

                let (mut tx, mut rx) = ws_stream.split();

                let initial_msg = state_manager.get_initial_message().await;
                if let Ok(json_msg) = serde_json::to_string(&initial_msg) {
                    if let Err(e) = tx.send(Message::Text(json_msg)).await {
                        debug!("Failed to send initial data: {}", e);
                        return;
                    }
                    debug!("Sent initial data dump at revision {}", initial_msg.rev);
                }

                let mut state_rx = state_manager.subscribe();
                let ping_interval = tokio::time::interval(tokio::time::Duration::from_secs(30));
                tokio::pin!(ping_interval);

                loop {
                    tokio::select! {
                        _ = ping_interval.tick() => {
                            if tx.send(Message::Ping(vec![])).await.is_err() {
                                debug!("Failed to send ping, connection likely closed");
                                break;
                            }
                        }
                        update = state_rx.recv() => {
                            match update {
                                Ok(msg) => {
                                    if let Ok(json_msg) = serde_json::to_string(&msg) {
                                        if let Err(e) = tx.send(Message::Text(json_msg)).await {
                                            debug!("Failed to send state update: {}", e);
                                            break;
                                        }
                                        debug!("Sent state update at revision {}", msg.rev);
                                    }
                                }
                                Err(broadcast::error::RecvError::Lagged(skipped)) => {
                                    debug!("Client lagged behind, skipped {} messages", skipped);
                                }
                                Err(broadcast::error::RecvError::Closed) => {
                                    debug!("Broadcast channel closed");
                                    break;
                                }
                            }
                        }
                        msg = rx.next() => {
                            match msg {
                                Some(Ok(Message::Close(_))) => break,
                                Some(Ok(Message::Pong(_))) => {
                                    debug!("Received pong");
                                }
                                Some(Ok(Message::Ping(data))) => {
                                    let _ = tx.send(Message::Pong(data)).await;
                                }
                                Some(Ok(_)) => {}
                                Some(Err(e)) => {
                                    debug!("WebSocket stream error: {}", e);
                                    break;
                                }
                                None => break,
                            }
                        }
                    }
                }
                info!("WebSocket /ws/db disconnected");
            });
        }

        Ok(response)
    }
}

/// Validate that an app ID matches the safe pattern: lowercase alphanumeric + hyphens.
fn is_valid_app_id(id: &str) -> bool {
    !id.is_empty()
        && id.len() <= 64
        && id.bytes().all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'-')
        && id.as_bytes()[0] != b'-'
}

/// Validate that a pubkey is a 64-char hex string.
fn is_valid_pubkey_hex(s: &str) -> bool {
    s.len() == 64 && s.bytes().all(|b| b.is_ascii_hexdigit())
}

/// Strip newlines and ANSI escape sequences from strings before logging.
fn sanitize_log_string(s: &str) -> String {
    s.replace('\n', "\\n")
        .replace('\r', "\\r")
        .replace('\x1b', "")
}

/// Strip HTML-sensitive characters to prevent XSS when stored/rendered.
fn sanitize_html(s: &str) -> String {
    s.replace('&', "&amp;")
        .replace('<', "&lt;")
        .replace('>', "&gt;")
        .replace('"', "&quot;")
        .replace('\'', "&#x27;")
}