Initial Whisper implementation (#6009)

* whisper skeleton

* basic message store

* rallying and message logic

* pass host info to network protocol handlers

* choose who starts rally based on node key

* module reshuffling

* mining messages

* prune messages by low PoW until below size target

* associated error type for ethkey generators and `OsRng` generator

* beginnings of RPC

* generic message handler for whisper

* reshuffle code order

* standard payload encoding and decoding

* basic crypto

* minor restructuring of net code

* implement shh_post

* merge?

* implement filters

* rand trait for hash types

* filter RPCs for whisper

* symmetric encryption of payload

* pub-sub

* filter tests

* use only secure random IDs

* attach arbitrary protocols to network

* basic integration of whisper into Parity

* eagerly prune low PoW entries

* broadcast messages with salted topics

* node info RPC

* fix import

* fix leading zeros calculation

* address minor grumbles

whisper/Cargo.toml

@@ -0,0 +1,32 @@
+[package]
+name = "parity-whisper"
+version = "0.1.0"
+authors = ["Parity Technologies <admin@parity.io>"]
+description = "Whisper Protocol implementation for Parity"
+
+[dependencies]
+bitflags = "0.9"
+byteorder = "1.0.0"
+ethcore-bigint = { path = "../util/bigint" }
+ethcore-network = { path = "../util/network" }
+ethcrypto = { path = "../ethcrypto" }
+ethkey = { path = "../ethkey" }
+futures = "0.1"
+hex = "0.2"
+log = "0.3"
+ordered-float = "0.5"
+parking_lot = "0.4"
+rand = "0.3"
+ring = "0.9.5"
+rlp = { path = "../util/rlp" }
+serde = "1.0"
+serde_derive = "1.0"
+serde_json = "1.0"
+slab = "0.3"
+smallvec = "0.4"
+time = "0.1"
+tiny-keccak = "1.2.1"
+
+jsonrpc-core = { git = "https://github.com/paritytech/jsonrpc.git", branch = "parity-1.7" }
+jsonrpc-macros = { git = "https://github.com/paritytech/jsonrpc.git", branch = "parity-1.7" }
+jsonrpc-pubsub = { git = "https://github.com/paritytech/jsonrpc.git", branch = "parity-1.7" }

whisper/README.md

@@ -0,0 +1,3 @@
+# Whisper
+
+Implementation of Whisper based on the Whisper-v2 PoC.

whisper/src/lib.rs

@@ -0,0 +1,59 @@
+// 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/>.
+
+//! Whisper P2P messaging system as a DevP2P subprotocol, with RPC and Rust
+//! interface.
+
+extern crate byteorder;
+extern crate ethcore_bigint as bigint;
+extern crate ethcore_network as network;
+extern crate ethcrypto;
+extern crate ethkey;
+extern crate futures;
+extern crate hex;
+extern crate ordered_float;
+extern crate parking_lot;
+extern crate rand;
+extern crate rlp;
+extern crate ring;
+extern crate serde;
+extern crate serde_json;
+extern crate slab;
+extern crate smallvec;
+extern crate time;
+extern crate tiny_keccak;
+
+extern crate jsonrpc_core;
+extern crate jsonrpc_pubsub;
+
+#[macro_use]
+extern crate bitflags;
+
+#[macro_use]
+extern crate log;
+
+#[macro_use]
+extern crate jsonrpc_macros;
+
+#[macro_use]
+extern crate serde_derive;
+
+pub use self::message::Message;
+pub use self::net::{Network, MessageHandler};
+
+pub mod message;
+pub mod net;
+pub mod rpc;

whisper/src/message.rs

@@ -0,0 +1,479 @@
+// 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/>.
+
+//! Whisper message parsing, handlers, and construction.
+
+use std::fmt;
+use std::time::{self, SystemTime, Duration};
+
+use bigint::hash::{H256, H512};
+use rlp::{self, DecoderError, RlpStream, UntrustedRlp};
+use smallvec::SmallVec;
+use tiny_keccak::{keccak256, Keccak};
+
+/// Work-factor proved. Takes 3 parameters: size of message, time to live,
+/// and hash.
+///
+/// Panics if size or TTL is zero.
+pub fn work_factor_proved(size: u64, ttl: u64, hash: H256) -> f64 {
+	assert!(size != 0 && ttl != 0);
+
+	let leading_zeros = {
+		let leading_zeros = hash.iter().take_while(|&&x| x == 0).count();
+		(leading_zeros * 8) + hash.get(leading_zeros + 1).map_or(0, |b| b.leading_zeros() as usize)
+	};
+	let spacetime = size as f64 * ttl as f64;
+
+	(1u64 << leading_zeros) as f64 / spacetime
+}
+
+/// A topic of a message.
+#[derive(Debug, Default, Clone, Copy, Hash, PartialEq, Eq, PartialOrd, Ord)]
+pub struct Topic(pub [u8; 4]);
+
+impl From<[u8; 4]> for Topic {
+	fn from(x: [u8; 4]) -> Self {
+		Topic(x)
+	}
+}
+
+impl Topic {
+	/// set up to three bits in the 64-byte bloom passed.
+	///
+	/// this takes 3 sets of 9 bits, treating each as an index in the range
+	/// 0..512 into the bloom and setting the corresponding bit in the bloom to 1.
+	pub fn bloom_into(&self, bloom: &mut H512) {
+		let mut set_bit = |idx: usize| {
+			let idx = idx & 511;
+			bloom[idx / 8] |= 1 << idx % 8;
+		};
+
+		let data = &self.0;
+		let mut combined = ((data[0] as usize) << 24) |
+			((data[1] as usize) << 16) |
+			((data[2] as usize) << 8) |
+			data[3] as usize;
+
+		// take off the last 5 bits as we only use 27.
+		combined >>= 5;
+
+		set_bit(combined);
+		set_bit(combined >> 9);
+		set_bit(combined >> 18);
+	}
+
+	/// Get bloom for single topic.
+	pub fn bloom(&self) -> H512 {
+		let mut bloom = Default::default();
+		self.bloom_into(&mut bloom);
+		bloom
+	}
+}
+
+impl rlp::Encodable for Topic {
+	fn rlp_append(&self, s: &mut RlpStream) {
+		s.encoder().encode_value(&self.0);
+	}
+}
+
+impl rlp::Decodable for Topic {
+	fn decode(rlp: &UntrustedRlp) -> Result<Self, DecoderError> {
+		use std::cmp;
+
+		rlp.decoder().decode_value(|bytes| match bytes.len().cmp(&4) {
+			cmp::Ordering::Less => Err(DecoderError::RlpIsTooShort),
+			cmp::Ordering::Greater => Err(DecoderError::RlpIsTooBig),
+			cmp::Ordering::Equal => {
+				let mut t = [0u8; 4];
+				t.copy_from_slice(bytes);
+				Ok(Topic(t))
+			}
+		})
+	}
+}
+
+/// Calculate union of blooms for given topics.
+pub fn bloom_topics(topics: &[Topic]) -> H512 {
+	let mut bloom = H512::default();
+	for topic in topics {
+		topic.bloom_into(&mut bloom);
+	}
+	bloom
+}
+
+/// Message errors.
+#[derive(Debug)]
+pub enum Error {
+	Decoder(DecoderError),
+	LivesTooLong,
+	IssuedInFuture,
+	ZeroTTL,
+}
+
+impl From<DecoderError> for Error {
+	fn from(err: DecoderError) -> Self {
+		Error::Decoder(err)
+	}
+}
+
+impl fmt::Display for Error {
+	fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+		match *self {
+			Error::Decoder(ref err) => write!(f, "Failed to decode message: {}", err),
+			Error::LivesTooLong => write!(f, "Message claims to be issued before the unix epoch."),
+			Error::IssuedInFuture => write!(f, "Message issued in future."),
+			Error::ZeroTTL => write!(f, "Message live for zero time."),
+		}
+	}
+}
+
+// Raw envelope struct.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct Envelope {
+	/// Expiry timestamp
+	pub expiry: u64,
+	/// Time-to-live in seconds
+	pub ttl: u64,
+	/// series of 4-byte topics.
+	pub topics: SmallVec<[Topic; 4]>,
+	/// The message contained within.
+	pub data: Vec<u8>,
+	/// Arbitrary value used to target lower PoW hash.
+	pub nonce: u64,
+}
+
+impl Envelope {
+	fn proving_hash(&self) -> H256 {
+		use byteorder::{BigEndian, ByteOrder};
+
+		let mut buf = [0; 32];
+
+		let mut stream = RlpStream::new_list(4);
+		stream.append(&self.expiry)
+			.append(&self.ttl)
+			.append_list(&self.topics)
+			.append(&self.data);
+
+		let mut digest = Keccak::new_keccak256();
+		digest.update(&*stream.drain());
+		digest.update(&{
+			let mut nonce_bytes = [0u8; 8];
+			BigEndian::write_u64(&mut nonce_bytes, self.nonce);
+
+			nonce_bytes
+		});
+
+		digest.finalize(&mut buf);
+		H256(buf)
+	}
+}
+
+impl rlp::Encodable for Envelope {
+	fn rlp_append(&self, s: &mut RlpStream) {
+		s.begin_list(5)
+			.append(&self.expiry)
+			.append(&self.ttl)
+			.append_list(&self.topics)
+			.append(&self.data)
+			.append(&self.nonce);
+	}
+}
+
+impl rlp::Decodable for Envelope {
+	fn decode(rlp: &UntrustedRlp) -> Result<Self, DecoderError> {
+		if rlp.item_count()? != 5 { return Err(DecoderError::RlpIncorrectListLen) }
+
+		Ok(Envelope {
+			expiry: rlp.val_at(0)?,
+			ttl: rlp.val_at(1)?,
+			topics: rlp.at(2)?.iter().map(|x| x.as_val()).collect::<Result<_, _>>()?,
+			data: rlp.val_at(3)?,
+			nonce: rlp.val_at(4)?,
+		})
+	}
+}
+
+/// Message creation parameters.
+/// Pass this to `Message::create` to make a message.
+pub struct CreateParams {
+	/// time-to-live in seconds.
+	pub ttl: u64,
+	/// payload data.
+	pub payload: Vec<u8>,
+	/// Topics.
+	pub topics: Vec<Topic>,
+	/// How many milliseconds to spend proving work.
+	pub work: u64,
+}
+
+/// A whisper message. This is a checked message carrying around metadata.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct Message {
+	envelope: Envelope,
+	bloom: H512,
+	hash: H256,
+	encoded_size: usize,
+}
+
+impl Message {
+	/// Create a message from creation parameters.
+	/// Panics if TTL is 0.
+	pub fn create(params: CreateParams) -> Self {
+		use byteorder::{BigEndian, ByteOrder};
+		use rand::{Rng, SeedableRng, XorShiftRng};
+
+		let mut rng = {
+			let mut thread_rng = ::rand::thread_rng();
+
+			XorShiftRng::from_seed(thread_rng.gen::<[u32; 4]>())
+		};
+
+		assert!(params.ttl > 0);
+
+		let expiry = {
+			let after_mining = SystemTime::now() + Duration::from_millis(params.work);
+			let since_epoch = after_mining.duration_since(time::UNIX_EPOCH)
+				.expect("time after now is after unix epoch; qed");
+
+			// round up the sub-second to next whole second.
+			since_epoch.as_secs() + if since_epoch.subsec_nanos() == 0 { 0 } else { 1 }
+		};
+
+		let start_digest = {
+			let mut stream = RlpStream::new_list(4);
+			stream.append(&expiry)
+				.append(&params.ttl)
+				.append_list(&params.topics)
+				.append(&params.payload);
+
+			let mut digest = Keccak::new_keccak256();
+			digest.update(&*stream.drain());
+			digest
+		};
+
+		let mut buf = [0; 32];
+		let mut try_nonce = move |nonce: &[u8; 8]| {
+			let mut digest = start_digest.clone();
+			digest.update(&nonce[..]);
+			digest.finalize(&mut buf[..]);
+
+			buf.clone()
+		};
+
+		let mut nonce: [u8; 8] = rng.gen();
+		let mut best_found = try_nonce(&nonce);
+
+		let start = ::time::precise_time_ns();
+
+		while ::time::precise_time_ns() <= start + params.work * 1_000_000 {
+			let temp_nonce = rng.gen();
+			let hash = try_nonce(&temp_nonce);
+
+			if hash < best_found {
+				nonce = temp_nonce;
+				best_found = hash;
+			}
+		}
+
+		let envelope = Envelope {
+			expiry: expiry,
+			ttl: params.ttl,
+			topics: params.topics.into_iter().collect(),
+			data: params.payload,
+			nonce: BigEndian::read_u64(&nonce[..]),
+		};
+
+		debug_assert_eq!(H256(best_found.clone()), envelope.proving_hash());
+
+		let encoded = ::rlp::encode(&envelope);
+
+		Message::from_components(
+			envelope,
+			encoded.len(),
+			H256(keccak256(&encoded)),
+			SystemTime::now(),
+		).expect("Message generated here known to be valid; qed")
+	}
+
+	/// Decode message from RLP and check for validity against system time.
+	pub fn decode(rlp: UntrustedRlp, now: SystemTime) -> Result<Self, Error> {
+		let envelope: Envelope = rlp.as_val()?;
+		let encoded_size = rlp.as_raw().len();
+		let hash = H256(keccak256(rlp.as_raw()));
+
+		Message::from_components(envelope, encoded_size, hash, now)
+	}
+
+	// create message from envelope, hash, and encoded size.
+	// does checks for validity.
+	fn from_components(envelope: Envelope, size: usize, hash: H256, now: SystemTime)
+		-> Result<Self, Error>
+	{
+		const LEEWAY_SECONDS: u64 = 2;
+
+		if envelope.expiry <= envelope.ttl { return Err(Error::LivesTooLong) }
+		if envelope.ttl == 0 { return Err(Error::ZeroTTL) }
+
+		let issue_time_adjusted = Duration::from_secs(
+			(envelope.expiry - envelope.ttl).saturating_sub(LEEWAY_SECONDS)
+		);
+
+		if time::UNIX_EPOCH + issue_time_adjusted > now {
+			return Err(Error::IssuedInFuture);
+		}
+
+		// other validity checks?
+		let bloom = bloom_topics(&envelope.topics);
+
+		Ok(Message {
+			envelope: envelope,
+			bloom: bloom,
+			hash: hash,
+			encoded_size: size,
+		})
+	}
+
+	/// Get a reference to the envelope.
+	pub fn envelope(&self) -> &Envelope {
+		&self.envelope
+	}
+
+	/// Get the encoded size of the envelope.
+	pub fn encoded_size(&self) -> usize {
+		self.encoded_size
+	}
+
+	/// Get a uniquely identifying hash for the message.
+	pub fn hash(&self) -> &H256 {
+		&self.hash
+	}
+
+	/// Get the bloom filter of the topics
+	pub fn bloom(&self) -> &H512 {
+		&self.bloom
+	}
+
+	/// Get the work proved by the hash.
+	pub fn work_proved(&self) -> f64 {
+		let proving_hash = self.envelope.proving_hash();
+
+		work_factor_proved(self.encoded_size as _, self.envelope.ttl, proving_hash)
+	}
+
+	/// Get the expiry time.
+	pub fn expiry(&self) -> SystemTime {
+		time::UNIX_EPOCH + Duration::from_secs(self.envelope.expiry)
+	}
+
+	/// Get the topics.
+	pub fn topics(&self) -> &[Topic] {
+		&self.envelope.topics
+	}
+
+	/// Get the message data.
+	pub fn data(&self) -> &[u8] {
+		&self.envelope.data
+	}
+}
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+	use std::time::{self, Duration, SystemTime};
+	use rlp::UntrustedRlp;
+
+	fn unix_time(x: u64) -> SystemTime {
+		time::UNIX_EPOCH + Duration::from_secs(x)
+	}
+
+	#[test]
+	fn create_message() {
+		let _ = Message::create(CreateParams {
+			ttl: 100,
+			payload: vec![1, 2, 3, 4],
+			topics: Vec::new(),
+			work: 50,
+		});
+	}
+
+	#[test]
+	fn round_trip() {
+		let envelope = Envelope {
+			expiry: 100_000,
+			ttl: 30,
+			data: vec![9; 256],
+			topics: Default::default(),
+			nonce: 1010101,
+		};
+
+		let encoded = ::rlp::encode(&envelope);
+		let decoded = ::rlp::decode(&encoded);
+
+		assert_eq!(envelope, decoded)
+	}
+
+	#[test]
+	fn passes_checks() {
+		let envelope = Envelope {
+			expiry: 100_000,
+			ttl: 30,
+			data: vec![9; 256],
+			topics: Default::default(),
+			nonce: 1010101,
+		};
+
+		let encoded = ::rlp::encode(&envelope);
+
+		for i in 0..30 {
+			let now = unix_time(100_000 - i);
+			Message::decode(UntrustedRlp::new(&*encoded), now).unwrap();
+		}
+	}
+
+	#[test]
+	#[should_panic]
+	fn future_message() {
+		let envelope = Envelope {
+			expiry: 100_000,
+			ttl: 30,
+			data: vec![9; 256],
+			topics: Default::default(),
+			nonce: 1010101,
+		};
+
+		let encoded = ::rlp::encode(&envelope);
+
+		let now = unix_time(100_000 - 1_000);
+		Message::decode(UntrustedRlp::new(&*encoded), now).unwrap();
+	}
+
+	#[test]
+	#[should_panic]
+	fn pre_epoch() {
+		let envelope = Envelope {
+			expiry: 100_000,
+			ttl: 200_000,
+			data: vec![9; 256],
+			topics: Default::default(),
+			nonce: 1010101,
+		};
+
+		let encoded = ::rlp::encode(&envelope);
+
+		let now = unix_time(95_000);
+		Message::decode(UntrustedRlp::new(&*encoded), now).unwrap();
+	}
+}

