openethereum/util/network-devp2p/src/connection.rs

// Copyright 2015-2020 Parity Technologies (UK) Ltd.
// This file is part of OpenEthereum.

// OpenEthereum 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.

// OpenEthereum 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 OpenEthereum.  If not, see <http://www.gnu.org/licenses/>.

use bytes::{Buf, BufMut};
use ethereum_types::{H128, H256, H512};
use ethkey::crypto;
use handshake::Handshake;
use hash::{keccak, write_keccak};
use io::{IoContext, StreamToken};
use mio::{
    deprecated::{EventLoop, Handler, TryRead, TryWrite},
    tcp::*,
    PollOpt, Ready, Token,
};
use network::{Error, ErrorKind};
use parity_bytes::*;
use rcrypto::{aessafe::*, blockmodes::*, buffer::*, symmetriccipher::*};
use rlp::{Rlp, RlpStream};
use std::{
    collections::VecDeque,
    io::{self, Cursor, Read, Write},
    net::SocketAddr,
    sync::atomic::{AtomicBool, Ordering as AtomicOrdering},
    time::Duration,
};
use tiny_keccak::Keccak;

const ENCRYPTED_HEADER_LEN: usize = 32;
const RECEIVE_PAYLOAD: Duration = Duration::from_secs(30);
pub const MAX_PAYLOAD_SIZE: usize = (1 << 24) - 1;

/// Network responses should try not to go over this limit.
/// This should be lower than MAX_PAYLOAD_SIZE
#[cfg(not(test))]
pub const PAYLOAD_SOFT_LIMIT: usize = (1 << 22) - 1;
#[cfg(test)]
pub const PAYLOAD_SOFT_LIMIT: usize = 100_000;

pub trait GenericSocket: Read + Write {}

impl GenericSocket for TcpStream {}

pub struct GenericConnection<Socket: GenericSocket> {
    /// Connection id (token)
    pub token: StreamToken,
    /// Network socket
    pub socket: Socket,
    /// Receive buffer
    rec_buf: Bytes,
    /// Expected size
    rec_size: usize,
    /// Send out packets FIFO
    send_queue: VecDeque<Cursor<Bytes>>,
    /// Event flags this connection expects
    interest: Ready,
    /// Registered flag
    registered: AtomicBool,
}

impl<Socket: GenericSocket> GenericConnection<Socket> {
    pub fn expect(&mut self, size: usize) {
        trace!(target:"network", "Expect to read {} bytes", size);
        if self.rec_size != self.rec_buf.len() {
            warn!(target:"network", "Unexpected connection read start");
        }
        self.rec_size = size;
    }

    /// Readable IO handler. Called when there is some data to be read.
    pub fn readable(&mut self) -> io::Result<Option<Bytes>> {
        if self.rec_size == 0 || self.rec_buf.len() >= self.rec_size {
            return Ok(None);
        }
        let sock_ref = <Socket as Read>::by_ref(&mut self.socket);
        loop {
            let max = self.rec_size - self.rec_buf.len();
            match sock_ref
                .take(max as u64)
                .try_read(unsafe { self.rec_buf.bytes_mut() })
            {
                Ok(Some(size)) if size != 0 => {
                    unsafe {
                        self.rec_buf.advance_mut(size);
                    }
                    trace!(target:"network", "{}: Read {} of {} bytes", self.token, self.rec_buf.len(), self.rec_size);
                    if self.rec_size != 0 && self.rec_buf.len() == self.rec_size {
                        self.rec_size = 0;
                        return Ok(Some(::std::mem::replace(&mut self.rec_buf, Bytes::new())));
                    } else if self.rec_buf.len() > self.rec_size {
                        warn!(target:"network", "Read past buffer {} bytes", self.rec_buf.len() - self.rec_size);
                        return Ok(Some(::std::mem::replace(&mut self.rec_buf, Bytes::new())));
                    }
                }
                Ok(_) => return Ok(None),
                Err(e) => {
                    debug!(target:"network", "Read error {} ({})", self.token, e);
                    return Err(e);
                }
            }
        }
    }

