Merge branch 'master' into csp-fix

dapps/src/api/api.rs

@@ -18,23 +18,36 @@ use std::sync::Arc;
 
 use hyper::{server, net, Decoder, Encoder, Next, Control};
 use hyper::method::Method;
+use hyper::status::StatusCode;
 
-use api::types::ApiError;
-use api::response;
+use api::{response, types};
+use api::time::TimeChecker;
 use apps::fetcher::Fetcher;
-
-use handlers::extract_url;
+use handlers::{self, extract_url};
 use endpoint::{Endpoint, Handler, EndpointPath};
+use parity_reactor::Remote;
+use {SyncStatus};
 
 #[derive(Clone)]
 pub struct RestApi {
 	fetcher: Arc<Fetcher>,
+	sync_status: Arc<SyncStatus>,
+	time: TimeChecker,
+	remote: Remote,
 }
 
 impl RestApi {
-	pub fn new(fetcher: Arc<Fetcher>) -> Box<Endpoint> {
+	pub fn new(
+		fetcher: Arc<Fetcher>,
+		sync_status: Arc<SyncStatus>,
+		time: TimeChecker,
+		remote: Remote,
+	) -> Box<Endpoint> {
 		Box::new(RestApi {
-			fetcher: fetcher,
+			fetcher,
+			sync_status,
+			time,
+			remote,
 		})
 	}
 }
@@ -58,11 +71,11 @@ impl RestApiRouter {
 			path: Some(path),
 			control: Some(control),
 			api: api,
-			handler: response::as_json_error(&ApiError {
+			handler: Box::new(response::as_json_error(StatusCode::NotFound, &types::ApiError {
 				code: "404".into(),
 				title: "Not Found".into(),
 				detail: "Resource you requested has not been found.".into(),
-			}),
+			})),
 		}
 	}
 
@@ -75,6 +88,78 @@ impl RestApiRouter {
 			_ => None
 		}
 	}