whisper/src/net.rs

@@ -0,0 +1,638 @@
+// 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/>.
+
+//! Whisper messaging system as a DevP2P subprotocol.
+
+use std::collections::{HashMap, HashSet};
+use std::cmp::Ordering;
+use std::fmt;
+use std::time::{Duration, SystemTime};
+use std::sync::Arc;
+
+use bigint::hash::{H256, H512};
+use network::{HostInfo, NetworkContext, NetworkError, NodeId, PeerId, TimerToken};
+use ordered_float::OrderedFloat;
+use parking_lot::{Mutex, RwLock};
+use rlp::{DecoderError, RlpStream, UntrustedRlp};
+
+use message::{Message, Error as MessageError};
+
+const RALLY_TOKEN: TimerToken = 1;
+const RALLY_TIMEOUT_MS: u64 = 750; // supposed to be at least once per second.
+
+const PROTOCOL_VERSION: usize = 2;
+
+/// Supported protocol versions.
+pub const SUPPORTED_VERSIONS: &'static [u8] = &[PROTOCOL_VERSION as u8];
+
+// maximum tolerated delay between messages packets.
+const MAX_TOLERATED_DELAY_MS: u64 = 2000;
+
+/// Number of packets.
+pub const PACKET_COUNT: u8 = 3;
+
+mod packet {
+	pub const STATUS: u8 = 0;
+	pub const MESSAGES: u8 = 1;
+	pub const TOPIC_FILTER: u8 = 2;
+}
+
+/// Handles messages within a single packet.
+pub trait MessageHandler: Send + Sync {
+	/// Evaluate the message and handle it.
+	///
+	/// The same message will not be passed twice.
+	/// Heavy handling should be done asynchronously.
+	/// If there is a significant overhead in this thread, then an attacker
+	/// can determine which kinds of messages we are listening for.
+	fn handle_messages(&self, message: &[Message]);
+}
+
+// errors in importing a whisper message.
+#[derive(Debug)]
+enum Error {
+	Decoder(DecoderError),
+	Network(NetworkError),
+	Message(MessageError),
+	UnknownPacket(u8),
+	UnknownPeer(PeerId),
+	ProtocolVersionMismatch(usize),
+	SameNodeKey,
+	UnexpectedMessage,
+}
+
+impl From<DecoderError> for Error {
+	fn from(err: DecoderError) -> Self {
+		Error::Decoder(err)
+	}
+}
+
+impl From<NetworkError> for Error {
+	fn from(err: NetworkError) -> Self {
+		Error::Network(err)
+	}
+}
+
+impl From<MessageError> for Error {
+	fn from(err: MessageError) -> Self {
+		Error::Message(err)
+	}
+}
+
+impl fmt::Display for Error {
+	fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+		match *self {
+			Error::Decoder(ref err) => write!(f, "Failed to decode packet: {}", err),
+			Error::Network(ref err) => write!(f, "Network error: {}", err),
+			Error::Message(ref err) => write!(f, "Error decoding message: {}", err),
+			Error::UnknownPacket(ref id) => write!(f, "Unknown packet kind: {}", id),
+			Error::UnknownPeer(ref id) => write!(f, "Message received from unknown peer: {}", id),
+			Error::ProtocolVersionMismatch(ref proto) =>
+				write!(f, "Unknown protocol version: {}", proto),
+			Error::UnexpectedMessage => write!(f, "Unexpected message."),
+			Error::SameNodeKey => write!(f, "Peer and us have same node key."),
+		}
+	}
+}
+
+// sorts by work proved, descending.
+#[derive(PartialEq, Eq)]
+struct SortedEntry {
+	slab_id: usize,
+	work_proved: OrderedFloat<f64>,
+	expiry: SystemTime,
+}
+
+impl Ord for SortedEntry {
+	fn cmp(&self, other: &SortedEntry) -> Ordering {
+		self.work_proved.cmp(&other.work_proved)
+	}
+}
+
+impl PartialOrd for SortedEntry {
+	fn partial_cmp(&self, other: &SortedEntry) -> Option<Ordering> {
+		Some(self.cmp(other))
+	}
+}
+
+// stores messages by two metrics: expiry and PoW rating
+// when full, will accept messages above the minimum stored.
+struct Messages {
+	slab: ::slab::Slab<Message>,
+	sorted: Vec<SortedEntry>,
+	known: HashSet<H256>,
+	removed_hashes: Vec<H256>,
+	cumulative_size: usize,
+	ideal_size: usize,
+}
+
+impl Messages {
+	fn new(ideal_size: usize) -> Self {
+		Messages {
+			slab: ::slab::Slab::with_capacity(0),
+			sorted: Vec::new(),
+			known: HashSet::new(),
+			removed_hashes: Vec::new(),
+			cumulative_size: 0,
+			ideal_size: ideal_size,
+		}
+	}
+
+	// reserve space for additional elements.
+	fn reserve(&mut self, additional: usize) {
+		self.slab.reserve_exact(additional);
+		self.sorted.reserve(additional);
+		self.known.reserve(additional);
+	}
+
+	// whether a message is not known and within the bounds of PoW.
+	fn may_accept(&self, message: &Message) -> bool {
+		!self.known.contains(message.hash()) && {
+			self.sorted.last().map_or(true, |entry| {
+				let work_proved = OrderedFloat(message.work_proved());
+				OrderedFloat(self.slab[entry.slab_id].work_proved()) < work_proved
+			})
+		}
+	}
+
+	// insert a message into the store. for best performance,
+	// call `reserve` before inserting a bunch.
+	//
+	fn insert(&mut self, message: Message) -> bool {
+		if !self.known.insert(message.hash().clone()) { return false }
+
+		let work_proved = OrderedFloat(message.work_proved());
+
+		// pop off entries by low PoW until we have enough space for the higher
+		// PoW message being inserted.
+		let size_upon_insertion = self.cumulative_size + message.encoded_size();
+		if size_upon_insertion >= self.ideal_size {
+			let diff = size_upon_insertion - self.ideal_size;
+			let mut found_diff = 0;
+			for entry in self.sorted.iter().rev() {
+				if found_diff >= diff { break }
+
+				// if we encounter a message with at least the PoW we're looking
+				// at, don't push that message out.
+				if entry.work_proved >= work_proved { return false }
+				found_diff += self.slab[entry.slab_id].encoded_size();
+			}
+
+			// message larger than ideal size.
+			if found_diff < diff { return false }
+
+			while found_diff > 0 {
+				let entry = self.sorted.pop()
+					.expect("found_diff built by traversing entries; therefore that many entries exist; qed");
+
+				let message = self.slab.remove(entry.slab_id)
+					.expect("sorted entry slab IDs always filled; qed");
+
+				found_diff -= message.encoded_size();
+
+				self.cumulative_size -= message.encoded_size();
+				self.known.remove(message.hash());
+				self.removed_hashes.push(message.hash().clone());
+			}
+		}
+
+		let expiry = message.expiry();
+
+		self.cumulative_size += message.encoded_size();
+
+		if !self.slab.has_available() { self.slab.reserve_exact(1) }
+		let id = self.slab.insert(message).expect("just ensured enough space in slab; qed");
+
+		let sorted_entry = SortedEntry {
+			slab_id: id,
+			work_proved: work_proved,
+			expiry: expiry,
+		};
+
+		match self.sorted.binary_search(&sorted_entry) {
+			Ok(idx) | Err(idx) => self.sorted.insert(idx, sorted_entry),
+		}
+
+		true
+	}
+
+	// prune expired messages, and then prune low proof-of-work messages
+	// until below ideal size.
+	fn prune(&mut self, now: SystemTime) -> Vec<H256> {
+		{
+			let slab = &mut self.slab;
+			let known = &mut self.known;
+			let cumulative_size = &mut self.cumulative_size;
+			let ideal_size = &self.ideal_size;
+			let removed = &mut self.removed_hashes;
+
+			// first pass, we look just at expired entries.
+			let all_expired = self.sorted.iter()
+				.filter(|entry| entry.expiry <= now)
+				.map(|x| (true, x));
+
+			// second pass, we look at entries which aren't expired but in order
+			// by PoW
+			let low_proof = self.sorted.iter().rev()
+				.filter(|entry| entry.expiry > now)
+				.map(|x| (false, x));
+
+			for (is_expired, entry) in all_expired.chain(low_proof) {
+				// break once we've removed all expired entries
+				// or have taken enough low-work entries.
+				if !is_expired && *cumulative_size <= *ideal_size {
+					break
+				}
+
+				let message = slab.remove(entry.slab_id)
+					.expect("references to ID kept upon creation; only destroyed upon removal; qed");
+
+				known.remove(message.hash());
+				removed.push(message.hash().clone());
+
+				*cumulative_size -= message.encoded_size();
+			}
+		}
+
+		// clear all the sorted entries we removed from slab.
+		let slab = &self.slab;
+		self.sorted.retain(|entry| slab.contains(entry.slab_id));
+
+		::std::mem::replace(&mut self.removed_hashes, Vec::new())
+	}
+
+	fn iter(&self) -> ::slab::Iter<Message, usize> {
+		self.slab.iter()
+	}
+
+	fn is_full(&self) -> bool {
+		self.cumulative_size >= self.ideal_size
+	}
+
+	fn status(&self) -> PoolStatus {
+		PoolStatus {
+			required_pow: if self.is_full() {
+				self.sorted.last().map(|entry| entry.work_proved.0)
+			} else {
+				None
+			},
+			message_count: self.sorted.len(),
+			cumulative_size: self.cumulative_size,
+			target_size: self.ideal_size,
+		}
+	}
+}
+
+enum State {
+	Unconfirmed(SystemTime), // awaiting status packet.
+	TheirTurn(SystemTime), // it has been their turn to send since stored time.
+	OurTurn,
+}
+
+struct Peer {
+	node_key: NodeId,
+	state: State,
+	known_messages: HashSet<H256>,
+	topic_filter: Option<H512>,
+}
+
+impl Peer {
+	// note that a message has been evicted from the queue.
+	fn note_evicted(&mut self, messages: &[H256]) {
+		for message_hash in messages {
+			self.known_messages.remove(message_hash);
+		}
+	}
+
+	// whether this peer will accept the message.
+	fn will_accept(&self, message: &Message) -> bool {
+		let known = self.known_messages.contains(message.hash());
+
+		let matches_bloom = self.topic_filter.as_ref()
+			.map_or(true, |topic| topic & message.bloom() == message.bloom().clone());
+
+		!known && matches_bloom
+	}
+
+	// note a message as known. returns true if it was already
+	// known, false otherwise.
+	fn note_known(&mut self, message: &Message) -> bool {
+		self.known_messages.insert(message.hash().clone())
+	}
+
+	fn set_topic_filter(&mut self, topic: H512) {
+		self.topic_filter = Some(topic);
+	}
+
+	fn can_send_messages(&self) -> bool {
+		match self.state {
+			State::Unconfirmed(_) | State::OurTurn => false,
+			State::TheirTurn(_) => true,
+		}
+	}
+}
+
+/// Pool status.
+pub struct PoolStatus {
+	/// Required PoW to be accepted into the pool
+	pub required_pow: Option<f64>,
+	/// Number of messages in the pool.
+	pub message_count: usize,
+	/// Cumulative size of the messages in the pool
+	pub cumulative_size: usize,
+	/// Target size of the pool.
+	pub target_size: usize,
+}
+
+/// Handle to the pool, for posting messages or getting info.
+#[derive(Clone)]
+pub struct PoolHandle {
+	messages: Arc<RwLock<Messages>>,
+}
+
+impl PoolHandle {
+	/// Post a message to the whisper network to be relayed.
+	pub fn post_message(&self, message: Message) -> bool {
+		self.messages.write().insert(message)
+	}
+
+	/// Get number of messages and amount of memory used by them.
+	pub fn pool_status(&self) -> PoolStatus {
+		self.messages.read().status()
+	}
+}
+
+/// The whisper network protocol handler.
+pub struct Network<T> {
+	messages: Arc<RwLock<Messages>>,
+	handler: T,
+	peers: RwLock<HashMap<PeerId, Mutex<Peer>>>,
+	node_key: RwLock<NodeId>,
+}
+
+// public API.
+impl<T> Network<T> {
+	/// Create a new network handler.
+	pub fn new(messages_size_bytes: usize, handler: T) -> Self {
+		Network {
+			messages: Arc::new(RwLock::new(Messages::new(messages_size_bytes))),
+			handler: handler,
+			peers: RwLock::new(HashMap::new()),
+			node_key: RwLock::new(Default::default()),
+		}
+	}
+
+	/// Acquire a sender to asynchronously feed messages to the whisper
+	/// network.
+	pub fn handle(&self) -> PoolHandle {
+		PoolHandle { messages: self.messages.clone() }
+	}
+}
+
+impl<T: MessageHandler> Network<T> {
+	fn rally(&self, io: &NetworkContext) {
+		// cannot be greater than 16MB (protocol limitation)
+		const MAX_MESSAGES_PACKET_SIZE: usize = 8 * 1024 * 1024;
+
+		// prune messages.
+		let now = SystemTime::now();
+		let pruned_hashes = self.messages.write().prune(now);
+
+		let messages = self.messages.read();
+		let peers = self.peers.read();
+
+		// send each peer a packet with new messages it may find relevant.
+		for (peer_id, peer) in peers.iter() {
+			let mut peer_data = peer.lock();
+			peer_data.note_evicted(&pruned_hashes);
+
+			let punish_timeout = |last_activity: &SystemTime| {
+				if *last_activity + Duration::from_millis(MAX_TOLERATED_DELAY_MS) <= now {
+					debug!(target: "whisper", "Disconnecting peer {} due to excessive timeout.", peer_id);
+					io.disconnect_peer(*peer_id);
+				}
+			};
+
+			// check timeouts and skip peers who we can't send a rally to.
+			match peer_data.state {
+				State::Unconfirmed(ref time) | State::TheirTurn(ref time) => {
+					punish_timeout(time);
+					continue;
+				}
+				State::OurTurn => {}
+			}
+
+			// construct packet, skipping messages the peer won't accept.
+			let mut stream = RlpStream::new();
+			stream.begin_unbounded_list();
+
+			for message in messages.iter() {
+				if !peer_data.will_accept(message) { continue }
+
+				if stream.estimate_size(message.encoded_size()) > MAX_MESSAGES_PACKET_SIZE {
+					break;
+				}
+
+				peer_data.note_known(message);
+				stream.append(message.envelope());
+			}
+
+			stream.complete_unbounded_list();
+
+			peer_data.state = State::TheirTurn(SystemTime::now());
+			if let Err(e) = io.send(*peer_id, packet::MESSAGES, stream.out()) {
+				debug!(target: "whisper", "Failed to send messages packet to peer {}: {}", peer_id, e);
+				io.disconnect_peer(*peer_id);
+			}
+		}
+	}
+
+	// handle status packet from peer.
+	fn on_status(&self, peer: &PeerId, status: UntrustedRlp)
+		-> Result<(), Error>
+	{
+		let proto: usize = status.as_val()?;
+		if proto != PROTOCOL_VERSION { return Err(Error::ProtocolVersionMismatch(proto)) }
+
+		let peers = self.peers.read();
+		match peers.get(peer) {
+			Some(peer) => {
+				let mut peer = peer.lock();
+				let our_node_key = self.node_key.read().clone();
+
+				// handle this basically impossible edge case gracefully.
+				if peer.node_key == our_node_key {
+					return Err(Error::SameNodeKey);
+				}
+
+				// peer with lower node key begins the rally.
+				if peer.node_key > our_node_key {
+					peer.state = State::OurTurn;
+				} else {
+					peer.state = State::TheirTurn(SystemTime::now());
+				}
+
+				Ok(())
+			}
+			None => {
+				debug!(target: "whisper", "Received message from unknown peer.");
+				Err(Error::UnknownPeer(*peer))
+			}
+		}
+	}
+
+	fn on_messages(&self, peer: &PeerId, message_packet: UntrustedRlp)
+		-> Result<(), Error>
+	{
+		let mut messages_vec = {
+			let peers = self.peers.read();
+			let peer = match peers.get(peer) {
+				Some(peer) => peer,
+				None => {
+					debug!(target: "whisper", "Received message from unknown peer.");
+					return Err(Error::UnknownPeer(*peer));
+				}
+			};
+
+			let mut peer = peer.lock();
+
+			if !peer.can_send_messages() {
+				return Err(Error::UnexpectedMessage);
+			}
+
+			peer.state = State::OurTurn;
+
+			let now = SystemTime::now();
+			let mut messages_vec = message_packet.iter().map(|rlp| Message::decode(rlp, now))
+				.collect::<Result<Vec<_>, _>>()?;
+
+			if messages_vec.is_empty() { return Ok(()) }
+
+			// disallow duplicates in packet.
+			messages_vec.retain(|message| peer.note_known(&message));
+			messages_vec
+		};
+
+		// import for relaying.
+		let mut messages = self.messages.write();
+
+		messages_vec.retain(|message| messages.may_accept(&message));
+		messages.reserve(messages_vec.len());
+
+		self.handler.handle_messages(&messages_vec);
+
+		for message in messages_vec {
+			messages.insert(message);
+		}
+
+		Ok(())
+	}
+
+	fn on_topic_filter(&self, peer: &PeerId, filter: UntrustedRlp)
+		-> Result<(), Error>
+	{
+		let peers = self.peers.read();
+		match peers.get(peer) {
+			Some(peer) => {
+				let mut peer = peer.lock();
+
+				if let State::Unconfirmed(_) = peer.state {
+					return Err(Error::UnexpectedMessage);
+				}
+
+				peer.set_topic_filter(filter.as_val()?)
+			}
+			None => {
+				debug!(target: "whisper", "Received message from unknown peer.");
+				return Err(Error::UnknownPeer(*peer));
+			}
+		}
+
+		Ok(())
+	}
+
+	fn on_connect(&self, io: &NetworkContext, peer: &PeerId) {
+		trace!(target: "whisper", "Connecting peer {}", peer);
+
+		let node_key = match io.session_info(*peer).and_then(|info| info.id) {
+			Some(node_key) => node_key,
+			None => {
+				debug!(target: "whisper", "Disconnecting peer {}, who has no node key.", peer);
+				io.disable_peer(*peer);
+				return;
+			}
+		};
+
+		self.peers.write().insert(*peer, Mutex::new(Peer {
+			node_key: node_key,
+			state: State::Unconfirmed(SystemTime::now()),
+			known_messages: HashSet::new(),
+			topic_filter: None,
+		}));
+
+		if let Err(e) = io.send(*peer, packet::STATUS, ::rlp::encode(&PROTOCOL_VERSION).to_vec()) {
+			debug!(target: "whisper", "Error sending status: {}", e);
+			io.disconnect_peer(*peer);
+		}
+	}
+
+	fn on_disconnect(&self, peer: &PeerId) {
+		trace!(target: "whisper", "Disconnecting peer {}", peer);
+		let _ = self.peers.write().remove(peer);
+	}
+}
+
+impl<T: MessageHandler> ::network::NetworkProtocolHandler for Network<T> {
+	fn initialize(&self, io: &NetworkContext, host_info: &HostInfo) {
+		// set up broadcast timer (< 1s)
+		io.register_timer(RALLY_TOKEN, RALLY_TIMEOUT_MS)
+			.expect("Failed to initialize message rally timer");
+
+		*self.node_key.write() = host_info.id().clone();
+	}
+
+	fn read(&self, io: &NetworkContext, peer: &PeerId, packet_id: u8, data: &[u8]) {
+		let rlp = UntrustedRlp::new(data);
+		let res = match packet_id {
+			packet::STATUS => self.on_status(peer, rlp),
+			packet::MESSAGES => self.on_messages(peer, rlp),
+			packet::TOPIC_FILTER => self.on_topic_filter(peer, rlp),
+			other => Err(Error::UnknownPacket(other)),
+		};
+
+		if let Err(e) = res {
+			trace!(target: "whisper", "Disabling peer due to misbehavior: {}", e);
+			io.disable_peer(*peer);
+		}
+	}
+
+	fn connected(&self, io: &NetworkContext, peer: &PeerId) {
+		// peer with higher ID should begin rallying.
+		self.on_connect(io, peer)
+	}
+
+	fn disconnected(&self, _io: &NetworkContext, peer: &PeerId) {
+		self.on_disconnect(peer)
+	}
+
+	fn timeout(&self, io: &NetworkContext, timer: TimerToken) {
+		// rally with each peer and handle timeouts.
+		match timer {
+			RALLY_TOKEN => self.rally(io),
+			other => debug!(target: "whisper", "Timout triggered on unknown token {}", other),
+		}
+	}
+}