    /// Add a packet to send queue.
    pub fn send<Message>(&mut self, io: &IoContext<Message>, data: Bytes)
    where
        Message: Send + Clone + Sync + 'static,
    {
        if !data.is_empty() {
            trace!(target:"network", "{}: Sending {} bytes", self.token, data.len());
            self.send_queue.push_back(Cursor::new(data));
            if !self.interest.is_writable() {
                self.interest.insert(Ready::writable());
            }
            io.update_registration(self.token).ok();
        }
    }

    /// Check if this connection has data to be sent.
    pub fn is_sending(&self) -> bool {
        self.interest.is_writable()
    }

    /// Writable IO handler. Called when the socket is ready to send.
    pub fn writable<Message>(&mut self, io: &IoContext<Message>) -> Result<WriteStatus, Error>
    where
        Message: Send + Clone + Sync + 'static,
    {
        {
            let buf = match self.send_queue.front_mut() {
                Some(buf) => buf,
                None => return Ok(WriteStatus::Complete),
            };
            let send_size = buf.get_ref().len();
            let pos = buf.position() as usize;
            if (pos as usize) >= send_size {
                warn!(target:"net", "Unexpected connection data");
                return Ok(WriteStatus::Complete);
            }

            match self.socket.try_write(Buf::bytes(&buf)) {
                Ok(Some(size)) if (pos + size) < send_size => {
                    buf.advance(size);
                    Ok(WriteStatus::Ongoing)
                }
                Ok(Some(size)) if (pos + size) == send_size => {
                    trace!(target:"network", "{}: Wrote {} bytes", self.token, send_size);
                    Ok(WriteStatus::Complete)
                }
                Ok(Some(_)) => {
                    panic!("Wrote past buffer");
                }
                Ok(None) => Ok(WriteStatus::Ongoing),
                Err(e) => Err(e)?,
            }
        }
        .and_then(|r| {
            if r == WriteStatus::Complete {
                self.send_queue.pop_front();
            }
            if self.send_queue.is_empty() {
                self.interest.remove(Ready::writable());
            }
            io.update_registration(self.token)?;
            Ok(r)
        })
    }
}

/// Low level tcp connection
pub type Connection = GenericConnection<TcpStream>;

impl Connection {
    /// Create a new connection with given id and socket.
    pub fn new(token: StreamToken, socket: TcpStream) -> Connection {
        Connection {
            token,
            socket,
            send_queue: VecDeque::new(),
            rec_buf: Bytes::new(),
            rec_size: 0,
            interest: Ready::hup() | Ready::readable(),
            registered: AtomicBool::new(false),
        }
    }

    /// Get socket token
    pub fn token(&self) -> StreamToken {
        self.token
    }

    /// Get remote peer address
    pub fn remote_addr(&self) -> io::Result<SocketAddr> {
        self.socket.peer_addr()
    }

    /// Get remote peer address string
    pub fn remote_addr_str(&self) -> String {
        self.socket
            .peer_addr()
            .map(|a| a.to_string())
            .unwrap_or_else(|_| "Unknown".to_owned())
    }

    /// Get local peer address string
    pub fn local_addr_str(&self) -> String {
        self.socket
            .local_addr()
            .map(|a| a.to_string())
            .unwrap_or_else(|_| "Unknown".to_owned())
    }

    /// Clone this connection. Clears the receiving buffer of the returned connection.
    pub fn try_clone(&self) -> io::Result<Self> {
        Ok(Connection {
            token: self.token,
            socket: self.socket.try_clone()?,
            rec_buf: Vec::new(),
            rec_size: 0,
            send_queue: self.send_queue.clone(),
            interest: Ready::hup(),
            registered: AtomicBool::new(false),
        })
    }

    /// Register this connection with the IO event loop.
    pub fn register_socket<Host: Handler>(
        &self,
        reg: Token,
        event_loop: &mut EventLoop<Host>,
    ) -> io::Result<()> {
        if self.registered.load(AtomicOrdering::SeqCst) {
            return Ok(());
        }
        trace!(target: "network", "connection register; token={:?}", reg);
        if let Err(e) = event_loop.register(
            &self.socket,
            reg,
            self.interest,
            PollOpt::edge(), /* | PollOpt::oneshot() */
        ) {
            // TODO: oneshot is broken on windows
            trace!(target: "network", "Failed to register {:?}, {:?}", reg, e);
        }
        self.registered.store(true, AtomicOrdering::SeqCst);
        Ok(())
    }

