wsb-daemon 0.248.6265

//! The control API: the command line's half of this daemon.
//!
//! Mounted on the same axum server that serves the workbench, outside the
//! permissive CORS layer and behind [`not_a_browser`]. These routes open
//! worktrees and stop the process; a page in the operator's browser must not
//! be able to drive them.
//!
//! # The shape of it
//!
//! * `GET /api/daemon/info` — open. A command line must be able to ask whether
//!   a daemon is here and which wsb it is before anything else. None of that
//!   is a secret.
//! * `POST /api/daemon/resolve` — a directory, in; the worktree it belongs
//!   to, out.
//! * `POST /api/daemon/shutdown` — stop the daemon, with the machine's control
//!   secret.
//! * `GET /api/daemon/plugins`, `POST /api/daemon/plugins/register`,
//!   `POST /api/daemon/plugins/send` — the plugins running on this machine,
//!   and their use of the peer bridge. All with the control secret: see
//!   [`wsb::plugin`].

use axum::{
    body::Body,
    extract::State,
    http::{HeaderMap, Request, StatusCode},
    middleware::Next,
    response::Response,
    routing::{get, post},
    Json, Router,
};

use wsb::daemon::protocol::{ApiError, Info};

use crate::machine::MachineState;
use std::sync::Arc;

/// Every control route, ready to be nested under `/`.
///
/// Returned unlayered by CORS on purpose: the caller mounts it after the
/// permissive layer so it does not inherit one. See `web::create_router`.
pub fn routes() -> Router<Arc<MachineState>> {
    Router::new()
        .route("/api/daemon/info", get(info))
        .route("/api/daemon/resolve", post(resolve))
        .route("/api/daemon/shutdown", post(shutdown))
        .route(wsb::plugin::route::LIST, get(plugins))
        .route(wsb::plugin::route::REGISTER, post(register_plugin))
        .route(wsb::plugin::route::SEND, post(send_for_plugin))
        .layer(axum::middleware::from_fn(not_a_browser))
}

type Refusal = (StatusCode, Json<ApiError>);

fn refuse(status: StatusCode, message: impl Into<String>) -> Refusal {
    (
        status,
        Json(ApiError {
            error: message.into(),
        }),
    )
}

/// Reject anything a browser sent.
///
/// This API belongs to the command line. A browser announces itself: it sets
/// `Sec-Fetch-Site` on every request and `Origin` on every cross-origin one,
/// and page script cannot suppress either. A request carrying neither is not
/// from a page.
pub async fn not_a_browser(request: Request<Body>, next: Next<Body>) -> Result<Response, Refusal> {
    let headers = request.headers();
    if let Some(site) = headers.get("sec-fetch-site").and_then(|v| v.to_str().ok()) {
        return Err(refuse(
            StatusCode::FORBIDDEN,
            format!(
                "the daemon's control API is for the wsb command line, not for a page \
                 (Sec-Fetch-Site: {site}). The workbench has its own routes under /api/v1."
            ),
        ));
    }
    if let Some(origin) = headers.get("origin").and_then(|v| v.to_str().ok()) {
        return Err(refuse(
            StatusCode::FORBIDDEN,
            format!(
                "the daemon's control API is for the wsb command line, not for {origin}"
            ),
        ));
    }
    Ok(next.run(request).await)
}

/// One open worktree, as the control API describes it.
pub(crate) fn worktree_info(runtime: &crate::machine::WorktreeRuntime) -> wsb::daemon::WorktreeInfo {
    let registered = &runtime.registered;
    wsb::daemon::WorktreeInfo {
        id: registered.id.to_string(),
        project: registered.project.clone(),
        name: registered.name.clone(),
        root: registered.super_root.display().to_string(),
        workspace: registered.workspace.as_ref().map(ToString::to_string),
        no_continuity: runtime.local.as_ref().err().cloned(),
    }
}

/// What this daemon is and what it holds.
async fn info(State(machine): State<Arc<MachineState>>) -> Json<Info> {
    let worktrees = machine.all().await.iter().map(|runtime| worktree_info(runtime)).collect();
    Json(Info {
        version: wsb::get_version().to_string(),
        pid: std::process::id(),
        http_address: machine.listen,
        http_port: machine.port,
        machine_id: machine.identity.id().to_string(),
        machine_name: machine.identity.name().to_string(),
        peer_address: machine.peers.peer_address(),
        peer_port: machine.peers.peer_port(),
        worktrees,
    })
}

