bui-backend 0.2.0

//! Implements a web server for a Browser User Interface (BUI).
use {serde_json, std, futures, jsonwebtoken};
#[cfg(feature = "bundle_files")]
use includedir;
use hyper;

use hyper::{Get, Post, StatusCode, mime};
use hyper::server::{Request, Response};
use hyper::header::{Accept, ContentLength};
use uuid::Uuid;

use futures::{Future, Stream, Sink};
use futures::sync::mpsc;

use std::sync::{Arc, Mutex};

#[cfg(feature = "serve_files")]
use std::io::Read;

// ---------------------------

/// Alias for `Uuid` indicating that sessions are tracked
/// by keys of this type (one per client browser).
pub type SessionKeyType = Uuid;

/// The claims validated using JSON Web Tokens.
#[derive(Serialize, Deserialize, Debug, Clone)]
struct JwtClaims {
    key: SessionKeyType,
}

/// Callback data from a connected client.
#[derive(Clone, Debug)]
pub struct CallbackDataAndSession {
    /// The name of the callback sent from the client.
    pub name: String,
    /// The arguments of the callback sent from the client.
    pub args: serde_json::Value,
    /// The session key associated with the client.
    pub session_key: SessionKeyType,
}

#[derive(Clone, Serialize, Deserialize, Debug)]
struct WireCallbackData {
    name: String,
    args: serde_json::Value,
}

/// Configuration settings for `BuiService`.
///
/// Defaults can be loaded using the function `get_default_config()`
/// generated by the `bui_backend_codegen` crate.
#[derive(Clone)]
pub struct Config {
    /// Location of the files to be served.
    pub serve_filepath: &'static std::path::Path,
    /// The bundled files.
    #[cfg(feature = "bundle_files")]
    pub bundled_files: &'static includedir::Files,
    /// The number of messages in the event stream channel before blocking.
    pub channel_size: usize,
    /// The name of the cookie stored in the clients browser.
    pub cookie_name: String,
}

/// Wrapper around `hyper::Chunk` to enable sending data to clients.
pub type EventChunkSender = mpsc::Sender<std::result::Result<hyper::Chunk, hyper::Error>>;

/// Wrap the sender to each connected event stream listener.
pub struct NewEventStreamConnection {
    /// A sink for messages send to each connection (one per client tab).
    pub chunk_sender: EventChunkSender,
    /// Identifier for each connected session (one per client browser).
    pub session_key: SessionKeyType,
    /// Identifier for each connection (one per client tab).
    pub connection_key: ConnectionKeyType,
    /// The path being requested (starts with `BuiService::events_prefix`).
    pub path: String,
}

type NewConnectionSender = mpsc::Sender<NewEventStreamConnection>;

/// Alias for `usize` to identify each connected event stream listener (one per client tab).
pub type ConnectionKeyType = usize;

/// Handle HTTP requests and coordinate responses to data updates.
///
/// Implements `hyper::server::Service` to act as HTTP server and handle requests.
#[derive(Clone)]
pub struct BuiService {
    config: Config,
    callback_senders: Arc<Mutex<Vec<mpsc::Sender<CallbackDataAndSession>>>>,
    next_connection_key: Arc<Mutex<ConnectionKeyType>>,
    jwt_secret: Arc<Mutex<Vec<u8>>>,
    tx_new_connection: NewConnectionSender,
    events_prefix: String,
}

impl BuiService {
    fn fullpath(&self, path: &str) -> String {
        assert!(path.starts_with("/")); // security check
        let path = std::path::PathBuf::from(path)
            .strip_prefix("/")
            .unwrap()
            .to_path_buf();
        assert!(!path.starts_with("..")); // security check

        let base = std::path::PathBuf::from(self.config.serve_filepath);
        let result = base.join(path);
        result.into_os_string().into_string().unwrap()
    }

    #[cfg(feature = "bundle_files")]
    fn get_file_content(&self, file_path: &str) -> Option<Vec<u8>> {
        let fullpath = self.fullpath(file_path);
        let r = self.config.bundled_files.get(&fullpath);
        match r {
            Ok(s) => Some(s.into_owned()),
            Err(_) => None,
        }
    }

    #[cfg(feature = "serve_files")]
    fn get_file_content(&self, file_path: &str) -> Option<Vec<u8>> {
        let fullpath = self.fullpath(file_path);
        let mut file = match std::fs::File::open(&fullpath) {
            Ok(f) => f,
            Err(e) => {
                error!("requested path {:?}, but got error {:?}", file_path, e);
                return None;
            }
        };
        let mut contents = Vec::new();
        match file.read_to_end(&mut contents) {
            Ok(_) => {}
            Err(e) => {
                error!("when reading path {:?}, got error {:?}", file_path, e);
                return None;
            }
        }
        Some(contents)
    }

