Versions + really asynchronous IO

Ignore-this: 32abf6faee2d4be9c3a29058a0ad762a

darcs-hash:9e8e17abf3ee2c5e396527ad7d93edfe5c96445d

Author: P-E-Meunier

Cargo.toml

@@ -10,25 +10,25 @@ documentation = "https://pijul.org/thrussh/doc/thrussh"
 license = "Apache-2.0"
 include = [
         "Cargo.toml",
-        "src/server/mod.rs",
-        "src/server/encrypted.rs",
-        "src/negociation.rs",
-        "src/kex.rs",
         "src/auth.rs",
-        "src/cryptobuf.rs",
-        "src/cipher",
-        "src/cipher/mod.rs",
-        "src/cipher/chacha20poly1305.rs",
-        "src/msg.rs",
-        "src/lib.rs",
         "src/encoding.rs",
-        "src/client",
-        "src/client/mod.rs",
-        "src/client/encrypted.rs",
+        "src/kex.rs",
         "src/key.rs",
+        "src/lib.rs",
+        "src/msg.rs",
+        "src/negociation.rs",
+        "src/pty.rs",
         "src/session.rs",
         "src/sshbuffer.rs",
-        "src/pty.rs"
+        "src/bin/client.rs",
+        "src/bin/server.rs",
+        "src/cipher/chacha20poly1305.rs",
+        "src/cipher/clear.rs",
+        "src/cipher/mod.rs",
+        "src/client/mod.rs",
+        "src/client/encrypted.rs",
+        "src/server/mod.rs",
+        "src/server/encrypted.rs"
         ]
 
 [dependencies]
@@ -37,19 +37,19 @@ byteorder = "0.5"
 bitflags = "0.7"
 libc = "0.2"
 log = "0.3"
-ring = { path = "../ring" }
+ring = "0.6.0-alpha"
 rustc-serialize = "0.3"
 time = "0.1"
 untrusted = "0.3.1"
 mio = "0.6"
 regex = "0.1"
 user = "0.1.1"
-cryptovec = "0.2"
+cryptovec = "0.3"
 
-tokio-core = "*"
-futures = "*"
+tokio-core = "0.1"
+futures = "0.1"
 
-env_logger = "*"
+env_logger = "0.3"
 
 [dev-dependencies]
 tempdir="0.3"

src/cipher/mod.rs

@@ -14,9 +14,7 @@
 //
 use byteorder::{ByteOrder, BigEndian};
 use Error;
-use std::io::BufRead;
 use std;
-use cryptovec::CryptoVec;
 use sshbuffer::SSHBuffer;
 use std::num::Wrapping;
 
@@ -106,53 +104,49 @@ pub trait SealingKey {
     fn seal(&self, seqn: u32, plaintext_in_ciphertext_out: &mut [u8], tag_out: &mut [u8]);
 }
 