/// A directory, in; the worktree it belongs to, out.
///
/// Opens it if this daemon has not. That is not a side effect to be
/// surprised by: a command run in a worktree IS the event that makes it one
/// this machine works with, and requiring a separate "add" would mean every
/// first command in a checkout failed once.
async fn resolve(
    State(machine): State<Arc<MachineState>>,
    Json(body): Json<wsb::daemon::protocol::ResolveRequest>,
) -> Result<Json<wsb::daemon::WorktreeInfo>, Refusal> {
    let cwd = std::path::PathBuf::from(&body.cwd);
    let runtime = machine
        .resolve(&cwd)
        .await
        .map_err(|e| refuse(StatusCode::NOT_FOUND, e))?;
    // A worktree's watchers start when it first becomes live, not at daemon
    // startup: the daemon may have been started before this worktree existed.
    crate::web::spawn_worktree_services_once(&runtime);
    Ok(Json(worktree_info(&runtime)))
}

/// Stop the daemon.
///
/// Answers first, exits after. A caller that gets `Ok` knows the request was
/// accepted; it does not know the process is already gone, and does not need
/// to — the next thing it does is start one, and that waits for a daemon that
/// answers.
///
/// Stopping the daemon is a MACHINE act, so what it has to establish is that
/// the caller is the account this daemon runs as. That account can already
/// send the process a signal. What it must NOT be is a request from somewhere
/// else — the workbench answers every interface when it is started with
/// `--listen`, and a passer-by should not be able to stop a machine's daemon.
/// The daemon mints a control secret at startup and publishes it in its
/// rendezvous file, which is written 0600: reading it is the proof.
async fn shutdown(
    State(machine): State<Arc<MachineState>>,
    headers: HeaderMap,
) -> Result<Json<serde_json::Value>, Refusal> {
    if !presents_control_secret(&machine, &headers) {
        return Err(refuse(
            StatusCode::UNAUTHORIZED,
            "stopping the daemon needs proof that you are this machine's operator: \
             the control secret from its rendezvous file",
        ));
    }
    tokio::spawn(async {
        // Long enough for this response to reach the socket, short enough that
        // nobody is waiting on it. The rendezvous file goes first: a caller
        // that finds it after the port stops answering treats it as stale and
        // starts a daemon, which is right but slower than not finding it.
        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
        crate::remove_rendezvous();
        tracing::info!("stopping at an operator's request");
        std::process::exit(0);
    });
    Ok(Json(serde_json::json!({ "stopping": true })))
}

/// Refuse a caller that did not present the control secret.
fn operator_only(machine: &MachineState, headers: &HeaderMap, act: &str) -> Result<(), Refusal> {
    if presents_control_secret(machine, headers) {
        Ok(())
    } else {
        Err(refuse(
            StatusCode::UNAUTHORIZED,
            format!(
                "{act} needs proof that you are this machine's operator: the control \
                 secret from its rendezvous file"
            ),
        ))
    }
}

/// The plugins running on this machine, and what they offer.
///
/// Endpoints and processes, never tokens: a token is what proves a call came
/// from the daemon, and it is shown to nobody.
async fn plugins(
    State(machine): State<Arc<MachineState>>,
    headers: HeaderMap,
) -> Result<Json<serde_json::Value>, Refusal> {
    operator_only(&machine, &headers, "listing plugins")?;
    let enabled = wsb::plugin::enabled(&machine.machine_dir)
        .map_err(|e| refuse(StatusCode::INTERNAL_SERVER_ERROR, e))?;
    let running: Vec<serde_json::Value> = machine
        .plugins
        .running()
        .map_err(|e| refuse(StatusCode::INTERNAL_SERVER_ERROR, e))?
        .into_iter()
        .map(|registration| {
            serde_json::json!({
                "plugin": registration.plugin,
                "services": registration.services,
                "endpoint": registration.endpoint,
                "pid": registration.pid,
            })
        })
        .collect();
    Ok(Json(serde_json::json!({ "enabled": enabled, "running": running })))
}

/// A running plugin, registering the services it answers.
async fn register_plugin(
    State(machine): State<Arc<MachineState>>,
    headers: HeaderMap,
    Json(registration): Json<wsb::plugin::Registration>,
) -> Result<Json<serde_json::Value>, Refusal> {
    operator_only(&machine, &headers, "registering a plugin")?;
    let plugin = registration.plugin.clone();
    let services = registration.services.clone();
    machine
        .plugins
        .register(registration)
        .map_err(|e| refuse(StatusCode::FORBIDDEN, e))?;
    tracing::info!(%plugin, services = ?services, "a plugin registered");
    Ok(Json(serde_json::json!({ "registered": services })))
}