    /// Update connection registration. Should be called at the end of the IO handler.
    pub fn update_socket<Host: Handler>(
        &self,
        reg: Token,
        event_loop: &mut EventLoop<Host>,
    ) -> io::Result<()> {
        trace!(target: "network", "connection reregister; token={:?}", reg);
        if !self.registered.load(AtomicOrdering::SeqCst) {
            self.register_socket(reg, event_loop)
        } else {
            event_loop
                .reregister(
                    &self.socket,
                    reg,
                    self.interest,
                    PollOpt::edge(), /* | PollOpt::oneshot() */
                )
                .unwrap_or_else(|e| {
                    // TODO: oneshot is broken on windows
                    trace!(target: "network", "Failed to reregister {:?}, {:?}", reg, e);
                });
            Ok(())
        }
    }

    /// Delete connection registration. Should be called at the end of the IO handler.
    pub fn deregister_socket<Host: Handler>(
        &self,
        event_loop: &mut EventLoop<Host>,
    ) -> io::Result<()> {
        trace!(target: "network", "connection deregister; token={:?}", self.token);
        event_loop.deregister(&self.socket).ok(); // ignore errors here
        Ok(())
    }
}

/// Connection write status.
#[derive(PartialEq, Eq)]
pub enum WriteStatus {
    /// Some data is still pending for current packet
    Ongoing,
    /// All data sent.
    Complete,
}

/// `RLPx` packet
pub struct Packet {
    pub protocol: u16,
    pub data: Bytes,
}

/// Encrypted connection receiving state.
enum EncryptedConnectionState {
    /// Reading a header.
    Header,
    /// Reading the rest of the packet.
    Payload,
}

/// Connection implementing `RLPx` framing
/// https://github.com/ethereum/devp2p/blob/master/rlpx.md#framing
pub struct EncryptedConnection {
    /// Underlying tcp connection
    pub connection: Connection,
    /// Egress data encryptor
    encoder: CtrMode<AesSafe256Encryptor>,
    /// Ingress data decryptor
    decoder: CtrMode<AesSafe256Encryptor>,
    /// Ingress data decryptor
    mac_encoder: EcbEncryptor<AesSafe256Encryptor, EncPadding<NoPadding>>,
    /// MAC for egress data
    egress_mac: Keccak,
    /// MAC for ingress data
    ingress_mac: Keccak,
    /// Read state
    read_state: EncryptedConnectionState,
    /// Protocol id for the last received packet
    protocol_id: u16,
    /// Payload expected to be received for the last header.
    payload_len: usize,
}

