launcher/src/web/mod.rs

//! `sm121 web` — axum + SSE dashboard.
//!
//! Read-only LED board (state snapshots every 30 s + on job finish),
//! actions — serve/stop/power run as tracked jobs calling the same
//! handlers the CLI uses — and live log windows (SSH `docker logs -f`
//! attaches with ring-buffer history). Data endpoints are token-gated
//! when `--token` is set (required when binding off-loopback).
//!
//! Design: `specs/tech-architecture/web-dashboard.md`.

mod api;
pub mod events;
pub mod jobs;
pub mod logs;
pub mod probe;
pub mod ssh_stream;
pub mod topo;

use crate::config::Config;
use crate::web::events::{Event, build_snapshot};
use axum::{
    Router,
    extract::State,
    http::{HeaderMap, Request, StatusCode},
    middleware::{self, Next},
    response::{
        Html, Json, Response,
        sse::{Event as SseEvent, KeepAlive, Sse},
    },
    routing::{get, post},
};
use std::collections::HashMap;
use std::net::SocketAddr;
use std::sync::{Arc, Mutex};
use tokio::sync::broadcast;
use tokio::time::Duration;

const UI_HTML: &str = include_str!("ui.html");
const POLL_SECS: u64 = 30;
const BUS_CAP: usize = 1024;
/// Dashboard session lifetime (the `sm121_session` cookie issued by
/// POST /api/login); sliding — refreshed on every authenticated use.
const SESSION_TTL_SECS: i64 = 24 * 3600;

#[derive(Clone)]
pub struct AppState {
    pub config: Arc<Config>,
    pub bus: broadcast::Sender<Event>,
    pub token: Option<String>,
    /// Issued dashboard sessions (id → expiry unix secs); the cookie
    /// covers the clients that cannot send headers (SSE, the recipes
    /// iframe) so the raw token never rides in a URL.
    pub sessions: Arc<Mutex<HashMap<String, i64>>>,
    /// Latest snapshot (written by the supervisor; `/api/state` serves it).
    pub last: Arc<Mutex<Option<events::Snapshot>>>,
    /// Tracked action jobs (serve/stop/power); capped at `jobs::MAX_JOBS`.
    pub jobs: Arc<Mutex<Vec<jobs::Job>>>,
    /// Tracked log streams; capped at `logs::MAX_ATTACHES`.
    pub log_attaches: Arc<Mutex<Vec<logs::LogAttach>>>,
}

/// Probe the fleet and broadcast a fresh state snapshot (supervisor tick
/// and job-finish both route through here).
pub(crate) async fn push_state(app: &AppState) {
    let fleet = probe::probe(&app.config).await;
    let jobs = app.jobs.lock().unwrap().clone();
    let attaches: Vec<events::AttachSnap> = app
        .log_attaches
        .lock()
        .unwrap()
        .iter()
        .map(|a| events::AttachSnap {
            id: a.id.clone(),
            cluster: a.cluster.clone(),
            node: a.node.clone(),
            recipe: a.recipe.clone(),
            rank: a.rank,
            done: a.done.load(std::sync::atomic::Ordering::Relaxed),
        })
        .collect();
    let snap = build_snapshot(&app.config, &fleet, &jobs, &attaches, &fleet.ha);
    let _ = app.bus.send(Event::State {
        ts: snap.ts,
        snapshot: snap.clone(),
    });
    *lock_last(app) = Some(snap);
}

/// `GET /events` — forward every bus event as an SSE message. A fresh
/// client gets the cached snapshot first, so it paints (and flips its
/// live indicator) at once instead of waiting for the next poll tick.
async fn events_stream(
    app: State<AppState>,
) -> Sse<impl futures::Stream<Item = Result<SseEvent, std::convert::Infallible>>> {
    let head = app
        .last
        .lock()
        .unwrap()
        .clone()
        .map(|snapshot| Event::State {
            ts: snapshot.ts,
            snapshot,
        });
    let rx = app.bus.subscribe();
    Sse::new(futures::stream::unfold(
        (rx, head),
        |(mut rx, mut head)| async move {
            if let Some(ev) = head.take() {
                return Some((sse_event(&ev), (rx, None)));
            }
            loop {
                match rx.recv().await {
                    Ok(ev) => return Some((sse_event(&ev), (rx, None))),
                    Err(broadcast::error::RecvError::Lagged(n)) => {
                        tracing::debug!("SSE client lagged {n} events");
                        // Same receiver is still valid — keep reading.
                    }
                    Err(broadcast::error::RecvError::Closed) => return None,
                }
            }
        },
    ))
    // comment lines between events: NAT/VPN/proxy idle timers (and frozen
    // tabs on reconnect) see traffic, so a dead stream is noticed in seconds,
    // not minutes
    .keep_alive(KeepAlive::new().interval(Duration::from_secs(15)))
}

