Implement /plaintext/2.0.0 (#1236)

* WIP

* plaintext/2.0.0

* Refactor protobuf related issues to compatible with the spec

* Rename: new PlainTextConfig -> PlainText2Config

* Keep plaintext/1.0.0 as PlainText1Config

* Config contains pubkey

* Rename: proposition -> exchange

* Add PeerId to Exchange

* Check the validity of the remote's `Exchange`

* Tweak

* Delete unused import

* Add debug log

* Delete unused field: public_key_encoded

* Delete unused field: local

* Delete unused field: exchange_bytes

* The inner instance should not be public

* identity::Publickey::Rsa is not available on wasm

* Delete PeerId from Config as it should be generated from the pubkey

* Catch up for #1240

* Tweak

* Update protocols/plaintext/src/error.rs

Co-Authored-By: Pierre Krieger <pierre.krieger1708@gmail.com>

* Update protocols/plaintext/src/handshake.rs

Co-Authored-By: Pierre Krieger <pierre.krieger1708@gmail.com>

* Update protocols/plaintext/src/error.rs

Co-Authored-By: Pierre Krieger <pierre.krieger1708@gmail.com>

* Update protocols/plaintext/src/error.rs

Co-Authored-By: Roman Borschel <romanb@users.noreply.github.com>

* Update protocols/plaintext/src/error.rs

Co-Authored-By: Roman Borschel <romanb@users.noreply.github.com>

* Rename: pubkey -> local_public_key

* Delete unused error

* Rename: PeerIdValidationFailed -> InvalidPeerId

* Fix: HandShake -> Handshake

* Use bytes insteadof Publickey to avoid code duplication

* Replace with ProtobufError

* Merge HandshakeContext<()> into HandshakeContext<Local>

* Improve the peer ID validation to simplify the handshake

* Propagate Remote to allow extracting the PeerId from the Remote

* Collapse the same kind of errors into the variant

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" }

protocols/plaintext/regen_structs_proto.sh

@@ -0,0 +1,8 @@
+#!/bin/sh
+
+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=./protocols/plaintext/src/pb ./protocols/plaintext/structs.proto;"
+

protocols/plaintext/src/error.rs

@@ -0,0 +1,75 @@
+// Copyright 2019 Parity Technologies (UK) Ltd.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a
+// copy of this software and associated documentation files (the "Software"),
+// to deal in the Software without restriction, including without limitation
+// the rights to use, copy, modify, merge, publish, distribute, sublicense,
+// and/or sell copies of the Software, and to permit persons to whom the
+// Software is furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
+// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+// DEALINGS IN THE SOFTWARE.
+
+use std::error;
+use std::fmt;
+use std::io::Error as IoError;
+use protobuf::error::ProtobufError;
+
+#[derive(Debug)]
+pub enum PlainTextError {
+    /// I/O error.
+    IoError(IoError),
+
+    /// Failed to parse the handshake protobuf message.
+    InvalidPayload(Option<ProtobufError>),
+
+    /// The peer id of the exchange isn't consistent with the remote public key.
+    InvalidPeerId,
+}
+
+impl error::Error for PlainTextError {
+    fn cause(&self) -> Option<&dyn error::Error> {
+        match *self {
+            PlainTextError::IoError(ref err) => Some(err),
+            PlainTextError::InvalidPayload(Some(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::InvalidPayload(protobuf_error) => {
+                match protobuf_error {
+                    Some(e) => write!(f, "Protobuf error: {}", e),
+                    None => f.write_str("Failed to parse one of the handshake protobuf messages")
+                }
+            },
+            PlainTextError::InvalidPeerId =>
+                f.write_str("The peer id of the exchange isn't consistent with the remote public key"),
+        }
+    }
+}
+
+impl From<IoError> for PlainTextError {
+    fn from(err: IoError) -> PlainTextError {
+        PlainTextError::IoError(err)
+    }
+}
+
+impl From<ProtobufError> for PlainTextError {
+    fn from(err: ProtobufError) -> PlainTextError {
+        PlainTextError::InvalidPayload(Some(err))
+    }
+}

protocols/plaintext/src/handshake.rs

@@ -0,0 +1,153 @@
+// Copyright 2019 Parity Technologies (UK) Ltd.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a
+// copy of this software and associated documentation files (the "Software"),
+// to deal in the Software without restriction, including without limitation
+// the rights to use, copy, modify, merge, publish, distribute, sublicense,
+// and/or sell copies of the Software, and to permit persons to whom the
+// Software is furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
+// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+// DEALINGS IN THE SOFTWARE.
+
+use bytes::BytesMut;
+use std::io::{Error as IoError, ErrorKind as IoErrorKind};
+use futures::Future;
+use futures::future;
+use futures::sink::Sink;
+use futures::stream::Stream;
+use libp2p_core::{PublicKey, PeerId};
+use log::{debug, trace};
+use crate::pb::structs::Exchange;
+use tokio_io::{AsyncRead, AsyncWrite};
+use tokio_io::codec::length_delimited;
+use tokio_io::codec::length_delimited::Framed;
+use protobuf::Message;
+use crate::error::PlainTextError;
+use crate::PlainText2Config;
+
+struct HandshakeContext<T> {
+    config: PlainText2Config,
+    state: T
+}
+
+// HandshakeContext<()> --with_local-> HandshakeContext<Local>
+struct Local {
+    // Our local exchange's raw bytes:
+    exchange_bytes: Vec<u8>,
+}
+
+// HandshakeContext<Local> --with_remote-> HandshakeContext<Remote>
+pub struct Remote {
+    // The remote's peer ID:
+    pub peer_id: PeerId,
+    // The remote's public key:
+    pub public_key: PublicKey,
+}
+
+impl HandshakeContext<Local> {
+    fn new(config: PlainText2Config) -> Result<Self, PlainTextError> {
+        let mut exchange = Exchange::new();
+        exchange.set_id(config.local_public_key.clone().into_peer_id().into_bytes());
+        exchange.set_pubkey(config.local_public_key.clone().into_protobuf_encoding());
+        let exchange_bytes = exchange.write_to_bytes()?;
+
+        Ok(Self {
+            config,
+            state: Local {
+                exchange_bytes
+            }
+        })
+    }
+
+    fn with_remote(self, exchange_bytes: BytesMut) -> Result<HandshakeContext<Remote>, PlainTextError> {
+        let mut prop = match protobuf::parse_from_bytes::<Exchange>(&exchange_bytes) {
+            Ok(prop) => prop,
+            Err(e) => {
+                debug!("failed to parse remote's exchange protobuf message");
+                return Err(PlainTextError::InvalidPayload(Some(e)));
+            },
+        };
+
+        let pb_pubkey = prop.take_pubkey();
+        let public_key = match PublicKey::from_protobuf_encoding(pb_pubkey.as_slice()) {
+            Ok(p) => p,
+            Err(_) => {
+                debug!("failed to parse remote's exchange's pubkey protobuf");
+                return Err(PlainTextError::InvalidPayload(None));
+            },
+        };
+        let peer_id = match PeerId::from_bytes(prop.take_id()) {
+            Ok(p) => p,
+            Err(_) => {
+                debug!("failed to parse remote's exchange's id protobuf");
+                return Err(PlainTextError::InvalidPayload(None));
+            },
+        };
+
+        // Check the validity of the remote's `Exchange`.
+        if peer_id != public_key.clone().into_peer_id() {
+            debug!("The remote's `PeerId` of the exchange isn't consist with the remote public key");
+            return Err(PlainTextError::InvalidPeerId)
+        }
+
+        Ok(HandshakeContext {
+            config: self.config,
+            state: Remote {
+                peer_id,
+                public_key,
+            }
+        })
+    }
+}
+
+pub fn handshake<S>(socket: S, config: PlainText2Config)
+    -> impl Future<Item = (Framed<S, BytesMut>, Remote), Error = PlainTextError>
+where
+    S: AsyncRead + AsyncWrite + Send,
+{
+    let socket = length_delimited::Builder::new()
+        .big_endian()
+        .length_field_length(4)
+        .new_framed(socket);
+
+    future::ok::<_, PlainTextError>(())
+        .and_then(|_| {
+            trace!("starting handshake");
+            Ok(HandshakeContext::new(config)?)
+        })
+        // Send our local `Exchange`.
+        .and_then(|context| {
+            trace!("sending exchange to remote");
+            socket.send(BytesMut::from(context.state.exchange_bytes.clone()))
+                .from_err()
+                .map(|s| (s, context))
+        })
+        // Receive the remote's `Exchange`.
+        .and_then(move |(socket, context)| {
+            trace!("receiving the remote's exchange");
+            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 exchange");
+                            let err = IoError::new(IoErrorKind::BrokenPipe, "unexpected eof");
+                            return Err(err.into());
+                        }
+                    };
+
+                    trace!("received exchange from remote; pubkey = {:?}", context.state.public_key);
+                    Ok((socket, context.state))
+                })
+        })
+}

