zzz

Cargo.toml

@@ -1,28 +1,33 @@
 [package]
 name = "dip"
 description = "Configurable webhook server."
-version = "0.1.4"
-authors = ["Michael Zhang <failed.down@gmail.com>"]
+version = "0.2.0"
+authors = ["Michael Zhang <mail@mzhang.io>"]
+edition = "2018"
+
+[features]
+github = []
+default = ["github"]
 
 [dependencies]
-failure = "0.1"
-futures = "0.1"
-generic-array = "0.9"
-hmac = "0.6"
-hyper = "0.12"
-mktemp = "0.3"
-lazy_static = "1.1"
-notify = "4.0"
-owning_ref = "0.3"
-regex = "1.0"
-secstr = "0.3"
-serde = "1.0"
-serde_derive = "1.0"
-serde_json = "1.0"
-sha-1 = "0.7"
-structopt = "0.2"
-tokio = "0.1"
-tokio-process = "0.2"
-toml = "0.4"
-walkdir = "2.2"
+anyhow = "1.0.34"
+futures = "0.3.7"
+generic-array = "0.14.4"
+hmac = "0.10.1"
+hyper = { version = "0.13", features = ["stream"] }
+lazy_static = "1.4.0"
+mktemp = "0.4.0"
+notify = "4.0.15"
+owning_ref = "0.4.1"
+parking_lot = "0.11.0"
+regex = "1.4.2"
+secstr = "0.4.0"
+serde = "1.0.117"
+serde_derive = "1.0.117"
+serde_json = "1.0.59"
+sha-1 = "0.9.1"
+structopt = "0.3.20"
+tokio = { version = "0.3", features = ["full"] }
+toml = "0.5.7"
+walkdir = "2.3.1"
 

src/builtins/github.rs

@@ -0,0 +1,13 @@
+use serde_json::Value;
+use anyhow::Result;
+
+use super::Builtin;
+
+#[derive(Default)]
+pub struct Github;
+
+impl Builtin for Github {
+    fn handle(&self, env: (), config: (), input: ()) -> Result<Value> {
+        todo!()
+    }
+}

src/builtins/mod.rs

@@ -0,0 +1,10 @@
+mod github;
+
+use serde_json::Value;
+use anyhow::Result;
+
+pub use self::github::Github;
+
+pub trait Builtin {
+    fn handle(&self, env: (), config: (), input: ()) -> Result<Value>;
+}

src/config.rs

@@ -6,7 +6,7 @@ use std::path::{Path, PathBuf};
 use std::sync::mpsc;
 use std::time::Duration;
 
-use failure::{err_msg, Error};
+use anyhow::Result;
 use notify::{self, RecommendedWatcher, RecursiveMode, Watcher};
 use walkdir::WalkDir;
 
