appmgr: add support for actions

appmgr/Cargo.lock

@@ -33,6 +33,12 @@ dependencies = [
  "winapi 0.3.9",
 ]
 
+[[package]]
+name = "anyhow"
+version = "1.0.38"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "afddf7f520a80dbf76e6f50a35bca42a2331ef227a28b3b6dc5c2e2338d114b1"
+
 [[package]]
 name = "appmgr"
 version = "0.2.9"
@@ -68,6 +74,7 @@ dependencies = [
  "tokio 0.3.5",
  "tokio-compat-02",
  "tokio-tar",
+ "yajrc",
 ]
 
 [[package]]
@@ -2264,6 +2271,26 @@ dependencies = [
  "unicode-width",
 ]
 
+[[package]]
+name = "thiserror"
+version = "1.0.23"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "76cc616c6abf8c8928e2fdcc0dbfab37175edd8fb49a4641066ad1364fdab146"
+dependencies = [
+ "thiserror-impl",
+]
+
+[[package]]
+name = "thiserror-impl"
+version = "1.0.23"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9be73a2caec27583d0046ef3796c3794f868a5bc813db689eed00c7631275cd1"
+dependencies = [
+ "proc-macro2 1.0.24",
+ "quote 1.0.7",
+ "syn",
+]
+
 [[package]]
 name = "thread-id"
 version = "3.3.0"
@@ -2748,6 +2775,16 @@ dependencies = [
  "libc",
 ]
 
+[[package]]
+name = "yajrc"
+version = "0.1.0"
+dependencies = [
+ "anyhow",
+ "serde",
+ "serde_json",
+ "thiserror",
+]
+
 [[package]]
 name = "yaml-rust"
 version = "0.4.4"

appmgr/Cargo.toml

@@ -49,3 +49,4 @@ simple-logging = "2.0"
 tokio = { version = "0.3.5", features = ["full"] }
 tokio-compat-02 = "0.1.2"
 tokio-tar = { version = "0.3.0", git = "https://github.com/dr-bonez/tokio-tar.git", rev = "1ba710f3" }
+yajrc = { path = "../../yajrc" }

appmgr/src/actions.rs

@@ -0,0 +1,105 @@
+use std::os::unix::process::ExitStatusExt;
+use std::process::Stdio;
+
+use linear_map::set::LinearSet;
+use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, Error as IoError};
+use yajrc::RpcError;
+
+use crate::apps::DockerStatus;
+
+pub const STATUS_NOT_ALLOWED: i32 = -2;
+
+#[derive(Debug, Clone, serde::Deserialize, serde::Serialize)]
+#[serde(rename_all = "kebab-case")]
+pub struct Action {
+    pub id: String,
+    pub name: String,
+    pub description: String,
+    pub allowed_statuses: LinearSet<DockerStatus>,
+    pub command: Vec<String>,
+}
+
+async fn tee<R: AsyncRead + Unpin, W: AsyncWrite + Unpin>(
+    mut r: R,
+    mut w: W,
+) -> Result<Vec<u8>, IoError> {
+    let mut res = Vec::new();
+    let mut buf = vec![0; 2048];
+    let mut bytes;
+    while {
+        bytes = r.read(&mut buf).await?;
+        bytes != 0
+    } {
+        res.extend_from_slice(&buf[..bytes]);
+        w.write_all(&buf[..bytes]).await?;
+    }
+    w.flush().await?;
+    Ok(res)
+}
+
+impl Action {
+    pub async fn perform(&self, app_id: &str) -> Result<String, RpcError> {
+        let man = crate::apps::manifest(app_id)
+            .await
+            .map_err(failure::Error::from)
+            .map_err(failure::Error::compat)?;
+        let status = crate::apps::status(app_id, true)
+            .await
+            .map_err(failure::Error::from)
+            .map_err(failure::Error::compat)?
+            .status;
+        if !self.allowed_statuses.contains(&status) {
+            return Err(RpcError {
+                code: STATUS_NOT_ALLOWED,
+                message: format!(
+                    "{} is in status {:?} which is not allowed by {}",
+                    app_id, status, self.id
+                ),
+                data: None,
+            });
+        }
+        let mut cmd = if status == DockerStatus::Running {
+            let mut cmd = tokio::process::Command::new("docker");
+            cmd.arg("exec").arg(&app_id).args(&self.command);
+            cmd
+        } else {
+            let mut cmd = tokio::process::Command::new("docker");
+            cmd.arg("run")
+                .arg("--rm")
+                .arg("--name")
+                .arg(format!("{}_{}", app_id, self.id))
+                .arg("--mount")
+                .arg(format!(
+                    "type=bind,src={}/{},dst={}",
+                    crate::VOLUMES,
+                    app_id,
+                    man.mount.display()
+                ))
+                .arg(format!("start9/{}", app_id))
+                .args(&self.command);
+            // TODO: 0.3.0: net, tor, shm
+            cmd
+        };
+        cmd.stdout(Stdio::piped());
+        cmd.stderr(Stdio::piped());
+        let mut child = cmd.spawn()?;
+
+        let (stdout, stderr) = futures::try_join!(
+            tee(child.stdout.take().unwrap(), tokio::io::sink()),
+            tee(child.stderr.take().unwrap(), tokio::io::sink())
+        )?;
+
+        let status = child.wait().await?;
+        if status.success() {
+            String::from_utf8(stdout).map_err(From::from)
+        } else {
+            Err(RpcError {
+                code: status
+                    .code()
+                    .unwrap_or_else(|| status.signal().unwrap_or(0) + 128),
+                message: String::from_utf8(stderr)?,
+                data: None,
+            })
+        }
+    }
+}

