fix message loss with requests with multiple responses

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

Cargo.lock

@@ -273,9 +273,9 @@ dependencies = [
 
 [[package]]
 name = "astral-tokio-tar"
-version = "0.5.6"
+version = "0.6.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ec179a06c1769b1e42e1e2cbe74c7dcdb3d6383c838454d063eaac5bbb7ebbe5"
+checksum = "4ce73b17c62717c4b6a9af10b43e87c578b0cac27e00666d48304d3b7d2c0693"
 dependencies = [
  "filetime",
  "futures-core",
@@ -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",
@@ -979,9 +982,9 @@ checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570"
 
 [[package]]
 name = "colorutils-rs"
-version = "0.7.6"
+version = "0.8.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6e2fc25857fa523662de5cae84225b0e7bfb24a2a3f9ed8802fecf03df7252b1"
+checksum = "69abc9a8ed011e2b7946769f460b9e76e8b659ece9ef4001b9d8bba3489f796d"
 dependencies = [
  "erydanos",
  "half",
@@ -1419,7 +1422,7 @@ dependencies = [
 
 [[package]]
 name = "deltachat"
-version = "2.44.0-dev"
+version = "2.50.0-dev"
 dependencies = [
  "anyhow",
  "astral-tokio-tar",
@@ -1472,7 +1475,6 @@ dependencies = [
  "ratelimit",
  "regex",
  "rusqlite",
- "rustls-pki-types",
  "sanitize-filename",
  "sdp",
  "serde",
@@ -1866,6 +1868,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 +3398,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 +3765,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 +4461,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 +4962,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"
@@ -5646,9 +5731,9 @@ dependencies = [
 
 [[package]]
 name = "sdp"
-version = "0.10.0"
+version = "0.17.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "32c374dceda16965d541c8800ce9cc4e1c14acfd661ddf7952feeedc3411e5c6"
+checksum = "22c3b0257608d7de4de4c4ea650ccc2e6e3e45e3cd80039fcdee768bcb449253"
 dependencies = [
  "rand 0.9.2",
  "substring",

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;
+  }
+}

rustfmt.toml

@@ -0,0 +1,10 @@
+max_width = 120
+hard_tabs = false
+tab_spaces = 4
+edition = "2021"
+reorder_imports = true
+reorder_modules = true
+format_strings = true
+normalize_comments = true
+wrap_comments = true
+comment_width = 80

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),
     }
 }
 
@@ -80,11 +80,7 @@ mod tests {
     fn parses_double_quoted_argument() {
         assert_eq!(
             split_command_line(r#"/cmd arg1 "arg2 value""#),
-            vec![
+            vec!["/cmd".to_string(), "arg1".to_string(), "arg2 value".to_string()]
-                "/cmd".to_string(),
-                "arg1".to_string(),
-                "arg2 value".to_string()
-            ]
         );
     }
 

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};
@@ -12,16 +12,13 @@ use deltachat::{
     message::Message,
 };
 use eui48::MacAddress;
-use russh::{
+use russh::{client::AuthResult, keys::PrivateKeyWithHashAlg};
-    client::{AuthResult, Handle},
-    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 +27,28 @@ 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>>,
@@ -43,12 +62,7 @@ pub async fn auth_command(
     let dchat_ctx_lock = dchat_ctx.lock().await;
 
     if !AUTH_REQUIRED {
-        chat::send_text_msg(
+        chat::send_text_msg(&dchat_ctx_lock, chat_id, "Authentication is disabled".to_owned()).await?;
-            &dchat_ctx_lock,
-            chat_id,
-            "Authentication is disabled".to_owned(),
-        )
-        .await?;
         return Ok(());
     }
 
@@ -58,23 +72,13 @@ pub async fn auth_command(
     }
 
     let Some(password) = args.first() else {
-        chat::send_text_msg(
+        chat::send_text_msg(&dchat_ctx_lock, chat_id, "Usage: /auth <password>".to_owned()).await?;
-            &dchat_ctx_lock,
-            chat_id,
-            "Usage: /auth <password>".to_owned(),
-        )
-        .await?;
         return Ok(());
     };
 
     if *password == ctx_lock.config.auth.password {
         ctx_lock.authed_contacts.insert(contact_id);
-        chat::send_text_msg(
+        chat::send_text_msg(&dchat_ctx_lock, chat_id, "Authentication successful!".to_owned()).await?;
-            &dchat_ctx_lock,
-            chat_id,
-            "Authentication successful!".to_owned(),
-        )
-        .await?;
     } else {
         chat::send_text_msg(&dchat_ctx_lock, chat_id, "Incorrect password".to_owned()).await?;
     }
@@ -179,12 +183,7 @@ async fn send_magic_packet(mac: MacAddress) -> AnyhowResult<()> {
 
 fn ssh_unlock_disk_usage(config: &BotConfig) -> String {
     if !config.machines.is_empty() {
-        let list = config
+        let list = config.machines.keys().cloned().collect::<Vec<_>>().join("|");
-            .machines
-            .keys()
-            .cloned()
-            .collect::<Vec<_>>()
-            .join("|");
         if config.default_machine().is_some() {
             format!("Usage: /ssh-unlock-disk [{list}] <password>")
         } else {
@@ -213,12 +212,7 @@ pub async fn ssh_unlock_disk_command(
 
     if args.is_empty() || args.len() > 2 {
         log::warn!("wrong number of args");
-        chat::send_text_msg(
+        chat::send_text_msg(&dchat_ctx_lock, chat_id, ssh_unlock_disk_usage(&ctx_lock.config)).await?;
-            &dchat_ctx_lock,
-            chat_id,
-            ssh_unlock_disk_usage(&ctx_lock.config),
-        )
-        .await?;
         return Ok(());
     }
 
@@ -233,12 +227,7 @@ pub async fn ssh_unlock_disk_command(
             None
         }
     }) else {
-        chat::send_text_msg(
+        chat::send_text_msg(&dchat_ctx_lock, chat_id, ssh_unlock_disk_usage(&ctx_lock.config)).await?;
-            &dchat_ctx_lock,
-            chat_id,
-            ssh_unlock_disk_usage(&ctx_lock.config),
-        )
-        .await?;
         return Ok(());
     };
 
@@ -262,8 +251,7 @@ pub async fn ssh_unlock_disk_command(
 
     let password = args[args.len() - 1].to_owned();
     let host = machine.static_ip.to_string();
-    let ssh_key =
+    let ssh_key = russh::keys::PrivateKey::from_openssh(ssh_key).context("Invalid remote unlock SSH key")?;
-        russh::keys::PrivateKey::from_openssh(ssh_key).context("Invalid remote unlock SSH key")?;
 
     drop(ctx_lock);
     drop(dchat_ctx_lock);
@@ -289,10 +277,7 @@ pub async fn ssh_unlock_disk_command(
             SSH_TIMEOUT,
             connection.authenticate_publickey(
                 "root",
-                PrivateKeyWithHashAlg::new(
+                PrivateKeyWithHashAlg::new(Arc::new(ssh_key), connection.best_supported_rsa_hash().await?.flatten()),
-                    Arc::new(ssh_key),
-                    connection.best_supported_rsa_hash().await?.flatten(),
-                ),
             ),
         )
         .await
@@ -328,10 +313,7 @@ pub async fn ssh_unlock_disk_command(
             .context("Cannot send password via SSH")?;
 
         loop {
-            let Some(msg) = timeout(SSH_TIMEOUT, channel.wait())
+            let Some(msg) = timeout(SSH_TIMEOUT, channel.wait()).await.context("Timed out")? else {
-                .await
-                .context("Timed out")?
-            else {
                 break;
             };
             log::info!("Received message from SSH server: {:?}", msg);
@@ -350,12 +332,7 @@ pub async fn ssh_unlock_disk_command(
 
 fn ssh_exec_usage(config: &BotConfig) -> String {
     if !config.machines.is_empty() {
-        let list = config
+        let list = config.machines.keys().cloned().collect::<Vec<_>>().join("|");
-            .machines
-            .keys()
-            .cloned()
-            .collect::<Vec<_>>()
-            .join("|");
         format!("Usage: /ssh-exec <username>@({list}|<hostname>|<IP address>) <command...>")
     } else {
         "Command /ssh-exec unavailable: no machines configured".to_string()
@@ -477,22 +454,14 @@ pub async fn ssh_exec_command(
                     russh::ChannelMsg::Data { data } => {
                         let dchat_ctx_lock = dchat_ctx.lock().await;
                         report_text += &String::from_utf8_lossy(&data);
-                        chat::send_edit_request(
+                        chat::send_edit_request(&dchat_ctx_lock, report_message_id, report_text.clone())
-                            &dchat_ctx_lock,
-                            report_message_id,
-                            report_text.clone(),
-                        )
                             .await
                             .context("Cannot edit message")?;
                     }
                     russh::ChannelMsg::ExtendedData { data, ext: 1 } => {
                         let dchat_ctx_lock = dchat_ctx.lock().await;
                         report_text += &String::from_utf8_lossy(&data);
-                        chat::send_edit_request(
+                        chat::send_edit_request(&dchat_ctx_lock, report_message_id, report_text.clone())
-                            &dchat_ctx_lock,
-                            report_message_id,
-                            report_text.clone(),
-                        )
                             .await
                             .context("Cannot edit message")?;
                     }
@@ -509,11 +478,7 @@ pub async fn ssh_exec_command(
 
                     let dchat_ctx_lock = dchat_ctx.lock().await;
                     report_text += "\n\nRead timed out";
-                    chat::send_edit_request(
+                    chat::send_edit_request(&dchat_ctx_lock, report_message_id, report_text.clone())
-                        &dchat_ctx_lock,
-                        report_message_id,
-                        report_text.clone(),
-                    )
                         .await
                         .context("Cannot edit message")?;
 

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

src/main.rs

@@ -1,33 +1,51 @@
 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,
+    contact::{Contact, ContactId},
     context::Context,
     message::{Message, MsgId},
     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 +53,9 @@ impl BotContext {
         BotContext {
             authed_contacts: HashSet::new(),
             config,
+            plugins: HashMap::new(),
+            plugin_commands: HashMap::new(),
+            plugin_cmd_aliases: HashMap::new(),
         }
     }
 }
@@ -51,61 +72,25 @@ 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()))?;
 
     let dchat_db_filename = dchat_db_dir.join("deltachat.sqlite");
 
-    let dchat_ctx = Context::new(
+    let dchat_ctx = Context::new(&dchat_db_filename, 0, Events::default(), StockStrings::default())
-        &dchat_db_filename,
-        0,
-        Events::default(),
-        StockStrings::default(),
-    )
         .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,17 +101,16 @@ 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);
             }
             if let Some(avatar_str) = avatar_path.to_str() {
-                dchat_ctx
+                dchat_ctx.set_config(Config::Selfavatar, Some(avatar_str)).await?;
-                    .set_config(Config::Selfavatar, Some(avatar_str))
-                    .await?;
             }
         }
     }
+    drop(ctx_lock);
 
     dchat_ctx.start_io().await;
 
@@ -141,8 +125,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 +140,16 @@ 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 +184,110 @@ 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 contact_id = msg.get_from_id();
+            let dchat_ctx_lock = dchat_ctx.lock().await;
+            let issuer_id = Contact::get_by_id(&dchat_ctx_lock, contact_id)
+                .await
+                .map(|contact| contact.get_addr().to_string())
+                .unwrap_or_else(|_| contact_id.to_string());
+            drop(dchat_ctx_lock);
+            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 {
+                            log::trace!("reply recv'ed: Some({:?})", reply);
+                            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;
+                                }
+                            }
+                        }
+                        log::debug!("{} :: /{} command execution finished", &plugin_name, &cmd_name);
+                    }
+                    Err(e) => {
+                        log::error!(
+                            "Plugin command ({} :: /{}) execution failed: {}",
+                            &plugin_name,
+                            &cmd_name,
+                            e
+                        );
+                    }
+                }
+            });
+        }
     }
 
     Ok(())
@@ -232,5 +327,35 @@ 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);
+            }
+        }
+        let ctx_lock = bot_context.lock().await;
+        log::debug!("Plugins: {:?}", &ctx_lock.plugins);
+        log::debug!("Plugin commands: {:?}", &ctx_lock.plugin_commands);
+        log::debug!("Plugin command aliases: {:?}", &ctx_lock.plugin_cmd_aliases);
+    }
+
+    run_bot(bot_context).await.expect("error");
 }