@@ -19,10 +19,12 @@ pub struct Config {
     /// The root configuration directory for dip. This argument is required.
     #[structopt(short = "d", long = "root", parse(from_os_str))]
     pub root: PathBuf,
+
     /// A string containing the address to bind to. This defaults to
     /// "0.0.0.0:5000".
     #[structopt(short = "b", long = "bind", default_value = "0.0.0.0:5000")]
     pub bind: String,
+
     /// If a hook is specified here, it will be triggered manually exactly once
     /// and then the program will exit rather than running as a server.
     #[structopt(short = "h", long = "hook")]
@@ -69,13 +71,7 @@ where
     println!("Reloading config...");
     // hold on to the lock while config is being reloaded
     {
-        let mut programs = match PROGRAMS.lock() {
-            Ok(programs) => programs,
-            Err(err) => {
-                eprintln!("Could not acquire programs lock: {}", err);
-                return;
-            }
-        };
+        let mut programs = PROGRAMS.lock();
         // TODO: some kind of smart diff
         programs.clear();
 
@@ -96,7 +92,7 @@ where
                 match path
                     .file_name()
                     .and_then(|s| s.to_str())
-                    .ok_or(err_msg("???"))
+                    .ok_or(anyhow!("???"))
                     .map(|s| {
                         let filename = s.to_owned();
                         programs.insert(filename, path.to_path_buf())
@@ -106,14 +102,9 @@ where
             }
         }
     }
+
     {
-        let mut hooks = match HOOKS.lock() {
-            Ok(hooks) => hooks,
-            Err(err) => {
-                eprintln!("Could not acquire hooks lock: {}", err);
-                return;
-            }
-        };
+        let mut hooks = HOOKS.lock();
         hooks.clear();
         let hooks_dir = {
             let mut p = root.as_ref().to_path_buf();
@@ -129,7 +120,7 @@ where
                 if !path.is_file() {
                     continue;
                 }
-                match (|path: &Path| -> Result<(), Error> {
+                match (|path: &Path| -> Result<()> {
                     let hook = Hook::from_file(path)?;
                     let name = hook.get_name();
                     hooks.insert(name, hook);

src/github.rs

@@ -3,9 +3,9 @@ use std::iter::FromIterator;
 use std::path::PathBuf;
 use std::process::Command;
 
-use failure::{err_msg, Error};
+use anyhow::Result;
 use generic_array::GenericArray;
-use hmac::{Hmac, Mac};
+use hmac::{Hmac, Mac, NewMac};
 use secstr::*;
 use serde::Serialize;
 use serde_json::{self, Serializer as JsonSerializer, Value as JsonValue};
@@ -55,7 +55,7 @@ pub(crate) fn main(
     env: &Environment,
     config: &TomlValue,
     input: &JsonValue,
-) -> Result<JsonValue, Error> {
+) -> Result<JsonValue> {
     let config_str = {
         let mut buf: Vec<u8> = Vec::new();
         {
@@ -72,10 +72,10 @@ pub(crate) fn main(
     if !config.disable_hmac_verify {
         let secret = GenericArray::from_iter(config.secret.bytes());
         let mut mac = Hmac::<Sha1>::new(&secret);
-        mac.input(payload.body.as_bytes());
+        mac.update(payload.body.as_bytes());
         let signature = mac
-            .result()
-            .code()
+            .finalize()
+            .into_bytes()
             .into_iter()
             .map(|b| format!("{:02x}", b))
             .collect::<Vec<_>>()
@@ -84,7 +84,7 @@ pub(crate) fn main(
         let auth = payload
             .headers
             .get("x-hub-signature")
-            .ok_or(err_msg("Missing auth header"))
+            .ok_or(anyhow!("Missing auth header"))
             .expect("Missing auth header");
 
         let left = SecStr::from(format!("sha1={}", signature));

src/handler.rs

@@ -1,36 +1,31 @@
 use std::fmt;
 use std::path::PathBuf;
-use std::process::{Command, Stdio};
+use std::sync::Arc;
+use std::process::Stdio;
 
-use failure::{err_msg, Error};
-use futures::{
-    future::{self, Either},
-    Future,
-};
+use anyhow::Result;
 use serde::Serialize;
 use serde_json::{Serializer as JsonSerializer, Value as JsonValue};
-use tokio::io::write_all;
-use tokio_process::CommandExt;
+use tokio::{io::AsyncWriteExt, process::Command};
 use toml::Value as TomlValue;
 
-use crate::github;
+use crate::builtins::{Builtin, Github};
 
 /// A single instance of handler as defined by the config.
-#[derive(Clone, Debug)]
 pub struct Handler {
     pub(crate) config: TomlValue,
     pub(crate) action: Action,
 }
 
 /// Describes an action that a hook can take.
-#[derive(Clone)]
 pub enum Action {
     /// A builtin function (for example, the Github handler).
-    Builtin(
-        fn(&Environment, &TomlValue, &JsonValue) -> Result<JsonValue, Error>,
-    ),
+    Builtin(Box<dyn Builtin + Send>),
+    // Fn(&Environment, &TomlValue, &JsonValue) -> Result<JsonValue>>),
+
     /// A command represents a string to be executed by `bash -c`.
     Command(String),
+
     /// A program represents one of the handlers specified in the `handlers`
     /// directory.
     Program(String),
@@ -53,22 +48,23 @@ impl fmt::Debug for Action {
 }
 
 impl Handler {
-    pub(crate) fn from(config: &TomlValue) -> Result<Self, Error> {
+    pub(crate) fn from(config: &TomlValue) -> Result<Self> {
         let handler = config
             .get("type")
-            .ok_or(err_msg("No 'type' found."))?
+            .ok_or(anyhow!("No 'type' found."))?
             .as_str()
-            .ok_or(err_msg("'type' is not a string."))?;
+            .ok_or(anyhow!("'type' is not a string."))?;
+
         let action = match handler {
             "command" => {
                 let command = config
                     .get("command")
-                    .ok_or(err_msg("No 'command' found"))?
+                    .ok_or(anyhow!("No 'command' found"))?
                     .as_str()
-                    .ok_or(err_msg("'command' is not a string."))?;
+                    .ok_or(anyhow!("'command' is not a string."))?;
                 Action::Command(command.to_owned())
             }
-            "github" => Action::Builtin(github::main),
+            "github" => Action::Builtin(Box::new(Github::default())),
             handler => Action::Program(handler.to_owned()),
         };
         let config = config.clone();
@@ -76,12 +72,12 @@ impl Handler {
     }
 
     /// Runs the given [action](Action) and produces a [Future](Future).
-    pub fn run(
+    pub async fn run(
         config: TomlValue,
         action: Action,
         temp_path: PathBuf,
         input: JsonValue,
-    ) -> impl Future<Item = (PathBuf, JsonValue), Error = Error> {
+    ) -> Result<(PathBuf, JsonValue)> {
         let temp_path_cp = temp_path.clone();
         let config_str = {
             let mut buf: Vec<u8> = Vec::new();
@@ -102,72 +98,57 @@ impl Handler {
                 .stderr(Stdio::piped());
         };
 
-        let output: Box<dyn Future<Item = JsonValue, Error = Error> + Send> =
-            match action {
-                Action::Builtin(ref func) => {
+        let output = match action {
+            Action::Builtin(ref builtin) => {
                 let workdir = temp_path_cp.clone();
                 let env = Environment { workdir };
-                    let result = func(&env, &config, &input);
-                    Box::new(future::result(result))
+                builtin.handle(&env, &config, &input)
             }
             Action::Command(ref cmd) => {
                 // TODO: allow some kind of simple variable replacement
                 let mut command = Command::new("/bin/bash");
                 command_helper(&mut command);
                 let child = command.arg("-c").arg(cmd);
-                    let result = child
-                        .output_async()
-                        .map_err(|err| {
-                            err_msg(format!("failed to spawn child: {}", err))
-                        })
-                        .and_then(|output| {
-                            let stdout = String::from_utf8(output.stdout)
-                                .unwrap_or_else(|_| String::new());
-                            let stderr = String::from_utf8(output.stderr)
-                                .unwrap_or_else(|_| String::new());
-                            future::ok(json!({
+                let output = child.output().await?;
+
+                let stdout = String::from_utf8(output.stdout)?;
+                let stderr = String::from_utf8(output.stderr)?;
+
+                Ok(json!({
                     "stdout": stdout,
                     "stderr": stderr,
                 }))
-                        });
-                    Box::new(result)
             }
             Action::Program(ref path) => {
                 let mut command = Command::new(&path);
                 command_helper(&mut command);
-                    let mut child = command
-                        .arg("--config")
-                        .arg(config_str)
-                        .spawn_async()
-                        .expect("could not spawn child");
-
-                    let stdin = child.stdin().take().unwrap();
+                let mut child =
+                    command.arg("--config").arg(config_str).spawn()?;
+                let mut stdin = child.stdin.take().unwrap();
 
                 let input = format!("{}", input);
-                    let result = write_all(stdin, input)
-                        .and_then(|_| child.wait_with_output())
-                        .map_err(|err| err_msg(format!("error: {}", err)))
-                        .and_then(|output| {
-                            let stdout = String::from_utf8(output.stdout)
-                                .unwrap_or_else(|_| String::new());
-                            let stderr = String::from_utf8(output.stderr)
-                                .unwrap_or_else(|_| String::new());
+                stdin.write_all(input.as_bytes()).await?;
+
+                let output = child.wait_with_output().await?;
+
+                let stdout = String::from_utf8(output.stdout)?;
+                let stderr = String::from_utf8(output.stderr)?;
+
                 if output.status.success() {
-                                Either::A(future::ok(json!({
+                    Ok(json!({
                         "stdout": stdout,
                         "stderr": stderr,
-                                })))
+                    }))
                 } else {
-                                Either::B(future::err(err_msg(format!(
+                    Err(anyhow!(
                         "Failed, stdout: '{}', stderr: '{}'",
-                                    stdout, stderr
-                                ))))
+                        stdout,
+                        stderr
+                    ))
                 }
-                        });
-
-                    Box::new(result)
             }
         };
+
         output.map(|x| (temp_path_cp, x))
     }
 }

src/hook.rs

@@ -4,10 +4,8 @@ use std::fs::File;
 use std::io::Read;
 use std::path::{Path, PathBuf};
 
-use failure::{err_msg, Error};
-use futures::{stream, Future};
+use anyhow::Result;
 use serde_json::Value as JsonValue;
-use tokio::{self, prelude::*};
 use toml::Value;
 
 use crate::Handler;
@@ -20,16 +18,13 @@ pub struct Hook {
 
 impl Hook {
     /// Creates a hook from a (name, config) pair.
-    pub fn from(
-        name: impl Into<String>,
-        config: &Value,
-    ) -> Result<Self, Error> {
+    pub fn from(name: impl Into<String>, config: &Value) -> Result<Self> {
         let name = name.into();
         let handlers = config
             .get("handlers")
-            .ok_or(err_msg("No 'handlers' found."))?
+            .ok_or(anyhow!("No 'handlers' found."))?
             .as_array()
-            .ok_or(err_msg("'handlers' is not an array."))?
+            .ok_or(anyhow!("'handlers' is not an array."))?
             .iter()
             .map(|value: &Value| Handler::from(value))
             .collect::<Result<Vec<_>, _>>()?;
@@ -37,16 +32,16 @@ impl Hook {
     }
 
     /// Creates a hook from a configuration file.
-    pub fn from_file<P>(path: P) -> Result<Hook, Error>
+    pub fn from_file<P>(path: P) -> Result<Hook>
     where
         P: AsRef<Path>,
     {
         let filename = path
             .as_ref()
             .file_name()
-            .ok_or(err_msg("what the fuck bro"))?
+            .ok_or(anyhow!("what the fuck bro"))?
             .to_str()
-            .ok_or(err_msg("???"))?
+            .ok_or(anyhow!("???"))?
             .to_owned();
         let mut file = File::open(path)?;
         let mut contents = String::new();
@@ -62,25 +57,25 @@ impl Hook {
         self.name.clone()
     }
 
-    pub(crate) fn handle(
+    pub(crate) async fn handle(
         &self,
         req: JsonValue,
         temp_path: PathBuf,
-    ) -> Result<String, String> {
+    ) -> Result<String> {
         let handlers = self
             .handlers
             .iter()
-            .map(|handler| (handler.config.clone(), handler.action.clone()))
+            .map(|handler| (handler.config, handler.action))
             .collect::<Vec<_>>();
-        let st = stream::iter_ok::<_, Error>(handlers.into_iter())
-            .fold((temp_path, req), |(path, prev), (config, action)| {
-                Handler::run(config, action, path, prev)
-            })
-            .map(|_| ())
-            .map_err(|err: Error| {
-                println!("Error from stream: {}", err);
-            });
-        tokio::executor::spawn(st);
+
+        let mut path = temp_path;
+        let mut prev = req;
+        for (config, action) in handlers {
+            let result = Handler::run(config, action, path, prev).await?;
+            path = result.0;
+            prev = result.1;
+        }
+
         Ok("success".to_owned())
     }
 }

src/lib.rs

@@ -6,31 +6,18 @@ The configurable webhook server. Latest stable binary releases for Linux are ava
 
 #![deny(missing_docs)]
 
-extern crate hmac;
-extern crate secstr;
-extern crate sha1;
+#[macro_use]
+extern crate anyhow;
 #[macro_use]
 extern crate serde_derive;
-extern crate failure;
-extern crate futures;
-extern crate generic_array;
-extern crate hyper;
-extern crate mktemp;
-extern crate owning_ref;
-extern crate serde;
-extern crate tokio;
-extern crate tokio_process;
 #[macro_use]
 extern crate serde_json;
 #[macro_use]
 extern crate lazy_static;
-extern crate notify;
 #[macro_use]
 extern crate structopt;
-extern crate regex;
-extern crate toml;
-extern crate walkdir;
 
+mod builtins;
 pub mod config;
 mod github;
 mod handler;
@@ -41,13 +28,15 @@ use std::collections::HashMap;
 use std::net::SocketAddrV4;
 use std::path::PathBuf;
 use std::str::FromStr;
-use std::sync::{Arc, Mutex};
+use std::sync::Arc;
 use std::thread;
 
-use failure::Error;
-use hyper::rt::Future;
-use hyper::service::service_fn;
-use hyper::Server;
+use anyhow::{Error, Result};
+use hyper::{
+    service::{make_service_fn, service_fn},
+    Server,
+};
+use parking_lot::Mutex;
 use regex::Regex;
 
 pub use crate::config::Config;
@@ -67,17 +56,23 @@ lazy_static! {
 }
 
 /// Main entry point of the entire application.
-pub fn run(config: &Config) -> Result<(), Error> {
+pub async fn run(config: &Config) -> Result<()> {
     config::load_config(&config.root);
 
     let v = config.root.clone();
     thread::spawn(|| config::watch(v));
 
     let addr: SocketAddrV4 = SocketAddrV4::from_str(config.bind.as_ref())?;
-    let server = Server::bind(&addr.into())
-        .serve(|| service_fn(dip_service))
-        .map_err(|e| eprintln!("server error: {}", e));
+    let make_svc = make_service_fn(|conn| async {
+        Ok::<_, Error>(service_fn(dip_service))
+    });
+
+    let server = Server::bind(&addr.into()).serve(make_svc);
     println!("Listening on {:?}", addr);
-    hyper::rt::run(server);
+
+    if let Err(e) = server.await {
+        eprintln!("server error: {}", e);
+    }
+
     Ok(())
 }

src/main.rs

@@ -1,12 +1,9 @@
-extern crate dip;
-extern crate failure;
-extern crate structopt;
-
+use anyhow::Result;
 use dip::Config;
-use failure::Error;
 use structopt::StructOpt;
 
-fn main() -> Result<(), Error> {
+#[tokio::main]
+async fn main() -> Result<()> {
     let config = Config::from_args();
     dip::run(&config)
 }

src/service.rs

@@ -1,35 +1,34 @@
 use std::collections::HashMap;
 
-use futures::{future, Future, Stream};
-use hyper::{Body, Error, Request, Response, StatusCode};
+use anyhow::Result;
+use futures::stream::TryStreamExt;
+use hyper::{
+    body::{to_bytes, Body},
+    Request, Response, StatusCode,
+};
 use mktemp::Temp;
 
 use crate::{HOOKS, URIPATTERN};
 
-pub(crate) fn dip_service(
-    req: Request<Body>,
-) -> Box<dyn Future<Item = Response<Body>, Error = Error> + Send> {
+pub(crate) async fn dip_service(req: Request<Body>) -> Result<Response<Body>> {
     let path = req.uri().path().to_owned();
     let captures = match URIPATTERN.captures(path.as_ref()) {
         Some(value) => value,
         None => {
-            return Box::new(future::ok(
-                Response::builder()
+            return Ok(Response::builder()
                 .status(StatusCode::NOT_FOUND)
                 .body(Body::from("not found"))
-                    .unwrap(),
-            ))
+                .unwrap());
         }
     };
+
     let name = match captures.name("name") {
         Some(value) => value.as_str().to_owned(),
         None => {
-            return Box::new(future::ok(
-                Response::builder()
+            return Ok(Response::builder()
                 .status(StatusCode::NOT_FOUND)
                 .body(Body::from("not found"))
-                    .unwrap(),
-            ))
+                .unwrap());
         }
     };
 
@@ -47,43 +46,45 @@ pub(crate) fn dip_service(
     let method = req.method().as_str().to_owned();
 
     // spawn job
-    Box::new(req.into_body().concat2().map(move |body| {
+    let body = to_bytes(req.into_body()).await?;
+
     let body = String::from_utf8(body.to_vec()).unwrap();
     let req_obj = json!({
         "body": body,
         "headers": headers,
         "method": method,
     });
-        let hooks = HOOKS.lock().unwrap();
-        {
+
     let mut temp_dir = Temp::new_dir().unwrap();
     let temp_path = temp_dir.to_path_buf();
     assert!(temp_path.exists());
 
+    let (code, msg) = {
+        let hooks = HOOKS.lock();
+
         let hook = match hooks.get(&name) {
             Some(hook) => hook,
             None => {
                 temp_dir.release();
-                    return Response::builder()
+                return Ok(Response::builder()
                     .status(StatusCode::NOT_FOUND)
                     .body(Body::from("not found"))
                     .unwrap_or_else(|_| {
                         Response::new(Body::from("not found"))
-                        });
+                    }));
             }
         };
-            let (code, msg) = match hook.handle(req_obj, temp_path) {
+
+        match hook.handle(req_obj, temp_path).await {
             Ok(msg) => (StatusCode::ACCEPTED, msg),
-                Err(msg) => (StatusCode::BAD_REQUEST, msg),
+            Err(msg) => (StatusCode::BAD_REQUEST, msg.to_string()),
+        }
     };
 
     temp_dir.release();
-            Response::builder()
+
+    Ok(Response::builder()
         .status(code)
         .body(Body::from(msg))
-                .unwrap_or_else(|err| {
-                    Response::new(Body::from(format!("{}", err)))
-                })
-        }
-    }))
+        .unwrap_or_else(|err| Response::new(Body::from(format!("{}", err)))))
 }