impl EncryptedConnection {
    /// Create an encrypted connection out of the handshake.
    pub fn new(handshake: &mut Handshake) -> Result<EncryptedConnection, Error> {
        let shared = crypto::ecdh::agree(handshake.ecdhe.secret(), &handshake.remote_ephemeral)?;
        let mut nonce_material = H512::new();
        if handshake.originated {
            handshake.remote_nonce.copy_to(&mut nonce_material[0..32]);
            handshake.nonce.copy_to(&mut nonce_material[32..64]);
        } else {
            handshake.nonce.copy_to(&mut nonce_material[0..32]);
            handshake.remote_nonce.copy_to(&mut nonce_material[32..64]);
        }
        let mut key_material = H512::new();
        shared.copy_to(&mut key_material[0..32]);
        write_keccak(&nonce_material, &mut key_material[32..64]);
        keccak(&key_material).copy_to(&mut key_material[32..64]);
        keccak(&key_material).copy_to(&mut key_material[32..64]);

        let iv = vec![0u8; 16];
        let encoder = CtrMode::new(AesSafe256Encryptor::new(&key_material[32..64]), iv);
        let iv = vec![0u8; 16];
        let decoder = CtrMode::new(AesSafe256Encryptor::new(&key_material[32..64]), iv);

        keccak(&key_material).copy_to(&mut key_material[32..64]);
        let mac_encoder =
            EcbEncryptor::new(AesSafe256Encryptor::new(&key_material[32..64]), NoPadding);

        let mut egress_mac = Keccak::new_keccak256();
        let mut mac_material = H256::from_slice(&key_material[32..64]) ^ handshake.remote_nonce;
        egress_mac.update(&mac_material);
        egress_mac.update(if handshake.originated {
            &handshake.auth_cipher
        } else {
            &handshake.ack_cipher
        });

        let mut ingress_mac = Keccak::new_keccak256();
        mac_material = H256::from_slice(&key_material[32..64]) ^ handshake.nonce;
        ingress_mac.update(&mac_material);
        ingress_mac.update(if handshake.originated {
            &handshake.ack_cipher
        } else {
            &handshake.auth_cipher
        });

        let old_connection = handshake.connection.try_clone()?;
        let connection = ::std::mem::replace(&mut handshake.connection, old_connection);
        let mut enc = EncryptedConnection {
            connection,
            encoder,
            decoder,
            mac_encoder,
            egress_mac,
            ingress_mac,
            read_state: EncryptedConnectionState::Header,
            protocol_id: 0,
            payload_len: 0,
        };
        enc.connection.expect(ENCRYPTED_HEADER_LEN);
        Ok(enc)
    }

    /// Send a packet
    pub fn send_packet<Message>(
        &mut self,
        io: &IoContext<Message>,
        payload: &[u8],
    ) -> Result<(), Error>
    where
        Message: Send + Clone + Sync + 'static,
    {
        let mut header = RlpStream::new();
        let len = payload.len();
        if len > MAX_PAYLOAD_SIZE {
            bail!(ErrorKind::OversizedPacket);
        }
        header.append_raw(&[(len >> 16) as u8, (len >> 8) as u8, len as u8], 1);
        header.append_raw(&[0xc2u8, 0x80u8, 0x80u8], 1);
        //TODO: get rid of vectors here
        let mut header = header.out();
        let padding = (16 - (payload.len() % 16)) % 16;
        header.resize(16, 0u8);

        let mut packet = vec![0u8; 32 + payload.len() + padding + 16];
        self.encoder
            .encrypt(
                &mut RefReadBuffer::new(&header),
                &mut RefWriteBuffer::new(&mut packet),
                false,
            )
            .expect("Invalid length or padding");
        EncryptedConnection::update_mac(
            &mut self.egress_mac,
            &mut self.mac_encoder,
            &packet[0..16],
        );
        self.egress_mac.clone().finalize(&mut packet[16..32]);
        self.encoder
            .encrypt(
                &mut RefReadBuffer::new(payload),
                &mut RefWriteBuffer::new(&mut packet[32..(32 + len)]),
                padding == 0,
            )
            .expect("Invalid length or padding");
        if padding != 0 {
            let pad = [0u8; 16];
            self.encoder
                .encrypt(
                    &mut RefReadBuffer::new(&pad[0..padding]),
                    &mut RefWriteBuffer::new(&mut packet[(32 + len)..(32 + len + padding)]),
                    true,
                )
                .expect("Invalid length or padding");
        }
        self.egress_mac.update(&packet[32..(32 + len + padding)]);
        EncryptedConnection::update_mac(&mut self.egress_mac, &mut self.mac_encoder, &[0u8; 0]);
        self.egress_mac
            .clone()
            .finalize(&mut packet[(32 + len + padding)..]);
        self.connection.send(io, packet);
        Ok(())
    }