whisper/src/rpc/crypto.rs

@@ -0,0 +1,316 @@
+// 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/>.
+
+//! Encryption schemes supported by RPC layer.
+
+use bigint::hash::H256;
+use ethkey::{self, Public, Secret};
+use ring::aead::{self, AES_256_GCM, SealingKey, OpeningKey};
+
+/// Length of AES key
+pub const AES_KEY_LEN: usize = 32;
+/// Length of AES nonce (IV)
+pub const AES_NONCE_LEN: usize = 12;
+
+// nonce used for encryption when broadcasting
+const BROADCAST_IV: [u8; AES_NONCE_LEN] = [0xff; AES_NONCE_LEN];
+
+// how to encode aes key/nonce.
+enum AesEncode {
+	AppendedNonce, // receiver known, random nonce appended.
+	OnTopics(Vec<H256>), // receiver knows topics but not key. nonce global.
+}
+
+enum EncryptionInner {
+	AES([u8; AES_KEY_LEN], [u8; AES_NONCE_LEN], AesEncode),
+	ECIES(Public),
+}
+
+/// Encryption good for single usage.
+pub struct EncryptionInstance(EncryptionInner);
+
+impl EncryptionInstance {
+	/// ECIES encryption using public key. Fails if invalid public key.
+	pub fn ecies(public: Public) -> Result<Self, &'static str> {
+		if !ethkey::public_is_valid(&public) {
+			return Err("Invalid public key");
+		}
+
+		Ok(EncryptionInstance(EncryptionInner::ECIES(public)))
+	}
+
+	/// 256-bit AES GCM encryption with given nonce.
+	/// It is extremely insecure to reuse nonces.
+	///
+	/// If generating nonces with a secure RNG, limit uses such that
+	/// the chance of collision is negligible.
+	pub fn aes(key: [u8; AES_KEY_LEN], nonce: [u8; AES_NONCE_LEN]) -> Self {
+		EncryptionInstance(EncryptionInner::AES(key, nonce, AesEncode::AppendedNonce))
+	}
+
+	/// Broadcast encryption for the message based on the given topics.
+	///
+	/// Key reuse here is extremely dangerous. It should be randomly generated
+	/// with a secure RNG.
+	pub fn broadcast(key: [u8; AES_KEY_LEN], topics: Vec<H256>) -> Self {
+		EncryptionInstance(EncryptionInner::AES(key, BROADCAST_IV, AesEncode::OnTopics(topics)))
+	}
+
+	/// Encrypt the supplied plaintext
+	pub fn encrypt(self, plain: &[u8]) -> Vec<u8> {
+		match self.0 {
+			EncryptionInner::AES(key, nonce, encode) => {
+				let sealing_key = SealingKey::new(&AES_256_GCM, &key)
+					.expect("key is of correct len; qed");
+
+				let encrypt_plain = move |buf: &mut Vec<u8>| {
+					let out_suffix_capacity = AES_256_GCM.tag_len();
+
+					let prepend_len = buf.len();
+					buf.extend(plain);
+
+					buf.resize(prepend_len + plain.len() + out_suffix_capacity, 0);
+
+					let out_size = aead::seal_in_place(
+						&sealing_key,
+						&nonce,
+						&[], // no authenticated data.
+						&mut buf[prepend_len..],
+						out_suffix_capacity,
+					).expect("key, nonce, buf are valid and out suffix large enough; qed");
+
+					// truncate to the output size and return.
+					buf.truncate(prepend_len + out_size);
+				};
+
+				match encode {
+					AesEncode::AppendedNonce => {
+						let mut buf = Vec::new();
+						encrypt_plain(&mut buf);
+						buf.extend(&nonce[..]);
+						buf
+					}
+					AesEncode::OnTopics(topics) => {
+						let mut buf = Vec::new();
+						let key = H256(key);
+
+						for topic in topics {
+							buf.extend(&*(topic ^ key));
+						}
+
+						encrypt_plain(&mut buf);
+						buf
+					}
+				}
+			}
+			EncryptionInner::ECIES(valid_public) => {
+				::ethcrypto::ecies::encrypt(&valid_public, &[], plain)
+					.expect("validity of public key an invariant of the type; qed")
+			}
+		}
+	}
+}
+
+enum AesExtract {
+	AppendedNonce([u8; AES_KEY_LEN]), // extract appended nonce.
+	OnTopics(usize, usize, H256), // number of topics, index we know, topic we know.
+}
+
+enum DecryptionInner {
+	AES(AesExtract),
+	ECIES(Secret),
+}
+
+/// Decryption instance good for single usage.
+pub struct DecryptionInstance(DecryptionInner);
+
+impl DecryptionInstance {
+	/// ECIES decryption using secret key. Fails if invalid secret.
+	pub fn ecies(secret: Secret) -> Result<Self, &'static str> {
+		secret.check_validity().map_err(|_| "Invalid secret key")?;
+
+		Ok(DecryptionInstance(DecryptionInner::ECIES(secret)))
+	}
+
+	/// 256-bit AES GCM decryption with appended nonce.
+	pub fn aes(key: [u8; AES_KEY_LEN]) -> Self {
+		DecryptionInstance(DecryptionInner::AES(AesExtract::AppendedNonce(key)))
+	}
+
+	/// Decode broadcast based on number of topics and known topic.
+	/// Known topic index may not be larger than num topics - 1.
+	pub fn broadcast(num_topics: usize, topic_idx: usize, known_topic: H256) -> Result<Self, &'static str> {
+		if topic_idx >= num_topics { return Err("topic index out of bounds") }
+
+		Ok(DecryptionInstance(DecryptionInner::AES(AesExtract::OnTopics(num_topics, topic_idx, known_topic))))
+	}
+
+	/// Decrypt ciphertext. Fails if it's an invalid message.
+	pub fn decrypt(self, ciphertext: &[u8]) -> Option<Vec<u8>> {
+		match self.0 {
+			DecryptionInner::AES(extract) => {
+				let decrypt = |
+					key: [u8; AES_KEY_LEN],
+					nonce: [u8; AES_NONCE_LEN],
+					ciphertext: &[u8]
+				| {
+					if ciphertext.len() < AES_256_GCM.tag_len() { return None }
+
+					let opening_key = OpeningKey::new(&AES_256_GCM, &key)
+						.expect("key length is valid for mode; qed");
+
+					let mut buf = ciphertext.to_vec();
+
+					// decrypted plaintext always ends up at the
+					// front of the buffer.
+					let maybe_decrypted = aead::open_in_place(
+						&opening_key,
+						&nonce,
+						&[], // no authenticated data
+						0, // no header.
+						&mut buf,
+					).ok().map(|plain_slice| plain_slice.len());
+
+					maybe_decrypted.map(move |len| { buf.truncate(len); buf })
+				};
+
+				match extract {
+					AesExtract::AppendedNonce(key) => {
+						if ciphertext.len() < AES_NONCE_LEN { return None }
+
+						// nonce is the suffix of ciphertext.
+						let mut nonce = [0; AES_NONCE_LEN];
+						let nonce_offset = ciphertext.len() - AES_NONCE_LEN;
+
+						nonce.copy_from_slice(&ciphertext[nonce_offset..]);
+						decrypt(key, nonce, &ciphertext[..nonce_offset])
+					}
+					AesExtract::OnTopics(num_topics, known_index, known_topic) => {
+						if ciphertext.len() < num_topics * 32 { return None }
+
+						let mut salted_topic = H256::new();
+						salted_topic.copy_from_slice(&ciphertext[(known_index * 32)..][..32]);
+
+						let key = (salted_topic ^ known_topic).0;
+
+						let offset = num_topics * 32;
+						decrypt(key, BROADCAST_IV, &ciphertext[offset..])
+					}
+				}
+			}
+			DecryptionInner::ECIES(secret) => {
+				// secret is checked for validity, so only fails on invalid message.
+				::ethcrypto::ecies::decrypt(&secret, &[], ciphertext).ok()
+			}
+		}
+	}
+}
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+
+	#[test]
+	fn aes_key_len_should_be_equal_to_constant() {
+		assert_eq!(::ring::aead::AES_256_GCM.key_len(), AES_KEY_LEN);
+	}
+
+	#[test]
+	fn aes_nonce_len_should_be_equal_to_constant() {
+		assert_eq!(::ring::aead::AES_256_GCM.nonce_len(), AES_NONCE_LEN);
+	}
+
+	#[test]
+	fn encrypt_asymmetric() {
+		use ethkey::{Generator, Random};
+
+		let key_pair = Random.generate().unwrap();
+		let test_message = move |message: &[u8]| {
+			let instance = EncryptionInstance::ecies(key_pair.public().clone()).unwrap();
+			let ciphertext = instance.encrypt(&message);
+
+			if !message.is_empty() {
+				assert!(&ciphertext[..message.len()] != message)
+			}
+
+			let instance = DecryptionInstance::ecies(key_pair.secret().clone()).unwrap();
+			let decrypted = instance.decrypt(&ciphertext).unwrap();
+
+			assert_eq!(message, &decrypted[..])
+		};
+
+		test_message(&[1, 2, 3, 4, 5]);
+		test_message(&[]);
+		test_message(&[255; 512]);
+	}
+
+	#[test]
+	fn encrypt_symmetric() {
+		use rand::{Rng, OsRng};
+
+		let mut rng = OsRng::new().unwrap();
+		let mut test_message = move |message: &[u8]| {
+			let key = rng.gen();
+
+			let instance = EncryptionInstance::aes(key, rng.gen());
+			let ciphertext = instance.encrypt(message);
+
+			if !message.is_empty() {
+				assert!(&ciphertext[..message.len()] != message)
+			}
+
+			let instance = DecryptionInstance::aes(key);
+			let decrypted = instance.decrypt(&ciphertext).unwrap();
+
+			assert_eq!(message, &decrypted[..])
+		};
+
+		test_message(&[1, 2, 3, 4, 5]);
+		test_message(&[]);
+		test_message(&[255; 512]);
+	}
+
+	#[test]
+	fn encrypt_broadcast() {
+		use rand::{Rng, OsRng};
+
+		let mut rng = OsRng::new().unwrap();
+
+		let mut test_message = move |message: &[u8]| {
+			let all_topics = (0..5).map(|_| rng.gen()).collect::<Vec<_>>();
+			let known_idx = 2;
+			let known_topic = all_topics[2];
+			let key = rng.gen();
+
+			let instance = EncryptionInstance::broadcast(key, all_topics);
+			let ciphertext = instance.encrypt(message);
+
+			if !message.is_empty() {
+				assert!(&ciphertext[..message.len()] != message)
+			}
+
+			let instance = DecryptionInstance::broadcast(5, known_idx, known_topic).unwrap();
+
+			let decrypted = instance.decrypt(&ciphertext).unwrap();
+
+			assert_eq!(message, &decrypted[..])
+		};
+
+		test_message(&[1, 2, 3, 4, 5]);
+		test_message(&[]);
+		test_message(&[255; 512]);
+	}
+}

