Implement /plaintext/2.0.0

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))
+                })
+        })
+}