    /// Decrypt and authenticate an incoming packet header. Prepare for receiving payload.
    fn read_header(&mut self, header: &[u8]) -> Result<(), Error> {
        if header.len() != ENCRYPTED_HEADER_LEN {
            return Err(ErrorKind::Auth.into());
        }
        EncryptedConnection::update_mac(
            &mut self.ingress_mac,
            &mut self.mac_encoder,
            &header[0..16],
        );
        let mac = &header[16..];
        let mut expected = H256::new();
        self.ingress_mac.clone().finalize(&mut expected);
        if mac != &expected[0..16] {
            return Err(ErrorKind::Auth.into());
        }

        let mut hdec = H128::new();
        self.decoder
            .decrypt(
                &mut RefReadBuffer::new(&header[0..16]),
                &mut RefWriteBuffer::new(&mut hdec),
                false,
            )
            .expect("Invalid length or padding");

        let length = ((((hdec[0] as u32) << 8) + (hdec[1] as u32)) << 8) + (hdec[2] as u32);
        let header_rlp = Rlp::new(&hdec[3..6]);
        let protocol_id = header_rlp.val_at::<u16>(0)?;

        self.payload_len = length as usize;
        self.protocol_id = protocol_id;
        self.read_state = EncryptedConnectionState::Payload;

        let padding = (16 - (length % 16)) % 16;
        let full_length = length + padding + 16;
        self.connection.expect(full_length as usize);
        Ok(())
    }

    /// Decrypt and authenticate packet payload.
    fn read_payload(&mut self, payload: &[u8]) -> Result<Packet, Error> {
        let padding = (16 - (self.payload_len % 16)) % 16;
        let full_length = self.payload_len + padding + 16;
        if payload.len() != full_length {
            return Err(ErrorKind::Auth.into());
        }
        self.ingress_mac.update(&payload[0..payload.len() - 16]);
        EncryptedConnection::update_mac(&mut self.ingress_mac, &mut self.mac_encoder, &[0u8; 0]);
        let mac = &payload[(payload.len() - 16)..];
        let mut expected = H128::new();
        self.ingress_mac.clone().finalize(&mut expected);
        if mac != &expected[..] {
            return Err(ErrorKind::Auth.into());
        }

        let mut packet = vec![0u8; self.payload_len];
        self.decoder
            .decrypt(
                &mut RefReadBuffer::new(&payload[0..self.payload_len]),
                &mut RefWriteBuffer::new(&mut packet),
                false,
            )
            .expect("Invalid length or padding");
        let mut pad_buf = [0u8; 16];
        self.decoder
            .decrypt(
                &mut RefReadBuffer::new(&payload[self.payload_len..(payload.len() - 16)]),
                &mut RefWriteBuffer::new(&mut pad_buf),
                false,
            )
            .expect("Invalid length or padding");
        Ok(Packet {
            protocol: self.protocol_id,
            data: packet,
        })
    }

    /// Update MAC after reading or writing any data.
    fn update_mac(
        mac: &mut Keccak,
        mac_encoder: &mut EcbEncryptor<AesSafe256Encryptor, EncPadding<NoPadding>>,
        seed: &[u8],
    ) {
        let mut prev = H128::new();
        mac.clone().finalize(&mut prev);
        let mut enc = H128::new();
        mac_encoder
            .encrypt(
                &mut RefReadBuffer::new(&prev),
                &mut RefWriteBuffer::new(&mut enc),
                true,
            )
            .expect("Error updating MAC");
        mac_encoder.reset();

        enc = enc
            ^ if seed.is_empty() {
                prev
            } else {
                H128::from_slice(seed)
            };
        mac.update(&enc);
    }

    /// Readable IO handler. Tracker receive status and returns decoded packet if available.
    pub fn readable<Message>(&mut self, io: &IoContext<Message>) -> Result<Option<Packet>, Error>
    where
        Message: Send + Clone + Sync + 'static,
    {
        io.clear_timer(self.connection.token)?;
        if let EncryptedConnectionState::Header = self.read_state {
            if let Some(data) = self.connection.readable()? {
                self.read_header(&data)?;
                io.register_timer(self.connection.token, RECEIVE_PAYLOAD)?;
            }
        };
        if let EncryptedConnectionState::Payload = self.read_state {
            match self.connection.readable()? {
                Some(data) => {
                    self.read_state = EncryptedConnectionState::Header;
                    self.connection.expect(ENCRYPTED_HEADER_LEN);
                    Ok(Some(self.read_payload(&data)?))
                }
                None => Ok(None),
            }
        } else {
            Ok(None)
        }
    }