whisper/src/rpc/filter.rs

@@ -0,0 +1,416 @@
+// 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/>.
+
+//! Abstraction over filters which works with polling and subscription.
+
+use std::collections::HashMap;
+use std::sync::{mpsc, Arc};
+use std::thread;
+
+use bigint::hash::{H256, H512};
+use ethkey::Public;
+use jsonrpc_macros::pubsub::{Subscriber, Sink};
+use parking_lot::{Mutex, RwLock};
+use rand::{Rng, OsRng};
+
+use message::{Message, Topic};
+use super::key_store::KeyStore;
+use super::types::{self, FilterItem, HexEncode};
+
+/// Kinds of filters,
+#[derive(PartialEq, Eq, Clone, Copy)]
+pub enum Kind {
+	/// Polled filter only returns data upon request
+	Poll,
+	/// Subscription filter pushes data to subscriber immediately.
+	Subscription,
+}
+
+pub type ItemBuffer = Arc<Mutex<Vec<FilterItem>>>;
+
+enum FilterEntry {
+	Poll(Arc<Filter>, ItemBuffer),
+	Subscription(Arc<Filter>, Sink<FilterItem>),
+}
+
+/// Filter manager. Handles filters as well as a thread for doing decryption
+/// and payload decoding.
+pub struct Manager {
+	key_store: Arc<RwLock<KeyStore>>,
+	filters: RwLock<HashMap<H256, FilterEntry>>,
+	tx: Mutex<mpsc::Sender<Box<Fn() + Send>>>,
+	join: Option<thread::JoinHandle<()>>,
+}
+
+impl Manager {
+	/// Create a new filter manager that will dispatch decryption tasks onto
+	/// the given thread pool.
+	pub fn new() -> ::std::io::Result<Self> {
+		let (tx, rx) = mpsc::channel::<Box<Fn() + Send>>();
+		let join_handle = thread::Builder::new()
+			.name("Whisper Decryption Worker".to_string())
+			.spawn(move || for item in rx { (item)() })?;
+
+		Ok(Manager {
+			key_store: Arc::new(RwLock::new(KeyStore::new()?)),
+			filters: RwLock::new(HashMap::new()),
+			tx: Mutex::new(tx),
+			join: Some(join_handle),
+		})
+	}
+
+	/// Get a handle to the key store.
+	pub fn key_store(&self) -> Arc<RwLock<KeyStore>> {
+		self.key_store.clone()
+	}
+
+	/// Get filter kind if it's known.
+	pub fn kind(&self, id: &H256) -> Option<Kind> {
+		self.filters.read().get(id).map(|filter| match *filter {
+			FilterEntry::Poll(_, _) => Kind::Poll,
+			FilterEntry::Subscription(_, _) => Kind::Subscription,
+		})
+	}
+
+	/// Remove filter by ID.
+	pub fn remove(&self, id: &H256) {
+		self.filters.write().remove(id);
+	}
+
+	/// Add a new polled filter.
+	pub fn insert_polled(&self, filter: Filter) -> Result<H256, &'static str> {
+		let buffer = Arc::new(Mutex::new(Vec::new()));
+		let entry = FilterEntry::Poll(Arc::new(filter), buffer);
+		let id = OsRng::new()
+			.map_err(|_| "unable to acquire secure randomness")?
+			.gen();
+
+		self.filters.write().insert(id, entry);
+		Ok(id)
+	}
+
+	/// Insert new subscription filter. Generates a secure ID and sends it to
+	/// the
+	pub fn insert_subscription(&self, filter: Filter, sub: Subscriber<FilterItem>)
+		-> Result<(), &'static str>
+	{
+		let id: H256 = OsRng::new()
+			.map_err(|_| "unable to acquire secure randomness")?
+			.gen();
+
+		sub.assign_id(::jsonrpc_pubsub::SubscriptionId::String(id.hex()))
+			.map(move |sink| {
+				let entry = FilterEntry::Subscription(Arc::new(filter), sink);
+				self.filters.write().insert(id, entry);
+			})
+			.map_err(|_| "subscriber disconnected")
+	}
+
+	/// Poll changes on filter identified by ID.
+	pub fn poll_changes(&self, id: &H256) -> Option<Vec<FilterItem>> {
+		self.filters.read().get(id).and_then(|filter| match *filter {
+			FilterEntry::Subscription(_, _) => None,
+			FilterEntry::Poll(_, ref changes)
+				=> Some(::std::mem::replace(&mut *changes.lock(), Vec::new())),
+		})
+	}
+}
+
+// machinery for attaching the manager to the network instance.
+impl ::net::MessageHandler for Arc<Manager> {
+	fn handle_messages(&self, messages: &[Message]) {
+		let filters = self.filters.read();
+		let filters_iter = filters
+			.values()
+			.flat_map(|filter| messages.iter().map(move |msg| (filter, msg))) ;
+
+		for	(filter, message) in filters_iter {
+			// if the message matches any of the possible bloom filters,
+			// send to thread pool to attempt decryption and avoid
+			// blocking the network thread for long.
+			let failed_send = match *filter {
+				FilterEntry::Poll(ref filter, _) | FilterEntry::Subscription(ref filter, _)
+					if !filter.basic_matches(message) => None,
+				FilterEntry::Poll(ref filter, ref buffer) => {
+					let (message, key_store) = (message.clone(), self.key_store.clone());
+					let (filter, buffer) = (filter.clone(), buffer.clone());
+
+					self.tx.lock().send(Box::new(move || {
+						filter.handle_message(
+							&message,
+							&*key_store,
+							|matched| buffer.lock().push(matched),
+						)
+					})).err().map(|x| x.0)
+				}
+				FilterEntry::Subscription(ref filter, ref sink) => {
+					let (message, key_store) = (message.clone(), self.key_store.clone());
+					let (filter, sink) = (filter.clone(), sink.clone());
+
+					self.tx.lock().send(Box::new(move || {
+						filter.handle_message(
+							&message,
+							&*key_store,
+							|matched| { let _ = sink.notify(Ok(matched)); },
+						)
+					})).err().map(|x| x.0)
+				}
+			};
+
+			// if we failed to send work, no option but to do it locally.
+			if let Some(local_work) = failed_send {
+				(local_work)()
+			}
+		}
+	}
+}
+
+impl Drop for Manager {
+	fn drop(&mut self) {
+		if let Some(guard) = self.join.take() {
+			let _ = guard.join();
+		}
+	}
+}
+
+/// Filter incoming messages by critera.
+pub struct Filter {
+	topics: Vec<(Vec<u8>, H512, Topic)>,
+	from: Option<Public>,
+	decrypt_with: Option<H256>,
+}
+
+impl Filter {
+	/// Create a new filter from filter request.
+	///
+	/// Fails if the topics vector is empty.
+	pub fn new(params: types::FilterRequest) -> Result<Self, &'static str> {
+		if params.topics.is_empty() {
+			return Err("no topics for filter");
+		}
+
+		let topics: Vec<_> = params.topics.into_iter()
+			.map(|x| x.into_inner())
+			.map(|topic| {
+				let abridged = super::abridge_topic(&topic);
+				(topic, abridged.bloom(), abridged)
+			})
+			.collect();
+
+		Ok(Filter {
+			topics: topics,
+			from: params.from.map(|x| x.into_inner()),
+			decrypt_with: params.decrypt_with.map(|x| x.into_inner()),
+		})
+	}
+
+	// does basic matching:
+	// whether the given message matches at least one of the topics of the
+	// filter.
+	// TODO: minimum PoW heuristic.
+	fn basic_matches(&self, message: &Message) -> bool {
+		self.topics.iter().any(|&(_, ref bloom, _)| {
+			&(bloom & message.bloom()) == bloom
+		})
+	}
+
+	// handle a message that matches the bloom.
+	fn handle_message<F: Fn(FilterItem)>(
+		&self,
+		message: &Message,
+		store: &RwLock<KeyStore>,
+		on_match: F,
+	) {
+		use rpc::crypto::DecryptionInstance;
+		use tiny_keccak::keccak256;
+
+		let matched_indices: Vec<_> = self.topics.iter()
+			.enumerate()
+			.filter_map(|(i, &(_, ref bloom, ref abridged))| {
+				let contains_topic = &(bloom & message.bloom()) == bloom
+					&& message.topics().contains(abridged);
+
+				if contains_topic { Some(i) } else { None }
+			})
+			.collect();
+
+		if matched_indices.is_empty() { return }
+
+		let decrypt = match self.decrypt_with {
+			Some(ref id) => match store.read().decryption_instance(id) {
+				Some(d) => d,
+				None => {
+					warn!(target: "whisper", "Filter attempted to decrypt with destroyed identity {}",
+						id);
+
+					return
+				}
+			},
+			None => {
+				let known_idx = matched_indices[0];
+				let known_topic = H256(keccak256(&self.topics[0].0));
+
+				DecryptionInstance::broadcast(message.topics().len(), known_idx, known_topic)
+					.expect("known idx is within the range 0..message.topics.len(); qed")
+			}
+		};
+
+		let decrypted = match decrypt.decrypt(message.data()) {
+			Some(d) => d,
+			None => {
+				trace!(target: "whisper", "Failed to decrypt message with {} matching topics",
+					matched_indices.len());
+
+				return
+			}
+		};
+
+		match ::rpc::payload::decode(&decrypted) {
+			Ok(decoded) => {
+				if decoded.from != self.from { return }
+
+				let matched_topics = matched_indices
+					.into_iter()
+					.map(|i| self.topics[i].0.clone())
+					.map(HexEncode)
+					.collect();
+
+				on_match(FilterItem {
+					from: decoded.from.map(HexEncode),
+					recipient: self.decrypt_with.map(HexEncode),
+					ttl: message.envelope().ttl,
+					topics: matched_topics,
+					timestamp: message.envelope().expiry - message.envelope().ttl,
+					payload: HexEncode(decoded.message.to_vec()),
+					padding: decoded.padding.map(|pad| HexEncode(pad.to_vec())),
+				})
+			}
+			Err(reason) =>
+				trace!(target: "whisper", "Bad payload in decrypted message with {} topics: {}",
+					matched_indices.len(), reason),
+		}
+	}
+}
+
+#[cfg(test)]
+mod tests {
+	use message::{CreateParams, Message};
+	use rpc::types::{FilterRequest, HexEncode};
+	use rpc::abridge_topic;
+	use super::*;
+
+	#[test]
+	fn rejects_empty_topics() {
+		let req = FilterRequest {
+			decrypt_with: Default::default(),
+			from: None,
+			topics: Vec::new(),
+		};
+
+		assert!(Filter::new(req).is_err());
+	}
+
+	#[test]
+	fn basic_match() {
+		let topics = vec![vec![1, 2, 3], vec![4, 5, 6]];
+		let req = FilterRequest {
+			decrypt_with: Default::default(),
+			from: None,
+			topics: topics.iter().cloned().map(HexEncode).collect(),
+		};
+
+		let filter = Filter::new(req).unwrap();
+		let message = Message::create(CreateParams {
+			ttl: 100,
+			payload: vec![1, 3, 5, 7, 9],
+			topics: topics.iter().map(|x| abridge_topic(&x)).collect(),
+			work: 0,
+		});
+
+		assert!(filter.basic_matches(&message));
+
+		let message = Message::create(CreateParams {
+			ttl: 100,
+			payload: vec![1, 3, 5, 7, 9],
+			topics: topics.iter().take(1).map(|x| abridge_topic(&x)).collect(),
+			work: 0,
+		});
+
+		assert!(filter.basic_matches(&message));
+
+		let message = Message::create(CreateParams {
+			ttl: 100,
+			payload: vec![1, 3, 5, 7, 9],
+			topics: Vec::new(),
+			work: 0,
+		});
+
+		assert!(!filter.basic_matches(&message));
+	}
+
+	#[test]
+	fn decrypt_and_decode() {
+		use rpc::payload::{self, EncodeParams};
+		use rpc::key_store::{Key, KeyStore};
+
+		let mut store = KeyStore::new().unwrap();
+		let signing_pair = Key::new_asymmetric(store.rng());
+		let encrypting_key = Key::new_symmetric(store.rng());
+
+		let decrypt_id = store.insert(encrypting_key);
+		let encryption_instance = store.encryption_instance(&decrypt_id).unwrap();
+
+		let store = ::parking_lot::RwLock::new(store);
+
+		let payload = payload::encode(EncodeParams {
+			message: &[1, 2, 3],
+			padding: Some(&[4, 5, 4, 5]),
+			sign_with: Some(signing_pair.secret().unwrap())
+		}).unwrap();
+
+		let encrypted = encryption_instance.encrypt(&payload);
+
+		let message = Message::create(CreateParams {
+			ttl: 100,
+			payload: encrypted,
+			topics: vec![abridge_topic(&[9; 32])],
+			work: 0,
+		});
+
+		let message2 = Message::create(CreateParams {
+			ttl: 100,
+			payload: vec![3, 5, 7, 9],
+			topics: vec![abridge_topic(&[9; 32])],
+			work: 0,
+		});
+
+		let filter = Filter::new(FilterRequest {
+			decrypt_with: Some(HexEncode(decrypt_id)),
+			from: Some(HexEncode(signing_pair.public().unwrap().clone())),
+			topics: vec![HexEncode(vec![9; 32])],
+		}).unwrap();
+
+		assert!(filter.basic_matches(&message));
+
+		let items = ::std::cell::Cell::new(0);
+		let on_match = |_| { items.set(items.get() + 1); };
+
+		filter.handle_message(&message, &store, &on_match);
+		filter.handle_message(&message2, &store, &on_match);
+
+		assert_eq!(items.get(), 1);
+	}
+}