protocols/plaintext/src/lib.rs

@@ -1,4 +1,4 @@
-// Copyright 2018 Parity Technologies (UK) Ltd.
+// Copyright 2019 Parity Technologies (UK) Ltd.
 //
 // Permission is hereby granted, free of charge, to any person obtaining a
 // copy of this software and associated documentation files (the "Software"),
@@ -18,15 +18,31 @@
 // FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
 // DEALINGS IN THE SOFTWARE.
 
-use futures::future::{self, FutureResult};
-use libp2p_core::{InboundUpgrade, OutboundUpgrade, UpgradeInfo, upgrade::Negotiated};
+use bytes::BytesMut;
+use futures::{Future, StartSend, Poll, future};
+use futures::sink::Sink;
+use futures::stream::MapErr as StreamMapErr;
+use futures::stream::Stream;
+use libp2p_core::{identity, InboundUpgrade, OutboundUpgrade, UpgradeInfo, upgrade::Negotiated, PeerId, PublicKey};
+use log::debug;
+use rw_stream_sink::RwStreamSink;
+use std::io;
 use std::iter;
+use tokio_io::{AsyncRead, AsyncWrite};
+use tokio_io::codec::length_delimited::Framed;
+use crate::error::PlainTextError;
 use void::Void;
+use futures::future::FutureResult;
+use crate::handshake::Remote;
+
+mod error;
+mod handshake;
+mod pb;
 
 #[derive(Debug, Copy, Clone)]