fn sse_event(ev: &Event) -> Result<SseEvent, std::convert::Infallible> {
    Ok(SseEvent::default().data(serde_json::to_string(ev).unwrap_or_default()))
}

/// Auth gate for data endpoints (`/` and `/api/auth/mode` stay public —
/// static HTML / a mode probe carry no secrets). Accepts:
/// - `Authorization: Bearer <token>` — fetch/scripts (the token never
///   lands in a URL: no history, log or Referer leakage),
/// - the `sm121_session` cookie from POST /api/login — covers the clients
///   that cannot set headers (SSE `EventSource`, the recipes iframe).
///
/// The legacy `?token=` query form is gone (that was the leak vector).
async fn token_guard(
    State(app): State<AppState>,
    headers: HeaderMap,
    req: Request<axum::body::Body>,
    next: Next,
) -> Result<Response, (StatusCode, String)> {
    if app.token.is_some() {
        let bearer = headers
            .get("authorization")
            .and_then(|v| v.to_str().ok())
            .and_then(|v| v.strip_prefix("Bearer "));
        let cookie_ok = bearer.is_none()
            && headers
                .get("cookie")
                .and_then(|v| v.to_str().ok())
                .and_then(|c| {
                    c.split(';').find_map(|p| {
                        let p = p.trim();
                        p.strip_prefix("sm121_session=")
                    })
                })
                .is_some_and(|id| session_valid(&app, id));
        if bearer != Some(app.token.as_deref().unwrap_or("")) && !cookie_ok {
            return Err((StatusCode::UNAUTHORIZED, "unauthorized".into()));
        }
    }
    Ok(next.run(req).await)
}

/// Issue a fresh session id (32 random bytes, hex) with a sliding expiry.
fn new_session(app: &AppState) -> String {
    let mut buf = [0u8; 32];
    getrandom::fill(&mut buf).expect("getrandom");
    let id: String = buf.iter().map(|b| format!("{b:02x}")).collect();
    let exp = crate::stamp::now_secs() + SESSION_TTL_SECS;
    {
        let mut s = app
            .sessions
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        s.retain(|_, exp| *exp > crate::stamp::now_secs()); // drop the stale ones
        s.insert(id.clone(), exp);
    }
    id
}

/// Whether a session id is live; sliding — the expiry moves forward on use.
fn session_valid(app: &AppState, id: &str) -> bool {
    let now = crate::stamp::now_secs();
    let mut s = app
        .sessions
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner);
    match s.get(id) {
        Some(exp) if *exp > now => {
            s.insert(id.to_string(), now + SESSION_TTL_SECS);
            true
        }
        _ => {
            s.remove(id);
            false
        }
    }
}

/// Supervisor: poll the fleet and broadcast a state snapshot every tick.
async fn supervisor(app: AppState) {
    loop {
        // isolate each tick: a probe bug or a poisoned lock must not kill
        // the poll loop for good (a dead supervisor means a dead live feed)
        let tick = app.clone();
        if let Err(e) = tokio::spawn(async move { push_state(&tick).await }).await {
            tracing::warn!("sm121 web: supervisor tick panicked: {e}");
        }
        tokio::time::sleep(Duration::from_secs(POLL_SECS)).await;
    }
}

/// `AppState.last` guard that survives a poisoned mutex (a panic while
/// holding the lock would otherwise take the whole live feed down).
fn lock_last(app: &AppState) -> std::sync::MutexGuard<'_, Option<events::Snapshot>> {
    app.last
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner)
}

/// POST /api/login: exchange the Bearer token for a session cookie —
/// the cookie is what SSE and the recipes iframe (which cannot set
/// headers) ride on. 24 h sliding TTL; HttpOnly + SameSite=Strict.
async fn login(State(app): State<AppState>) -> Response {
    let id = new_session(&app);
    let mut res = axum::response::Response::builder()
        .status(StatusCode::NO_CONTENT)
        .header(
            "set-cookie",
            format!(
                "sm121_session={id}; Path=/; HttpOnly; SameSite=Strict; Max-Age={SESSION_TTL_SECS}"
            ),
        )
        .body(axum::body::Body::empty())
        .unwrap();
    *res.status_mut() = StatusCode::NO_CONTENT;
    res
}