whisper/src/rpc/key_store.rs

@@ -0,0 +1,197 @@
+// 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/>.
+
+//! Identity and keystore for Whisper sessions.
+//!
+//! Can handle symmetric and asymmetric keys.
+//! Symmetric encryption is done via AES-256 in GCM mode.
+
+use std::collections::HashMap;
+
+use bigint::hash::H256;
+use ethkey::{KeyPair, Public, Secret};
+use rand::{Rng, OsRng};
+use ring::error::Unspecified;
+
+use rpc::crypto::{AES_KEY_LEN, EncryptionInstance, DecryptionInstance};
+
+/// A symmetric or asymmetric key used for encryption, decryption, and signing
+/// of payloads.
+pub enum Key {
+	/// ECIES key pair for Secp2561k curve. Suitable for encryption, decryption,
+	/// and signing.
+	Asymmetric(KeyPair),
+	/// AES-256 GCM mode. Suitable for encryption, decryption, but not signing.
+	Symmetric([u8; AES_KEY_LEN]),
+}
+
+impl Key {
+	/// Generate a random asymmetric key with the given cryptographic RNG.
+	pub fn new_asymmetric(rng: &mut OsRng) -> Self {
+		match ::ethkey::Generator::generate(rng) {
+			Ok(pair) => Key::Asymmetric(pair),
+			Err(void) => match void {},
+		}
+	}
+
+	/// Generate a random symmetric key with the given cryptographic RNG.
+	pub fn new_symmetric(rng: &mut OsRng) -> Self {
+		Key::Symmetric(rng.gen())
+	}
+
+	/// From secret asymmetric key. Fails if secret is invalid.
+	pub fn from_secret(secret: Secret) -> Result<Self, Unspecified> {
+		KeyPair::from_secret(secret)
+			.map(Key::Asymmetric)
+			.map_err(|_| Unspecified)
+	}
+
+	/// From raw symmetric key.
+	pub fn from_raw_symmetric(key: [u8; AES_KEY_LEN]) -> Self {
+		Key::Symmetric(key)
+	}
+
+	/// Get a handle to the public key if this is an asymmetric key.
+	pub fn public(&self) -> Option<&Public> {
+		match *self {
+			Key::Asymmetric(ref pair) => Some(pair.public()),
+			Key::Symmetric(_) => None,
+		}
+	}
+
+	/// Get a handle to the secret key if this is an asymmetric key.
+	pub fn secret(&self) -> Option<&Secret> {
+		match *self {
+			Key::Asymmetric(ref pair) => Some(pair.secret()),
+			Key::Symmetric(_) => None,
+		}
+	}
+
+	/// Get a handle to the symmetric key.
+	pub fn symmetric(&self) -> Option<&[u8; AES_KEY_LEN]>  {
+		match *self {
+			Key::Asymmetric(_) => None,
+			Key::Symmetric(ref key) => Some(key),
+		}
+	}
+}
+
+/// Key store.
+pub struct KeyStore {
+	rng: OsRng,
+	identities: HashMap<H256, Key>,
+}
+
+impl KeyStore {
+	/// Create the key store. Returns any error in accessing the system's secure
+	/// RNG.
+	pub fn new() -> Result<Self, ::std::io::Error> {
+		Ok(KeyStore {
+			rng: OsRng::new()?,
+			identities: HashMap::new(),
+		})
+	}
+
+	/// Import a key, generating a random identity for it.
+	pub fn insert(&mut self, key: Key) -> H256 {
+		let id = self.rng().gen();
+		self.identities.insert(id, key);
+
+		id
+	}
+
+	/// Get a key by ID.
+	pub fn get<'a>(&'a self, id: &H256) -> Option<&'a Key> {
+		self.identities.get(id)
+	}
+
+	/// Get asymmetric ID's public key.
+	pub fn public<'a>(&'a self, id: &H256) -> Option<&'a Public> {
+		self.get(id).and_then(Key::public)
+	}
+
+	/// Get asymmetric ID's secret key.
+	pub fn secret<'a>(&'a self, id: &H256) -> Option<&'a Secret> {
+		self.get(id).and_then(Key::secret)
+	}
+
+	/// Get symmetric ID's key.
+	pub fn symmetric<'a>(&'a self, id: &H256) -> Option<&'a [u8; AES_KEY_LEN]> {
+		self.get(id).and_then(Key::symmetric)
+	}
+
+	/// Get encryption instance for identity.
+	pub fn encryption_instance(&self, id: &H256) -> Result<EncryptionInstance, &'static str> {
+		self.get(id).ok_or("no such identity").and_then(|key| match *key {
+			Key::Asymmetric(ref pair) => EncryptionInstance::ecies(pair.public().clone())
+				.map_err(|_| "could not create encryption instance for id"),
+			Key::Symmetric(ref key) =>
+				 OsRng::new()
+					.map(|mut rng| EncryptionInstance::aes(key.clone(), rng.gen()))
+				 	.map_err(|_| "unable to get secure randomness")
+		})
+	}
+
+	/// Get decryption instance for identity.
+	/// If the identity is known, always succeeds.
+	pub fn decryption_instance(&self, id: &H256) -> Option<DecryptionInstance> {
+		self.get(id).map(|key| match *key {
+			Key::Asymmetric(ref pair) => DecryptionInstance::ecies(pair.secret().clone())
+				.expect("all keys stored are valid; qed"),
+			Key::Symmetric(ref key) => DecryptionInstance::aes(key.clone()),
+		})
+	}
+
+	/// Whether the store contains a key by this ID.
+	pub fn contains(&self, id: &H256) -> bool {
+		self.identities.contains_key(id)
+	}
+
+	/// Remove a key by ID.
+	pub fn remove(&mut self, id: &H256) -> bool {
+		self.identities.remove(id).is_some()
+	}
+
+	/// Get RNG.
+	pub fn rng(&mut self) -> &mut OsRng {
+		&mut self.rng
+	}
+}
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+
+	#[test]
+	fn rejects_invalid_secret() {
+		let bad_secret = ::ethkey::Secret::from_slice(&[0xff; 32]);
+		assert!(Key::from_secret(bad_secret).is_err());
+	}
+
+	#[test]
+	fn generated_key_should_exist() {
+		let mut store = KeyStore::new().unwrap();
+		let key = Key::new_asymmetric(store.rng());
+
+		assert!(key.public().is_some());
+		assert!(key.secret().is_some());
+
+		let id = store.insert(key);
+
+		assert!(store.contains(&id));
+		assert!(store.get(&id).is_some());
+	}
+}

