diff --git a/protocols/plaintext/Cargo.toml b/protocols/plaintext/Cargo.toml
index 5b04674b5162..4c3d501f4d50 100644
--- a/protocols/plaintext/Cargo.toml
+++ b/protocols/plaintext/Cargo.toml
@@ -10,7 +10,11 @@ keywords = ["peer-to-peer", "libp2p", "networking"]
 categories = ["network-programming", "asynchronous"]
 
 [dependencies]
+bytes = "0.4"
 futures = "0.1"
 libp2p-core = { version = "0.12.0", path = "../../core" }
+log = "0.4.6"
 void = "1"
-
+tokio-io = "0.1.12"
+protobuf = "2.3"
+rw-stream-sink = { version = "0.1.1", path = "../../misc/rw-stream-sink" }
diff --git a/protocols/plaintext/regen_structs_proto.sh b/protocols/plaintext/regen_structs_proto.sh
new file mode 100755
index 000000000000..0071cf0169a8
--- /dev/null
+++ b/protocols/plaintext/regen_structs_proto.sh
@@ -0,0 +1,13 @@
+#!/bin/sh
+
+# This script regenerates the `src/structs_proto.rs` file from `structs.proto`.
+
+sudo docker run --rm -v `pwd`:/usr/code:z -w /usr/code rust /bin/bash -c " \
+    apt-get update; \
+    apt-get install -y protobuf-compiler; \
+    cargo install --version 2.3.0 protobuf-codegen; \
+    protoc --rust_out . structs.proto"
+
+#sudo chown $USER:$USER *.rs
+
+mv -f structs.rs ./src/structs_proto.rs
diff --git a/protocols/plaintext/src/lib.rs b/protocols/plaintext/src/lib.rs
index c8c6aafbaa83..f29918ea3e78 100644
--- a/protocols/plaintext/src/lib.rs
+++ b/protocols/plaintext/src/lib.rs
@@ -18,40 +18,313 @@
 // FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
 // DEALINGS IN THE SOFTWARE.
 
+use futures::{Future, StartSend, Poll};
+use futures::sink::Sink;
+use futures::stream::MapErr as StreamMapErr;
+use futures::stream::Stream;
 use futures::future::{self, FutureResult};
-use libp2p_core::{InboundUpgrade, OutboundUpgrade, UpgradeInfo, upgrade::Negotiated};
+use log::{debug, trace};
+use libp2p_core::{identity, InboundUpgrade, OutboundUpgrade, UpgradeInfo, upgrade::Negotiated, PeerId, PublicKey};
+use rw_stream_sink::RwStreamSink;
 use std::iter;
+use std::io::{Error as IoError, ErrorKind as IoErrorKind};
+use std::error;
+use std::fmt;
 use void::Void;
+use bytes::BytesMut;
+use tokio_io::{AsyncRead, AsyncWrite};
+use tokio_io::codec::length_delimited;
+use tokio_io::codec::length_delimited::Framed;
+use crate::structs_proto::Propose;
+use protobuf::Message;
+use protobuf::error::ProtobufError;
 