    /// Writable IO handler. Processes send queue.
    pub fn writable<Message>(&mut self, io: &IoContext<Message>) -> Result<(), Error>
    where
        Message: Send + Clone + Sync + 'static,
    {
        self.connection.writable(io)?;
        Ok(())
    }
}

#[test]
pub fn test_encryption() {
    use ethereum_types::{H128, H256};
    use std::str::FromStr;
    let key =
        H256::from_str("2212767d793a7a3d66f869ae324dd11bd17044b82c9f463b8a541a4d089efec5").unwrap();
    let before = H128::from_str("12532abaec065082a3cf1da7d0136f15").unwrap();
    let before2 = H128::from_str("7e99f682356fdfbc6b67a9562787b18a").unwrap();
    let after = H128::from_str("89464c6b04e7c99e555c81d3f7266a05").unwrap();
    let after2 = H128::from_str("85c070030589ef9c7a2879b3a8489316").unwrap();

    let mut got = H128::new();

    let mut encoder = EcbEncryptor::new(AesSafe256Encryptor::new(&key), NoPadding);
    encoder
        .encrypt(
            &mut RefReadBuffer::new(&before),
            &mut RefWriteBuffer::new(&mut got),
            true,
        )
        .unwrap();
    encoder.reset();
    assert_eq!(got, after);
    got = H128::new();
    encoder
        .encrypt(
            &mut RefReadBuffer::new(&before2),
            &mut RefWriteBuffer::new(&mut got),
            true,
        )
        .unwrap();
    encoder.reset();
    assert_eq!(got, after2);
}

#[cfg(test)]
mod tests {
    use std::{
        cmp,
        collections::VecDeque,
        io::{Cursor, Error, ErrorKind, Read, Result, Write},
        sync::atomic::AtomicBool,
    };

    use super::*;
    use io::*;
    use mio::Ready;
    use parity_bytes::Bytes;

    pub struct TestSocket {
        pub read_buffer: Vec<u8>,
        pub write_buffer: Vec<u8>,
        pub cursor: usize,
        pub buf_size: usize,
    }

    impl Default for TestSocket {
        fn default() -> Self {
            TestSocket::new()
        }
    }

    impl TestSocket {
        pub fn new() -> Self {
            TestSocket {
                read_buffer: vec![],
                write_buffer: vec![],
                cursor: 0,
                buf_size: 0,
            }
        }

        pub fn new_buf(buf_size: usize) -> TestSocket {
            TestSocket {
                read_buffer: vec![],
                write_buffer: vec![],
                cursor: 0,
                buf_size,
            }
        }
    }

    impl Read for TestSocket {
        fn read(&mut self, buf: &mut [u8]) -> Result<usize> {
            let end_position = cmp::min(self.read_buffer.len(), self.cursor + buf.len());
            if self.cursor > end_position {
                return Ok(0);
            }
            let len = cmp::max(end_position - self.cursor, 0);
            match len {
                0 => Ok(0),
                _ => {
                    for i in self.cursor..end_position {
                        buf[i - self.cursor] = self.read_buffer[i];
                    }
                    self.cursor = end_position;
                    Ok(len)
                }
            }
        }
    }

    impl Write for TestSocket {
        fn write(&mut self, buf: &[u8]) -> Result<usize> {
            if self.buf_size == 0 || buf.len() < self.buf_size {
                self.write_buffer.extend(buf.iter().cloned());
                Ok(buf.len())
            } else {
                self.write_buffer
                    .extend(buf.iter().take(self.buf_size).cloned());
                Ok(self.buf_size)
            }
        }

        fn flush(&mut self) -> Result<()> {
            unimplemented!();
        }
    }

    impl GenericSocket for TestSocket {}

    struct TestBrokenSocket {
        error: String,
    }

    impl Read for TestBrokenSocket {
        fn read(&mut self, _: &mut [u8]) -> Result<usize> {
            Err(Error::new(ErrorKind::Other, self.error.clone()))
        }
    }

    impl Write for TestBrokenSocket {
        fn write(&mut self, _: &[u8]) -> Result<usize> {
            Err(Error::new(ErrorKind::Other, self.error.clone()))
        }