whisper/src/rpc/mod.rs

@@ -0,0 +1,402 @@
+// 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/>.
+
+//! JSONRPC interface for Whisper.
+//!
+//! Manages standard message format decoding, ephemeral identities, signing,
+//! encryption, and decryption.
+//!
+//! Provides an interface for using whisper to transmit data securely.
+
+use std::sync::Arc;
+
+use jsonrpc_core::{Error, ErrorCode, Metadata};
+use jsonrpc_pubsub::{Session, PubSubMetadata, SubscriptionId};
+use jsonrpc_macros::pubsub;
+
+use bigint::hash::H256;
+use futures::{future, BoxFuture};
+use parking_lot::RwLock;
+
+use self::filter::Filter;
+use self::key_store::{Key, KeyStore};
+use self::types::HexEncode;
+
+use message::{CreateParams, Message, Topic};
+
+mod crypto;
+mod filter;
+mod key_store;
+mod payload;
+mod types;
+
+pub use self::filter::Manager as FilterManager;
+
+// create whisper RPC error.
+fn whisper_error<T: Into<String>>(message: T) -> Error {
+	const ERROR_CODE: i64 = -32085;
+
+	Error {
+		code: ErrorCode::ServerError(ERROR_CODE),
+		message: message.into(),
+		data: None,
+	}
+}
+
+fn topic_hash(topic: &[u8]) -> H256 {
+	H256(::tiny_keccak::keccak256(topic))
+}
+
+// abridge topic using first four bytes of hash.
+fn abridge_topic(topic: &[u8]) -> Topic {
+	let mut abridged = [0; 4];
+	let hash = topic_hash(topic).0;
+	abridged.copy_from_slice(&hash[..4]);
+	abridged.into()
+}
+
+build_rpc_trait! {
+	/// Whisper RPC interface.
+	pub trait Whisper {
+		/// Info about the node.
+		#[rpc(name = "shh_info")]
+		fn info(&self) -> Result<types::NodeInfo, Error>;
+
+		/// Generate a new asymmetric key pair and return an identity.
+		#[rpc(name = "shh_newKeyPair")]
+		fn new_key_pair(&self) -> Result<types::Identity, Error>;
+
+		/// Import the given SECP2561k private key and return an identity.
+		#[rpc(name = "shh_addPrivateKey")]
+		fn add_private_key(&self, types::Private) -> Result<types::Identity, Error>;
+
+		/// Generate a new symmetric key and return an identity.
+		#[rpc(name = "shh_newSymKey")]
+		fn new_sym_key(&self) -> Result<types::Identity, Error>;
+
+		/// Import the given symmetric key and return an identity.
+		#[rpc(name = "shh_addSymKey")]
+		fn add_sym_key(&self, types::Symmetric) -> Result<types::Identity, Error>;
+
+		/// Get public key. Succeeds if identity is stored and asymmetric.
+		#[rpc(name = "shh_getPublicKey")]
+		fn get_public(&self, types::Identity) -> Result<types::Public, Error>;
+
+		/// Get private key. Succeeds if identity is stored and asymmetric.
+		#[rpc(name = "shh_getPrivateKey")]
+		fn get_private(&self, types::Identity) -> Result<types::Private, Error>;
+
+		#[rpc(name = "shh_getSymKey")]
+		fn get_symmetric(&self, types::Identity) -> Result<types::Symmetric, Error>;
+
+		/// Delete key pair denoted by given identity.
+		///
+		/// Return true if successfully removed, false if unknown,
+		/// and error otherwise.
+		#[rpc(name = "shh_deleteKey")]
+		fn remove_key(&self, types::Identity) -> Result<bool, Error>;
+
+		/// Post a message to the network with given parameters.
+		#[rpc(name = "shh_post")]
+		fn post(&self, types::PostRequest) -> Result<bool, Error>;
+
+		/// Create a new polled filter.
+		#[rpc(name = "shh_newMessageFilter")]
+		fn new_filter(&self, types::FilterRequest) -> Result<types::Identity, Error>;
+
+		/// Poll changes on a polled filter.
+		#[rpc(name = "shh_getFilterMessages")]
+		fn poll_changes(&self, types::Identity) -> Result<Vec<types::FilterItem>, Error>;
+
+		/// Delete polled filter. Return bool indicating success.
+		#[rpc(name = "shh_deleteMessageFilter")]
+		fn delete_filter(&self, types::Identity) -> Result<bool, Error>;
+	}
+}
+
+build_rpc_trait! {
+	/// Whisper RPC pubsub.
+	pub trait WhisperPubSub {
+		type Metadata;
+
+		#[pubsub(name = "hello")] {
+			/// Subscribe to messages matching the filter.
+			#[rpc(name = "ssh_subscribe")]
+			fn subscribe(&self, Self::Metadata, pubsub::Subscriber<types::FilterItem>, types::FilterRequest);
+
+			/// Unsubscribe from filter matching given ID. Return
+			/// true on success, error otherwise.
+			#[rpc(name = "shh_unsubscribe")]
+			fn unsubscribe(&self, SubscriptionId) -> BoxFuture<bool, Error>;
+		}
+	}
+}
+
+/// Something which can send messages to the network.
+pub trait PoolHandle: Send + Sync {
+	/// Give message to the whisper network for relay.
+	/// Returns false if PoW too low.
+	fn relay(&self, message: Message) -> bool;
+
+	/// Number of messages and memory used by resident messages.
+	fn pool_status(&self) -> ::net::PoolStatus;
+}
+
+impl PoolHandle for ::net::PoolHandle {
+	fn relay(&self, message: Message) -> bool {
+		self.post_message(message)
+	}
+
+	fn pool_status(&self) -> ::net::PoolStatus {
+		::net::PoolHandle::pool_status(self)
+	}
+}
+
+/// Default, simple metadata implementation.
+#[derive(Clone, Default)]
+pub struct Meta {
+	session: Option<Arc<Session>>,
+}
+
+impl Metadata for Meta {}
+impl PubSubMetadata for Meta {
+	fn session(&self) -> Option<Arc<Session>> {
+		self.session.clone()
+	}
+}
+
+/// Implementation of whisper RPC.
+pub struct WhisperClient<P, M = Meta> {
+	store: Arc<RwLock<KeyStore>>,
+	pool: P,
+	filter_manager: Arc<filter::Manager>,
+	_meta: ::std::marker::PhantomData<M>,
+}
+
+impl<P> WhisperClient<P> {
+	/// Create a new whisper client with basic metadata.
+	pub fn with_simple_meta(pool: P, filter_manager: Arc<filter::Manager>) -> Self {
+		WhisperClient::new(pool, filter_manager)
+	}
+}
+
+impl<P, M> WhisperClient<P, M> {
+	/// Create a new whisper client.
+	pub fn new(pool: P, filter_manager: Arc<filter::Manager>) -> Self {
+		WhisperClient {
+			store: filter_manager.key_store(),
+			pool: pool,
+			filter_manager: filter_manager,
+			_meta: ::std::marker::PhantomData,
+		}
+	}
+
+	fn delete_filter_kind(&self, id: H256, kind: filter::Kind) -> bool {
+		match self.filter_manager.kind(&id) {
+			Some(k) if k == kind => {
+				self.filter_manager.remove(&id);
+				true
+			}
+			None | Some(_) => false,
+		}
+	}
+}
+
+impl<P: PoolHandle + 'static, M: Send + Sync + 'static> Whisper for WhisperClient<P, M> {
+	fn info(&self) -> Result<types::NodeInfo, Error> {
+		let status = self.pool.pool_status();
+
+		Ok(types::NodeInfo {
+			required_pow: status.required_pow,
+			messages: status.message_count,
+			memory: status.cumulative_size,
+			target_memory: status.target_size,
+		})
+	}
+
+	fn new_key_pair(&self) -> Result<types::Identity, Error> {
+		let mut store = self.store.write();
+		let key_pair = Key::new_asymmetric(store.rng());
+
+		Ok(HexEncode(store.insert(key_pair)))
+	}
+
+	fn add_private_key(&self, private: types::Private) -> Result<types::Identity, Error> {
+		let key_pair = Key::from_secret(private.into_inner().into())
+			.map_err(|_| whisper_error("Invalid private key"))?;
+
+		Ok(HexEncode(self.store.write().insert(key_pair)))
+	}
+
+	fn new_sym_key(&self) -> Result<types::Identity, Error> {
+		let mut store = self.store.write();
+		let key = Key::new_symmetric(store.rng());
+
+		Ok(HexEncode(store.insert(key)))
+	}
+
+	fn add_sym_key(&self, raw_key: types::Symmetric) -> Result<types::Identity, Error> {
+		let raw_key = raw_key.into_inner().0;
+		let key = Key::from_raw_symmetric(raw_key);
+
+		Ok(HexEncode(self.store.write().insert(key)))
+	}
+
+	fn get_public(&self, id: types::Identity) -> Result<types::Public, Error> {
+		self.store.read().public(&id.into_inner())
+			.cloned()
+			.map(HexEncode)
+			.ok_or_else(|| whisper_error("Unknown identity"))
+	}
+
+	fn get_private(&self, id: types::Identity) -> Result<types::Private, Error> {
+		self.store.read().secret(&id.into_inner())
+			.map(|x| (&**x).clone())
+			.map(HexEncode)
+			.ok_or_else(|| whisper_error("Unknown identity"))
+	}
+
+	fn get_symmetric(&self, id: types::Identity) -> Result<types::Symmetric, Error> {
+		self.store.read().symmetric(&id.into_inner())
+			.cloned()
+			.map(H256)
+			.map(HexEncode)
+			.ok_or_else(|| whisper_error("Unknown identity"))
+	}
+
+	fn remove_key(&self, id: types::Identity) -> Result<bool, Error> {
+		Ok(self.store.write().remove(&id.into_inner()))
+	}
+
+	fn post(&self, req: types::PostRequest) -> Result<bool, Error> {
+		use self::crypto::EncryptionInstance;
+
+		let encryption = match req.to {
+			Some(types::Receiver::Public(public)) => EncryptionInstance::ecies(public.into_inner())
+				.map_err(whisper_error)?,
+			Some(types::Receiver::Identity(id)) => self.store.read().encryption_instance(&id.into_inner())
+				.map_err(whisper_error)?,
+			None => {
+				use rand::{Rng, OsRng};
+
+				// broadcast mode: use fixed nonce and fresh key each time.
+
+				let mut rng = OsRng::new()
+					.map_err(|_| whisper_error("unable to acquire secure randomness"))?;
+
+				let key = rng.gen();
+				if req.topics.is_empty() {
+					return Err(whisper_error("must supply at least one topic for broadcast message"));
+				}
+
+				EncryptionInstance::broadcast(
+					key,
+					req.topics.iter().map(|x| topic_hash(&x)).collect()
+				)
+			}
+		};
+
+		let sign_with = match req.from {
+			Some(from) => {
+				Some(
+					self.store.read().secret(&from.into_inner())
+						.cloned()
+						.ok_or_else(|| whisper_error("Unknown identity `from`"))?
+				)
+			}
+			None => None,
+		};
+
+		let encrypted = {
+			let payload = payload::encode(payload::EncodeParams {
+				message: &req.payload.into_inner(),
+				padding: req.padding.map(|p| p.into_inner()).as_ref().map(|x| &x[..]),
+				sign_with: sign_with.as_ref(),
+			}).map_err(whisper_error)?;
+
+			encryption.encrypt(&payload)
+		};
+
+		// mining the packet is the heaviest item of work by far.
+		// there may be a benefit to dispatching this onto the CPU pool
+		// and returning a future. but then things get _less_ efficient
+		// if the server infrastructure has more threads than the CPU pool.
+		let message = Message::create(CreateParams {
+			ttl: req.ttl,
+			payload: encrypted,
+			topics: req.topics.into_iter().map(|x| abridge_topic(&x.into_inner())).collect(),
+			work: req.priority,
+		});
+
+		if !self.pool.relay(message) {
+			Err(whisper_error("PoW too low to compete with other messages"))
+		} else {
+			Ok(true)
+		}
+	}
+
+	fn new_filter(&self, req: types::FilterRequest) -> Result<types::Identity, Error> {
+		let filter = Filter::new(req).map_err(whisper_error)?;
+
+		self.filter_manager.insert_polled(filter)
+			.map(HexEncode)
+			.map_err(whisper_error)
+	}
+
+	fn poll_changes(&self, id: types::Identity) -> Result<Vec<types::FilterItem>, Error> {
+		match self.filter_manager.poll_changes(&id.into_inner()) {
+			None => Err(whisper_error("no such message filter")),
+			Some(items) => Ok(items),
+		}
+	}
+
+	fn delete_filter(&self, id: types::Identity) -> Result<bool, Error> {
+		Ok(self.delete_filter_kind(id.into_inner(), filter::Kind::Poll))
+	}
+}
+
+impl<P: PoolHandle + 'static, M: Send + Sync + PubSubMetadata> WhisperPubSub for WhisperClient<P, M> {
+	type Metadata = M;
+
+	fn subscribe(
+		&self,
+		_meta: Self::Metadata,
+		subscriber: pubsub::Subscriber<types::FilterItem>,
+		req: types::FilterRequest,
+	) {
+		match Filter::new(req) {
+			Ok(filter) => {
+				if let Err(e) = self.filter_manager.insert_subscription(filter, subscriber) {
+					debug!(target: "whisper", "Failed to add subscription: {}", e);
+				}
+			}
+			Err(reason) => { let _ = subscriber.reject(whisper_error(reason)); }
+		}
+	}
+
+	fn unsubscribe(&self, id: SubscriptionId) -> BoxFuture<bool, Error> {
+		use std::str::FromStr;
+
+		let res = match id {
+			SubscriptionId::String(s) => H256::from_str(&s)
+				.map_err(|_| "unrecognized ID")
+				.map(|id| self.delete_filter_kind(id, filter::Kind::Subscription)),
+			SubscriptionId::Number(_) => Err("unrecognized ID"),
+		};
+
+		Box::new(future::done(res.map_err(whisper_error)))
+	}
+}

