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

mod content;
mod dwn;
mod node_message;
mod proxy;
mod remote_input;
mod remote_relay;
mod websocket;

use crate::api::rpc::RpcHandler;
use crate::config::Config;
use crate::monitoring::MetricsStore;
use crate::session::{self, SessionStore};
use crate::state::StateManager;
use anyhow::Result;
use hyper::{Method, Request, Response, StatusCode};
use std::sync::Arc;
use tokio::sync::broadcast;
use tracing::debug;

/// Build an HTTP response without unwrap. Falls back to a plain 500 if builder fails.
// Used by handler submodules after unwrap elimination
#[allow(dead_code)]
pub(super) fn build_response(status: StatusCode, content_type: &str, body: hyper::Body) -> Response<hyper::Body> {
    Response::builder()
        .status(status)
        .header("Content-Type", content_type)
        .body(body)
        .unwrap_or_else(|_| Response::new(hyper::Body::from("Internal error")))
}

pub struct ApiHandler {
    config: Config,
    rpc_handler: Arc<RpcHandler>,
    state_manager: Arc<StateManager>,
    metrics_store: Arc<MetricsStore>,
    session_store: SessionStore,
    /// Broadcast channel for relaying companion app input to remote browsers.
    input_relay_tx: broadcast::Sender<String>,
}

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

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

    /// Access the RPC handler (for service initialization after construction).
    pub fn rpc_handler(&self) -> &Arc<RpcHandler> {
        &self.rpc_handler
    }

    /// 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()
    }

    /// Allowed CORS origins derived from the config host IP.
    fn allowed_origins(&self) -> Vec<String> {
        let mut origins = vec![
            format!("http://{}", self.config.host_ip),
            format!("https://{}", self.config.host_ip),
        ];
        if self.config.dev_mode {
            origins.push("http://localhost:8100".to_string()); // Vite dev server
        }
        origins
    }

    /// Validate the Origin header against allowed origins.
    /// Returns the matched origin if valid, None if cross-origin is not allowed.
    fn validate_origin(&self, headers: &hyper::HeaderMap) -> Option<String> {
        let origin = headers
            .get("origin")
            .and_then(|v| v.to_str().ok())?;
        let allowed = self.allowed_origins();
        if allowed.iter().any(|a| a == origin) {
            Some(origin.to_string())
        } else {
            None
        }
    }

    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 mut builder = Response::builder()
                .status(StatusCode::NO_CONTENT)
                .header("Vary", "Origin");
            if let Some(origin) = self.validate_origin(req.headers()) {
                builder = builder
                    .header("Access-Control-Allow-Origin", &origin)
                    .header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
                    .header("Access-Control-Allow-Headers", "Content-Type, X-CSRF-Token")
                    .header("Access-Control-Allow-Credentials", "true");
            }
            return Ok(builder.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 {
                tracing::warn!("401 WebSocket /ws/db — session invalid or missing");
                return Ok(Self::unauthorized());
            }
            return Self::handle_websocket(req, self.state_manager.clone(), self.metrics_store.clone()).await;
        }

        // Remote input WebSocket — companion app sends keyboard/mouse events
        if method == Method::GET && path == "/ws/remote-input" {
            if !self.is_authenticated(req.headers()).await {
                tracing::warn!("401 WebSocket /ws/remote-input — session invalid or missing");
                return Ok(Self::unauthorized());
            }
            return Self::handle_remote_input(req, self.input_relay_tx.clone()).await;
        }

        // Remote relay WebSocket — browser receives companion input events
        if method == Method::GET && path == "/ws/remote-relay" {
            if !self.is_authenticated(req.headers()).await {
                tracing::warn!("401 WebSocket /ws/remote-relay — session invalid or missing");
                return Ok(Self::unauthorized());
            }
            return Self::handle_remote_relay(req, self.input_relay_tx.subscribe()).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, returns JSON with service status
            (Method::GET, "/health") => {
                let recovery_complete = crate::crash_recovery::is_recovery_complete();
                let uptime = crate::crash_recovery::uptime_seconds();
                let health_status = if recovery_complete { "ok" } else { "degraded" };
                let status = serde_json::json!({
                    "status": health_status,
                    "crash_recovery_complete": recovery_complete,
                    "uptime_seconds": uptime,
                    "version": env!("CARGO_PKG_VERSION"),
                    "services": {
                        "rpc": true,
                        "sessions": true,
                    }
                });
                Ok(Response::builder()
                    .status(StatusCode::OK)
                    .header("Content-Type", "application/json")
                    .body(hyper::Body::from(serde_json::to_vec(&status).unwrap_or_default()))
                    .unwrap())
            }

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

            // Content preview — degraded previews for paid content (no auth, no payment)
            (Method::GET, p) if p.starts_with("/content/") && p.ends_with("/preview") => {
                Self::handle_content_preview(p, &self.config).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,

            // LND connect info — nginx validates session cookie (presence check),
            // backend is bound to 127.0.0.1 so only nginx can reach it.
            // No backend auth check here because the LND UI iframe fetches this
            // endpoint and the session cookie flow is validated at the nginx layer.
            (Method::GET, "/lnd-connect-info") => {
                Self::handle_lnd_connect_info(self.rpc_handler.clone()).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());
                }
                let origin = self.validate_origin(&headers).unwrap_or_default();
                Self::handle_container_logs_http(self.rpc_handler.clone(), path, &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());
                }
                let origin = self.validate_origin(&headers).unwrap_or_default();
                Self::handle_lnd_proxy(path, &origin).await
            }

            // DWN health — unauthenticated
            (Method::GET, "/dwn/health") => {
                Self::handle_dwn_health(&self.config).await
            }

            // DWN message processing — peers access over Tor for sync (no session auth)
            (Method::POST, "/dwn") => {
                Self::handle_dwn_message(body_bytes, &self.config).await
            }

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

/// 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;")
}