-/// Fills the read buffer, and returns whether a complete message has been read.
-fn read(stream: &mut BufRead,
-        read_buffer: &mut CryptoVec,
-        read_len: usize,
-        bytes_read: &mut usize)
-        -> Result<(), Error> {
-    // This loop consumes something or returns, it cannot loop forever.
-    loop {
-        let consumed_len = {
-            let buf = try!(stream.fill_buf());
-            if read_buffer.len() + buf.len() < read_len + 4 {
-                read_buffer.extend(buf);
-                buf.len()
-            } else {
-                let consumed_len = read_len + 4 - read_buffer.len();
-                read_buffer.extend(&buf[0..consumed_len]);
-                consumed_len
-            }
-        };
-        stream.consume(consumed_len);
-        *bytes_read += consumed_len;
-        if read_buffer.len() >= 4 + read_len {
-            return Ok(());
-        }
-    }
-}
-
-
 impl CipherPair {
-    pub fn read<'a>(&self,
-                    stream: &mut BufRead,
-                    buffer: &'a mut SSHBuffer)
-                    -> Result<&'a [u8], Error> {
+    pub fn read<'a, R:std::io::Read>(&self,
+                                     mut stream: R,
+                                     buffer: &'a mut SSHBuffer)
+                                     -> Result<&'a [u8], Error> {
 
         let key = self.remote_to_local.as_opening_key();
 
         let seqn = buffer.seqn.0;
 
         if buffer.len == 0 {
+
             buffer.buffer.clear();
-            let mut len = [0; 4];
-            try!(stream.read_exact(&mut len));
-            buffer.buffer.extend(&len);
-            let len = key.decrypt_packet_length(seqn, len);
+
+            while buffer.read_len_bytes < 4 {
+                debug!("cipherpair: reading");
+                let r = buffer.read_len_bytes;
+                let extra = try!(stream.read(&mut buffer.len_bytes[r..]));
+                if extra == 0 {
+                    return Ok(&[])
+                } else {
+                    buffer.read_len_bytes += extra
+                }
+            }
+
+            buffer.buffer.extend(&buffer.len_bytes);
+            let len = key.decrypt_packet_length(seqn, buffer.len_bytes);
             buffer.len = BigEndian::read_u32(&len) as usize + key.tag_len();
+            buffer.read_len_bytes = 0;
+
+        }
+
+        while buffer.buffer.len() < 4 + buffer.len {
+            // try!(read(stream, &mut buffer.buffer, buffer.len, &mut buffer.bytes));
+            debug!("cipherpair: reading ciphertext");
+            let current_len = buffer.buffer.len();
+            let n = try!(buffer.buffer.read(4 + buffer.len - current_len, &mut stream));
+            if n == 0 {
+                return Ok(&[])
+            } else {
+                buffer.bytes += n;
+            }
         }
-        try!(read(stream, &mut buffer.buffer, buffer.len, &mut buffer.bytes));
 
         let ciphertext_len = buffer.buffer.len() - key.tag_len();
         let (ciphertext, tag) = buffer.buffer.split_at_mut(ciphertext_len);
@@ -190,8 +184,8 @@ impl CipherPair {
         assert!(padding_length <= std::u8::MAX as usize);
         buffer.buffer.push(padding_length as u8);
         buffer.buffer.extend(payload);
-        key.fill_padding(buffer.buffer.reserve(padding_length));
-        buffer.buffer.reserve(key.tag_len());
+        key.fill_padding(buffer.buffer.resize_mut(padding_length));
+        buffer.buffer.resize_mut(key.tag_len());
 
         let (plaintext, tag) = buffer.buffer[offset..]
             .split_at_mut(PACKET_LENGTH_LEN + packet_length);

src/client/mod.rs

@@ -274,7 +274,7 @@ impl KexInit {
                                                  &mut self.exchange.client_kex_init));
 
         cipher.write(&self.exchange.client_kex_init[i0..], write_buffer);
-        self.exchange.client_kex_init.truncate(i0);
+        self.exchange.client_kex_init.resize(i0);
 
 
         Ok(KexDhDone {
@@ -339,7 +339,10 @@ impl<H: Handler> Future for Connection<TcpStream, H> {
         // If timeout, shutdown the socket.
         try_ready!(self.poll_timeout());
         loop {
-            try_ready!(self.atomic_poll())
+            debug!("client polling");
+            if ! try_ready!(self.atomic_poll()) {
+                return Ok(Async::Ready(()))
+            }
         }
     }
 }
@@ -362,6 +365,7 @@ impl<H:Handler> Connection<TcpStream, H> {
             if let Async::Ready(()) = try!(timeout.poll()) {
                 debug!("Timeout, shutdown");
                 try_nb!(self.stream.get_mut().shutdown(std::net::Shutdown::Both));
+                self.session.0.disconnected = true;
                 return Err(Error::ConnectionTimeout)
             }
         }
@@ -437,18 +441,21 @@ impl<H:Handler> Connection<TcpStream, H> {
     /// whether at least one complete packet was read.  `buffer` and
     /// `buffer2` are work spaces mostly used to compute keys. They
     /// are cleared before using, hence nothing is expected from them.
-    fn atomic_poll(&mut self) -> Poll<(), Error> {
+    fn atomic_poll(&mut self) -> Poll<bool, Error> {
 
         try_ready!(self.pending_poll());
 
         // Special case for the beginning.
         match std::mem::replace(&mut self.state, None) {
+            None if self.session.0.disconnected => Ok(Async::Ready(false)),
             None => {
-                Ok(Async::Ready(()))
+                try_nb!(self.stream.get_mut().shutdown(std::net::Shutdown::Both));
+                self.session.0.disconnected = true;
+                Ok(Async::Ready(false))
             }
             Some(ConnectionState::WriteSshId) => {
                 self.state = Some(ConnectionState::ReadSshId { sshid: read_ssh_id() });
-                Ok(Async::Ready(()))
+                Ok(Async::Ready(true))
             }
             Some(ConnectionState::ReadSshId { mut sshid }) => {
 
@@ -479,7 +486,7 @@ impl<H:Handler> Connection<TcpStream, H> {
                                              &mut self.session.0.write_buffer);
                         self.session.0.kex = Some(Kex::KexInit(kexinit));
                         self.state = Some(ConnectionState::Write);
-                        Ok(Async::Ready(()))
+                        Ok(Async::Ready(true))
                     }
                 }
             },
@@ -488,24 +495,24 @@ impl<H:Handler> Connection<TcpStream, H> {
                 self.session.flush();
                 try_nb!(self.session.0.write_buffer.write_all(self.stream.get_mut()));
                 self.state = Some(ConnectionState::Read);
-                Ok(Async::Ready(()))
+                Ok(Async::Ready(true))
             },
             Some(ConnectionState::Read) => {
 
                 self.state = Some(ConnectionState::Read);
                 // In all other cases:
                 let buf = try_nb!(self.session.0.cipher.read(&mut self.stream, &mut self.read_buffer));
                 // Handle the transport layer.
-                if buf[0] == msg::DISCONNECT {
+                if buf.len() == 0 || buf[0] == msg::DISCONNECT {
                     // transport
-                    return Ok(Async::Ready(()));
+                    return Ok(Async::Ready(false));
                 }
                 // If we don't disconnect, keep the state.
                 self.state = Some(ConnectionState::Write);
 
                 // Handle transport layer packets.
                 if buf[0] <= 4 {
-                    return Ok(Async::Ready(()))
+                    return Ok(Async::Ready(true))
                 }
 
                 // Handle key exchange/re-exchange.
@@ -521,7 +528,7 @@ impl<H:Handler> Connection<TcpStream, H> {
                                 match kexdhdone {
                                     Ok(kexdhdone) => {
                                         self.session.0.kex = Some(Kex::KexDhDone(kexdhdone));
-                                        return Ok(Async::Ready(()));
+                                        return Ok(Async::Ready(true));
                                     }
                                     Err(e) => return Err(e),
                                 }
@@ -563,7 +570,7 @@ impl<H:Handler> Connection<TcpStream, H> {
                         try_nb!(self.session.client_read_encrypted(&mut self.handler, buf, &mut self.buffer));
                     }
                 }
-                Ok(Async::Ready(()))
+                Ok(Async::Ready(true))
             }
         }
     }
@@ -634,6 +641,7 @@ impl<H: Handler> Future for Authenticate<TcpStream, H> {
             try_ready!(c.poll_timeout());
         }
         loop {
+            debug!("authenticated loop");
             let is_authenticated = if let Some(ref c) = self.0 { c.is_authenticated() }  else { false };
             if is_authenticated {
                 return Ok(Async::Ready(std::mem::replace(&mut self.0, None).unwrap()))
@@ -675,6 +683,7 @@ impl<H: Handler, ChannelType> Future for ChannelOpen<TcpStream, H, ChannelType>
         }
 
         loop {
+            debug!("channelopen loop");
             let is_open = if let Some(ref c) = self.connection {
                 c.channel_is_open(self.channel)
             }  else { false };
@@ -699,6 +708,7 @@ impl<H: Handler> Future for ChannelClose<TcpStream, H> {
         }
 
         loop {
+            debug!("channelclose loop");
             let is_open = if let Some(ref c) = self.connection {
                 c.channel_is_open(self.channel)
             }  else { false };
@@ -722,6 +732,7 @@ impl<H: Handler> Future for Flush<TcpStream, H> {
             try_ready!(c.poll_timeout());
         }
         loop {
+            debug!("flush loop");
             let completely_written = if let Some(ref mut c) = self.connection {
                 try_nb!(c.session.0.write_buffer.write_all(c.stream.get_mut()))
             } else {

src/server/mod.rs

@@ -169,7 +169,9 @@ pub trait Handler {
                                  user: &str,
                                  submethods: &str,
                                  response: Option<Response>)
-                                 -> Self::FutureAuth;
+                                 -> Self::FutureAuth {
+        Self::FutureAuth::finished(Auth::Reject)
+    }
 
     /// Called when the client closes a channel.
     #[allow(unused_variables)]
@@ -502,8 +504,7 @@ impl<H: Handler> Future for Connection<TcpStream, H> {
                 }
             }
             debug!("polling, state = {:?}", self.state);
-            try_ready!(self.atomic_poll());
-            if self.state.is_none() {
+            if let Status::Disconnect = try_ready!(self.atomic_poll()) {
                 return Ok(Async::Ready(()))
             }
         }
@@ -601,7 +602,7 @@ impl<H:Handler> Connection<TcpStream, H> {
         debug!("connection state: {:?}", self.state);
 
         match std::mem::replace(&mut self.state, None) {
-            None if self.session.0.disconnected => Ok(Async::Ready(Status::Ok)),
+            None if self.session.0.disconnected => Ok(Async::Ready(Status::Disconnect)),
             None => {
                 try_nb!(self.stream.get_mut().shutdown(std::net::Shutdown::Both));
                 self.session.0.disconnected = true;
@@ -661,7 +662,7 @@ impl<H:Handler> Connection<TcpStream, H> {
                 let buf = try_nb!(self.session.0.cipher.read(&mut self.stream, &mut self.read_buffer));
                 debug!("read buf = {:?}", buf);
                 // Handle the transport layer.
-                if buf[0] == msg::DISCONNECT {
+                if buf.len() == 0 || buf[0] == msg::DISCONNECT {
                     // transport
                     self.state = None;
                     debug!("Received disconnect, shutdown");

src/sshbuffer.rs

@@ -23,6 +23,9 @@ pub struct SSHBuffer {
     pub len: usize, // next packet length.
     pub bytes: usize,
 
+    pub len_bytes: [u8;4],
+    pub read_len_bytes: usize,
+
     // Sequence numbers are on 32 bits and wrap.
     // https://tools.ietf.org/html/rfc4253#section-6.4
     pub seqn: Wrapping<u32>,
@@ -34,6 +37,8 @@ impl SSHBuffer {
             buffer: CryptoVec::new(),
             len: 0,
             bytes: 0,
+            len_bytes: [0;4],
+            read_len_bytes: 0,
             seqn: Wrapping(0),
         }
     }
@@ -46,9 +51,9 @@ impl SSHBuffer {
     }
 
     /// Returns true iff the write buffer has been completely written.
-    pub fn write_all<W: Write>(&mut self, stream: &mut W) -> Result<bool, Error> {
+    pub fn write_all<W:Write>(&mut self, mut stream: W) -> Result<bool, Error> {
         while self.len < self.buffer.len() {
-            let s = try!(self.buffer.write_all_from(self.len, stream));
+            let s = try!(self.buffer.write_all_from(self.len, &mut stream));
             self.len += s;
             self.bytes += s;
             try!(stream.flush());