whisper/src/rpc/payload.rs

@@ -0,0 +1,357 @@
+// 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/>.
+
+//! Common payload format definition, construction, and decoding.
+//!
+//! Format:
+//! flags: 1 byte
+//!
+//! payload size: 0..4 bytes, BE, determined by flags.
+//! optional padding: byte array up to 2^24 bytes in length. encoded in payload size.
+//! optional signature: 65 bytes (r, s, v)
+//!
+//! payload: byte array of length of arbitrary size.
+//!
+//! flag bits used:
+//!   0, 1 => how many bytes indicate padding length (up to 3)
+//!   2 => whether signature is present
+//!
+//! padding is used to mask information about size of message.
+//!
+//! AES-256-GCM will append 12 bytes of metadata to the front of the message.
+
+use bigint::hash::H256;
+use byteorder::{BigEndian, ByteOrder, WriteBytesExt};
+use ethkey::{Public, Secret};
+use tiny_keccak::keccak256;
+
+const SIGNATURE_LEN: usize = 65;
+
+const STANDARD_PAYLOAD_VERSION: u8 = 1;
+
+bitflags! {
+	struct Flags: u8 {
+		const FLAG_PAD_LEN_HIGH = 0b10000000;
+		const FLAG_PAD_LEN_LOW  = 0b01000000;
+		const FLAG_SIGNED       = 0b00100000;
+	}
+}
+
+// number of bytes of padding length (in the range 0..4)
+fn padding_length_bytes(flags: Flags) -> usize {
+	match (flags & FLAG_PAD_LEN_HIGH, flags & FLAG_PAD_LEN_LOW) {
+		(FLAG_PAD_LEN_HIGH, FLAG_PAD_LEN_LOW) => 3,
+		(FLAG_PAD_LEN_HIGH, _) => 2,
+		(_, FLAG_PAD_LEN_LOW) => 1,
+		(_, _) => 0,
+	}
+}
+
+// how many bytes are necessary to encode the given length. Range 0..4.
+// `None` if too large.
+fn num_padding_length_bytes(padding_len: usize) -> Option<usize> {
+	let bits = 64 - (padding_len as u64).leading_zeros();
+	match bits {
+		0 => Some(0),
+		0 ... 8 => Some(1),
+		0 ... 16 => Some(2),
+		0 ... 24 => Some(3),
+		_ => None,
+	}
+}
+
+/// Parameters for encoding a standard payload.
+pub struct EncodeParams<'a> {
+	/// Message to encode.
+	pub message: &'a [u8],
+	/// Padding bytes. Maximum padding allowed is 65536 bytes.
+	pub padding: Option<&'a [u8]>,
+	/// Private key to sign with.
+	pub sign_with: Option<&'a Secret>,
+}
+
+impl<'a> Default for EncodeParams<'a> {
+	fn default() -> Self {
+		EncodeParams {
+			message: &[],
+			padding: None,
+			sign_with: None,
+		}
+	}
+}
+
+/// Parameters for decoding a standard payload.
+pub struct Decoded<'a> {
+	/// Decoded message.
+	pub message: &'a [u8],
+	/// optional padding.
+	pub padding: Option<&'a [u8]>,
+	/// Recovered signature.
+	pub from: Option<Public>,
+}
+
+/// Encode using provided parameters.
+pub fn encode(params: EncodeParams) -> Result<Vec<u8>, &'static str> {
+	const VEC_WRITE_INFALLIBLE: &'static str = "writing to a Vec<u8> can never fail; qed";
+
+	let padding_len = params.padding.map_or(0, |x| x.len());
+	let padding_len_bytes = num_padding_length_bytes(padding_len)
+		.ok_or_else(|| "padding size too long")?;
+
+	let signature = params.sign_with.map(|secret| {
+		let hash = H256(keccak256(params.message));
+		::ethkey::sign(secret, &hash)
+	});
+
+	let signature = match signature {
+		Some(Ok(sig)) => Some(sig),
+		Some(Err(_)) => return Err("invalid signing key provided"),
+		None => None,
+	};
+
+	let (flags, plaintext_size) = {
+		let mut flags = Flags::empty();
+
+		// 1 byte each for flags and version.
+		let mut plaintext_size = 2
+			+ padding_len_bytes
+			+ padding_len
+			+ params.message.len();
+
+		flags.bits = (padding_len_bytes << 6) as u8;
+		debug_assert_eq!(padding_length_bytes(flags), padding_len_bytes);
+
+		if let Some(ref sig) = signature {
+			plaintext_size += sig.len();
+			flags |= FLAG_SIGNED;
+		}
+
+		(flags, plaintext_size)
+	};
+
+	let mut plaintext = Vec::with_capacity(plaintext_size);
+
+	plaintext.push(STANDARD_PAYLOAD_VERSION);
+	plaintext.push(flags.bits);
+
+	if let Some(padding) = params.padding {
+		plaintext.write_uint::<BigEndian>(padding_len as u64, padding_len_bytes)
+			.expect(VEC_WRITE_INFALLIBLE);
+
+		plaintext.extend(padding)
+	}
+
+	if let Some(signature) = signature {
+		plaintext.extend(signature.r());
+		plaintext.extend(signature.s());
+		plaintext.push(signature.v());
+	}
+
+	plaintext.extend(params.message);
+
+	Ok(plaintext)
+}
+
+/// Decode using provided parameters
+pub fn decode(payload: &[u8]) -> Result<Decoded, &'static str> {
+	let mut offset = 0;
+
+	let (padding, signature) = {
+		// use a closure for reading slices since std::io::Read would require
+		// us to copy.
+		let mut next_slice = |len| {
+			let end = offset + len;
+			if payload.len() >= end {
+				let slice = &payload[offset .. end];
+				offset = end;
+
+				Ok(slice)
+			} else {
+				return Err("unexpected end of payload")
+			}
+		};
+
+
+		if next_slice(1)?[0] != STANDARD_PAYLOAD_VERSION {
+			return Err("unknown payload version.");
+		}
+
+		let flags = Flags::from_bits_truncate(next_slice(1)?[0]);
+
+		let padding_len_bytes = padding_length_bytes(flags);
+		let padding = if padding_len_bytes != 0 {
+			let padding_len = BigEndian::read_uint(
+				next_slice(padding_len_bytes)?,
+				padding_len_bytes,
+			);
+
+			Some(next_slice(padding_len as usize)?)
+		} else {
+			None
+		};
+
+		let signature = if flags & FLAG_SIGNED == FLAG_SIGNED {
+			let slice = next_slice(SIGNATURE_LEN)?;
+			let mut arr = [0; SIGNATURE_LEN];
+
+			arr.copy_from_slice(slice);
+			let signature = ::ethkey::Signature::from(arr);
+
+			let not_rsv = signature.r() != &slice[..32]
+				|| signature.s() != &slice[32..64]
+				|| signature.v() != slice[64];
+
+			if not_rsv {
+				return Err("signature not in RSV format");
+			} else {
+				Some(signature)
+			}
+		} else {
+			None
+		};
+
+		(padding, signature)
+	};
+
+	// remaining data is the message.
+	let message = &payload[offset..];
+
+	let from = match signature {
+		None => None,
+		Some(sig) => {
+			let hash = H256(keccak256(message));
+			Some(::ethkey::recover(&sig, &hash).map_err(|_| "invalid signature")?)
+		}
+	};
+
+	Ok(Decoded {
+		message: message,
+		padding: padding,
+		from: from,
+	})
+}
+
+#[cfg(test)]
+mod tests {
+	use ethkey::{Generator, Random};
+	use super::*;
+
+	#[test]
+	fn padding_len_bytes_sanity() {
+		const U24_MAX: usize = (1 << 24) - 1;
+
+		assert_eq!(padding_length_bytes(FLAG_PAD_LEN_HIGH | FLAG_PAD_LEN_LOW), 3);
+		assert_eq!(padding_length_bytes(FLAG_PAD_LEN_HIGH), 2);
+		assert_eq!(padding_length_bytes(FLAG_PAD_LEN_LOW), 1);
+		assert_eq!(padding_length_bytes(Flags::empty()), 0);
+
+		assert!(num_padding_length_bytes(u32::max_value() as _).is_none());
+		assert!(num_padding_length_bytes(U24_MAX + 1).is_none());
+
+		assert_eq!(num_padding_length_bytes(U24_MAX), Some(3));
+
+		assert_eq!(num_padding_length_bytes(u16::max_value() as usize + 1), Some(3));
+		assert_eq!(num_padding_length_bytes(u16::max_value() as usize), Some(2));
+
+		assert_eq!(num_padding_length_bytes(u8::max_value() as usize + 1), Some(2));
+		assert_eq!(num_padding_length_bytes(u8::max_value() as usize), Some(1));
+
+		assert_eq!(num_padding_length_bytes(1), Some(1));
+		assert_eq!(num_padding_length_bytes(0), Some(0));
+	}
+
+	#[test]
+	fn encode_decode_roundtrip() {
+		let message = [1, 2, 3, 4, 5];
+		let encoded = encode(EncodeParams {
+			message: &message,
+			padding: None,
+			sign_with: None,
+		}).unwrap();
+
+		let decoded = decode(&encoded).unwrap();
+
+		assert_eq!(message, decoded.message);
+	}
+
+	#[test]
+	fn encode_empty() {
+		let encoded = encode(EncodeParams {
+			message: &[],
+			padding: None,
+			sign_with: None,
+		}).unwrap();
+
+		let decoded = decode(&encoded).unwrap();
+
+		assert!(decoded.message.is_empty());
+	}
+
+	#[test]
+	fn encode_with_signature() {
+		let key_pair = Random.generate().unwrap();
+		let message = [1, 3, 5, 7, 9];
+
+		let encoded = encode(EncodeParams {
+			message: &message,
+			padding: None,
+			sign_with: Some(key_pair.secret()),
+		}).unwrap();
+
+		let decoded = decode(&encoded).unwrap();
+
+		assert_eq!(decoded.message, message);
+		assert_eq!(decoded.from, Some(key_pair.public().clone()));
+		assert!(decoded.padding.is_none());
+	}
+
+	#[test]
+	fn encode_with_padding() {
+		let message = [1, 3, 5, 7, 9];
+		let padding = [0xff; 1024 - 5];
+
+		let encoded = encode(EncodeParams {
+			message: &message,
+			padding: Some(&padding),
+			sign_with: None,
+		}).unwrap();
+
+		let decoded = decode(&encoded).unwrap();
+
+		assert_eq!(decoded.message, message);
+		assert_eq!(decoded.padding, Some(&padding[..]));
+		assert!(decoded.from.is_none());
+	}
+
+	#[test]
+	fn encode_with_padding_and_signature() {
+		let key_pair = Random.generate().unwrap();
+		let message = [1, 3, 5, 7, 9];
+		let padding = [0xff; 1024 - 5];
+
+		let encoded = encode(EncodeParams {
+			message: &message,
+			padding: Some(&padding),
+			sign_with: Some(key_pair.secret()),
+		}).unwrap();
+
+		let decoded = decode(&encoded).unwrap();
+
+		assert_eq!(decoded.message, message);
+		assert_eq!(decoded.padding, Some(&padding[..]));
+		assert_eq!(decoded.from, Some(key_pair.public().clone()));
+	}
+}