+
+	fn health(&self, control: Control) -> Box<Handler> {
+		use self::types::{HealthInfo, HealthStatus, Health};
+
+		trace!(target: "dapps", "Checking node health.");
+		// Check timediff
+		let sync_status = self.api.sync_status.clone();
+		let map = move |time| {
+			// Check peers
+			let peers = {
+				let (connected, max) = sync_status.peers();
+				let (status, message) = match connected {
+					0 => {
+						(HealthStatus::Bad, "You are not connected to any peers. There is most likely some network issue. Fix connectivity.".into())
+					},
+					1 => (HealthStatus::NeedsAttention, "You are connected to only one peer. Your node might not be reliable. Check your network connection.".into()),
+					_ => (HealthStatus::Ok, "".into()),
+				};
+				HealthInfo { status, message, details: (connected, max) }
+			};
+
+			// Check sync
+			let sync = {
+				let is_syncing = sync_status.is_major_importing();
+				let (status, message) = if is_syncing {
+					(HealthStatus::NeedsAttention, "Your node is still syncing, the values you see might be outdated. Wait until it's fully synced.".into())
+				} else {
+					(HealthStatus::Ok, "".into())
+				};
+				HealthInfo { status, message, details: is_syncing }
+			};
+
+			// Check time
+			let time = {
+				const MAX_DRIFT: i64 = 500;
+				let (status, message, details) = match time {
+					Ok(Ok(diff)) if diff < MAX_DRIFT && diff > -MAX_DRIFT => {
+						(HealthStatus::Ok, "".into(), diff)
+					},
+					Ok(Ok(diff)) => {
+						(HealthStatus::Bad, format!(
+							"Your clock is not in sync. Detected difference is too big for the protocol to work: {}ms. Synchronize your clock.",
+							diff,
+						), diff)
+					},
+					Ok(Err(err)) => {
+						(HealthStatus::NeedsAttention, format!(
+							"Unable to reach time API: {}. Make sure that your clock is synchronized.",
+							err,
+						), 0)
+					},
+					Err(_) => {
+						(HealthStatus::NeedsAttention, "Time API request timed out. Make sure that the clock is synchronized.".into(), 0)
+					},
+				};
+
+				HealthInfo { status, message, details, }
+			};
+
+			let status = if [&peers.status, &sync.status, &time.status].iter().any(|x| *x != &HealthStatus::Ok) {
+				StatusCode::PreconditionFailed // HTTP 412
+			} else {
+				StatusCode::Ok // HTTP 200
+			};
+
+			response::as_json(status, &Health { peers, sync, time })
+		};
+
+		let time = self.api.time.time_drift();
+		let remote = self.api.remote.clone();
+		Box::new(handlers::AsyncHandler::new(time, map, remote, control))
+	}
 }
 
 impl server::Handler<net::HttpStream> for RestApiRouter {
@@ -103,6 +188,7 @@ impl server::Handler<net::HttpStream> for RestApiRouter {
 
 		let handler = endpoint.and_then(|v| match v {
 			"ping" => Some(response::ping()),
+			"health" => Some(self.health(control)),
 			"content" => self.resolve_content(hash, path, control),
 			_ => None
 		});

dapps/src/api/mod.rs

@@ -18,6 +18,8 @@
 
 mod api;
 mod response;
+mod time;
 mod types;
 
 pub use self::api::RestApi;
+pub use self::time::TimeChecker;

dapps/src/api/response.rs

@@ -16,6 +16,8 @@
 
 use serde::Serialize;
 use serde_json;
+use hyper::status::StatusCode;
+
 use endpoint::Handler;
 use handlers::{ContentHandler, EchoHandler};
 
@@ -23,10 +25,16 @@ pub fn empty() -> Box<Handler> {
 	Box::new(ContentHandler::ok("".into(), mime!(Text/Plain)))
 }
 
-pub fn as_json_error<T: Serialize>(val: &T) -> Box<Handler> {
+pub fn as_json<T: Serialize>(status: StatusCode, val: &T) -> ContentHandler {
 	let json = serde_json::to_string(val)
 		.expect("serialization to string is infallible; qed");
-	Box::new(ContentHandler::not_found(json, mime!(Application/Json)))
+	ContentHandler::new(status, json, mime!(Application/Json))
+}
+
+pub fn as_json_error<T: Serialize>(status: StatusCode, val: &T) -> ContentHandler {
+	let json = serde_json::to_string(val)
+		.expect("serialization to string is infallible; qed");
+	ContentHandler::new(status, json, mime!(Application/Json))
 }
 
 pub fn ping() -> Box<Handler> {

dapps/src/api/time.rs

@@ -0,0 +1,264 @@
+// Copyright 2015-2017 Parity Technologies (UK) Ltd.
+// This file is part of Parity.
+
+// Parity is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// Parity is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+
+// You should have received a copy of the GNU General Public License
+// along with Parity.  If not, see <http://www.gnu.org/licenses/>.
+
+//! Periodically checks node's time drift using [SNTP](https://tools.ietf.org/html/rfc1769).
+//!
+//! An NTP packet is sent to the server with a local timestamp, the server then completes the packet, yielding the
+//! following timestamps:
+//!
+//!    Timestamp Name          ID   When Generated
+//!   ------------------------------------------------------------
+//!    Originate Timestamp     T1   time request sent by client
+//!    Receive Timestamp       T2   time request received at server
+//!    Transmit Timestamp      T3   time reply sent by server
+//!    Destination Timestamp   T4   time reply received at client
+//!
+//! The drift is defined as:
+//!
+//! drift = ((T2 - T1) + (T3 - T4)) / 2.
+//!
+
+use std::io;
+use std::{fmt, mem, time};
+
+use std::collections::VecDeque;
+use futures::{self, Future, BoxFuture};
+use futures_cpupool::CpuPool;
+use ntp;
+use time::{Duration, Timespec};
+use util::{Arc, RwLock};
+
+/// Time checker error.
+#[derive(Debug, Clone, PartialEq)]
+pub enum Error {
+	/// There was an error when trying to reach the NTP server.
+	Ntp(String),
+	/// IO error when reading NTP response.
+	Io(String),
+}
+
+impl fmt::Display for Error {
+	fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
+		use self::Error::*;
+
+		match *self {
+			Ntp(ref err) => write!(fmt, "NTP error: {}", err),
+			Io(ref err) => write!(fmt, "Connection Error: {}", err),
+		}
+	}
+}
+
+impl From<io::Error> for Error {
+	fn from(err: io::Error) -> Self { Error::Io(format!("{}", err)) }
+}
+
+impl From<ntp::errors::Error> for Error {
+	fn from(err: ntp::errors::Error) -> Self { Error::Ntp(format!("{}", err)) }
+}
+
+/// NTP time drift checker.
+pub trait Ntp {
+	/// Returns the current time drift.
+	fn drift(&self) -> BoxFuture<Duration, Error>;
+}
+
+/// NTP client using the SNTP algorithm for calculating drift.
+#[derive(Clone)]
+pub struct SimpleNtp {
+	address: Arc<String>,
+	pool: CpuPool,
+}
+
+impl fmt::Debug for SimpleNtp {
+	fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+		write!(f, "Ntp {{ address: {} }}", self.address)
+	}
+}
+
+impl SimpleNtp {
+	fn new(address: &str, pool: CpuPool) -> SimpleNtp {
+		SimpleNtp {
+			address: Arc::new(address.to_owned()),
+			pool: pool,
+		}
+	}
+}
+
+impl Ntp for SimpleNtp {
+	fn drift(&self) -> BoxFuture<Duration, Error> {
+		let address = self.address.clone();
+		self.pool.spawn_fn(move || {
+			let packet = ntp::request(&*address)?;
+			let dest_time = ::time::now_utc().to_timespec();
+			let orig_time = Timespec::from(packet.orig_time);
+			let recv_time = Timespec::from(packet.recv_time);
+			let transmit_time = Timespec::from(packet.transmit_time);
+
+			let drift = ((recv_time - orig_time) + (transmit_time - dest_time)) / 2;
+
+			Ok(drift)
+		}).boxed()
+	}
+}
+
+const MAX_RESULTS: usize = 4;
+const UPDATE_TIMEOUT_OK_SECS: u64 = 30;
+const UPDATE_TIMEOUT_ERR_SECS: u64 = 2;
+
+#[derive(Debug, Clone)]
+/// A time checker.
+pub struct TimeChecker<N: Ntp = SimpleNtp> {
+	ntp: N,
+	last_result: Arc<RwLock<(time::Instant, VecDeque<Result<i64, Error>>)>>,
+}
+
+impl TimeChecker<SimpleNtp> {
+	/// Creates new time checker given the NTP server address.
+	pub fn new(ntp_address: String, pool: CpuPool) -> Self {
+		let last_result = Arc::new(RwLock::new(
+			// Assume everything is ok at the very beginning.
+			(time::Instant::now(), vec![Ok(0)].into())
+		));
+
+		let ntp = SimpleNtp::new(&ntp_address, pool);
+
+		TimeChecker {
+			ntp,
+			last_result,
+		}
+	}
+}
+
+impl<N: Ntp> TimeChecker<N> {
+	/// Updates the time
+	pub fn update(&self) -> BoxFuture<i64, Error> {
+		let last_result = self.last_result.clone();
+		self.ntp.drift().then(move |res| {
+			let mut results = mem::replace(&mut last_result.write().1, VecDeque::new());
+			let valid_till = time::Instant::now() + time::Duration::from_secs(
+				if res.is_ok() && results.len() == MAX_RESULTS {
+					UPDATE_TIMEOUT_OK_SECS
+				} else {
+					UPDATE_TIMEOUT_ERR_SECS
+				}
+			);
+
+			// Push the result.
+			results.push_back(res.map(|d| d.num_milliseconds()));
+			while results.len() > MAX_RESULTS {
+				results.pop_front();
+			}
+
+			// Select a response and update last result.
+			let res = select_result(results.iter());
+			*last_result.write() = (valid_till, results);
+			res
+		}).boxed()
+	}
+
+	/// Returns a current time drift or error if last request to NTP server failed.
+	pub fn time_drift(&self) -> BoxFuture<i64, Error> {
+		// return cached result
+		{
+			let res = self.last_result.read();
+			if res.0 > time::Instant::now() {
+				return futures::done(select_result(res.1.iter())).boxed();
+			}
+		}
+		// or update and return result
+		self.update()
+	}
+}
+
+fn select_result<'a, T: Iterator<Item=&'a Result<i64, Error>>>(results: T) -> Result<i64, Error> {
+	let mut min = None;
+	for res in results {
+		min = Some(match (min.take(), res) {
+			(Some(Ok(min)), &Ok(ref new)) => Ok(::std::cmp::min(min, *new)),
+			(Some(Ok(old)), &Err(_)) => Ok(old),
+			(_, ref new) => (*new).clone(),
+		})
+	}
+
+	min.unwrap_or_else(|| Err(Error::Ntp("NTP server unavailable.".into())))
+}
+
+#[cfg(test)]
+mod tests {
+	use std::sync::Arc;
+	use std::cell::{Cell, RefCell};
+	use std::time::Instant;
+	use time::Duration;
+	use futures::{self, BoxFuture, Future};
+	use super::{Ntp, TimeChecker, Error};
+	use util::RwLock;
+
+	#[derive(Clone)]
+	struct FakeNtp(RefCell<Vec<Duration>>, Cell<u64>);
+	impl FakeNtp {
+		fn new() -> FakeNtp {
+			FakeNtp(
+				RefCell::new(vec![Duration::milliseconds(150)]),
+				Cell::new(0))
+		}
+	}
+
+	impl Ntp for FakeNtp {
+		fn drift(&self) -> BoxFuture<Duration, Error> {
+			self.1.set(self.1.get() + 1);
+			futures::future::ok(self.0.borrow_mut().pop().expect("Unexpected call to drift().")).boxed()
+		}
+	}
+
+	fn time_checker() -> TimeChecker<FakeNtp> {
+		let last_result = Arc::new(RwLock::new(
+			(Instant::now(), vec![Err(Error::Ntp("NTP server unavailable.".into()))].into())
+		));
+
+		TimeChecker {
+			ntp: FakeNtp::new(),
+			last_result: last_result,
+		}
+	}
+
+	#[test]
+	fn should_fetch_time_on_start() {
+		// given
+		let time = time_checker();
+
+		// when
+		let diff = time.time_drift().wait().unwrap();
+
+		// then
+		assert_eq!(diff, 150);
+		assert_eq!(time.ntp.1.get(), 1);
+	}
+
+	#[test]
+	fn should_not_fetch_twice_if_timeout_has_not_passed() {
+		// given
+		let time = time_checker();
+
+		// when
+		let diff1 = time.time_drift().wait().unwrap();
+		let diff2 = time.time_drift().wait().unwrap();
+
+		// then
+		assert_eq!(diff1, 150);
+		assert_eq!(diff2, 150);
+		assert_eq!(time.ntp.1.get(), 1);
+	}
+}

dapps/src/api/types.rs

@@ -14,11 +14,54 @@
 // You should have received a copy of the GNU General Public License
 // along with Parity.  If not, see <http://www.gnu.org/licenses/>.
 
+/// A structure representing any error in REST API.
 #[derive(Debug, PartialEq, Serialize, Deserialize)]
 #[serde(deny_unknown_fields)]
 pub struct ApiError {
+	/// Error code.
 	pub code: String,
+	/// Human-readable error summary.
 	pub title: String,
+	/// More technical error details.
 	pub detail: String,
 }
 
+/// Health API endpoint status.
+#[derive(Debug, PartialEq, Serialize)]
+pub enum HealthStatus {
+	/// Everything's OK.
+	#[serde(rename = "ok")]
+	Ok,
+	/// Node health need attention
+	/// (the issue is not critical, but may need investigation)
+	#[serde(rename = "needsAttention")]
+	NeedsAttention,
+	/// There is something bad detected with the node.
+	#[serde(rename = "bad")]
+	Bad
+}
+
+/// Represents a single check in node health.
+/// Cointains the status of that check and apropriate message and details.
+#[derive(Debug, PartialEq, Serialize)]
+#[serde(deny_unknown_fields)]
+pub struct HealthInfo<T> {
+	/// Check status.
+	pub status: HealthStatus,
+	/// Human-readable message.
+	pub message: String,
+	/// Technical details of the check.
+	pub details: T,
+}
+
+/// Node Health status.
+#[derive(Debug, PartialEq, Serialize)]
+#[serde(deny_unknown_fields)]
+pub struct Health {
+	/// Status of peers.
+	pub peers: HealthInfo<(usize, usize)>,
+	/// Sync status.
+	pub sync: HealthInfo<bool>,
+	/// Time diff info.
+	pub time: HealthInfo<i64>,
+}

dapps/src/apps/fetcher/mod.rs

@@ -48,7 +48,7 @@ pub trait Fetcher: Send + Sync + 'static {
 }
 
 pub struct ContentFetcher<F: Fetch = FetchClient, R: URLHint + 'static = URLHintContract> {
-	dapps_path: PathBuf,
+	cache_path: PathBuf,
 	resolver: R,
 	cache: Arc<Mutex<ContentCache>>,
 	sync: Arc<SyncStatus>,
@@ -61,7 +61,7 @@ pub struct ContentFetcher<F: Fetch = FetchClient, R: URLHint + 'static = URLHint
 impl<R: URLHint + 'static, F: Fetch> Drop for ContentFetcher<F, R> {
 	fn drop(&mut self) {
 		// Clear cache path
-		let _ = fs::remove_dir_all(&self.dapps_path);
+		let _ = fs::remove_dir_all(&self.cache_path);
 	}
 }
 