-pub struct PlainTextConfig;
+pub struct PlainText1Config;
 
-impl UpgradeInfo for PlainTextConfig {
+impl UpgradeInfo for PlainText1Config {
     type Info = &'static [u8];
     type InfoIter = iter::Once<Self::Info>;
 
@@ -35,7 +51,7 @@ impl UpgradeInfo for PlainTextConfig {
     }
 }
 
-impl<C> InboundUpgrade<C> for PlainTextConfig {
+impl<C> InboundUpgrade<C> for PlainText1Config {
     type Output = Negotiated<C>;
     type Error = Void;
     type Future = FutureResult<Negotiated<C>, Self::Error>;
@@ -45,7 +61,7 @@ impl<C> InboundUpgrade<C> for PlainTextConfig {
     }
 }
 
-impl<C> OutboundUpgrade<C> for PlainTextConfig {
+impl<C> OutboundUpgrade<C> for PlainText1Config {
     type Output = Negotiated<C>;
     type Error = Void;
     type Future = FutureResult<Negotiated<C>, Self::Error>;
@@ -55,3 +71,160 @@ impl<C> OutboundUpgrade<C> for PlainTextConfig {
     }
 }
 
+#[derive(Clone)]
+pub struct PlainText2Config {
+    pub local_public_key: identity::PublicKey,
+}
+
+impl UpgradeInfo for PlainText2Config {
+    type Info = &'static [u8];
+    type InfoIter = iter::Once<Self::Info>;
+
+    fn protocol_info(&self) -> Self::InfoIter {
+        iter::once(b"/plaintext/2.0.0")
+    }
+}
+
+impl<C> InboundUpgrade<C> for PlainText2Config
+where
+    C: AsyncRead + AsyncWrite + Send + 'static
+{
+    type Output = (PeerId, 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))
+    }
+}
+
+impl<C> OutboundUpgrade<C> for PlainText2Config
+where
+    C: AsyncRead + AsyncWrite + Send + 'static
+{
+    type Output = (PeerId, 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))
+    }
+}
+
+impl PlainText2Config {
+    fn handshake<T>(self, socket: T) -> impl Future<Item = (PeerId, PlainTextOutput<T>), Error = PlainTextError>
+    where
+        T: AsyncRead + AsyncWrite + Send + 'static
+    {
+        debug!("Starting plaintext upgrade");
+        PlainTextMiddleware::handshake(socket, self)
+            .map(|(stream_sink, remote)| {
+                let mapped = stream_sink.map_err(map_err as fn(_) -> _);
+                (
+                    remote.peer_id,
+                    PlainTextOutput {
+                        stream: RwStreamSink::new(mapped),
+                        remote_key: remote.public_key,
+                    }
+                )
+            })
+    }
+}
+
+#[inline]
+fn map_err(err: io::Error) -> io::Error {
+    debug!("error during plaintext handshake {:?}", err);
+    io::Error::new(io::ErrorKind::InvalidData, err)
+}
+
+pub struct PlainTextMiddleware<S> {
+    inner: Framed<S, BytesMut>,
+}
+
+impl<S> PlainTextMiddleware<S>
+where
+    S: AsyncRead + AsyncWrite + Send,
+{
+    fn handshake(socket: S, config: PlainText2Config)
+        -> impl Future<Item = (PlainTextMiddleware<S>, Remote), Error = PlainTextError>
+    {
+        handshake::handshake(socket, config).map(|(inner, remote)| {
+            (PlainTextMiddleware { inner }, remote)
+        })
+    }
+}
+
+impl<S> Sink for PlainTextMiddleware<S>
+where
+    S: AsyncRead + AsyncWrite,
+{
+    type SinkItem = BytesMut;
+    type SinkError = io::Error;
+
+    #[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 = io::Error;
+
+    #[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,
+{
+    /// The plaintext stream.
+    pub stream: RwStreamSink<StreamMapErr<PlainTextMiddleware<S>, fn(io::Error) -> io::Error>>,
+    /// The public key of the remote.
+    pub remote_key: PublicKey,
+}
+
+impl<S: AsyncRead + AsyncWrite> std::io::Read for PlainTextOutput<S> {
+    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
+        self.stream.read(buf)
+    }
+}
+
+impl<S: AsyncRead + AsyncWrite> AsyncRead for PlainTextOutput<S> {
+    unsafe fn prepare_uninitialized_buffer(&self, buf: &mut [u8]) -> bool {
+        self.stream.prepare_uninitialized_buffer(buf)
+    }
+}
+
+impl<S: AsyncRead + AsyncWrite> std::io::Write for PlainTextOutput<S> {
+    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
+        self.stream.write(buf)
+    }
+
+    fn flush(&mut self) -> std::io::Result<()> {
+        self.stream.flush()
+    }
+}
+
+impl<S: AsyncRead + AsyncWrite> AsyncWrite for PlainTextOutput<S> {
+    fn shutdown(&mut self) -> Poll<(), io::Error> {
+        self.stream.shutdown()
+    }
+}