    /// Get the event stream path prefix.
    pub fn events_prefix(&self) -> &str {
        &self.events_prefix
    }

    fn get_next_connection_key(&self) -> ConnectionKeyType {
        let mut nk = self.next_connection_key.lock().unwrap();
        let result = *nk;
        *nk += 1;
        result
    }

    /// Get a stream of callback events.
    pub fn add_callback_listener(&mut self,
                                 channel_size: usize)
                                 -> mpsc::Receiver<CallbackDataAndSession> {
        let (tx, rx) = mpsc::channel(channel_size);
        {
            let mut cb_tx_vec = self.callback_senders.lock().unwrap();
            cb_tx_vec.push(tx);
        }
        rx
    }
}

fn into_bytes(body: hyper::Body) -> Box<Future<Item = Vec<u8>, Error = hyper::Error>> {
    Box::new(body.fold(vec![], |mut buf, chunk| {
        buf.extend_from_slice(&*chunk);
        futures::future::ok::<_, hyper::Error>(buf)
    }))
}

fn get_client_key(headers: &hyper::Headers,
                  cookie_name: &str,
                  jwt_secret: &[u8])
                  -> Option<SessionKeyType> {
    if let Some(ref cookie) = headers.get::<hyper::header::Cookie>() {
        match cookie.get(&cookie_name) {
            Some(k) => {
                match jsonwebtoken::decode::<JwtClaims>(&k,
                                                        jwt_secret,
                                                        &jsonwebtoken::Validation::default())
                              .map(|token| token.claims.key) {
                    Ok(k) => Some(k),
                    Err(e) => {
                        warn!("client passed token {:?}, resulting in error: {:?}", k, e);
                        None
                    }
                }
            }
            None => None,
        }
    } else {
        None
    }
}

impl hyper::server::Service for BuiService {
    type Request = Request;
    type Response = Response;
    type Error = hyper::Error;
    type Future = Box<Future<Item = Self::Response, Error = Self::Error>>;