@@ -73,11 +73,11 @@ impl<R: URLHint + 'static, F: Fetch> ContentFetcher<F, R> {
 		remote: Remote,
 		fetch: F,
 	) -> Self {
-		let mut dapps_path = env::temp_dir();
-		dapps_path.push(random_filename());
+		let mut cache_path = env::temp_dir();
+		cache_path.push(random_filename());
 
 		ContentFetcher {
-			dapps_path: dapps_path,
+			cache_path: cache_path,
 			resolver: resolver,
 			sync: sync_status,
 			cache: Arc::new(Mutex::new(ContentCache::default())),
@@ -200,7 +200,7 @@ impl<R: URLHint + 'static, F: Fetch> Fetcher for ContentFetcher<F, R> {
 								control,
 								installers::Dapp::new(
 									content_id.clone(),
-									self.dapps_path.clone(),
+									self.cache_path.clone(),
 									Box::new(on_done),
 									self.embeddable_on.clone(),
 								),
@@ -219,7 +219,7 @@ impl<R: URLHint + 'static, F: Fetch> Fetcher for ContentFetcher<F, R> {
 								installers::Content::new(
 									content_id.clone(),
 									content.mime,
-									self.dapps_path.clone(),
+									self.cache_path.clone(),
 									Box::new(on_done),
 								),
 								self.embeddable_on.clone(),
@@ -271,6 +271,7 @@ mod tests {
 	use endpoint::EndpointInfo;
 	use page::LocalPageEndpoint;
 	use super::{ContentFetcher, Fetcher};
+	use {SyncStatus};
 
 	#[derive(Clone)]
 	struct FakeResolver;
@@ -280,11 +281,17 @@ mod tests {
 		}
 	}
 
+	struct FakeSync(bool);
+	impl SyncStatus for FakeSync {
+		fn is_major_importing(&self) -> bool { self.0 }
+		fn peers(&self) -> (usize, usize) { (0, 5) }
+	}
+
 	#[test]
 	fn should_true_if_contains_the_app() {
 		// given
 		let path = env::temp_dir();
-		let fetcher = ContentFetcher::new(FakeResolver, Arc::new(|| false), Remote::new_sync(), Client::new().unwrap())
+		let fetcher = ContentFetcher::new(FakeResolver, Arc::new(FakeSync(false)), Remote::new_sync(), Client::new().unwrap())
 			.allow_dapps(true);
 		let handler = LocalPageEndpoint::new(path, EndpointInfo {
 			name: "fake".into(),

dapps/src/apps/mod.rs

@@ -26,7 +26,7 @@ use fetch::Fetch;
 use parity_dapps::WebApp;
 use parity_reactor::Remote;
 use parity_ui;
-use {WebProxyTokens, ParentFrameSettings, as_embeddable};
+use {WebProxyTokens, ParentFrameSettings};
 
 mod app;
 mod cache;
@@ -52,20 +52,19 @@ pub fn ui() -> Box<Endpoint> {
 	Box::new(PageEndpoint::with_fallback_to_index(parity_ui::App::default()))
 }
 
-pub fn ui_redirection(ui_address: Option<(String, u16)>, dapps_domain: String) -> Box<Endpoint> {
-	Box::new(ui::Redirection::new(as_embeddable(ui_address, dapps_domain)))
+pub fn ui_redirection(embeddable: Option<ParentFrameSettings>) -> Box<Endpoint> {
+	Box::new(ui::Redirection::new(embeddable))
 }
 
 pub fn all_endpoints<F: Fetch>(
 	dapps_path: PathBuf,
 	extra_dapps: Vec<PathBuf>,
-	dapps_domain: String,
-	ui_address: Option<(String, u16)>,
+	dapps_domain: &str,
+	embeddable: Option<ParentFrameSettings>,
 	web_proxy_tokens: Arc<WebProxyTokens>,
 	remote: Remote,
 	fetch: F,
 ) -> Endpoints {
-	let embeddable = as_embeddable(ui_address.clone(), dapps_domain.clone());
 	// fetch fs dapps at first to avoid overwriting builtins
 	let mut pages = fs::local_endpoints(dapps_path, embeddable.clone());
 	for path in extra_dapps {
@@ -78,7 +77,7 @@ pub fn all_endpoints<F: Fetch>(
 
 	// NOTE [ToDr] Dapps will be currently embeded on 8180
 	insert::<parity_ui::App>(&mut pages, "ui", Embeddable::Yes(embeddable.clone()));
-	pages.insert("proxy".into(), ProxyPac::boxed(ui_address.clone(), dapps_domain));
+	pages.insert("proxy".into(), ProxyPac::boxed(embeddable.clone(), dapps_domain.to_owned()));
 	pages.insert(WEB_PATH.into(), Web::boxed(embeddable.clone(), web_proxy_tokens.clone(), remote.clone(), fetch.clone()));
 
 	Arc::new(pages)

dapps/src/handlers/async.rs

@@ -0,0 +1,112 @@
+// Copyright 2015-2017 Parity Technologies (UK) Ltd.
+// This file is part of Parity.
+
+// Parity is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// Parity is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+
+// You should have received a copy of the GNU General Public License
+// along with Parity.  If not, see <http://www.gnu.org/licenses/>.
+
+//! Async Content Handler
+//! Temporary solution until we switch to future-based server.
+//! Wraps a future and converts it to hyper::server::Handler;
+
+use std::{mem, time};
+use std::sync::mpsc;
+use futures::Future;
+use hyper::{server, Decoder, Encoder, Next, Control};
+use hyper::net::HttpStream;
+
+use handlers::ContentHandler;
+use parity_reactor::Remote;
+
+const TIMEOUT_SECS: u64 = 15;
+
+enum State<F, T, M> {
+	Initial(F, M, Remote, Control),
+	Waiting(mpsc::Receiver<Result<T, ()>>, M),
+	Done(ContentHandler),
+	Invalid,
+}
+
+pub struct AsyncHandler<F, T, M> {
+	state: State<F, T, M>,
+}
+
+impl<F, T, M> AsyncHandler<F, T, M> {
+	pub fn new(future: F, map: M, remote: Remote, control: Control) -> Self {
+		AsyncHandler {
+			state: State::Initial(future, map, remote, control),
+		}
+	}
+}
+
+impl<F, T, E, M> server::Handler<HttpStream> for AsyncHandler<F, Result<T, E>, M> where
+	F: Future<Item=T, Error=E> + Send + 'static,
+	M: FnOnce(Result<Result<T, E>, ()>) -> ContentHandler,
+	T: Send + 'static,
+	E: Send + 'static,
+{
+	fn on_request(&mut self, _request: server::Request<HttpStream>) -> Next {
+		if let State::Initial(future, map, remote, control) = mem::replace(&mut self.state, State::Invalid) {
+			let (tx, rx) = mpsc::sync_channel(1);
+			let control2 = control.clone();
+			let tx2 = tx.clone();
+			remote.spawn_with_timeout(move || future.then(move |result| {
+				// Send a result (ignore errors if the connection was dropped)
+				let _ = tx.send(Ok(result));
+				// Resume handler
+				let _ = control.ready(Next::read());
+
+				Ok(())
+			}), time::Duration::from_secs(TIMEOUT_SECS), move || {
+				// Notify about error
+				let _ = tx2.send(Err(()));
+				// Resume handler
+				let _ = control2.ready(Next::read());
+			});
+
+			self.state = State::Waiting(rx, map);
+		}
+
+		Next::wait()
+	}
+
+	fn on_request_readable(&mut self, _decoder: &mut Decoder<HttpStream>) -> Next {
+		if let State::Waiting(rx, map) = mem::replace(&mut self.state, State::Invalid) {
+			match rx.try_recv() {
+				Ok(result) => {
+					self.state = State::Done(map(result));
+				},
+				Err(err) => {
+					warn!("Resuming handler in incorrect state: {:?}", err);
+				}
+			}
+		}
+
+		Next::write()
+	}
+
+	fn on_response(&mut self, res: &mut server::Response) -> Next {
+		if let State::Done(ref mut handler) = self.state {
+			handler.on_response(res)
+		} else {
+			Next::end()
+		}
+	}
+
+	fn on_response_writable(&mut self, encoder: &mut Encoder<HttpStream>) -> Next {
+		if let State::Done(ref mut handler) = self.state {
+			handler.on_response_writable(encoder)
+		} else {
+			Next::end()
+		}
+	}
+}

dapps/src/handlers/content.rs

@@ -40,10 +40,6 @@ impl ContentHandler {
 		Self::new(StatusCode::Ok, content, mimetype)
 	}
 
-	pub fn not_found(content: String, mimetype: Mime) -> Self {
-		Self::new(StatusCode::NotFound, content, mimetype)
-	}
-
 	pub fn html(code: StatusCode, content: String, embeddable_on: Embeddable) -> Self {
 		Self::new_embeddable(code, content, mime!(Text/Html), embeddable_on)
 	}

dapps/src/handlers/mod.rs

@@ -16,18 +16,23 @@
 
 //! Hyper handlers implementations.
 
+mod async;
 mod content;
 mod echo;
 mod fetch;
 mod redirect;
 mod streaming;
 
+pub use self::async::AsyncHandler;
 pub use self::content::ContentHandler;
 pub use self::echo::EchoHandler;
 pub use self::fetch::{ContentFetcherHandler, ContentValidator, FetchControl, ValidatorResponse};
 pub use self::redirect::Redirection;
 pub use self::streaming::StreamingHandler;
 
+use std::iter;
+use util::Itertools;
+
 use url::Url;
 use hyper::{server, header, net, uri};
 use {apps, address, Embeddable};
@@ -78,21 +83,21 @@ pub fn add_security_headers(headers: &mut header::Headers, embeddable_on: Embedd
 				let proxy = format!("{}.{}", apps::HOME_PAGE, embed.dapps_domain);
 				let domain = format!("*.{}:{}", embed.dapps_domain, embed.port);
 
-				if embed.host == "127.0.0.1" {
+				let mut ancestors = vec![std, domain, proxy]
+					.into_iter()
+					.chain(embed.extra_embed_on
+						.iter()
+						.map(|&(ref host, port)| format!("{}:{}", host, port))
+					);
+
+				let ancestors = if embed.host == "127.0.0.1" {
 					let localhost = address("localhost", embed.port);
-					format!("frame-ancestors {} {} {} {};",
-						std,
-						localhost,
-						domain,
-						proxy,
-					)
+					ancestors.chain(iter::once(localhost)).join(" ")
 				} else {
-					format!("frame-ancestors {} {} {};",
-						std,
-						domain,
-						proxy,
-					)
-				}
+					ancestors.join(" ")
+				};
+
+				format!("frame-ancestors {};", ancestors)
 			},
 			None => format!("frame-ancestors 'self';"),
 		}.into_bytes(),

dapps/src/lib.rs

@@ -21,8 +21,10 @@
 
 extern crate base32;
 extern crate futures;
+extern crate futures_cpupool;
 extern crate linked_hash_map;
 extern crate mime_guess;
+extern crate ntp;
 extern crate rand;
 extern crate rustc_hex;
 extern crate serde;
@@ -74,6 +76,7 @@ use std::collections::HashMap;
 use jsonrpc_http_server::{self as http, hyper, Origin};
 
 use fetch::Fetch;
+use futures_cpupool::CpuPool;
 use parity_reactor::Remote;
 
 pub use hash_fetch::urlhint::ContractClient;
@@ -82,10 +85,9 @@ pub use hash_fetch::urlhint::ContractClient;
 pub trait SyncStatus: Send + Sync {
 	/// Returns true if there is a major sync happening.
 	fn is_major_importing(&self) -> bool;
-}
 
-impl<F> SyncStatus for F where F: Fn() -> bool + Send + Sync {
-	fn is_major_importing(&self) -> bool { self() }
+	/// Returns number of connected and ideal peers.
+	fn peers(&self) -> (usize, usize);
 }
 
 /// Validates Web Proxy tokens
@@ -127,21 +129,29 @@ impl Middleware {
 	}
 
 	/// Creates new middleware for UI server.
-	pub fn ui<F: Fetch + Clone>(
+	pub fn ui<F: Fetch>(
+		ntp_server: &str,
+		pool: CpuPool,
 		remote: Remote,
+		dapps_domain: &str,
 		registrar: Arc<ContractClient>,
 		sync_status: Arc<SyncStatus>,
 		fetch: F,
-		dapps_domain: String,
 	) -> Self {
 		let content_fetcher = Arc::new(apps::fetcher::ContentFetcher::new(
 			hash_fetch::urlhint::URLHintContract::new(registrar),
-			sync_status,
+			sync_status.clone(),
 			remote.clone(),
 			fetch.clone(),
 		).embeddable_on(None).allow_dapps(false));
 		let special = {
-			let mut special = special_endpoints(content_fetcher.clone());
+			let mut special = special_endpoints(
+				ntp_server,
+				pool,
+				content_fetcher.clone(),
+				remote.clone(),
+				sync_status.clone(),
+			);
 			special.insert(router::SpecialEndpoint::Home, Some(apps::ui()));
 			special
 		};
@@ -150,7 +160,7 @@ impl Middleware {
 			None,
 			special,
 			None,
-			dapps_domain,
+			dapps_domain.to_owned(),
 		);
 
 		Middleware {
@@ -160,39 +170,48 @@ impl Middleware {
 	}
 
 	/// Creates new Dapps server middleware.
-	pub fn dapps<F: Fetch + Clone>(
+	pub fn dapps<F: Fetch>(
+		ntp_server: &str,
+		pool: CpuPool,
 		remote: Remote,
 		ui_address: Option<(String, u16)>,
+		extra_embed_on: Vec<(String, u16)>,
 		dapps_path: PathBuf,
 		extra_dapps: Vec<PathBuf>,
-		dapps_domain: String,
+		dapps_domain: &str,
 		registrar: Arc<ContractClient>,
 		sync_status: Arc<SyncStatus>,
 		web_proxy_tokens: Arc<WebProxyTokens>,
 		fetch: F,
 	) -> Self {
-		let embeddable = as_embeddable(ui_address.clone(), dapps_domain.clone());
+		let embeddable = as_embeddable(ui_address, extra_embed_on, dapps_domain);
 		let content_fetcher = Arc::new(apps::fetcher::ContentFetcher::new(
 			hash_fetch::urlhint::URLHintContract::new(registrar),
-			sync_status,
+			sync_status.clone(),
 			remote.clone(),
 			fetch.clone(),
 		).embeddable_on(embeddable.clone()).allow_dapps(true));
 		let endpoints = apps::all_endpoints(
 			dapps_path,
 			extra_dapps,
-			dapps_domain.clone(),
-			ui_address.clone(),
+			dapps_domain,
+			embeddable.clone(),
 			web_proxy_tokens,
 			remote.clone(),
 			fetch.clone(),
 		);
 
 		let special = {
-			let mut special = special_endpoints(content_fetcher.clone());
+			let mut special = special_endpoints(
+				ntp_server,
+				pool,
+				content_fetcher.clone(),
+				remote.clone(),
+				sync_status,
+			);
 			special.insert(
 				router::SpecialEndpoint::Home,
-				Some(apps::ui_redirection(ui_address.clone(), dapps_domain.clone())),
+				Some(apps::ui_redirection(embeddable.clone())),
 			);
 			special
 		};
@@ -202,7 +221,7 @@ impl Middleware {
 			Some(endpoints.clone()),
 			special,
 			embeddable,
-			dapps_domain,
+			dapps_domain.to_owned(),
 		);
 
 		Middleware {
@@ -218,11 +237,22 @@ impl http::RequestMiddleware for Middleware {
 	}
 }
 
-fn special_endpoints(content_fetcher: Arc<apps::fetcher::Fetcher>) -> HashMap<router::SpecialEndpoint, Option<Box<endpoint::Endpoint>>> {
+fn special_endpoints(
+	ntp_server: &str,
+	pool: CpuPool,
+	content_fetcher: Arc<apps::fetcher::Fetcher>,
+	remote: Remote,
+	sync_status: Arc<SyncStatus>,
+) -> HashMap<router::SpecialEndpoint, Option<Box<endpoint::Endpoint>>> {
 	let mut special = HashMap::new();
 	special.insert(router::SpecialEndpoint::Rpc, None);
 	special.insert(router::SpecialEndpoint::Utils, Some(apps::utils()));
-	special.insert(router::SpecialEndpoint::Api, Some(api::RestApi::new(content_fetcher)));
+	special.insert(router::SpecialEndpoint::Api, Some(api::RestApi::new(
+		content_fetcher,
+		sync_status,
+		api::TimeChecker::new(ntp_server.into(), pool),
+		remote,
+	)));
 	special
 }
 
@@ -230,8 +260,17 @@ fn address(host: &str, port: u16) -> String {
 	format!("{}:{}", host, port)
 }
 
-fn as_embeddable(ui_address: Option<(String, u16)>, dapps_domain: String) -> Option<ParentFrameSettings> {
-	ui_address.map(|(host, port)| ParentFrameSettings { host, port, dapps_domain, })
+fn as_embeddable(
+	ui_address: Option<(String, u16)>,
+	extra_embed_on: Vec<(String, u16)>,
+	dapps_domain: &str,
+) -> Option<ParentFrameSettings> {
+	ui_address.map(|(host, port)| ParentFrameSettings {
+		host,
+		port,
+		extra_embed_on,
+		dapps_domain: dapps_domain.to_owned(),
+	})
 }
 
 /// Random filename
@@ -250,6 +289,8 @@ pub struct ParentFrameSettings {
 	pub host: String,
 	/// Port
 	pub port: u16,
+	/// Additional pages the pages can be embedded on.
+	pub extra_embed_on: Vec<(String, u16)>,
 	/// Dapps Domain (web3.site)
 	pub dapps_domain: String,
 }

dapps/src/proxypac.rs

@@ -19,27 +19,24 @@
 use endpoint::{Endpoint, Handler, EndpointPath};
 use handlers::ContentHandler;
 use apps::HOME_PAGE;
-use address;
+use {address, Embeddable};
 
 pub struct ProxyPac {
-	signer_address: Option<(String, u16)>,
+	embeddable: Embeddable,
 	dapps_domain: String,
 }
 
 impl ProxyPac {
-	pub fn boxed(signer_address: Option<(String, u16)>, dapps_domain: String) -> Box<Endpoint> {
-		Box::new(ProxyPac {
-			signer_address: signer_address,
-			dapps_domain: dapps_domain,
-		})
+	pub fn boxed(embeddable: Embeddable, dapps_domain: String) -> Box<Endpoint> {
+		Box::new(ProxyPac { embeddable, dapps_domain })
 	}
 }
 
 impl Endpoint for ProxyPac {
 	fn to_handler(&self, path: EndpointPath) -> Box<Handler> {
-		let signer = self.signer_address
+		let ui = self.embeddable
 			.as_ref()
-			.map(|&(ref host, port)| address(host, port))
+			.map(|ref parent| address(&parent.host, parent.port))
 			.unwrap_or_else(|| format!("{}:{}", path.host, path.port));
 
 		let content = format!(
@@ -58,7 +55,7 @@ function FindProxyForURL(url, host) {{
 	return "DIRECT";
 }}
 "#,
-		HOME_PAGE, self.dapps_domain, path.host, path.port, signer);
+		HOME_PAGE, self.dapps_domain, path.host, path.port, ui);
 
 		Box::new(ContentHandler::ok(content, mime!(Application/Javascript)))
 	}

dapps/src/tests/helpers/mod.rs

@@ -26,6 +26,7 @@ use jsonrpc_http_server::{self as http, Host, DomainsValidation};
 use devtools::http_client;
 use hash_fetch::urlhint::ContractClient;
 use fetch::{Fetch, Client as FetchClient};
+use futures_cpupool::CpuPool;
 use parity_reactor::Remote;
 
 use {Middleware, SyncStatus, WebProxyTokens};
@@ -38,6 +39,12 @@ use self::fetch::FakeFetch;
 
 const SIGNER_PORT: u16 = 18180;
 
+struct FakeSync(bool);
+impl SyncStatus for FakeSync {
+	fn is_major_importing(&self) -> bool { self.0 }
+	fn peers(&self) -> (usize, usize) { (0, 5) }
+}
+
 fn init_logger() {
 	// Initialize logger
 	if let Ok(log) = env::var("RUST_LOG") {
@@ -82,7 +89,7 @@ pub fn serve_with_registrar() -> (Server, Arc<FakeRegistrar>) {
 
 pub fn serve_with_registrar_and_sync() -> (Server, Arc<FakeRegistrar>) {
 	init_server(|builder| {
-		builder.sync_status(Arc::new(|| true))
+		builder.sync_status(Arc::new(FakeSync(true)))
 	}, Default::default(), Remote::new_sync())
 }
 
@@ -148,7 +155,7 @@ impl ServerBuilder {
 		ServerBuilder {
 			dapps_path: dapps_path.as_ref().to_owned(),
 			registrar: registrar,
-			sync_status: Arc::new(|| false),
+			sync_status: Arc::new(FakeSync(false)),
 			web_proxy_tokens: Arc::new(|_| None),
 			signer_address: None,
 			allowed_hosts: DomainsValidation::Disabled,
@@ -248,8 +255,11 @@ impl Server {
 		fetch: F,
 	) -> Result<Server, http::Error> {
 		let middleware = Middleware::dapps(
+			"pool.ntp.org:123",
+			CpuPool::new(4),
 			remote,
 			signer_address,
+			vec![],
 			dapps_path,
 			extra_dapps,
 			DAPPS_DOMAIN.into(),
@@ -290,4 +300,3 @@ impl Drop for Server {
 		self.server.take().unwrap().close()
 	}
 }
-