added bases for nonblocking packet sending and receiving. packet_send

and packet_read may return SSH_AGAIN if the session is nonblocking and 
there is not enough data to be read.
I also added a socket buffering through session->in_socket_buffer. It is 
more low-level than the packet buffer.
I should rename in_buffer with in_packet_buffer.
There is still work to do with the socket status, (opened, error, 
closed, ...) and much more work to extend the nonblocking to session 
opening, messages sending and such.
I find the switch(session->packet_state) solution very nice (especially 
when the nonblocking function may have 10 differents states, like in a 
connection.)


git-svn-id: svn+ssh://svn.berlios.de/svnroot/repos/libssh/trunk@49 7dcaeef0-15fb-0310-b436-a5af3365683c

include/libssh/libssh.h

@@ -90,6 +90,10 @@ typedef uint8_t u8;
 #define SSH_FATAL 2
 #define SSH_EINTR 3
 
+/* error return codes */
+#define SSH_OK 0     /* No error */
+#define SSH_ERROR -1 /* error of some kind */
+#define SSH_AGAIN 1  /* the nonblocking call must be repeated */
 
 char *ssh_get_error(void *error); 
 int ssh_get_error_code(void *error);

include/libssh/priv.h

@@ -304,7 +304,7 @@ struct ssh_session {
                          not block */
     int data_to_write;
     int data_except;
-    int blocking; // functions should not block
+    int blocking; // functions should block
     
     STRING *banner; /* that's the issue banner from 
                        the server */
@@ -315,6 +315,14 @@ struct ssh_session {
     BUFFER *in_buffer;
     PACKET in_packet;
     BUFFER *out_buffer;
+    
+    BUFFER *out_socket_buffer;
+    BUFFER *in_socket_buffer;
+    
+    /* the states are used by the nonblocking stuff to remember */
+    /* where it was before being interrupted */
+    int packet_state;
+    
     KEX server_kex;
     KEX client_kex;
     BUFFER *in_hashbuf;

libssh/Makefile.in

@@ -45,7 +45,7 @@ install: all
 	$(top_srcdir)/mkinstalldirs $(libdir)
 	$(LIBTOOL) --mode=install $(INSTALL) libssh.la $(libdir)
 clean:
-	$(LIBTOOL) --mode=clean rm -f *~ libssh.la *.lo
+	$(LIBTOOL) --mode=clean rm -f *~ libssh.la *.lo *.o
 distclean: clean
 	rm -f Makefile
 

libssh/packet.c

@@ -41,7 +41,7 @@ static int completeread(int fd, void *buffer, int len){
     int toread=len;
     while((r=read(fd,buffer+total,toread))){
         if(r==-1)
-            return -1;
+            return SSH_ERROR;
         total += r;
         toread-=r;
         if(total==len)
@@ -52,6 +52,63 @@ static int completeread(int fd, void *buffer, int len){
     return total ; /* connection closed */
 }
 
+/* in nonblocking mode, socket_read will read as much as it can, and return */
+/* SSH_OK if it has read at least len bytes, otherwise, SSH_AGAIN. */
+/* in blocking mode, it will read at least len bytes and will block until it's ok. */
+
+static int socket_read(SSH_SESSION *session,int len){
+    int except, can_write;
+    int ret;
+    int to_read;
+    int r;
+    char *buf;
+    char buffer[4096];
+    if(!session->in_socket_buffer)
+        session->in_socket_buffer=buffer_new();
+    to_read=len - buffer_get_rest_len(session->in_socket_buffer);
+    if(to_read <= 0)
+        return SSH_OK;
+    if(session->blocking){
+        buf=malloc(to_read);
+        ret=completeread(session->fd,buf,to_read);
+        session->data_to_read=0;
+        if(ret==SSH_ERROR || ret ==0){
+            ssh_set_error(session,SSH_FATAL,
+                (ret==0)?"Connection closed by remote host" : "Error reading socket");
+            close(session->fd);
+            session->fd=-1;
+            session->data_except=1;
+            return SSH_ERROR;
+        }
+            
+        buffer_add_data(session->in_socket_buffer,buf,to_read);
+        free(buf);
+        return SSH_OK;
+    }
+    /* nonblocking read */
+    do {
+        ssh_fd_poll(session,&can_write,&except); /* internally sets data_to_read */
+        if(!session->data_to_read)
+            return SSH_AGAIN;
+        session->data_to_read=0;
+        /* read as much as we can */
+        r=read(session->fd,buffer,sizeof(buffer));
+        if(r<=0){
+            ssh_set_error(session,SSH_FATAL,
+                (ret==0)?"Connection closed by remote host" : "Error reading socket");
+            close(session->fd);
+            session->fd=-1;
+            session->data_except=1;
+            return SSH_ERROR;
+        }
+        buffer_add_data(session->in_socket_buffer,buffer,r);
+    } while(buffer_get_rest_len(session->in_socket_buffer)<len);
+    return SSH_OK;
+}
+
+#define PACKET_STATE_INIT 0
+#define PACKET_STATE_SIZEREAD 1
+
 static int packet_read2(SSH_SESSION *session){
     u32 len;
     void *packet=NULL;
@@ -59,103 +116,92 @@ static int packet_read2(SSH_SESSION *session){
     char buffer[16];
     int be_read,i;
     int to_be_read;
+    int ret;
     u8 padding;
     unsigned int blocksize=(session->current_crypto?session->current_crypto->in_cipher->blocksize:8);
-    session->data_to_read=0; /* clear the dataavailable flag */
-    memset(&session->in_packet,0,sizeof(PACKET));
-    if(session->in_buffer)
-        buffer_free(session->in_buffer);
-    session->in_buffer=buffer_new();
-
-    be_read=completeread(session->fd,buffer,blocksize);
-    if(be_read!=blocksize){
-        if(be_read<=0){
-            session->alive=0;
-            close(session->fd);
-            session->fd=-1;
-            ssh_set_error(session,SSH_FATAL,
-            (be_read==0)?"Connection closed by remote host" : "Error reading socket");
-            return -1;
-        }
-        ssh_set_error(session,SSH_FATAL,"read_packet(): asked %d bytes, received %d",blocksize,be_read);
-        return -1;
-    }
-    len=packet_decrypt_len(session,buffer);
-    buffer_add_data(session->in_buffer,buffer,blocksize);
-    
-    if(len> MAX_PACKET_LEN){
-        ssh_set_error(session,SSH_FATAL,"read_packet(): Packet len too high(%uld %.8lx)",len,len);
-        return -1;
-    }
-    to_be_read=len-be_read+sizeof(u32);
-    if(to_be_read<0){
-        /* remote sshd is trying to get me ?*/
-        ssh_set_error(session,SSH_FATAL,"given numbers of bytes left to be read <0 (%d)!",to_be_read);
-        return -1;
-    }
-    /* handle the case in which the whole packet size = blocksize */
-    if(to_be_read !=0){
-        packet=malloc(to_be_read);
-        i=completeread(session->fd,packet,to_be_read);
-        if(i<=0){
-            session->alive=0;
-            close(session->fd);
-            session->fd=-1;
-            ssh_set_error(session,SSH_FATAL,"Server closed connection");
-            return -1;
-        }
-        if(i!=to_be_read){
-            free(packet);
-            packet=NULL;
-            ssh_say(3,"Read only %d, wanted %d\n",i,to_be_read);
-            ssh_set_error(session,SSH_FATAL,"read_packet(): read only %d, wanted %d",i,to_be_read);
-            return -1;
-        }
-        ssh_say(3,"Read a %d bytes packet\n",len);
-        buffer_add_data(session->in_buffer,packet,to_be_read);
-        free(packet);
-    }
-    if(session->current_crypto){
-        packet_decrypt(session,buffer_get(session->in_buffer)+blocksize,buffer_get_len(session->in_buffer)-blocksize);
-        if((i=completeread(session->fd,mac,macsize))!=macsize){
-            if(i<=0){
-                session->alive=0;
-                close(session->fd);
-                session->fd=-1;
-                ssh_set_error(session,SSH_FATAL,"Server closed connection");
-                return -1;
+    int current_macsize=session->current_crypto?macsize:0;
+    if(!session->alive || session->data_except)
+        return SSH_ERROR; // the error message was already set into this session
+    switch(session->packet_state){
+        case PACKET_STATE_INIT:    
+            memset(&session->in_packet,0,sizeof(PACKET));
+            if(session->in_buffer)
+                buffer_reinit(session->in_buffer);
+            else
+                session->in_buffer=buffer_new();
+            ret=socket_read(session,blocksize);
+            if(ret != SSH_OK)
+                return ret; // can be SSH_ERROR or SSH_AGAIN
+//    be_read=completeread(session->fd,buffer,blocksize);
+            memcpy(buffer,buffer_get_rest(session->in_socket_buffer),blocksize);
+            buffer_pass_bytes(session->in_socket_buffer,blocksize); // mark them as read
+            len=packet_decrypt_len(session,buffer);
+            buffer_add_data(session->in_buffer,buffer,blocksize);
+            if(len> MAX_PACKET_LEN){
+                ssh_set_error(session,SSH_FATAL,"read_packet(): Packet len too high(%uld %.8lx)",len,len);
+                return SSH_ERROR;
             }
-            ssh_set_error(session,SSH_FATAL,"read_packet(): wanted %d, had %d",i,macsize);
-            return -1;
-        }
-        if(packet_hmac_verify(session,session->in_buffer,mac)){
-            ssh_set_error(session,SSH_FATAL,"HMAC error");
-            return -1;
-        }
-    }
-    buffer_pass_bytes(session->in_buffer,sizeof(u32));   /*pass the size which has been processed before*/
-    if(!buffer_get_u8(session->in_buffer,&padding)){
-        ssh_set_error(session,SSH_FATAL,"Packet too short to read padding");
-        return -1;
-    }
-    ssh_say(3,"%hhd bytes padding, %d bytes left in buffer\n",padding,buffer_get_rest_len(session->in_buffer));
-    if(padding > buffer_get_rest_len(session->in_buffer)){
-        ssh_set_error(session,SSH_FATAL,"invalid padding: %d (%d resting)",padding,buffer_get_rest_len(session->in_buffer));
+            to_be_read=len-blocksize+sizeof(u32);
+            if(to_be_read<0){
+                /* remote sshd sends invalid sizes?*/
+                ssh_set_error(session,SSH_FATAL,"given numbers of bytes left to be read <0 (%d)!",to_be_read);
+                return SSH_ERROR;
+            }
+            /* saves the status of the current operations */
+            session->in_packet.len=len;
+            session->packet_state=PACKET_STATE_SIZEREAD;
+        case PACKET_STATE_SIZEREAD:
+            len=session->in_packet.len;
+            to_be_read=len-blocksize+sizeof(u32) + current_macsize;
+            /* if to_be_read is zero, the whole packet was blocksize bytes. */
+            if(to_be_read != 0){ 
+                ret=socket_read(session,to_be_read);
+                if(ret!=SSH_OK)
+                    return ret;
+                packet=malloc(to_be_read);
+                memcpy(packet,buffer_get_rest(session->in_socket_buffer),to_be_read-current_macsize);
+                buffer_pass_bytes(session->in_socket_buffer,to_be_read-current_macsize);
+                ssh_say(3,"Read a %d bytes packet\n",len);
+                buffer_add_data(session->in_buffer,packet,to_be_read-current_macsize);
+                free(packet);
+            }
+            if(session->current_crypto){
+                /* decrypt the rest of the packet (blocksize bytes already have been decrypted */
+                packet_decrypt(session,buffer_get(session->in_buffer)+blocksize,buffer_get_len(session->in_buffer)-blocksize);
+                memcpy(mac,buffer_get_rest(session->in_socket_buffer),macsize);
+                buffer_pass_bytes(session->in_socket_buffer,macsize);
+                if(packet_hmac_verify(session,session->in_buffer,mac)){
+                    ssh_set_error(session,SSH_FATAL,"HMAC error");
+                    return SSH_ERROR;
+                }
+            }
+            buffer_pass_bytes(session->in_buffer,sizeof(u32));   /*pass the size which has been processed before*/
+            if(!buffer_get_u8(session->in_buffer,&padding)){
+                ssh_set_error(session,SSH_FATAL,"Packet too short to read padding");
+                return SSH_ERROR;
+            }
+            ssh_say(3,"%hhd bytes padding, %d bytes left in buffer\n",padding,buffer_get_rest_len(session->in_buffer));
+            if(padding > buffer_get_rest_len(session->in_buffer)){
+                ssh_set_error(session,SSH_FATAL,"invalid padding: %d (%d resting)",padding,buffer_get_rest_len(session->in_buffer));
 #ifdef DEBUG_CRYPTO
-        ssh_print_hexa("incrimined packet",buffer_get(session->in_buffer),buffer_get_len(session->in_buffer));
+                ssh_print_hexa("incrimined packet",buffer_get(session->in_buffer),buffer_get_len(session->in_buffer));
 #endif
-        return -1;
-    }
-    buffer_pass_bytes_end(session->in_buffer,padding);
-    ssh_say(3,"After padding, %d bytes left in buffer\n",buffer_get_rest_len(session->in_buffer));
+                return SSH_ERROR;
+            }
+            buffer_pass_bytes_end(session->in_buffer,padding);
+            ssh_say(3,"After padding, %d bytes left in buffer\n",buffer_get_rest_len(session->in_buffer));
 #ifdef HAVE_LIBZ
-    if(session->current_crypto && session->current_crypto->do_compress_in){
-        ssh_say(3,"Decompressing ...\n");
-        decompress_buffer(session,session->in_buffer);
-    }
+            if(session->current_crypto && session->current_crypto->do_compress_in){
+                ssh_say(3,"Decompressing ...\n");
+                decompress_buffer(session,session->in_buffer);
+            }
 #endif
-    session->recv_seq++;
-    return 0;
+            session->recv_seq++;
+            session->packet_state=PACKET_STATE_INIT;
+            return SSH_OK;
+    }
+    ssh_set_error(session,SSH_FATAL,"Invalid state into packet_read2() : %d",session->packet_state);
+    return SSH_ERROR;
 }
 
 #ifdef HAVE_SSH1
@@ -171,6 +217,8 @@ static int packet_read1(SSH_SESSION *session){
     ssh_say(3,"packet_read1()\n");
 //    unsigned int blocksize=8;
     session->data_to_read=0; /* clear the dataavailable flag */
+    if(!session->alive || session->data_except)
+        return -1; // the error message was already set
     memset(&session->in_packet,0,sizeof(PACKET));
     if(session->in_buffer)
         buffer_free(session->in_buffer);
@@ -180,6 +228,7 @@ static int packet_read1(SSH_SESSION *session){
     if(be_read!=sizeof(u32)){
         if(be_read<=0){
             session->alive=0;
+            session->data_except=1;
             close(session->fd);
             session->fd=-1;
             ssh_set_error(session,SSH_FATAL,
@@ -206,6 +255,7 @@ static int packet_read1(SSH_SESSION *session){
         i=completeread(session->fd,packet,to_be_read);
         if(i<=0){
             session->alive=0;
+            session->data_except=1;
             close(session->fd);
             session->fd=-1;
             ssh_set_error(session,SSH_FATAL,"Server closed connection");
@@ -297,18 +347,68 @@ int packet_translate(SSH_SESSION *session){
 
 static int atomic_write(int fd, void *buffer, int len){
     int written;
-    int total=0;
     do {
         written=write(fd,buffer,len);
-        if(written==0)
-            return 0;
-        if(written==-1)
-            return total;
-        total+=written;
+        if(written==0 || written==-1)
+            return -1;
         len-=written;
         buffer+=written;
     } while (len > 0);
-    return total;
+    return SSH_OK;
+}
+
+int packet_nonblocking_flush(SSH_SESSION *session){
+    int except, can_write;
+    int w;
+    ssh_fd_poll(session,&can_write,&except); /* internally sets data_to_write */
+    while(session->data_to_write && buffer_get_rest_len(session->out_socket_buffer)>0){
+        w=write(session->fd,buffer_get_rest(session->out_socket_buffer),
+                buffer_get_rest_len(session->out_socket_buffer));
+        if(w<0){
+            session->data_to_write=0;
+            session->data_except=1;
+            session->alive=0;
+            close(session->fd);
+            session->fd=-1;
+            ssh_set_error(session,SSH_FATAL,"Writing packet : error on socket (or connection closed): %s",
+                          strerror(errno));
+            return SSH_ERROR;
+        }
+        buffer_pass_bytes(session->out_socket_buffer,w);
+    }
+    if(buffer_get_rest_len(session->out_socket_buffer)>0)
+        return SSH_AGAIN;  /* there is data pending */
+    return SSH_OK; // all data written
+}
+
+int packet_flush(SSH_SESSION *session){
+    if(atomic_write(session->fd,buffer_get_rest(session->out_socket_buffer),
+       buffer_get_rest_len(session->out_socket_buffer))){
+        session->data_to_write=0;
+        session->data_except=1;
+        session->alive=0;
+        close(session->fd);
+        session->fd=-1;
+        ssh_set_error(session,SSH_FATAL,"Writing packet : error on socket (or connection closed): %s",
+                         strerror(errno));
+        return SSH_ERROR;
+    }
+    session->data_to_write=0;
+    buffer_reinit(session->out_socket_buffer);
+    return SSH_OK; // no data pending
+}
+
+/* this function places the outgoing packet buffer into an outgoing socket buffer */
+static int socket_write(SSH_SESSION *session){
+    if(!session->out_socket_buffer){
+        session->out_socket_buffer=buffer_new();
+    }
+    buffer_add_data(session->out_socket_buffer,
+               buffer_get(session->out_buffer),buffer_get_len(session->out_buffer));
+    if(!session->blocking){
+        return packet_nonblocking_flush(session);
+    } else
+        return packet_flush(session);
 }
 
 static int packet_send2(SSH_SESSION *session){
@@ -342,17 +442,10 @@ static int packet_send2(SSH_SESSION *session){
     hmac=packet_encrypt(session,buffer_get(session->out_buffer),buffer_get_len(session->out_buffer));
     if(hmac)
         buffer_add_data(session->out_buffer,hmac,20);
-    if(atomic_write(session->fd,buffer_get(session->out_buffer),buffer_get_len(session->out_buffer))!=buffer_get_len(session->out_buffer)){
-        session->alive=0;
-        close(session->fd);
-        session->fd=-1;
-        ssh_set_error(session,SSH_FATAL,"Writing packet : error on socket (or connection closed): %s",
-            strerror(errno));
-        ret=-1;
-    }
+    ret=socket_write(session);
     session->send_seq++;
     buffer_reinit(session->out_buffer);
-    return ret;
+    return ret; /* SSH_OK, AGAIN or ERROR */
 }
 
 #ifdef HAVE_SSH1
@@ -393,17 +486,10 @@ static int packet_send1(SSH_SESSION *session){
     ssh_print_hexa("encrypted packet",buffer_get(session->out_buffer),
             buffer_get_len(session->out_buffer));
 #endif
-    if(atomic_write(session->fd,buffer_get(session->out_buffer),buffer_get_len(session->out_buffer))!=buffer_get_len(session->out_buffer)){
-        session->alive=0;
-        close(session->fd);
-        session->fd=-1;
-        ssh_set_error(session,SSH_FATAL,"Writing packet : error on socket (or connection closed): %s",
-            strerror(errno));
-        ret=-1;
-    }
+    ret=socket_write(session);
     session->send_seq++;
     buffer_reinit(session->out_buffer);
-    return ret;
+    return ret;     /* SSH_OK, AGAIN or ERROR */
 }
 
 #endif /* HAVE_SSH1 */

libssh/session.c

@@ -34,6 +34,7 @@ SSH_SESSION *ssh_new() {
     session->next_crypto=crypto_new();
     session->maxchannel=FIRST_CHANNEL;
     session->fd=-1;
+    session->blocking=1;
     return session;
 }