-#[derive(Debug, Copy, Clone)]
-pub struct PlainTextConfig;
+mod structs_proto;
+
+#[derive(Clone)]
+pub struct PlainTextConfig {
+//    peerId: PeerId,
+    pub key: identity::Keypair,
+}
 
 impl UpgradeInfo for PlainTextConfig {
     type Info = &'static [u8];
     type InfoIter = iter::Once<Self::Info>;
 
     fn protocol_info(&self) -> Self::InfoIter {
-        iter::once(b"/plaintext/1.0.0")
+        iter::once(b"/plaintext/2.0.0")
+    }
+}
+
+impl<C> InboundUpgrade<C> for PlainTextConfig
+where
+    C: AsyncRead + AsyncWrite + Send + 'static
+{
+    type Output = PlainTextOutput<Negotiated<C>>;
+    type Error = PlainTextError;
+    type Future = Box<dyn Future<Item = Self::Output, Error = Self::Error> + Send>;
+
+    fn upgrade_inbound(self, socket: Negotiated<C>, _: Self::Info) -> Self::Future {
+        Box::new(self.handshake(socket))
+//        future::ok(i)
+
+    }
+}
+
+impl<C> OutboundUpgrade<C> for PlainTextConfig
+where
+    C: AsyncRead + AsyncWrite + Send + 'static
+{
+    type Output = PlainTextOutput<Negotiated<C>>;
+    type Error = PlainTextError;
+    type Future = Box<dyn Future<Item = Self::Output, Error = Self::Error> + Send>;
+
+    fn upgrade_outbound(self, socket: Negotiated<C>, _: Self::Info) -> Self::Future {
+        Box::new(self.handshake(socket))
+//        future::ok(i)
+    }
+}
+
+impl PlainTextConfig {
+    fn handshake<T>(self, socket: T) -> impl Future<Item = PlainTextOutput<T>, Error = PlainTextError>
+    where
+        T: AsyncRead + AsyncWrite + Send + 'static
+    {
+        debug!("Starting plaintext upgrade");
+        PlainTextMiddleware::handshake(socket, self)
+            .map(|(stream_sink, public_key)| {
+                let mapped = stream_sink.map_err(map_err as fn(_) -> _);
+                PlainTextOutput {
+                    stream: RwStreamSink::new(mapped),
+                    remote_key: public_key,
+                }
+            })
+    }
+}
+
+#[derive(Debug)]
+pub enum PlainTextError {
+    /// I/O error.
+    IoError(IoError),
+
+    /// Protocol buffer error.
+    ProtobufError(ProtobufError),
+
+    /// Failed to parse one of the handshake protobuf messages.
+    HandshakeParsingFailure,
+}
+
+impl error::Error for PlainTextError {
+    fn cause(&self) -> Option<&dyn error::Error> {
+        match *self {
+            PlainTextError::IoError(ref err) => Some(err),
+            PlainTextError::ProtobufError(ref err) => Some(err),
+            _ => None,
+        }
+    }
+}
+
+impl fmt::Display for PlainTextError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> Result<(), fmt::Error> {
+        match self {
+            PlainTextError::IoError(e) => write!(f, "I/O error: {}", e),
+            PlainTextError::ProtobufError(e) => write!(f, "Protobuf error: {}", e),
+            PlainTextError::HandshakeParsingFailure => f.write_str("Failed to parse one of the handshake protobuf messages"),
+        }
+    }
+}
+
+impl From<IoError> for PlainTextError {
+    #[inline]
+    fn from(err: IoError) -> PlainTextError {
+        PlainTextError::IoError(err)
+    }
+}
+
+impl From<ProtobufError> for PlainTextError {
+    #[inline]
+    fn from(err: ProtobufError) -> PlainTextError {
+        PlainTextError::ProtobufError(err)
+    }
+}
+
+struct HandShakeContext<T> {
+    config: PlainTextConfig,
+    state: T
+}
+
+// HandshakeContext<()> --with_local-> HandshakeContext<Local>
+struct Local {
+    // Our encoded local public key
+    public_key_encoded: Vec<u8>,
+    // Our local proposition's raw bytes:
+    proposition_bytes: Vec<u8>,
+}
+
+struct Remote {
+    local: Local,
+    // The remote's proposition's raw bytes:
+    proposition_bytes: BytesMut,
+    // The remote's public key:
+    public_key: PublicKey,
+}
+
+impl HandShakeContext<()> {
+    fn new(config: PlainTextConfig) -> Self {
+        Self {
+            config,
+            state: (),
+        }
+    }
+
+    fn with_local(self) -> Result<HandShakeContext<Local>, PlainTextError> {
+        let public_key_encoded = self.config.key.public().into_protobuf_encoding();
+        let mut proposition = Propose::new();
+        proposition.set_pubkey(public_key_encoded.clone());
+
+        let proposition_bytes = proposition.write_to_bytes()?;
+
+        Ok(HandShakeContext {
+            config: self.config,
+            state: Local {
+                public_key_encoded,
+                proposition_bytes,
+            }
+        })
     }
 }
 
