wsb-daemon 0.248.6265

//! The plugins running on this machine, and the calls the daemon makes to
//! them.
//!
//! A plugin registers its services with the daemon (see `control.rs`) and
//! is called when a paired machine asks for one of them over the peer
//! bridge (see `peer/connection.rs`). What may register is decided by the
//! owner's list of enabled plugins, read at registration; how long a
//! registration stands is decided by the plugin's process, which is checked
//! at every call.

use std::collections::BTreeMap;
use std::path::PathBuf;
use std::sync::Mutex;
use std::time::Duration;

use wsb::plugin::{Answer, Call, PluginService, Registration, MAX_BODY};

/// How long a plugin has to answer one call.
///
/// A plugin answers a question from its own state — a secret it holds, what
/// machine it is on — and needs nothing like this long. Bounded because the
/// peer that asked is waiting on the other end of a connection, and a plugin
/// that never answered would hold that connection, and the thread serving
/// it, for ever.
pub const CALL_BOUND: Duration = Duration::from_secs(30);

/// What the registry knows about one service.
pub enum Lookup {
    /// No running plugin has registered it.
    NotOffered,
    /// A plugin registered it, and its process has since gone.
    Gone,
    /// A running plugin answers it.
    Offered(Registration),
}

/// Why a call to a plugin did not come back with an answer.
#[derive(Debug)]
pub struct CallFailure(pub String);

/// Every plugin service registered on this machine.
pub struct PluginRegistry {
    machine_dir: PathBuf,
    registered: Mutex<BTreeMap<PluginService, Registration>>,
}

impl PluginRegistry {
    pub fn new(machine_dir: PathBuf) -> Self {
        Self {
            machine_dir,
            registered: Mutex::new(BTreeMap::new()),
        }
    }

    /// Take a plugin's registration, or refuse it with the reason.
    ///
    /// The plugin must be one the owner enabled, its process must be running,
    /// and the registration must pass its own checks (its own services, a
    /// loopback endpoint, a real token). A plugin that registers again — after
    /// a restart, with a new endpoint and token — replaces what it had; a
    /// service it no longer lists is no longer offered.
    pub fn register(&self, registration: Registration) -> Result<(), String> {
        registration.check()?;
        let enabled = wsb::plugin::enabled(&self.machine_dir)?;
        if !enabled.contains(&registration.plugin) {
            return Err(format!(
                "plugin {} is not enabled on this machine; its owner enables it with \
                 `wsb plugin add {}`",
                registration.plugin, registration.plugin
            ));
        }
        if !wsb::ops::process::is_alive(registration.pid) {
            return Err(format!(
                "plugin {} registered for process {}, which is not running",
                registration.plugin, registration.pid
            ));
        }
        let mut registered = self.registered.lock().map_err(|_| "the plugin registry is poisoned")?;
        registered.retain(|_, held| held.plugin != registration.plugin);
        for service in &registration.services {
            registered.insert(service.clone(), registration.clone());
        }
        Ok(())
    }

    /// The registrations of enabled plugins whose processes are running, one
    /// per plugin.
    pub fn running(&self) -> Result<Vec<Registration>, String> {
        let enabled = wsb::plugin::enabled(&self.machine_dir)?;
        let registered = self.registered.lock().map_err(|_| "the plugin registry is poisoned")?;
        let mut seen = BTreeMap::new();
        for registration in registered.values() {
            if enabled.contains(&registration.plugin) && wsb::ops::process::is_alive(registration.pid) {
                seen.insert(registration.plugin.clone(), registration.clone());
            }
        }
        Ok(seen.into_values().collect())
    }

    /// Whether, and by whom, a service is offered here now.
    ///
    /// Asked at each call rather than trusted from registration, on both
    /// counts. A plugin's process can end at any moment, and a registration
    /// that outlived it would send a peer's request to a port somebody else may
    /// now hold. And the owner can disable a plugin at any moment
    /// (`wsb plugin remove`), which must stop it answering then — not when it
    /// next happens to register.
    pub fn lookup(&self, service: &PluginService) -> Lookup {
        let enabled = match wsb::plugin::enabled(&self.machine_dir) {
            Ok(enabled) => enabled,
            Err(error) => {
                tracing::warn!(%error, "the enabled plugin list could not be read; answering for none");
                return Lookup::NotOffered;
            }
        };
        if !enabled.contains(&service.plugin()) {
            return Lookup::NotOffered;
        }
        let Ok(registered) = self.registered.lock() else {
            return Lookup::NotOffered;
        };
        match registered.get(service) {
            None => Lookup::NotOffered,
            Some(registration) if !wsb::ops::process::is_alive(registration.pid) => Lookup::Gone,
            Some(registration) => Lookup::Offered(registration.clone()),
        }
    }

    /// Hand one call to the plugin that offers its service, and bring back
    /// what it answered.
    ///
    /// Blocking, by design: it is made from the thread that serves one peer's
    /// connection, which is waiting for exactly this. Never from the
    /// runtime's threads.
    pub fn call(registration: &Registration, call: &Call) -> Result<Answer, CallFailure> {
        let client = reqwest::blocking::Client::builder()
            .connect_timeout(Duration::from_secs(5))
            .timeout(CALL_BOUND)
            // A plugin is on this machine. A redirect would be a plugin
            // sending the daemon, and the token, somewhere else.
            .redirect(reqwest::redirect::Policy::none())
            .no_proxy()
            .build()
            .map_err(|error| CallFailure(format!("cannot make a client for the plugin: {error}")))?;
        let response = client
            .post(&registration.endpoint)
            .bearer_auth(&registration.token)
            .json(call)
            .send()
            .map_err(|error| {
                CallFailure(format!(
                    "plugin {} did not answer at {}: {error}",
                    registration.plugin, registration.endpoint
                ))
            })?;
        let status = response.status();
        if !status.is_success() {
            return Err(CallFailure(format!(
                "plugin {} answered {status} at {}",
                registration.plugin, registration.endpoint
            )));
        }
        let text = response.text().map_err(|error| {
            CallFailure(format!("plugin {}'s answer could not be read: {error}", registration.plugin))
        })?;
        if text.len() > encoded_bound() + 1024 {
            return Err(CallFailure(format!(
                "plugin {} answered {} bytes, more than a plugin reply may carry",
                registration.plugin,
                text.len()
            )));
        }
        let answer: Answer = serde_json::from_str(&text).map_err(|error| {
            CallFailure(format!("plugin {}'s answer is not one: {error}", registration.plugin))
        })?;
        if let Answer::Reply { body } = &answer {
            if body.len() > encoded_bound() {
                return Err(CallFailure(format!(
                    "plugin {} replied with more than a plugin reply may carry",
                    registration.plugin
                )));
            }
        }
        Ok(answer)
    }
}

/// The longest base64 a body of [`MAX_BODY`] bytes encodes to.
pub fn encoded_bound() -> usize {
    MAX_BODY.div_ceil(3) * 4
}