/// A plugin here, asking the same plugin on a paired machine.
///
/// Refused before anything is sent when this machine's owner has not
/// enabled the plugin, or has not switched the service on with that peer:
/// what a machine asks its peers for is decided here, as what it answers
/// them is decided there. A machine that cannot be reached is an error, not
/// a refusal — the peer said nothing.
async fn send_for_plugin(
    State(machine): State<Arc<MachineState>>,
    headers: HeaderMap,
    Json(request): Json<wsb::plugin::SendRequest>,
) -> Result<Json<wsb::plugin::Sent>, Refusal> {
    operator_only(&machine, &headers, "sending for a plugin")?;
    if request.body.len() > crate::plugin::encoded_bound() {
        return Err(refuse(
            StatusCode::PAYLOAD_TOO_LARGE,
            format!("a plugin request may carry at most {} bytes", wsb::plugin::MAX_BODY),
        ));
    }
    let plugin = request.service.plugin();
    let enabled = wsb::plugin::enabled(&machine.machine_dir)
        .map_err(|e| refuse(StatusCode::INTERNAL_SERVER_ERROR, e))?;
    if !enabled.contains(&plugin) {
        return Err(refuse(
            StatusCode::FORBIDDEN,
            format!("plugin {plugin} is not enabled on this machine; `wsb plugin add {plugin}`"),
        ));
    }
    let peer = wsb::peer::model::PeerId::parse(&request.peer)
        .map_err(|e| refuse(StatusCode::BAD_REQUEST, e))?;
    let service = wsb::peer::model::Service::Plugin(request.service.clone());
    let switch = machine
        .store
        .service(&peer, service)
        .map_err(|e| refuse(StatusCode::INTERNAL_SERVER_ERROR, e))?;
    if !switch.is_on() {
        return Ok(Json(wsb::plugin::Sent::Refused {
            code: wsb::peer::protocol::RefusalCode::ServiceOff.as_str().to_string(),
            detail: format!(
                "this machine does not ask {peer} for {}: `wsb peer service {peer} {} on`",
                request.service, request.service
            ),
        }));
    }

    let peers = machine.peers.clone();
    let wsb::plugin::SendRequest { service, body, .. } = request;
    let answer = tokio::task::spawn_blocking(move || {
        let mut session = peers.connect(&peer)?;
        session.set_timeout(crate::plugin::CALL_BOUND + std::time::Duration::from_secs(15))?;
        let answer = session.exchange::<_, wsb::peer::protocol::Response>(
            &wsb::peer::protocol::Request::Plugin { service, body },
        );
        session.shutdown();
        answer
    })
    .await
    .map_err(|e| refuse(StatusCode::INTERNAL_SERVER_ERROR, format!("the send did not finish: {e}")))?
    .map_err(|e| refuse(StatusCode::SERVICE_UNAVAILABLE, e))?;

    match answer {
        wsb::peer::protocol::Response::Plugin { body } => Ok(Json(wsb::plugin::Sent::Reply { body })),
        wsb::peer::protocol::Response::Refused { code, detail } => Ok(Json(wsb::plugin::Sent::Refused {
            code: code.as_str().to_string(),
            detail,
        })),
        other => Err(refuse(
            StatusCode::BAD_GATEWAY,
            format!("the peer answered a plugin request with {other:?}"),
        )),
    }
}

/// Whether the caller presented this daemon's control secret.
///
/// Compared in constant time. The comparison is against a value an attacker
/// is trying to guess, and an early return on the first wrong byte is a
/// measurement of how much of it they have right.
fn presents_control_secret(machine: &MachineState, headers: &HeaderMap) -> bool {
    let presented = headers
        .get(wsb::daemon::protocol::route::CONTROL_HEADER)
        .and_then(|value| value.to_str().ok())
        .map(str::trim)
        .unwrap_or_default();
    let expected = machine.control_token.as_str();
    if presented.is_empty() || expected.is_empty() {
        return false;
    }
    if presented.len() != expected.len() {
        return false;
    }
    presented
        .bytes()
        .zip(expected.bytes())
        .fold(0u8, |differences, (a, b)| differences | (a ^ b))
        == 0
}

#[cfg(test)]
#[path = "control_tests.rs"]
mod tests;