-impl<C> InboundUpgrade<C> for PlainTextConfig {
-    type Output = Negotiated<C>;
-    type Error = Void;
-    type Future = FutureResult<Negotiated<C>, Self::Error>;
+impl HandShakeContext<Local> {
+    fn with_remote(self, proposition_bytes: BytesMut) -> Result<HandShakeContext<Remote>, PlainTextError> {
+        let mut prop = match protobuf::parse_from_bytes::<Propose>(&proposition_bytes) {
+            Ok(prop) => prop,
+            Err(_) => {
+                debug!("failed to parse remote's proposition protobuf message");
+                return Err(PlainTextError::HandshakeParsingFailure);
+            },
+        };
 
-    fn upgrade_inbound(self, i: Negotiated<C>, _: Self::Info) -> Self::Future {
-        future::ok(i)
+        let public_key_encoded = prop.take_pubkey();
+        let public_key = match PublicKey::from_protobuf_encoding(&public_key_encoded) {
+            Ok(p) => p,
+            Err(_) => {
+                debug!("failed to parse remote's proposition's pubkey protobuf");
+                return Err(PlainTextError::HandshakeParsingFailure);
+            },
+        };
+
+        Ok(HandShakeContext {
+            config: self.config,
+            state: Remote {
+                local: self.state,
+                proposition_bytes,
+                public_key,
+            }
+        })
     }
 }
 
-impl<C> OutboundUpgrade<C> for PlainTextConfig {
-    type Output = Negotiated<C>;
-    type Error = Void;
-    type Future = FutureResult<Negotiated<C>, Self::Error>;
+#[inline]
+fn map_err(err: IoError) -> IoError {
+    debug!("error during plaintext handshake {:?}", err);
+    IoError::new(IoErrorKind::InvalidData, err)
+}
+
+pub struct PlainTextMiddleware<S> {
+    pub inner: Framed<S, BytesMut>,
+}
 
-    fn upgrade_outbound(self, i: Negotiated<C>, _: Self::Info) -> Self::Future {
-        future::ok(i)
+impl<T> PlainTextMiddleware<T>
+where
+    T: AsyncRead + AsyncWrite + Send,
+{
+    fn handshake(socket: T, config: PlainTextConfig) -> impl Future<Item = (PlainTextMiddleware<T>, PublicKey), Error = PlainTextError>
+        where
+            T: AsyncRead + AsyncWrite + Send + 'static
+    {
+        let socket = length_delimited::Builder::new()
+            .big_endian()
+            .length_field_length(4)
+            .new_framed(socket);
+
+        future::ok::<_, PlainTextError>(HandShakeContext::new(config))
+            .and_then(|context| {
+                trace!("starting handshake");
+                Ok(context.with_local()?)
+            })
+            .and_then(|context| {
+                trace!("sending proposition to remote");
+                socket.send(BytesMut::from(context.state.proposition_bytes.clone()))
+                    .from_err()
+                    .map(|s| (s, context))
+            })
+            .and_then(move |(socket, context)| {
+                trace!("receiving the remote's proposition");
+                socket.into_future()
+                    .map_err(|(e, _)| e.into())
+                    .and_then(move |(prop_raw, socket)| {
+                        let context = match prop_raw {
+                            Some(p) => context.with_remote(p)?,
+                            None => {
+                                debug!("unexpected eof while waiting for remote's proposition");
+                                let err = IoError::new(IoErrorKind::BrokenPipe, "unexpected eof");
+                                return Err(err.into());
+                            }
+                        };
+
+                        trace!("received proposition from remote; pubkey = {:?}", context.state.public_key);
+                        Ok((socket, context))
+                    })
+            })
+            .and_then(|(socket, context)| {
+                let middleware = PlainTextMiddleware { inner: socket };
+                Ok((middleware, context.state.public_key))
+            })
     }
 }
 
+impl<S> Sink for PlainTextMiddleware<S>
+where
+    S: AsyncRead + AsyncWrite,
+{
+    type SinkItem = BytesMut;
+    type SinkError = IoError;
+
+    #[inline]
+    fn start_send(&mut self, item: Self::SinkItem) -> StartSend<Self::SinkItem, Self::SinkError> {
+        self.inner.start_send(item)
+    }
+
+    #[inline]
+    fn poll_complete(&mut self) -> Poll<(), Self::SinkError> {
+        self.inner.poll_complete()
+    }
+
+    #[inline]
+    fn close(&mut self) -> Poll<(), Self::SinkError> {
+        self.inner.close()
+    }
+}
+
+impl<S> Stream for PlainTextMiddleware<S>
+where
+    S: AsyncRead + AsyncWrite,
+{
+    type Item = BytesMut;
+    type Error = IoError;
+
+    #[inline]
+    fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
+        self.inner.poll()
+    }
+}
+
+/// Output of the plaintext protocol.
+pub struct PlainTextOutput<S>
+where
+    S: AsyncRead + AsyncWrite,
+{
+    pub stream: RwStreamSink<StreamMapErr<PlainTextMiddleware<S>, fn(IoError) -> IoError>>,
+    /// The public key of the remote.
+    pub remote_key: PublicKey,
+}
diff --git a/protocols/plaintext/src/structs_proto.rs b/protocols/plaintext/src/structs_proto.rs
new file mode 100644
index 000000000000..a9b4d2692a8d
--- /dev/null
+++ b/protocols/plaintext/src/structs_proto.rs
@@ -0,0 +1,222 @@
+// This file is generated by rust-protobuf 2.3.0. Do not edit
+// @generated
+
+// https://github.com/Manishearth/rust-clippy/issues/702
+#![allow(unknown_lints)]
+#![allow(clippy)]
+
+#![cfg_attr(rustfmt, rustfmt_skip)]
+
+#![allow(box_pointers)]
+#![allow(dead_code)]
+#![allow(missing_docs)]
+#![allow(non_camel_case_types)]
+#![allow(non_snake_case)]
+#![allow(non_upper_case_globals)]
+#![allow(trivial_casts)]
+#![allow(unsafe_code)]
+#![allow(unused_imports)]
+#![allow(unused_results)]
+
+use protobuf::Message as Message_imported_for_functions;
+use protobuf::ProtobufEnum as ProtobufEnum_imported_for_functions;
+
+#[derive(PartialEq,Clone,Default)]
+pub struct Propose {
+    // message fields
+    pubkey: ::protobuf::SingularField<::std::vec::Vec<u8>>,
+    // special fields
+    pub unknown_fields: ::protobuf::UnknownFields,
+    pub cached_size: ::protobuf::CachedSize,
+}
+
+impl Propose {
+    pub fn new() -> Propose {
+        ::std::default::Default::default()
+    }
+
+    // optional bytes pubkey = 1;
+
+    pub fn clear_pubkey(&mut self) {
+        self.pubkey.clear();
+    }
+
+    pub fn has_pubkey(&self) -> bool {
+        self.pubkey.is_some()
+    }
+
+    // Param is passed by value, moved
+    pub fn set_pubkey(&mut self, v: ::std::vec::Vec<u8>) {
+        self.pubkey = ::protobuf::SingularField::some(v);
+    }
+
+    // Mutable pointer to the field.
+    // If field is not initialized, it is initialized with default value first.
+    pub fn mut_pubkey(&mut self) -> &mut ::std::vec::Vec<u8> {
+        if self.pubkey.is_none() {
+            self.pubkey.set_default();
+        }
+        self.pubkey.as_mut().unwrap()
+    }
+
+    // Take field
+    pub fn take_pubkey(&mut self) -> ::std::vec::Vec<u8> {
+        self.pubkey.take().unwrap_or_else(|| ::std::vec::Vec::new())
+    }
+
+    pub fn get_pubkey(&self) -> &[u8] {
+        match self.pubkey.as_ref() {
+            Some(v) => &v,
+            None => &[],
+        }
+    }
+}
+
+impl ::protobuf::Message for Propose {
+    fn is_initialized(&self) -> bool {
+        true
+    }
+
+    fn merge_from(&mut self, is: &mut ::protobuf::CodedInputStream) -> ::protobuf::ProtobufResult<()> {
+        while !is.eof()? {
+            let (field_number, wire_type) = is.read_tag_unpack()?;
+            match field_number {
+                1 => {
+                    ::protobuf::rt::read_singular_bytes_into(wire_type, is, &mut self.pubkey)?;
+                },
+                _ => {
+                    ::protobuf::rt::read_unknown_or_skip_group(field_number, wire_type, is, self.mut_unknown_fields())?;
+                },
+            };
+        }
+        ::std::result::Result::Ok(())
+    }
+
+    // Compute sizes of nested messages
+    #[allow(unused_variables)]
+    fn compute_size(&self) -> u32 {
+        let mut my_size = 0;
+        if let Some(ref v) = self.pubkey.as_ref() {
+            my_size += ::protobuf::rt::bytes_size(1, &v);
+        }
+        my_size += ::protobuf::rt::unknown_fields_size(self.get_unknown_fields());
+        self.cached_size.set(my_size);
+        my_size
+    }
+
+    fn write_to_with_cached_sizes(&self, os: &mut ::protobuf::CodedOutputStream) -> ::protobuf::ProtobufResult<()> {
+        if let Some(ref v) = self.pubkey.as_ref() {
+            os.write_bytes(1, &v)?;
+        }
+        os.write_unknown_fields(self.get_unknown_fields())?;
+        ::std::result::Result::Ok(())
+    }
+
+    fn get_cached_size(&self) -> u32 {
+        self.cached_size.get()
+    }
+
+    fn get_unknown_fields(&self) -> &::protobuf::UnknownFields {
+        &self.unknown_fields
+    }
+
+    fn mut_unknown_fields(&mut self) -> &mut ::protobuf::UnknownFields {
+        &mut self.unknown_fields
+    }
+
+    fn as_any(&self) -> &::std::any::Any {
+        self as &::std::any::Any
+    }
+    fn as_any_mut(&mut self) -> &mut ::std::any::Any {
+        self as &mut ::std::any::Any
+    }
+    fn into_any(self: Box<Self>) -> ::std::boxed::Box<::std::any::Any> {
+        self
+    }
+
+    fn descriptor(&self) -> &'static ::protobuf::reflect::MessageDescriptor {
+        Self::descriptor_static()
+    }
+
+    fn new() -> Propose {
+        Propose::new()
+    }
+
+    fn descriptor_static() -> &'static ::protobuf::reflect::MessageDescriptor {
+        static mut descriptor: ::protobuf::lazy::Lazy<::protobuf::reflect::MessageDescriptor> = ::protobuf::lazy::Lazy {
+            lock: ::protobuf::lazy::ONCE_INIT,
+            ptr: 0 as *const ::protobuf::reflect::MessageDescriptor,
+        };
+        unsafe {
+            descriptor.get(|| {
+                let mut fields = ::std::vec::Vec::new();
+                fields.push(::protobuf::reflect::accessor::make_singular_field_accessor::<_, ::protobuf::types::ProtobufTypeBytes>(
+                    "pubkey",
+                    |m: &Propose| { &m.pubkey },
+                    |m: &mut Propose| { &mut m.pubkey },
+                ));
+                ::protobuf::reflect::MessageDescriptor::new::<Propose>(
+                    "Propose",
+                    fields,
+                    file_descriptor_proto()
+                )
+            })
+        }
+    }
+
+    fn default_instance() -> &'static Propose {
+        static mut instance: ::protobuf::lazy::Lazy<Propose> = ::protobuf::lazy::Lazy {
+            lock: ::protobuf::lazy::ONCE_INIT,
+            ptr: 0 as *const Propose,
+        };
+        unsafe {
+            instance.get(Propose::new)
+        }
+    }
+}
+
+impl ::protobuf::Clear for Propose {
+    fn clear(&mut self) {
+        self.clear_pubkey();
+        self.unknown_fields.clear();
+    }
+}
+
+impl ::std::fmt::Debug for Propose {
+    fn fmt(&self, f: &mut ::std::fmt::Formatter) -> ::std::fmt::Result {
+        ::protobuf::text_format::fmt(self, f)
+    }
+}
+
+impl ::protobuf::reflect::ProtobufValue for Propose {
+    fn as_ref(&self) -> ::protobuf::reflect::ProtobufValueRef {
+        ::protobuf::reflect::ProtobufValueRef::Message(self)
+    }
+}
+
+static file_descriptor_proto_data: &'static [u8] = b"\
+    \n\rstructs.proto\x12\x08spipe.pb\"!\n\x07Propose\x12\x16\n\x06pubkey\
+    \x18\x01\x20\x01(\x0cR\x06pubkeyJo\n\x06\x12\x04\0\0\x04\x01\n\x08\n\x01\
+    \x02\x12\x03\0\x08\x10\n\n\n\x02\x04\0\x12\x04\x02\0\x04\x01\n\n\n\x03\
+    \x04\0\x01\x12\x03\x02\x08\x0f\n\x0b\n\x04\x04\0\x02\0\x12\x03\x03\x08\"\
+    \n\x0c\n\x05\x04\0\x02\0\x04\x12\x03\x03\x08\x10\n\x0c\n\x05\x04\0\x02\0\
+    \x05\x12\x03\x03\x11\x16\n\x0c\n\x05\x04\0\x02\0\x01\x12\x03\x03\x17\x1d\
+    \n\x0c\n\x05\x04\0\x02\0\x03\x12\x03\x03\x20!\
+";
+
+static mut file_descriptor_proto_lazy: ::protobuf::lazy::Lazy<::protobuf::descriptor::FileDescriptorProto> = ::protobuf::lazy::Lazy {
+    lock: ::protobuf::lazy::ONCE_INIT,
+    ptr: 0 as *const ::protobuf::descriptor::FileDescriptorProto,
+};
+
+fn parse_descriptor_proto() -> ::protobuf::descriptor::FileDescriptorProto {
+    ::protobuf::parse_from_bytes(file_descriptor_proto_data).unwrap()
+}
+
+pub fn file_descriptor_proto() -> &'static ::protobuf::descriptor::FileDescriptorProto {
+    unsafe {
+        file_descriptor_proto_lazy.get(|| {
+            parse_descriptor_proto()
+        })
+    }
+}
diff --git a/protocols/plaintext/structs.proto b/protocols/plaintext/structs.proto
new file mode 100644
index 000000000000..651bc9480073
--- /dev/null
+++ b/protocols/plaintext/structs.proto
@@ -0,0 +1,5 @@
+package spipe.pb;
+
+message Propose {
+	optional bytes pubkey = 1;
+}