appmgr/src/lib.rs

@@ -20,6 +20,7 @@ lazy_static::lazy_static! {
     pub static ref QUIET: tokio::sync::RwLock<bool> = tokio::sync::RwLock::new(!std::env::var("APPMGR_QUIET").map(|a| a == "0").unwrap_or(true));
 }
 
+pub mod actions;
 pub mod apps;
 pub mod backup;
 pub mod config;

appmgr/src/main.rs

@@ -820,6 +820,16 @@ async fn inner_main() -> Result<(), Error> {
         )
         .subcommand(
             SubCommand::with_name("repair-app-status").about("Restarts crashed apps"), // TODO: remove
+        )
+        .subcommand(
+            SubCommand::with_name("actions")
+                .about("Perform an action for a service")
+                .arg(
+                    Arg::with_name("SERVICE")
+                        .help("ID of the service to perform an action on")
+                        .required(true),
+                )
+                .arg(Arg::with_name("ACTION").help("ID of the action to perform")),
         );
 
     let matches = app.clone().get_matches();
@@ -1547,6 +1557,34 @@ async fn inner_main() -> Result<(), Error> {
         ("repair-app-status", _) => {
             control::repair_app_status().await?;
         }
+        #[cfg(not(feature = "portable"))]
+        ("actions", Some(sub_m)) => {
+            use yajrc::{GenericRpcMethod, RpcResponse};
+
+            let man = apps::manifest(sub_m.value_of("SERVICE").unwrap()).await?;
+            let action_id = sub_m.value_of("ACTION").unwrap();
+            println!(
+                "{}",
+                serde_json::to_string(&RpcResponse::<GenericRpcMethod>::from_result(
+                    man.actions
+                        .iter()
+                        .filter(|a| &a.id == &action_id)
+                        .next()
+                        .ok_or_else(|| {
+                            failure::format_err!(
+                                "action {} does not exist for {}",
+                                action_id,
+                                man.id
+                            )
+                        })
+                        .with_code(error::NOT_FOUND)?
+                        .perform(&man.id)
+                        .await
+                        .map(serde_json::Value::String)
+                ))
+                .with_code(error::SERDE_ERROR)?
+            )
+        }
         ("pack", Some(sub_m)) => {
             pack(
                 sub_m.value_of("PATH").unwrap(),

appmgr/src/manifest.rs

@@ -2,6 +2,7 @@ use std::path::PathBuf;
 
 use linear_map::LinearMap;
 
+use crate::actions::Action;
 use crate::dependencies::Dependencies;
 use crate::tor::HiddenServiceVersion;
 use crate::tor::PortMapping;
@@ -63,6 +64,8 @@ pub struct ManifestV0 {
     pub hidden_service_version: HiddenServiceVersion,
     #[serde(default)]
     pub dependencies: Dependencies,
+    #[serde(default)]
+    pub actions: Vec<Action>,
     #[serde(flatten)]
     pub extra: LinearMap<String, serde_yaml::Value>,
 }