implement plugin command execution

- change .proto file for plugins
- add global usage guide for all commands
- handle commands not included in the built-in set

Cargo.lock

@@ -653,11 +653,14 @@ name = "bot"
 version = "0.1.0"
 dependencies = [
  "anyhow",
+ "async-trait",
  "clap",
  "deltachat",
  "env_logger",
  "eui48",
  "log",
+ "prost",
+ "prost-build",
  "russh",
  "serde",
  "serde_yaml",
@@ -1866,6 +1869,12 @@ dependencies = [
  "zeroize",
 ]
 
+[[package]]
+name = "either"
+version = "1.15.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "48c757948c5ede0e46177b7add2e67155f70e33c07fea8284df6576da70b3719"
+
 [[package]]
 name = "elliptic-curve"
 version = "0.13.8"
@@ -3390,6 +3399,15 @@ version = "1.70.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "a6cb138bb79a146c1bd460005623e142ef0181e3d0219cb493e02f7d08a35695"
 
+[[package]]
+name = "itertools"
+version = "0.14.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2b192c782037fadd9cfa75548310488aabdbf3d2da73885b31bd0abd03351285"
+dependencies = [
+ "either",
+]
+
 [[package]]
 name = "itoa"
 version = "1.0.17"
@@ -3748,6 +3766,12 @@ dependencies = [
  "pxfm",
 ]
 
+[[package]]
+name = "multimap"
+version = "0.10.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1d87ecb2933e8aeadb3e3a02b828fed80a7528047e68b4f424523a0981a3a084"
+
 [[package]]
 name = "mutate_once"
 version = "0.1.2"
@@ -4438,6 +4462,17 @@ dependencies = [
  "sha2 0.10.9",
 ]
 
+[[package]]
+name = "petgraph"
+version = "0.8.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8701b58ea97060d5e5b155d383a69952a60943f0e6dfe30b04c287beb0b27455"
+dependencies = [
+ "fixedbitset",
+ "hashbrown 0.15.5",
+ "indexmap",
+]
+
 [[package]]
 name = "pgp"
 version = "0.19.0"
@@ -4928,6 +4963,57 @@ dependencies = [
  "unicode-ident",
 ]
 
+[[package]]
+name = "prost"
+version = "0.14.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d2ea70524a2f82d518bce41317d0fae74151505651af45faf1ffbd6fd33f0568"
+dependencies = [
+ "bytes",
+ "prost-derive",
+]
+
+[[package]]
+name = "prost-build"
+version = "0.14.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "343d3bd7056eda839b03204e68deff7d1b13aba7af2b2fd16890697274262ee7"
+dependencies = [
+ "heck",
+ "itertools",
+ "log",
+ "multimap",
+ "petgraph",
+ "prettyplease",
+ "prost",
+ "prost-types",
+ "regex",
+ "syn",
+ "tempfile",
+]
+
+[[package]]
+name = "prost-derive"
+version = "0.14.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "27c6023962132f4b30eb4c172c91ce92d933da334c59c23cddee82358ddafb0b"
+dependencies = [
+ "anyhow",
+ "itertools",
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
+[[package]]
+name = "prost-types"
+version = "0.14.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8991c4cbdb8bc5b11f0b074ffe286c30e523de90fee5ba8132f1399f23cb3dd7"
+dependencies = [
+ "prost",
+]
+
 [[package]]
 name = "pxfm"
 version = "0.1.28"

Cargo.toml

@@ -5,12 +5,17 @@ edition = "2024"
 
 [dependencies]
 anyhow = "1.0.102"
+async-trait = "0.1.89"
 clap = { version = "4", features = [ "derive" ] }
 deltachat = { path = "./chatmail-core" }
 env_logger = "0.11.9"
 eui48 = { version = "1.1.0", features = [ "serde" ] }
 log = { version = "0.4.29" }
+prost = "0.14.3"
 russh = { version = "0.60.0" }
 serde = { version = "1", features = [ "derive" ] }
 serde_yaml = { version = "0.9" }
 tokio = { version = "1.50.0", features = ["full"] }
+
+[build-dependencies]
+prost-build = "0.14.3"

bot.example.yml

@@ -16,3 +16,7 @@ deltaChat:
   email: "0000000000000000@email.com"
   password: "kek-a-kek"
   avatar: "avatar.png"
+
+plugins:
+  - name: additional-commands-plus
+    enabled: true

build.rs

@@ -0,0 +1,5 @@
+use std::io::Result;
+
+fn main() -> Result<()> {
+    prost_build::compile_protos(&["protobuf/plugin.proto"], &["protobuf/"])
+}

protobuf/plugin.proto

@@ -0,0 +1,58 @@
+syntax = "proto3";
+package deltachat_remotecontrol_bot.plugin;
+
+message Request {
+  uint32 request_id = 1;
+  oneof req {
+    PluginInitializeRequest initialize_req = 10;
+    PluginCommandListRequest command_list_req = 11;
+    PluginExecuteRequest execute_req = 12;
+  }
+}
+
+message PluginInitializeRequest {
+  string config = 1;
+}
+
+message PluginCommandListRequest {}
+
+message PluginExecuteRequest {
+  string command_id = 1;
+  string issuer_id = 2;
+  repeated string arg_vector = 5;
+}
+
+message Response {
+  uint32 request_id = 1;
+  oneof res {
+    PluginInitializeResponse initialize_res = 10;
+    PluginCommandListResponse command_list_res = 11;
+    CommandReply cmd_reply = 12;
+  }
+}
+
+message PluginInitializeResponse {
+  string unique_name = 1;
+  string name = 2;
+  string version = 3;
+  string authors = 4;
+}
+
+message PluginCommandListResponse {
+  repeated CommandSpec commands = 1;
+}
+
+message CommandSpec {
+  string name = 1;
+  repeated string aliases = 2;
+  string usage = 3;
+  string description = 4;
+}
+
+message CommandReply {
+  bool edit = 1;
+  oneof reply {
+    string text = 2;
+    bool end = 8;
+  }
+}

src/commands/args.rs

@@ -2,9 +2,9 @@ pub(super) fn append_port_if_needed(input: &str, default_port: u16) -> String {
     match input.rsplit_once(':') {
         Some((_, port)) => match port.parse::<u16>() {
             Ok(_) => input.to_owned(),
-            Err(_) => format!("{}:{}", input.to_owned(), default_port.to_string()),
+            Err(_) => format!("{}:{}", input.to_owned(), default_port),
         },
-        None => format!("{}:{}", input.to_owned(), default_port.to_string()),
+        None => format!("{}:{}", input.to_owned(), default_port),
     }
 }
 

src/commands/mod.rs

@@ -1,6 +1,6 @@
 pub(crate) mod args;
 
-use std::{fmt::Debug, fs::File, io::Read, ops::Add, path::PathBuf, sync::Arc, time::Duration};
+use std::{fs::File, io::Read, ops::Add, path::PathBuf, sync::Arc, time::Duration};
 
 use crate::{commands::args::append_port_if_needed, ssh};
 use anyhow::{Context as _, Result as AnyhowResult};
@@ -13,15 +13,15 @@ use deltachat::{
 };
 use eui48::MacAddress;
 use russh::{
-    client::{AuthResult, Handle},
+    client::AuthResult,
     keys::PrivateKeyWithHashAlg,
 };
 use tokio::{
     sync::Mutex,
-    time::{Instant, error::Elapsed, timeout, timeout_at},
+    time::{Instant, timeout, timeout_at},
 };
 
-use crate::{AUTH_REQUIRED, BotContext, config::BotConfig, data_path, ssh::ClientHandler};
+use crate::{AUTH_REQUIRED, BotContext, config::BotConfig, data_path};
 
 pub async fn echo(dchat_ctx: Arc<Mutex<Context>>, msg: Message) -> AnyhowResult<()> {
     let chat_id = msg.get_chat_id();
@@ -30,6 +30,31 @@ pub async fn echo(dchat_ctx: Arc<Mutex<Context>>, msg: Message) -> AnyhowResult<
     Ok(())
 }
 
+pub async fn global_usage(ctx: Arc<Mutex<BotContext>>) -> String {
+    let ctx_lock = ctx.lock().await;
+    let config = &ctx_lock.config;
+    let mut usage = String::from("DeltaChat Remote Control Bot\n\nCommands:\n");
+    usage += "/auth <password> - Authenticate (prove you are the bot owner or a trusted person)\n";
+    usage += &wol_usage(config);
+    usage += " - send Wake-on-LAN magic packet to a pre-configured (or an arbitrary) machine in the local network\n";
+    usage += &ssh_unlock_disk_usage(config);
+    usage += " - log into a pre-configured machine via SSH (as root) and send the password to unlock a LUKS-encrypted root partition\n";
+    usage += &ssh_exec_usage(config);
+    usage += " - log into a machine via SSH and execute one command. Waits until the process finishes or forcefully exits after 20 seconds\n";
+
+    usage += "\nCommands from plugins:\n";
+    for cmd in ctx_lock.plugin_commands.values() {
+        usage += &format!(
+            "{} (from {}) - {}\n",
+            &cmd.usage, &cmd.plugin_id, &cmd.description
+        );
+    }
+
+    usage += "\nDeltaChat Remote Control Bot by slavasil, 2026\n";
+
+    usage
+}
+
 pub async fn auth_command(
     dchat_ctx: Arc<Mutex<Context>>,
     ctx: Arc<Mutex<BotContext>>,

src/config/mod.rs

@@ -1,5 +1,8 @@
+pub mod util;
+
 use eui48::MacAddress;
 use serde::Deserialize;
+use serde_yaml::{Mapping, Value};
 use std::{
     collections::HashMap,
     fs, io,
@@ -13,6 +16,8 @@ pub struct BotConfig {
     pub machines: HashMap<String, BotTargetMachineConfig>,
     #[serde(rename = "deltaChat")]
     pub delta_chat: BotDeltaChatConfig,
+    #[serde(default)]
+    pub plugins: Vec<PluginConfig>,
 }
 
 #[derive(Deserialize, Debug)]
@@ -47,6 +52,23 @@ pub struct BotDeltaChatConfig {
     pub avatar: Option<PathBuf>,
 }
 
+#[derive(Deserialize, Debug, Clone)]
+pub struct PluginConfig {
+    pub name: String,
+    #[serde(default = "default_plugin_enabled")]
+    pub enabled: bool,
+    #[serde(default = "default_plugin_config")]
+    pub config: Value,
+}
+
+fn default_plugin_enabled() -> bool {
+    true
+}
+
+fn default_plugin_config() -> Value {
+    Value::Mapping(Mapping::new())
+}
+
 #[derive(Debug)]
 pub enum ConfigError {
     Io(io::Error),

src/config/util.rs

@@ -0,0 +1,84 @@
+use serde_yaml::Value;
+
+pub(crate) fn yaml_to_json(yaml: &Value) -> String {
+    let mut json = String::new();
+    write_yaml_to_json(yaml, &mut json);
+    json
+}
+
+fn write_yaml_to_json(yaml: &Value, out: &mut String) {
+    match yaml {
+        Value::Null => {
+            out.push_str("null");
+        }
+        Value::Bool(b) => {
+            out.push_str(if *b { "true" } else { "false" });
+        }
+        Value::Number(n) => {
+            out.push_str(&n.to_string());
+        }
+        Value::String(s) => {
+            out.push('"');
+            for c in s.encode_utf16() {
+                match c {
+                    0x005C | 0x0022 => {
+                        out.push('\\');
+                        out.push(char::from_u32(c as u32).unwrap());
+                    }
+                    0x0008 => {
+                        out.push('\\');
+                        out.push('b');
+                    }
+                    0x0009 => {
+                        out.push('\\');
+                        out.push('t');
+                    }
+                    0x000a => {
+                        out.push('\\');
+                        out.push('n');
+                    }
+                    0x000C => {
+                        out.push('\\');
+                        out.push('f');
+                    }
+                    0x0000..=0x001F | 0x0080..=0xFFFF => {
+                        out.push_str(&format!("\\u{c:04x}"));
+                    }
+                    _ => {
+                        out.push(char::from_u32(c as u32).unwrap());
+                    }
+                }
+            }
+            out.push_str(s); // TODO escape
+            out.push('"');
+        }
+        Value::Tagged(_) => {}
+        Value::Sequence(list) => {
+            out.push('[');
+            if !list.is_empty() {
+                write_yaml_to_json(&list[0], out);
+                for i in 1..list.len() {
+                    out.push(',');
+                    write_yaml_to_json(&list[i], out);
+                }
+            }
+            out.push(']');
+        }
+        Value::Mapping(map) => {
+            out.push('{');
+            let mut iter = map.iter();
+            if let Some(kv) = iter.next() {
+                write_yaml_to_json(&Value::String(yaml_to_json(kv.0)), out);
+                out.push(':');
+                write_yaml_to_json(kv.1, out);
+            }
+            for kv in iter {
+                out.push(',');
+                write_yaml_to_json(&Value::String(yaml_to_json(kv.0)), out);
+                out.push(':');
+                write_yaml_to_json(kv.1, out);
+            }
+            out.push('}');
+        }
+    }
+}

src/main.rs

@@ -1,12 +1,27 @@
 mod commands;
 mod config;
+mod paths;
+mod plugin;
 mod ssh;
 
-use anyhow::{Context as _, Result as AnyhowResult};
+mod proto {
+    pub(crate) mod deltachat_remotecontrol_bot {
+        pub(crate) mod plugin {
+            include!(concat!(
+                env!("OUT_DIR"),
+                "/deltachat_remotecontrol_bot.plugin.rs"
+            ));
+        }
+    }
+    pub(crate) use deltachat_remotecontrol_bot::plugin::*;
+}
+
+use anyhow::{Context as _, Result as AnyhowResult, bail};
 use clap::Parser;
 use config::BotConfig;
 use deltachat::{
-    EventType, Events, chat,
+    EventType, Events,
+    chat::{self, ChatId},
     config::Config,
     contact::ContactId,
     context::Context,
@@ -14,20 +29,26 @@ use deltachat::{
     securejoin,
     stock_str::StockStrings,
 };
-use std::collections::HashSet;
+use std::collections::{HashMap, HashSet};
 use std::{path::PathBuf, sync::Arc};
 use tokio::sync::Mutex;
 
-const CONFIG_FILENAME: &str = "bot.yml";
+use crate::{
+    config::PluginConfig,
+    paths::{data_path, default_config_paths},
+    plugin::{LoadedPlugin, PluginCommand, try_load_plugin},
+};
+
 const APP_NAME: &str = "deltachat-remotecontrol-bot";
-const APP_CONFIG_DIR: &str = APP_NAME;
-const APP_DATA_DIR: &str = APP_NAME;
 const BOT_DISPLAY_NAME: &str = "🤖Remote🖲️";
 const AUTH_REQUIRED: bool = false;
 
 pub struct BotContext {
     authed_contacts: HashSet<ContactId>,
     config: BotConfig,
+    plugins: HashMap<String, Arc<Mutex<LoadedPlugin>>>, // maps plugin's unique name (id) to plugin object
+    plugin_commands: HashMap<String, Arc<PluginCommand>>, // maps command name to command object
+    plugin_cmd_aliases: HashMap<String, Arc<PluginCommand>>, // maps command alias to command object
 }
 
 impl BotContext {
@@ -35,6 +56,9 @@ impl BotContext {
         BotContext {
             authed_contacts: HashSet::new(),
             config,
+            plugins: HashMap::new(),
+            plugin_commands: HashMap::new(),
+            plugin_cmd_aliases: HashMap::new(),
         }
     }
 }
@@ -51,40 +75,7 @@ struct Args {
     config: Option<PathBuf>,
 }
 
-fn default_config_paths() -> Vec<PathBuf> {
+async fn run_bot(bot_context: Arc<Mutex<BotContext>>) -> AnyhowResult<()> {
-    let mut paths = vec![PathBuf::from(CONFIG_FILENAME)];
-
-    if let Ok(config_home) = std::env::var("XDG_CONFIG_HOME")
-        .map(PathBuf::from)
-        .or(std::env::var("HOME").map(|home| PathBuf::from(home).join(".config")))
-    {
-        paths.push(config_home.join(APP_CONFIG_DIR).join(CONFIG_FILENAME));
-    }
-
-    paths.push(
-        PathBuf::from("/etc")
-            .join(APP_CONFIG_DIR)
-            .join(CONFIG_FILENAME),
-    );
-
-    paths
-}
-
-fn data_path() -> PathBuf {
-    std::env::var("XDG_DATA_HOME")
-        .map(|dh| PathBuf::from(dh).join(APP_DATA_DIR))
-        .or_else(|_| {
-            std::env::var("HOME").map(|h| {
-                PathBuf::from(h)
-                    .join(".local")
-                    .join("share")
-                    .join(APP_DATA_DIR)
-            })
-        })
-        .unwrap_or(PathBuf::from("."))
-}
-
-async fn run_bot(cfg: config::BotConfig) -> AnyhowResult<()> {
     let dchat_db_dir = data_path();
     std::fs::create_dir_all(&dchat_db_dir)
         .with_context(|| format!("Failed to create data directory {}", dchat_db_dir.display()))?;
@@ -100,12 +91,14 @@ async fn run_bot(cfg: config::BotConfig) -> AnyhowResult<()> {
     .await
     .context("Failed to open Delta Chat client DB")?;
 
+    let ctx_lock = bot_context.lock().await;
+
     if !dchat_ctx.is_configured().await? {
         dchat_ctx
-            .set_config(Config::Addr, Some(&cfg.delta_chat.email))
+            .set_config(Config::Addr, Some(&ctx_lock.config.delta_chat.email))
             .await?;
         dchat_ctx
-            .set_config(Config::MailPw, Some(&cfg.delta_chat.password))
+            .set_config(Config::MailPw, Some(&ctx_lock.config.delta_chat.password))
             .await?;
         dchat_ctx.set_config(Config::Bot, Some("1")).await?;
         dchat_ctx
@@ -116,7 +109,7 @@ async fn run_bot(cfg: config::BotConfig) -> AnyhowResult<()> {
         dchat_ctx
             .set_config(Config::Displayname, Some(BOT_DISPLAY_NAME))
             .await?;
-        if let Some(mut avatar_path) = cfg.delta_chat.avatar.clone() {
+        if let Some(mut avatar_path) = ctx_lock.config.delta_chat.avatar.clone() {
             if avatar_path.is_relative() {
                 avatar_path = data_path().join(&avatar_path);
             }
@@ -127,6 +120,7 @@ async fn run_bot(cfg: config::BotConfig) -> AnyhowResult<()> {
             }
         }
     }
+    drop(ctx_lock);
 
     dchat_ctx.start_io().await;
 
@@ -141,8 +135,6 @@ async fn run_bot(cfg: config::BotConfig) -> AnyhowResult<()> {
 
     let dchat_ctx = Arc::new(Mutex::new(dchat_ctx));
 
-    let bot_context = Arc::new(Mutex::new(BotContext::new(cfg)));
-
     let ev_emitter = dchat_ctx.lock().await.get_event_emitter();
 
     while let Some(ev) = ev_emitter.recv().await {
@@ -158,6 +150,21 @@ async fn run_bot(cfg: config::BotConfig) -> AnyhowResult<()> {
     Ok(())
 }
 
+async fn send_command_not_found(
+    dchat_ctx: Arc<Mutex<Context>>,
+    bot_ctx: Arc<Mutex<BotContext>>,
+    chat_id: ChatId,
+) -> AnyhowResult<()> {
+    let dchat_ctx_lock = dchat_ctx.lock().await;
+    chat::send_text_msg(
+        &dchat_ctx_lock,
+        chat_id,
+        commands::global_usage(bot_ctx).await,
+    )
+    .await?;
+    Ok(())
+}
+
 async fn handle_message(
     dchat_ctx: Arc<Mutex<Context>>,
     bot_ctx: Arc<Mutex<BotContext>>,
@@ -192,7 +199,136 @@ async fn handle_message(
         "/ssh-exec" => {
             commands::ssh_exec_command(dchat_ctx, bot_ctx, msg, &args).await?;
         }
-        _ => {}
+        _ => {
+            let ctx_lock = bot_ctx.lock().await;
+            let Some(cmd) = ctx_lock.plugin_cmd_aliases.get(&cmd[1..]) else {
+                drop(ctx_lock);
+                send_command_not_found(dchat_ctx, bot_ctx, msg.get_chat_id()).await?;
+                return Ok(());
+            };
+
+            let Some(plugin) = ctx_lock.plugins.get(&cmd.plugin_id) else {
+                bail!(
+                    "Inconsistency in plugin_cmd_aliases hashmap: command references a plugin that is not loaded"
+                );
+            };
+            let (plugin_conn, plugin_name) = {
+                let lock = plugin.lock().await;
+                let Some(conn) = lock.connection.clone() else {
+                    bail!("Plugin disconnected");
+                };
+                let plugin_name = lock.plugin_id.clone();
+                log::info!(
+                    "Delegating command /{} to plugin {}",
+                    &cmd.name,
+                    &plugin_name
+                );
+                (conn, plugin_name)
+            };
+            let issuer_id = msg.get_from_id().to_string();
+            let cmd_name = cmd.name.clone();
+            let chat_id = msg.get_chat_id();
+            let dchat_ctx = Arc::clone(&dchat_ctx);
+
+            tokio::spawn(async move {
+                log::debug!("Started plugin command task");
+                match plugin_conn
+                    .execute_command(
+                        cmd_name.clone(),
+                        issuer_id,
+                        commands::args::split_command_line(&text),
+                    )
+                    .await
+                {
+                    Ok(mut replies) => {
+                        let mut last_message_id = None;
+                        while let Some(reply) = replies.recv().await {
+                            match reply {
+                                Ok(reply) => {
+                                    log::debug!(
+                                        "Reply to /{} command from plugin {}: {:?}",
+                                        &cmd_name,
+                                        &plugin_name,
+                                        &reply
+                                    );
+                                    if reply.edit {
+                                        if let Some(msg_id) = last_message_id {
+                                            if let Some(proto::command_reply::Reply::Text(text)) =
+                                                reply.reply
+                                            {
+                                                let dchat_ctx_lock = dchat_ctx.lock().await;
+                                                log::debug!(
+                                                    "Command {} :: /{}: editing last message",
+                                                    &plugin_name,
+                                                    &cmd_name
+                                                );
+                                                if let Err(e) = chat::send_edit_request(
+                                                    &dchat_ctx_lock,
+                                                    msg_id,
+                                                    text,
+                                                )
+                                                .await
+                                                {
+                                                    log::error!("Cannot edit message: {e}");
+                                                    continue;
+                                                }
+                                            }
+                                        } else {
+                                            log::warn!(
+                                                "Plugin {} requested to edit a message but no messages sent yet",
+                                                &plugin_name
+                                            );
+                                        }
+                                    } else {
+                                        if let Some(proto::command_reply::Reply::Text(text)) =
+                                            reply.reply
+                                        {
+                                            let dchat_ctx_lock = dchat_ctx.lock().await;
+                                            log::debug!(
+                                                "Command {} :: /{}: sending message",
+                                                &plugin_name,
+                                                &cmd_name
+                                            );
+                                            let msg_id = match chat::send_text_msg(
+                                                &dchat_ctx_lock,
+                                                chat_id,
+                                                text,
+                                            )
+                                            .await
+                                            {
+                                                Ok(msg_id) => msg_id,
+                                                Err(e) => {
+                                                    log::error!("Cannot send new message: {e}");
+                                                    continue;
+                                                }
+                                            };
+                                            last_message_id = Some(msg_id);
+                                        }
+                                    }
+                                }
+                                Err(e) => {
+                                    log::error!(
+                                        "Error while executing {} :: /{}: {}",
+                                        &plugin_name,
+                                        &cmd_name,
+                                        e
+                                    );
+                                    break;
+                                }
+                            }
+                        }
+                    }
+                    Err(e) => {
+                        log::error!(
+                            "Plugin command ({} :: /{}) execution failed: {}",
+                            &plugin_name,
+                            &cmd_name,
+                            e
+                        );
+                    }
+                }
+            });
+        }
     }
 
     Ok(())
@@ -232,5 +368,36 @@ async fn main() {
         }
     };
 
-    run_bot(config).await.expect("error");
+    let requested_plugins: Vec<PluginConfig> = config
+        .plugins
+        .clone()
+        .into_iter()
+        .filter(|p| p.enabled)
+        .collect();
+
+    let bot_context = Arc::new(Mutex::new(BotContext::new(config)));
+
+    if !requested_plugins.is_empty() {
+        log::info!(
+            "Loading plugins ({})",
+            requested_plugins
+                .iter()
+                .map(|p| p.name.as_str())
+                .collect::<Vec<&str>>()
+                .join(", ")
+        );
+        for plugin in requested_plugins {
+            if let Err(e) = try_load_plugin(
+                Arc::clone(&bot_context),
+                plugin.name.clone(),
+                config::util::yaml_to_json(&plugin.config),
+            )
+            .await
+            {
+                log::error!("Failed to load plugin \"{}\": {e}", &plugin.name);
+            }
+        }
+    }
+
+    run_bot(bot_context).await.expect("error");
 }

src/paths.rs

@@ -0,0 +1,44 @@
+use std::path::PathBuf;
+
+use crate::APP_NAME;
+
+const APP_CONFIG_DIR: &str = APP_NAME;
+const APP_DATA_DIR: &str = APP_NAME;
+const CONFIG_FILENAME: &str = "bot.yml";
+
+pub(crate) fn default_config_paths() -> Vec<PathBuf> {
+    let mut paths = vec![PathBuf::from(CONFIG_FILENAME)];
+
+    if let Ok(config_home) = std::env::var("XDG_CONFIG_HOME")
+        .map(PathBuf::from)
+        .or(std::env::var("HOME").map(|home| PathBuf::from(home).join(".config")))
+    {
+        paths.push(config_home.join(APP_CONFIG_DIR).join(CONFIG_FILENAME));
+    }
+
+    paths.push(
+        PathBuf::from("/etc")
+            .join(APP_CONFIG_DIR)
+            .join(CONFIG_FILENAME),
+    );
+
+    paths
+}
+
+pub(crate) fn data_path() -> PathBuf {
+    std::env::var("XDG_DATA_HOME")
+        .map(|dh| PathBuf::from(dh).join(APP_DATA_DIR))
+        .or_else(|_| {
+            std::env::var("HOME").map(|h| {
+                PathBuf::from(h)
+                    .join(".local")
+                    .join("share")
+                    .join(APP_DATA_DIR)
+            })
+        })
+        .unwrap_or(PathBuf::from("."))
+}
+
+pub(crate) fn plugins_path() -> PathBuf {
+    data_path().join("plugins")
+}

src/plugin/mod.rs

@@ -0,0 +1,171 @@
+mod stdio;
+
+use anyhow::{Context as _, Result as AnyhowResult, bail};
+use async_trait::async_trait;
+use prost::DecodeError;
+use std::{
+    error::Error,
+    fmt::{Debug, Display},
+    process::Stdio,
+    sync::Arc,
+};
+
+use tokio::sync::{Mutex, mpsc};
+
+use crate::{BotContext, paths::plugins_path, proto};
+
+pub(crate) struct PluginCommand {
+    pub plugin_id: String,
+    pub name: String,
+    pub aliases: Vec<String>,
+    pub usage: String,
+    pub description: String,
+}
+
+#[derive(Default)]
+pub(crate) struct LoadedPlugin {
+    pub plugin_id: String,
+    pub name: String,
+    pub version: String,
+    pub authors: String,
+    pub commands: Vec<Arc<PluginCommand>>,
+    pub connection: Option<Arc<dyn PluginConnection>>,
+}
+
+#[derive(Debug)]
+pub(crate) enum PluginRequestType {
+    Initialize,
+    CommandList,
+    Execute,
+}
+
+impl From<PluginRequestType> for i32 {
+    fn from(value: PluginRequestType) -> Self {
+        match value {
+            PluginRequestType::Initialize => 1,
+            PluginRequestType::CommandList => 2,
+            PluginRequestType::Execute => 3,
+        }
+    }
+}
+
+impl Display for PluginRequestType {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        <PluginRequestType as Debug>::fmt(self, f)
+    }
+}
+
+#[async_trait]
+pub(crate) trait PluginConnection: Send + Sync {
+    async fn initialize_plugin(
+        &self,
+        config: String,
+    ) -> Result<proto::PluginInitializeResponse, PluginConnectionError>;
+
+    async fn request_plugin_command_list(
+        &self,
+    ) -> Result<proto::PluginCommandListResponse, PluginConnectionError>;
+
+    async fn execute_command(
+        self: Arc<Self>,
+        command_id: String,
+        issuer_id: String,
+        argv: Vec<String>,
+    ) -> Result<
+        mpsc::Receiver<Result<proto::CommandReply, PluginConnectionError>>,
+        PluginConnectionError,
+    >;
+}
+
+#[derive(Debug)]
+pub(crate) enum PluginConnectionError {
+    SendRequest(String, PluginRequestType, Box<dyn Error + Send + Sync>),
+    ReadResponse(String, PluginRequestType, Box<dyn Error + Send + Sync>),
+    DecodeResponse(String, PluginRequestType, Option<DecodeError>),
+    InvalidMessageLength,
+}
+
+impl Display for PluginConnectionError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::SendRequest(plugin_name, req_type, e) => f.write_fmt(format_args!(
+                "Can't send request ({req_type}) to plugin {plugin_name}: {e}"
+            )),
+            Self::ReadResponse(plugin_name, resp_type, e) => f.write_fmt(format_args!(
+                "Can't read response ({resp_type}) from plugin {plugin_name}: {e}"
+            )),
+            Self::DecodeResponse(plugin_name, resp_type, e) => f.write_fmt(format_args!(
+                "Can't decode response ({resp_type}) from plugin {plugin_name}{}",
+                e.as_ref().map(|e| format!(": {e}")).unwrap_or_default()
+            )),
+            Self::InvalidMessageLength => f.write_str("Plugin response length is invalid"),
+        }
+    }
+}
+
+impl Error for PluginConnectionError {
+    fn source(&self) -> Option<&(dyn Error + 'static)> {
+        match self {
+            Self::SendRequest(_, _, e) => Some(e.as_ref()),
+            Self::ReadResponse(_, _, e) => Some(e.as_ref()),
+            Self::DecodeResponse(_, _, e) => e.as_ref().map(|e| e as &dyn Error),
+            Self::InvalidMessageLength => None,
+        }
+    }
+}
+
+pub(crate) async fn try_load_plugin(
+    ctx: Arc<Mutex<BotContext>>,
+    unique_name: String,
+    config_json: String,
+) -> AnyhowResult<()> {
+    let plugin_dir = plugins_path().join(&unique_name);
+    if ctx.lock().await.plugins.contains_key(&unique_name) {
+        bail!("plugin unique name is not unique");
+    }
+    if !std::fs::metadata(&plugin_dir)?.is_dir() {
+        bail!("Plugin directory doesn't exist");
+    }
+    log::debug!("Loading plugin with JSON config: {}", &config_json);
+    let plugin_executable_path = plugin_dir.join("plugin_run");
+    log::debug!("Starting plugin executable {:?}", &plugin_executable_path);
+    let mut cmd = tokio::process::Command::new(plugin_executable_path);
+    cmd.stdin(Stdio::piped())
+        .stdout(Stdio::piped())
+        .stderr(Stdio::inherit());
+    cmd.current_dir(&plugin_dir);
+    cmd.env("DCRCBOT_PLUGIN_TRANSPORT", "stdio");
+
+    let plugin_process = cmd.spawn().context("Failed to start the plugin")?;
+    let plugin =
+        stdio::initialize_stdio_plugin(plugin_process, unique_name.clone(), config_json).await?;
+
+    let mut ctx_lock = ctx.lock().await;
+    for cmd in plugin.lock().await.commands.iter().cloned() {
+        log::debug!("adding command /{} of plugin {}", &cmd.name, &unique_name);
+        if ctx_lock.plugin_commands.contains_key(&cmd.name) {
+            bail!("duplicate command specification");
+        }
+        ctx_lock
+            .plugin_commands
+            .insert(cmd.name.clone(), Arc::clone(&cmd));
+        ctx_lock
+            .plugin_cmd_aliases
+            .insert(cmd.name.clone(), Arc::clone(&cmd));
+        for alias in cmd.aliases.iter() {
+            log::debug!(
+                "adding command alias /{} -> /{} of plugin {}",
+                alias,
+                &cmd.name,
+                &unique_name
+            );
+            ctx_lock
+                .plugin_cmd_aliases
+                .insert(alias.to_owned(), Arc::clone(&cmd));
+        }
+    }
+    drop(ctx_lock);
+
+    ctx.lock().await.plugins.insert(unique_name, plugin);
+    Ok(())
+}

src/plugin/stdio.rs

@@ -0,0 +1,401 @@
+use std::{collections::HashMap, error::Error, ops::DerefMut, sync::Arc, time::Duration};
+
+use anyhow::Result as AnyhowResult;
+use async_trait::async_trait;
+use prost::Message;
+use tokio::{
+    io::{AsyncRead, AsyncReadExt, AsyncWriteExt, BufReader},
+    process::{Child, ChildStdout},
+    sync::{Mutex, mpsc, oneshot},
+};
+
+use crate::{
+    plugin::{
+        LoadedPlugin, PluginCommand, PluginConnection, PluginConnectionError, PluginRequestType,
+    },
+    proto,
+};
+
+pub(super) async fn initialize_stdio_plugin(
+    process: Child,
+    unique_name: String,
+    config_json: String,
+) -> AnyhowResult<Arc<Mutex<LoadedPlugin>>> {
+    let plugin = Arc::new(Mutex::new(LoadedPlugin::default()));
+    log::info!("Connecting to plugin {} using standard I/O", &unique_name);
+    plugin.lock().await.plugin_id = unique_name;
+    let connection = Arc::new(StdioPluginConnection::new(Arc::clone(&plugin), process));
+    Arc::clone(&connection).run_stdio_loops();
+
+    let plugin_info = connection.initialize_plugin(config_json).await?;
+    log::debug!("received plugin identification: {:?}", plugin_info);
+    let mut plugin_lock = plugin.lock().await;
+    plugin_lock.name = plugin_info.name;
+    plugin_lock.plugin_id = plugin_info.unique_name.clone();
+    plugin_lock.version = plugin_info.version;
+    plugin_lock.authors = plugin_info.authors;
+    drop(plugin_lock);
+
+    let command_list = connection.request_plugin_command_list().await?;
+    let mut plugin_lock = plugin.lock().await;
+    plugin_lock.commands = command_list
+        .commands
+        .into_iter()
+        .map(|cmd| {
+            Arc::new(PluginCommand {
+                name: cmd.name,
+                plugin_id: plugin_info.unique_name.clone(),
+                aliases: cmd.aliases,
+                usage: cmd.usage,
+                description: cmd.description,
+            })
+        })
+        .collect();
+
+    plugin_lock.connection = Some(connection);
+    drop(plugin_lock);
+
+    Ok(plugin)
+}
+
+struct StdioPluginConnection {
+    plugin: Arc<Mutex<LoadedPlugin>>,
+    process: Mutex<Child>,
+    buffered_stdout: Mutex<BufReader<ChildStdout>>,
+    next_request_id: Mutex<u32>,
+    pending_requests: Mutex<HashMap<u32, oneshot::Sender<proto::Response>>>,
+}
+
+impl StdioPluginConnection {
+    pub fn new(plugin: Arc<Mutex<LoadedPlugin>>, mut process: Child) -> StdioPluginConnection {
+        let stdout = process.stdout.take().unwrap();
+        
+        StdioPluginConnection {
+            plugin,
+            process: Mutex::new(process),
+            buffered_stdout: Mutex::new(BufReader::new(stdout)),
+            next_request_id: Mutex::new(0),
+            pending_requests: Mutex::new(HashMap::new()),
+        }
+    }
+
+    fn run_stdio_loops(self: Arc<Self>) {
+        tokio::spawn(self.stdout_reader_loop());
+    }
+
+    async fn stdout_reader_loop(self: Arc<Self>) {
+        loop {
+            let frame =
+                match Self::read_length_delimited(self.buffered_stdout.lock().await.deref_mut())
+                    .await
+                {
+                    Ok(frame) => frame,
+                    Err(e) => {
+                        log::error!(
+                            "Error while reading STDOUT of stdio plugin {}: {e}",
+                            &self.plugin.lock().await.plugin_id
+                        );
+                        break;
+                    }
+                };
+            let response = match proto::Response::decode(frame.as_slice()) {
+                Ok(response) => response,
+                Err(_) => {
+                    log::error!(
+                        "Invalid response received from stdio plugin {}",
+                        &self.plugin.lock().await.plugin_id
+                    );
+                    continue;
+                }
+            };
+            match self
+                .pending_requests
+                .lock()
+                .await
+                .remove(&response.request_id)
+            {
+                Some(sender) => match sender.send(response) {
+                    Ok(()) => {}
+                    Err(response) => {
+                        log::warn!(
+                            "Dropping response with request_id {} from plugin {}",
+                            response.request_id,
+                            self.plugin.lock().await.plugin_id
+                        );
+                    }
+                },
+                None => {
+                    continue;
+                }
+            }
+        }
+
+        self.pending_requests.lock().await.clear();
+    }
+
+    async fn read_length_delimited<R>(
+        reader: &mut R,
+    ) -> Result<Vec<u8>, Box<dyn Error + Send + Sync>>
+    where
+        R: AsyncRead + Unpin,
+    {
+        const MAX_FRAME_LENGTH: usize = 0x40000000;
+        let mut length: usize = 0;
+        let mut bit = 0;
+        let mut byte = [0u8; 1];
+        let mut complete = false;
+        for _ in 0..10 {
+            reader.read_exact(&mut byte).await?;
+            length += ((byte[0] & 0x7f) as usize).unbounded_shl(bit);
+            bit += 7;
+            if (byte[0] & 0x80) == 0 {
+                complete = true;
+                break;
+            }
+        }
+
+        if !complete || length > MAX_FRAME_LENGTH {
+            return Err(Box::new(PluginConnectionError::InvalidMessageLength));
+        }
+
+        let mut message = vec![0u8; length];
+        reader.read_exact(&mut message).await?;
+        Ok(message)
+    }
+
+    async fn await_response_to(
+        &self,
+        request_id: u32,
+        timeout: Duration,
+    ) -> Result<proto::Response, Box<dyn Error + Send + Sync>> {
+        let (tx, rx) = oneshot::channel();
+        self.pending_requests.lock().await.insert(request_id, tx);
+
+        match tokio::time::timeout(timeout, rx).await {
+            Ok(Ok(response)) => Ok(response),
+            Ok(Err(e)) => Err(Box::new(e)),
+            Err(e) => Err(Box::new(e)),
+        }
+    }
+}
+
+#[async_trait]
+impl PluginConnection for StdioPluginConnection {
+    async fn initialize_plugin(
+        &self,
+        config: String,
+    ) -> Result<proto::PluginInitializeResponse, PluginConnectionError> {
+        let request_id = {
+            let mut r = self.next_request_id.lock().await;
+            let id = *r;
+            *r += 1;
+            id
+        };
+        let request = proto::Request {
+            request_id,
+            req: Some(proto::request::Req::InitializeReq(
+                proto::PluginInitializeRequest { config },
+            )),
+        }
+        .encode_length_delimited_to_vec();
+
+        if let Err(e) = self
+            .process
+            .lock()
+            .await
+            .stdin
+            .as_mut()
+            .unwrap()
+            .write_all(&request)
+            .await
+        {
+            return Err(PluginConnectionError::SendRequest(
+                self.plugin.lock().await.plugin_id.clone(),
+                PluginRequestType::Initialize,
+                Box::new(e),
+            ));
+        }
+
+        let response = match self
+            .await_response_to(request_id, Duration::from_secs(10))
+            .await
+        {
+            Ok(response) => response,
+            Err(e) => {
+                return Err(PluginConnectionError::ReadResponse(
+                    self.plugin.lock().await.plugin_id.clone(),
+                    PluginRequestType::Initialize,
+                    e,
+                ));
+            }
+        };
+
+        match response.res {
+            Some(proto::response::Res::InitializeRes(resp)) => Ok(resp),
+            _ => Err(PluginConnectionError::DecodeResponse(
+                self.plugin.lock().await.plugin_id.clone(),
+                PluginRequestType::Initialize,
+                None,
+            )),
+        }
+    }
+
+    async fn request_plugin_command_list(
+        &self,
+    ) -> Result<proto::PluginCommandListResponse, PluginConnectionError> {
+        let request_id = {
+            let mut r = self.next_request_id.lock().await;
+            let id = *r;
+            *r += 1;
+            id
+        };
+        let request = proto::Request {
+            request_id,
+            req: Some(proto::request::Req::CommandListReq(
+                proto::PluginCommandListRequest {},
+            )),
+        }
+        .encode_length_delimited_to_vec();
+
+        if let Err(e) = self
+            .process
+            .lock()
+            .await
+            .stdin
+            .as_mut()
+            .unwrap()
+            .write_all(&request)
+            .await
+        {
+            return Err(PluginConnectionError::SendRequest(
+                self.plugin.lock().await.plugin_id.clone(),
+                PluginRequestType::CommandList,
+                Box::new(e),
+            ));
+        }
+
+        let response = match self
+            .await_response_to(request_id, Duration::from_secs(10))
+            .await
+        {
+            Ok(response) => response,
+            Err(e) => {
+                return Err(PluginConnectionError::ReadResponse(
+                    self.plugin.lock().await.plugin_id.clone(),
+                    PluginRequestType::CommandList,
+                    e,
+                ));
+            }
+        };
+
+        match response.res {
+            Some(proto::response::Res::CommandListRes(resp)) => Ok(resp),
+            _ => Err(PluginConnectionError::DecodeResponse(
+                self.plugin.lock().await.plugin_id.clone(),
+                PluginRequestType::CommandList,
+                None,
+            )),
+        }
+    }
+
+    async fn execute_command(
+        self: Arc<Self>,
+        command_id: String,
+        issuer_id: String,
+        argv: Vec<String>,
+    ) -> Result<
+        mpsc::Receiver<Result<proto::CommandReply, PluginConnectionError>>,
+        PluginConnectionError,
+    > {
+        let request_id = {
+            let mut r = self.next_request_id.lock().await;
+            let id = *r;
+            *r += 1;
+            id
+        };
+        let request = proto::Request {
+            request_id,
+            req: Some(proto::request::Req::ExecuteReq(
+                proto::PluginExecuteRequest {
+                    command_id,
+                    issuer_id,
+                    arg_vector: argv,
+                },
+            )),
+        }
+        .encode_length_delimited_to_vec();
+
+        if let Err(e) = self
+            .process
+            .lock()
+            .await
+            .stdin
+            .as_mut()
+            .unwrap()
+            .write_all(&request)
+            .await
+        {
+            return Err(PluginConnectionError::SendRequest(
+                self.plugin.lock().await.plugin_id.clone(),
+                PluginRequestType::Execute,
+                Box::new(e),
+            ));
+        }
+
+        let (tx, rx) = mpsc::channel(4);
+
+        tokio::spawn(async move {
+            loop {
+                let response = match self
+                    .await_response_to(request_id, Duration::from_secs(600))
+                    .await
+                {
+                    Ok(response) => response,
+                    Err(e) => {
+                        if let Err(e) = tx
+                            .send(Err(PluginConnectionError::ReadResponse(
+                                self.plugin.lock().await.plugin_id.clone(),
+                                PluginRequestType::Execute,
+                                e,
+                            )))
+                            .await
+                        {
+                            log::error!("Cannot send error notification to another task: {e}");
+                        }
+                        break;
+                    }
+                };
+
+                match response.res {
+                    Some(proto::response::Res::CmdReply(reply)) => {
+                        let end = match reply.reply {
+                            Some(proto::command_reply::Reply::End(_)) => true,
+                            _ => false,
+                        };
+                        if let Err(e) = tx.send(Ok(reply)).await {
+                            log::error!("Cannot send command reply to another task: {e}");
+                        }
+                        if end {
+                            break;
+                        }
+                    }
+                    _ => {
+                        if let Err(e) = tx
+                            .send(Err(PluginConnectionError::DecodeResponse(
+                                self.plugin.lock().await.plugin_id.clone(),
+                                PluginRequestType::Execute,
+                                None,
+                            )))
+                            .await
+                        {
+                            log::error!("Cannot send error notification to another task: {e}");
+                        }
+                        break;
+                    }
+                }
+            }
+        });
+
+        Ok(rx)
+    }
+}

src/ssh.rs

@@ -14,7 +14,7 @@ impl Handler for ClientHandler {
 
     async fn check_server_key(
         &mut self,
-        server_public_key: &russh::keys::ssh_key::PublicKey,
+        _server_public_key: &russh::keys::ssh_key::PublicKey,
     ) -> Result<bool, Self::Error> {
         Ok(true)
     }