src/paths.rs

@@ -0,0 +1,33 @@
+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,163 @@
+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};
+
+#[derive(Debug)]
+pub(crate) struct PluginCommand {
+    pub plugin_id: String,
+    pub name: String,
+    pub aliases: Vec<String>,
+    pub usage: String,
+    pub description: String,
+}
+
+#[derive(Default, Debug)]
+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 + Debug {
+    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>),
+    Timeout(String, PluginRequestType),
+    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::Timeout(plugin_name, req_type) => f.write_fmt(format_args!(
+                "Timed out waiting for response to {req_type} request to plugin {plugin_name}"
+            )),
+            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::Timeout(_, _) => None,
+            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));
+        }
+    }
+
+    let real_plugin_id = plugin.lock().await.plugin_id.clone();
+    ctx_lock.plugins.insert(real_plugin_id, plugin);
+    Ok(())
+}

src/plugin/stdio.rs

@@ -0,0 +1,419 @@
+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);
+    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)
+}
+
+#[derive(Debug)]
+enum ResponseListener {
+    Oneshot(oneshot::Sender<proto::Response>),
+    Multishot(mpsc::Sender<proto::Response>),
+}
+
+#[derive(Debug)]
+struct StdioPluginConnection {
+    plugin: Arc<Mutex<LoadedPlugin>>,
+    process: Mutex<Child>,
+    buffered_stdout: Mutex<BufReader<ChildStdout>>,
+    next_request_id: Mutex<u32>,
+    pending_requests: Mutex<HashMap<u32, ResponseListener>>,
+}
+
+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;
+                }
+            };
+            let mut pending_req_lock = self.pending_requests.lock().await;
+            match pending_req_lock.get(&response.request_id) {
+                Some(ResponseListener::Oneshot(_)) => {
+                    // We must remove to be able to send the response to the oneshot channel
+                    if let Some(ResponseListener::Oneshot(sender)) = pending_req_lock.remove(&response.request_id) {
+                        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
+                                );
+                            }
+                        }
+                    }
+                }
+                Some(ResponseListener::Multishot(sender)) => match sender.send(response).await {
+                    Ok(()) => {}
+                    Err(e) => {
+                        log::warn!(
+                            "Dropping response with request_id {} from plugin {}",
+                            e.0.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_trait]
+impl PluginConnection for StdioPluginConnection {
+    async fn initialize_plugin(
+        &self,
+        config: String,
+    ) -> Result<proto::PluginInitializeResponse, PluginConnectionError> {
+        // Generate request id
+        let request_id = {
+            let mut r = self.next_request_id.lock().await;
+            let id = *r;
+            *r += 1;
+            id
+        };
+
+        // Encode request
+        let request = proto::Request {
+            request_id,
+            req: Some(proto::request::Req::InitializeReq(proto::PluginInitializeRequest {
+                config,
+            })),
+        }
+        .encode_length_delimited_to_vec();
+
+        // Add response listener
+        let (tx, rx) = oneshot::channel();
+        self.pending_requests
+            .lock()
+            .await
+            .insert(request_id, ResponseListener::Oneshot(tx));
+
+        // Send request
+        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),
+            ));
+        }
+
+        // Receive response
+        let response = match tokio::time::timeout(Duration::from_secs(10), rx).await {
+            Ok(Ok(response)) => response,
+            Ok(Err(e)) => {
+                return Err(PluginConnectionError::ReadResponse(
+                    self.plugin.lock().await.plugin_id.clone(),
+                    PluginRequestType::Initialize,
+                    Box::new(e),
+                ));
+            }
+            Err(_) => {
+                return Err(PluginConnectionError::Timeout(
+                    self.plugin.lock().await.plugin_id.clone(),
+                    PluginRequestType::Initialize,
+                ));
+            }
+        };
+
+        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> {
+        // Generate request id
+        let request_id = {
+            let mut r = self.next_request_id.lock().await;
+            let id = *r;
+            *r += 1;
+            id
+        };
+
+        // Encode request
+        let request = proto::Request {
+            request_id,
+            req: Some(proto::request::Req::CommandListReq(proto::PluginCommandListRequest {})),
+        }
+        .encode_length_delimited_to_vec();
+
+        // Add response listener
+        let (tx, rx) = oneshot::channel();
+        self.pending_requests
+            .lock()
+            .await
+            .insert(request_id, ResponseListener::Oneshot(tx));
+
+        // Send request
+        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),
+            ));
+        }
+
+        // Receive response
+        let response = match tokio::time::timeout(Duration::from_secs(10), rx).await {
+            Ok(Ok(response)) => response,
+            Ok(Err(e)) => {
+                return Err(PluginConnectionError::ReadResponse(
+                    self.plugin.lock().await.plugin_id.clone(),
+                    PluginRequestType::CommandList,
+                    Box::new(e),
+                ));
+            }
+            Err(_) => {
+                return Err(PluginConnectionError::Timeout(
+                    self.plugin.lock().await.plugin_id.clone(),
+                    PluginRequestType::CommandList,
+                ));
+            }
+        };
+
+        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> {
+        // Generate request id
+        let request_id = {
+            let mut r = self.next_request_id.lock().await;
+            let id = *r;
+            *r += 1;
+            id
+        };
+
+        // Encode request
+        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();
+
+        // Add response listener
+        let (tx, mut rx) = mpsc::channel(4);
+        self.pending_requests
+            .lock()
+            .await
+            .insert(request_id, ResponseListener::Multishot(tx));
+
+        // Send request
+        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),
+            ));
+        }
+
+        // Receive responses
+        let (tx_out, rx_out) = mpsc::channel(4);
+        tokio::spawn(async move {
+            loop {
+                let response = match rx.recv().await {
+                    Some(response) => response,
+                    None => {
+                        log::error!("Response channel for request {request_id} closed unexpectedly");
+                        break;
+                    }
+                };
+
+                match response.res {
+                    Some(proto::response::Res::CmdReply(reply)) => {
+                        let end = match reply.reply {
+                            Some(proto::command_reply::Reply::End(_)) => true,
+                            _ => false,
+                        };
+                        log::trace!("reply sent: Ok({:?})", &reply);
+                        if let Err(e) = tx_out.send(Ok(reply)).await {
+                            log::error!("Cannot send command reply to another task: {e}");
+                        }
+                        if end {
+                            break;
+                        }
+                    }
+                    _ => {
+                        if let Err(e) = tx_out
+                            .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;
+                    }
+                }
+            }
+            // Remove response listener
+            self.pending_requests.lock().await.remove_entry(&request_id);
+        });
+
+        Ok(rx_out)
+    }
+}

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)
     }