protocols/plaintext/src/pb.rs

@@ -0,0 +1,21 @@
+// Copyright 2019 Parity Technologies (UK) Ltd.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a
+// copy of this software and associated documentation files (the "Software"),
+// to deal in the Software without restriction, including without limitation
+// the rights to use, copy, modify, merge, publish, distribute, sublicense,
+// and/or sell copies of the Software, and to permit persons to whom the
+// Software is furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
+// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+// DEALINGS IN THE SOFTWARE.
+
+pub mod structs;

protocols/plaintext/src/pb/structs.rs

@@ -0,0 +1,278 @@
+// 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 Exchange {
+    // message fields
+    id: ::protobuf::SingularField<::std::vec::Vec<u8>>,
+    pubkey: ::protobuf::SingularField<::std::vec::Vec<u8>>,
+    // special fields
+    pub unknown_fields: ::protobuf::UnknownFields,
+    pub cached_size: ::protobuf::CachedSize,
+}
+
+impl Exchange {
+    pub fn new() -> Exchange {
+        ::std::default::Default::default()
+    }
+
+    // optional bytes id = 1;
+
+    pub fn clear_id(&mut self) {
+        self.id.clear();
+    }
+
+    pub fn has_id(&self) -> bool {
+        self.id.is_some()
+    }
+
+    // Param is passed by value, moved
+    pub fn set_id(&mut self, v: ::std::vec::Vec<u8>) {
+        self.id = ::protobuf::SingularField::some(v);
+    }
+
+    // Mutable pointer to the field.
+    // If field is not initialized, it is initialized with default value first.
+    pub fn mut_id(&mut self) -> &mut ::std::vec::Vec<u8> {
+        if self.id.is_none() {
+            self.id.set_default();
+        }
+        self.id.as_mut().unwrap()
+    }
+
+    // Take field
+    pub fn take_id(&mut self) -> ::std::vec::Vec<u8> {
+        self.id.take().unwrap_or_else(|| ::std::vec::Vec::new())
+    }
+
+    pub fn get_id(&self) -> &[u8] {
+        match self.id.as_ref() {
+            Some(v) => &v,
+            None => &[],
+        }
+    }
+
+    // optional bytes pubkey = 2;
+
+    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 Exchange {
+    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.id)?;
+                },
+                2 => {
+                    ::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.id.as_ref() {
+            my_size += ::protobuf::rt::bytes_size(1, &v);
+        }
+        if let Some(ref v) = self.pubkey.as_ref() {
+            my_size += ::protobuf::rt::bytes_size(2, &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.id.as_ref() {
+            os.write_bytes(1, &v)?;
+        }
+        if let Some(ref v) = self.pubkey.as_ref() {
+            os.write_bytes(2, &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() -> Exchange {
+        Exchange::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>(
+                    "id",
+                    |m: &Exchange| { &m.id },
+                    |m: &mut Exchange| { &mut m.id },
+                ));
+                fields.push(::protobuf::reflect::accessor::make_singular_field_accessor::<_, ::protobuf::types::ProtobufTypeBytes>(
+                    "pubkey",
+                    |m: &Exchange| { &m.pubkey },
+                    |m: &mut Exchange| { &mut m.pubkey },
+                ));
+                ::protobuf::reflect::MessageDescriptor::new::<Exchange>(
+                    "Exchange",
+                    fields,
+                    file_descriptor_proto()
+                )
+            })
+        }
+    }
+
+    fn default_instance() -> &'static Exchange {
+        static mut instance: ::protobuf::lazy::Lazy<Exchange> = ::protobuf::lazy::Lazy {
+            lock: ::protobuf::lazy::ONCE_INIT,
+            ptr: 0 as *const Exchange,
+        };
+        unsafe {
+            instance.get(Exchange::new)
+        }
+    }
+}
+
+impl ::protobuf::Clear for Exchange {
+    fn clear(&mut self) {
+        self.clear_id();
+        self.clear_pubkey();
+        self.unknown_fields.clear();
+    }
+}
+
+impl ::std::fmt::Debug for Exchange {
+    fn fmt(&self, f: &mut ::std::fmt::Formatter) -> ::std::fmt::Result {
+        ::protobuf::text_format::fmt(self, f)
+    }
+}
+
+impl ::protobuf::reflect::ProtobufValue for Exchange {
+    fn as_ref(&self) -> ::protobuf::reflect::ProtobufValueRef {
+        ::protobuf::reflect::ProtobufValueRef::Message(self)
+    }
+}
+
+static file_descriptor_proto_data: &'static [u8] = b"\
+    \n!protocols/plaintext/structs.proto\"2\n\x08Exchange\x12\x0e\n\x02id\
+    \x18\x01\x20\x01(\x0cR\x02id\x12\x16\n\x06pubkey\x18\x02\x20\x01(\x0cR\
+    \x06pubkeyJ\xb4\x01\n\x06\x12\x04\0\0\x05\x01\n\x08\n\x01\x0c\x12\x03\0\
+    \0\x12\n\n\n\x02\x04\0\x12\x04\x02\0\x05\x01\n\n\n\x03\x04\0\x01\x12\x03\
+    \x02\x08\x10\n\x0b\n\x04\x04\0\x02\0\x12\x03\x03\x02\x18\n\x0c\n\x05\x04\
+    \0\x02\0\x04\x12\x03\x03\x02\n\n\x0c\n\x05\x04\0\x02\0\x05\x12\x03\x03\
+    \x0b\x10\n\x0c\n\x05\x04\0\x02\0\x01\x12\x03\x03\x11\x13\n\x0c\n\x05\x04\
+    \0\x02\0\x03\x12\x03\x03\x16\x17\n\x0b\n\x04\x04\0\x02\x01\x12\x03\x04\
+    \x02\x1c\n\x0c\n\x05\x04\0\x02\x01\x04\x12\x03\x04\x02\n\n\x0c\n\x05\x04\
+    \0\x02\x01\x05\x12\x03\x04\x0b\x10\n\x0c\n\x05\x04\0\x02\x01\x01\x12\x03\
+    \x04\x11\x17\n\x0c\n\x05\x04\0\x02\x01\x03\x12\x03\x04\x1a\x1b\
+";
+
+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()
+        })
+    }
+}

protocols/plaintext/structs.proto

@@ -0,0 +1,6 @@
+syntax = "proto2";
+
+message Exchange {
+  optional bytes id = 1;
+  optional bytes pubkey = 2;
+}