    fn call(&self, req: Request) -> Self::Future {

        // Parse cookies.
        let opt_client_key = {
            let jwt_secret = self.jwt_secret.lock().unwrap();
            get_client_key(&req.headers(), &self.config.cookie_name, &*jwt_secret)
        };

        trace!("got request from key {:?}: {:?}", opt_client_key, req);

        if req.method() == &Post {
            if req.path() == "/callback" {
                let session_key = if let Some(session_key) = opt_client_key {
                    session_key
                } else {
                    error!("no client key in callback");
                    return Box::new(futures::future::ok(Response::new()
                        .with_header( hyper::header::ContentType::plaintext() )
                        .with_status(StatusCode::BadRequest)));
                };

                let bytes_future = into_bytes(req.body()).map_err(|e| e.into());
                let cbsenders = self.callback_senders.clone();

                let resp_future = bytes_future
                    .and_then(move |data| -> futures::future::FutureResult<_, hyper::Error> {
                        let resp = match serde_json::from_slice::<WireCallbackData>(&data) {
                            Ok(data) => {
                                {

                                    let mut cb_tx_vec = cbsenders.lock().unwrap();
                                    let mut restore_tx = Vec::new();

                                    let cmd_name = data.name.clone();
                                    let args = CallbackDataAndSession {
                                        name: data.name,
                                        args: data.args,
                                        session_key: session_key,
                                    };
                                    for tx in cb_tx_vec.drain(..) {
                                        // TODO can we somehow do this without waiting?
                                        match tx.send(args.clone()).wait() {
                                            Ok(t) => restore_tx.push(t),
                                            Err(e) => {
                                                // listener failed
                                                warn!("when sending callback {:?}, error: {:?}",
                                                      cmd_name,
                                                      e);
                                            }
                                        };
                                    }

                                    for tx in restore_tx.into_iter() {
                                        cb_tx_vec.push(tx);
                                    }

                                }

                                Response::new().with_header(hyper::header::ContentType::plaintext())
                            }
                            Err(e) => {
                                error!("Failed parsing JSON to WireCallbackData: {:?}", e);
                                Response::new()
                                    .with_header(hyper::header::ContentType::plaintext())
                                    .with_status(StatusCode::BadRequest)
                            }
                        };
                        futures::future::ok(resp)
                    });
                return Box::new(resp_future);
            }
        }

        let (session_key, resp) = if let Some(key) = opt_client_key {
            (key, Response::<hyper::Body>::new())
        } else {
            // There was no valid client key in the HTTP header, so generate a
            // new one and set it on client.
            let session_key = Uuid::new_v4();
            let claims = JwtClaims { key: session_key.clone() };

            let token = {
                let jwt_secret = self.jwt_secret.lock().unwrap();
                jsonwebtoken::encode(&jsonwebtoken::Header::default(), &claims, &*jwt_secret)
                    .unwrap()
            };
            let cookie = format!("{}={}", self.config.cookie_name, token);
            (session_key,
             Response::<hyper::Body>::new().with_header(hyper::header::SetCookie(vec![cookie])))
        };

        let resp_final = match (req.method(), req.path()) {
            (&Get, path) => {

                let path = if path == "/" { "/index.html" } else { path };

                if path.starts_with(&self.events_prefix) {
                    match req.headers().get::<Accept>() {
                        Some(accept_headers) => {
                            let (_, is_eventsource) = accept_headers
                                .as_slice()
                                .iter()
                                .fold((hyper::header::q(0), false), |prev, quality_item| {
                                    let (mut best_qual, mut is_eventsource) = prev;
                                    let this_quality = quality_item.quality;
                                    if this_quality > best_qual {
                                        best_qual = this_quality;
                                        let (ref top_level, ref sub_level) =
                                            (quality_item.item.type_(),
                                             quality_item.item.subtype());
                                        is_eventsource = top_level == &mime::TEXT &&
                                                         sub_level == &mime::EVENT_STREAM;
                                    }
                                    (best_qual, is_eventsource)
                                });
                            if !is_eventsource {
                                warn!("HTTP GET for \"{}\" does not list text/event-stream \
                                      in accepted header",
                                      self.events_prefix);
                            }

                            let connection_key = self.get_next_connection_key();
                            let (tx_event_stream, rx_event_stream) =
                                mpsc::channel(self.config.channel_size);

                            {
                                let tx_new_conn = self.tx_new_connection.clone();
                                let conn_info = NewEventStreamConnection {
                                    chunk_sender: tx_event_stream,
                                    session_key: session_key,
                                    connection_key: connection_key,
                                    path: path.to_string(),
                                };

                                match tx_new_conn.send(conn_info).wait() {
                                    Ok(_tx) => {} // Cloned above, so don't need to keep _tx here.
                                    Err(e) => {
                                        error!("failed to send new connection info: {:?}", e);
                                        // should we panic here?
                                    }
                                };
                            }

                            resp.with_header(hyper::header::ContentType(mime::TEXT_EVENT_STREAM))
                                .with_body(rx_event_stream)
                        }
                        None => {
                            error!("Event request does specify accept header");
                            resp.with_status(StatusCode::BadRequest)
                        }
                    }
                } else {
                    match self.get_file_content(path) {
                        Some(buf) => {
                            resp.with_header(ContentLength(buf.len() as u64))
                                .with_body(buf)
                        }
                        None => resp.with_status(StatusCode::NotFound),
                    }
                }
            }
            _ => resp.with_status(StatusCode::NotFound),
        };
        Box::new(futures::future::ok(resp_final))
    }
}

/// Implement the `hyper::server::NewService` trait to return a `BuiService`.
pub struct NewBuiService {
    value: Box<Fn() -> std::result::Result<BuiService, std::io::Error> + Send + Sync>,
}

impl NewBuiService {
    /// Return a NewBuiService for the given boxed function.
    pub fn new(value: Box<Fn() -> std::result::Result<BuiService, std::io::Error> + Send + Sync>)
               -> NewBuiService {
        Self { value }
    }
}

impl hyper::server::NewService for NewBuiService {
    type Request = Request;
    type Response = Response;
    type Error = hyper::Error;
    type Instance = BuiService;

    fn new_service(&self) -> std::result::Result<Self::Instance, std::io::Error> {
        (self.value)()
    }
}

/// Create a stream of connection events and a `BuiService`.
pub fn launcher(config: Config,
                jwt_secret: &[u8],
                channel_size: usize,
                events_prefix: &str)
                -> (mpsc::Receiver<NewEventStreamConnection>, BuiService) {
    let next_connection_key = Arc::new(Mutex::new(0));

    let callback_senders = Arc::new(Mutex::new(Vec::new()));

    let jwt_secret = jwt_secret.to_vec();
    let (tx_new_connection, rx_new_connection) = mpsc::channel(channel_size);

    let service = BuiService {
        config: config,
        callback_senders: callback_senders,
        next_connection_key: next_connection_key,
        jwt_secret: Arc::new(Mutex::new(jwt_secret.clone())),
        tx_new_connection: tx_new_connection,
        events_prefix: events_prefix.to_string(),
    };

    (rx_new_connection, service)
}