whisper/src/rpc/types.rs

@@ -0,0 +1,298 @@
+// 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/>.
+
+//! Types for Whisper RPC.
+
+use std::fmt;
+use std::ops::Deref;
+
+use bigint::hash::*;
+use hex::{ToHex, FromHex};
+
+use serde::{Serialize, Serializer, Deserialize, Deserializer};
+use serde::de::{Error, Visitor};
+
+/// Helper trait for generic hex bytes encoding.
+pub trait HexEncodable: Sized + ::std::ops::Deref<Target=[u8]> {
+	fn from_bytes(bytes: Vec<u8>) -> Option<Self>;
+}
+
+impl HexEncodable for Vec<u8> {
+	fn from_bytes(bytes: Vec<u8>) -> Option<Self> { Some(bytes) }
+}
+
+macro_rules! impl_hex_for_hash {
+	($($t: ident)*) => {
+		$(
+			impl HexEncodable for $t {
+				fn from_bytes(bytes: Vec<u8>) -> Option<Self> {
+					if bytes.len() != $t::len() {
+						None
+					} else {
+						Some($t::from_slice(&bytes))
+					}
+				}
+			}
+		)*
+	}
+}
+
+impl_hex_for_hash!(
+	H32 H64 H128 H256 H264 H512 H1024
+);
+
+/// Wrapper structure around hex-encoded data.
+#[derive(Debug, PartialEq, Eq, Default, Hash, Clone)]
+pub struct HexEncode<T>(pub T);
+
+impl<T> From<T> for HexEncode<T> {
+	fn from(x: T) -> Self {
+		HexEncode(x)
+	}
+}
+
+impl<T> HexEncode<T> {
+	/// Create a new wrapper from the inner value.
+	pub fn new(x: T) -> Self { HexEncode(x) }
+
+	/// Consume the wrapper, yielding the inner value.
+	pub fn into_inner(self) -> T { self.0 }
+}
+
+impl<T> Deref for HexEncode<T> {
+	type Target = T;
+
+	fn deref(&self) -> &T { &self.0 }
+}
+
+/// Hex-encoded arbitrary-byte vector.
+pub type Bytes = HexEncode<Vec<u8>>;
+
+/// 32-byte local identity
+pub type Identity = HexEncode<H256>;
+
+/// Public key for ECIES, SECP256k1
+pub type Public = HexEncode<::ethkey::Public>;
+
+/// Unvalidated private key for ECIES, SECP256k1
+pub type Private = HexEncode<H256>;
+
+/// Abridged topic is four bytes.
+// only used in tests for now.
+#[cfg(test)]
+pub type AbridgedTopic = HexEncode<H32>;
+
+/// 32-byte AES key.
+pub type Symmetric = HexEncode<H256>;
+
+impl<T: HexEncodable> Serialize for HexEncode<T> {
+	fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
+		let data = &self.0[..];
+		let serialized = "0x".to_owned() + &data.to_hex();
+
+		serializer.serialize_str(serialized.as_ref())
+	}
+}
+
+impl<'a, T: 'a + HexEncodable> Deserialize<'a> for HexEncode<T> {
+	fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+		where D: Deserializer<'a>
+	{
+		deserializer.deserialize_any(HexEncodeVisitor::<T>(::std::marker::PhantomData))
+	}
+}
+
+// helper type for decoding anything from hex.
+struct HexEncodeVisitor<T>(::std::marker::PhantomData<T>);
+
+impl<'a, T: HexEncodable> Visitor<'a> for HexEncodeVisitor<T> {
+	type Value = HexEncode<T>;
+
+	fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
+		write!(formatter, "a 0x-prefixed, hex-encoded vector of bytes")
+	}
+
+	fn visit_str<E: Error>(self, value: &str) -> Result<Self::Value, E> {
+		let decoded = if value.len() >= 2 && &value[0..2] == "0x" && value.len() & 1 == 0 {
+			Ok(Vec::from_hex(&value[2..]).map_err(|_| Error::custom("invalid hex"))?)
+		} else {
+			Err(Error::custom("invalid format"))
+		};
+
+		decoded
+			.and_then(|x| T::from_bytes(x).ok_or(Error::custom("invalid format")))
+			.map(HexEncode)
+	}
+
+	fn visit_string<E>(self, value: String) -> Result<Self::Value, E> where E: Error {
+		self.visit_str(value.as_ref())
+	}
+}
+
+/// Receiver of a message. Either a public key, identity (presumably symmetric),
+/// or broadcast over the topics.
+#[derive(Deserialize)]
+pub enum Receiver {
+	#[serde(rename="public")]
+	Public(Public),
+	#[serde(rename="identity")]
+	Identity(Identity),
+}
+
+/// A request to post a message to the whisper network.
+#[derive(Deserialize)]
+pub struct PostRequest {
+	/// Receiver of the message. Either a public key or
+	/// an identity. If the identity is symmetric, it will
+	/// encrypt to that identity.
+	///
+	/// If the receiver is missing, this will be a broadcast message.
+	pub to: Option<Receiver>,
+
+	/// Sender of the message.
+	///
+	/// If present, the payload will be signed by this
+	/// identity. The call will fail if the whisper node doesn't store the
+	/// signing key for this identity.
+	#[serde(skip_serializing_if = "Option::is_none")]
+	pub from: Option<Identity>,
+
+	/// Full topics to identify a message by.
+	/// At least one topic must be specified if the receiver is
+	/// not specified.
+	pub topics: Vec<Bytes>,
+
+	/// Payload of the message
+	pub payload: Bytes,
+
+	/// Optional padding of the message. No larger than 2^24 - 1.
+	pub padding: Option<Bytes>,
+
+	/// Priority of the message: how many milliseconds to spend doing PoW
+	pub priority: u64,
+
+	/// Time-To-Live of the message in seconds.
+	pub ttl: u64,
+}
+
+/// Request for filter or subscription creation.
+#[derive(Deserialize)]
+pub struct FilterRequest {
+	/// ID of key used for decryption.
+	///
+	/// If this identity is removed, then no further messages will be returned.
+	///
+	/// If optional, this will listen for broadcast messages.
+	#[serde(rename = "decryptWith")]
+	pub decrypt_with: Option<Identity>,
+
+	/// Accept only messages signed by given public key.
+	pub from: Option<Public>,
+
+	/// Possible topics. Cannot be empty if the identity is `None`
+	pub topics: Vec<Bytes>,
+}
+
+/// A message captured by a filter or subscription.
+#[derive(Serialize, Clone)]
+pub struct FilterItem {
+	/// Public key that signed this message.
+	#[serde(skip_serializing_if = "Option::is_none")]
+	pub from: Option<Public>,
+
+	/// Identity of recipient. If the filter wasn't registered with a
+	/// recipient, this will be `None`.
+	#[serde(skip_serializing_if = "Option::is_none")]
+	pub recipient: Option<Identity>,
+
+	/// Time to live in seconds.
+	pub ttl: u64,
+
+	/// Abridged topics that matched the filter.
+	pub topics: Vec<Bytes>,
+
+	/// Unix timestamp of the message generation.
+	pub timestamp: u64,
+
+	/// Decrypted/Interpreted payload.
+	pub payload: Bytes,
+
+	/// Optional padding data.
+	#[serde(skip_serializing_if = "Option::is_none")]
+	pub padding: Option<Bytes>,
+}
+
+/// Whisper node info.
+#[derive(Serialize)]
+pub struct NodeInfo {
+	/// min PoW to be accepted into the local pool.
+	#[serde(skip_serializing_if = "Option::is_none")]
+	#[serde(rename = "minPow")]
+	pub required_pow: Option<f64>,
+
+	/// Number of messages in the pool.
+	pub messages: usize,
+
+	/// Memory used by messages in the pool.
+	pub memory: usize,
+
+	/// Target memory of the pool.
+	#[serde(rename = "targetMemory")]
+	pub target_memory: usize,
+}
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+	use serde_json;
+	use hex::FromHex;
+
+	#[test]
+	fn test_bytes_serialize() {
+		let bytes = Bytes::new(Vec::from_hex("0123456789abcdef").unwrap());
+		let serialized = serde_json::to_string(&bytes).unwrap();
+		assert_eq!(serialized, r#""0x0123456789abcdef""#);
+	}
+
+	#[test]
+	fn test_bytes_deserialize() {
+		let bytes2: Result<Bytes, serde_json::Error> = serde_json::from_str(r#""0x123""#);
+		let bytes3: Result<Bytes, serde_json::Error> = serde_json::from_str(r#""0xgg""#);
+
+		let bytes4: Bytes = serde_json::from_str(r#""0x""#).unwrap();
+		let bytes5: Bytes = serde_json::from_str(r#""0x12""#).unwrap();
+		let bytes6: Bytes = serde_json::from_str(r#""0x0123""#).unwrap();
+
+		assert!(bytes2.is_err());
+		assert!(bytes3.is_err());
+		assert_eq!(bytes4, Bytes::new(vec![]));
+		assert_eq!(bytes5, Bytes::new(vec![0x12]));
+		assert_eq!(bytes6, Bytes::new(vec![0x1, 0x23]));
+	}
+
+	#[test]
+	fn deserialize_topic() {
+		let topic = AbridgedTopic::new([1, 2, 3, 15].into());
+
+		let topic1: Result<AbridgedTopic, _> = serde_json::from_str(r#""0x010203""#);
+		let topic2: Result<AbridgedTopic, _> = serde_json::from_str(r#""0102030F""#);
+		let topic3: AbridgedTopic = serde_json::from_str(r#""0x0102030F""#).unwrap();
+
+		assert!(topic1.is_err());
+		assert!(topic2.is_err());
+		assert_eq!(topic3, topic);
+	}
+}