@@ -36,11 +36,7 @@ pub(crate) async fn connect_ssh<A: ToSocketAddrs + Debug, K: AsRef<[u8]>, P: Int
         "Connecting to SSH {:?} {} key & {} password",
         &addr,
         if key.is_some() { "with" } else { "without" },
-        if password.is_some() {
+        if password.is_some() { "with" } else { "without" }
-            "with"
-        } else {
-            "without"
-        }
     );
     let mut connection = tokio::time::timeout(
         connect_timeout,
@@ -65,12 +61,7 @@ pub(crate) async fn connect_ssh<A: ToSocketAddrs + Debug, K: AsRef<[u8]>, P: Int
     if !auth_success {
         if let Some(key) = key {
             log::info!("Trying to authenticate using key");
-            match tokio::time::timeout(
+            match tokio::time::timeout(auth_timeout, connection.authenticate_publickey(username, key)).await {
-                auth_timeout,
-                connection.authenticate_publickey(username, key),
-            )
-            .await
-            {
                 Ok(Ok(AuthResult::Success)) => {
                     auth_success = true;
                 }
@@ -78,8 +69,7 @@ pub(crate) async fn connect_ssh<A: ToSocketAddrs + Debug, K: AsRef<[u8]>, P: Int
                     remaining_methods,
                     partial_success,
                 })) => {
-                    if partial_success || !remaining_methods.contains(&russh::MethodKind::Password)
+                    if partial_success || !remaining_methods.contains(&russh::MethodKind::Password) {
-                    {
                         if partial_success {
                             anyhow::bail!("SSH auth failed: multi-factor auth is not supported");
                         } else {
@@ -102,12 +92,7 @@ pub(crate) async fn connect_ssh<A: ToSocketAddrs + Debug, K: AsRef<[u8]>, P: Int
     if !auth_success {
         if let Some(password) = password {
             log::info!("Trying to authenticate using password");
-            match tokio::time::timeout(
+            match tokio::time::timeout(auth_timeout, connection.authenticate_password(username, password)).await {
-                auth_timeout,
-                connection.authenticate_password(username, password),
-            )
-            .await
-            {
                 Ok(Ok(AuthResult::Success)) => {
                     auth_success = true;
                 }