        fn flush(&mut self) -> Result<()> {
            unimplemented!();
        }
    }

    impl GenericSocket for TestBrokenSocket {}

    type TestConnection = GenericConnection<TestSocket>;

    impl Default for TestConnection {
        fn default() -> Self {
            TestConnection::new()
        }
    }

    impl TestConnection {
        pub fn new() -> Self {
            TestConnection {
                token: 999998888usize,
                socket: TestSocket::new(),
                send_queue: VecDeque::new(),
                rec_buf: Bytes::new(),
                rec_size: 0,
                interest: Ready::hup() | Ready::readable(),
                registered: AtomicBool::new(false),
            }
        }
    }

    type TestBrokenConnection = GenericConnection<TestBrokenSocket>;

    impl Default for TestBrokenConnection {
        fn default() -> Self {
            TestBrokenConnection::new()
        }
    }

    impl TestBrokenConnection {
        pub fn new() -> Self {
            TestBrokenConnection {
                token: 999998888usize,
                socket: TestBrokenSocket {
                    error: "test broken socket".to_owned(),
                },
                send_queue: VecDeque::new(),
                rec_buf: Bytes::new(),
                rec_size: 0,
                interest: Ready::hup() | Ready::readable(),
                registered: AtomicBool::new(false),
            }
        }
    }

    fn test_io() -> IoContext<i32> {
        IoContext::new(IoChannel::disconnected(), 0)
    }

    #[test]
    fn connection_expect() {
        let mut connection = TestConnection::new();
        connection.expect(1024);
        assert_eq!(1024, connection.rec_size);
    }

    #[test]
    fn connection_write_empty() {
        let mut connection = TestConnection::new();
        let status = connection.writable(&test_io());
        assert!(status.is_ok());
        assert!(WriteStatus::Complete == status.unwrap());
    }

    #[test]
    fn connection_write() {
        let mut connection = TestConnection::new();
        let data = Cursor::new(vec![0; 10240]);
        connection.send_queue.push_back(data);

        let status = connection.writable(&test_io());
        assert!(status.is_ok());
        assert!(WriteStatus::Complete == status.unwrap());
        assert_eq!(10240, connection.socket.write_buffer.len());
    }

    #[test]
    fn connection_write_is_buffered() {
        let mut connection = TestConnection::new();
        connection.socket = TestSocket::new_buf(1024);
        let data = Cursor::new(vec![0; 10240]);
        connection.send_queue.push_back(data);

        let status = connection.writable(&test_io());

        assert!(status.is_ok());
        assert!(WriteStatus::Ongoing == status.unwrap());
        assert_eq!(1024, connection.socket.write_buffer.len());
    }

    #[test]
    fn connection_write_to_broken() {
        let mut connection = TestBrokenConnection::new();
        let data = Cursor::new(vec![0; 10240]);
        connection.send_queue.push_back(data);

        let status = connection.writable(&test_io());

        assert!(!status.is_ok());
        assert_eq!(1, connection.send_queue.len());
    }

    #[test]
    fn connection_read() {
        let mut connection = TestConnection::new();
        connection.rec_size = 2048;
        connection.rec_buf = vec![10; 1024];
        connection.socket.read_buffer = vec![99; 2048];

        let status = connection.readable();

        assert!(status.is_ok());
        assert_eq!(1024, connection.socket.cursor);
    }

    #[test]
    fn connection_read_from_broken() {
        let mut connection = TestBrokenConnection::new();
        connection.rec_size = 2048;

        let status = connection.readable();
        assert!(!status.is_ok());
        assert_eq!(0, connection.rec_buf.len());
    }

    #[test]
    fn connection_read_nothing() {
        let mut connection = TestConnection::new();
        connection.rec_size = 2048;

        let status = connection.readable();

        assert!(status.is_ok());
        assert_eq!(0, connection.rec_buf.len());
    }

    #[test]
    fn connection_read_full() {
        let mut connection = TestConnection::new();
        connection.rec_size = 1024;
        connection.rec_buf = vec![76; 1024];

        let status = connection.readable();

        assert!(status.is_ok());
        assert_eq!(0, connection.socket.cursor);
    }
}