/// GET /api/auth/mode (public): `open` (no token — loopback or --insecure)
/// or `token` — the UI decides whether to prompt.
async fn auth_mode(State(app): State<AppState>) -> Json<serde_json::Value> {
    Json(serde_json::json!({ "mode": if app.token.is_some() { "token" } else { "open" } }))
}

fn is_loopback(addr: SocketAddr) -> bool {
    addr.ip().is_loopback()
}

/// Build the router (separate from `run` so tests can drive it without
/// binding a port).
pub fn router(app: AppState) -> Router {
    let protected = Router::new()
        .route("/api/state", get(api::state))
        .route("/api/serve", post(api::serve))
        .route("/api/stop", post(api::stop))
        .route("/api/power", post(api::power))
        .route("/api/recipes", get(api::recipes))
        .route("/api/login", post(login))
        .route(
            "/api/logs",
            get(api::logs_history)
                .post(api::logs_attach)
                .delete(api::logs_detach),
        )
        .route("/api/raw/config", get(api::raw_config))
        .route("/api/raw/recipe/{name}", get(api::raw_recipe))
        .route("/api/raw/effective", get(api::raw_effective))
        .route("/events", get(events_stream))
        .layer(middleware::from_fn_with_state(app.clone(), token_guard));
    // Recipes panel: the very page `sm121 report --serve` serves, nested at
    // `/recipes` — its own SSE stream and `/r/<recipe>/<file>` viewer move
    // under the prefix (see `report::html_page_at`), live refresh included.
    let rows = crate::report::collect(&app.config, false);
    let recipes = crate::report::serve::router(
        app.config.clone(),
        crate::report::html_page_at(&rows, "/recipes"),
        serde_json::to_string(&rows).unwrap_or_default(),
        false,
        "/recipes",
    )
    .layer(middleware::from_fn_with_state(app.clone(), token_guard));
    Router::new()
        .route("/", get(|| async { Html(UI_HTML) }))
        .route("/api/auth/mode", get(auth_mode))
        .merge(protected)
        .with_state(app)
        .nest("/recipes", recipes)
}

/// Run the dashboard until the process exits.
pub async fn run(
    config: Config,
    port: u16,
    bind: String,
    token: Option<String>,
    open: bool,
    insecure: bool,
) -> anyhow::Result<()> {
    let addr: SocketAddr = format!("{bind}:{port}").parse()?;
    if !is_loopback(addr) && token.is_none() && !insecure {
        anyhow::bail!(
            "refusing to bind {bind} (non-loopback) without --token — the dashboard can act on the cluster (pass --insecure to accept the risk)"
        );
    }
    if insecure {
        tracing::warn!(
            "sm121 web: INSECURE MODE on {bind}:{port} — no authentication; anyone who can reach the port can serve/stop/power the cluster"
        );
    }
    let (tx, _) = broadcast::channel(BUS_CAP);
    let app = AppState {
        config: Arc::new(config),
        bus: tx,
        token: token.filter(|_| !insecure),
        sessions: Arc::new(Mutex::new(HashMap::new())),
        last: Arc::new(Mutex::new(None)),
        jobs: Arc::new(Mutex::new(Vec::new())),
        log_attaches: Arc::new(Mutex::new(Vec::new())),
    };
    jobs::load(&app); // history survives restarts; in-flight jobs -> failed

    tokio::spawn(supervisor(app.clone()));
    let listener = tokio::net::TcpListener::bind(addr).await?;
    if open {
        let url = format!("http://{addr}/");
        tokio::spawn(async move {
            tokio::time::sleep(Duration::from_millis(400)).await;
            let prog = if cfg!(target_os = "macos") {
                "open"
            } else {
                "xdg-open"
            };
            let _ = std::process::Command::new(prog).arg(&url).spawn();
        });
    }
    tracing::info!("sm121 web: http://{addr}/ (poll every {POLL_SECS}s)");
    axum::serve(listener, router(app)).await?;
    Ok(())
}

#[cfg(test)]
mod api_tests;
#[cfg(test)]
pub(crate) mod tests;
#[cfg(test)]
mod ui_js_tests;