]> jfr.im git - solanum.git/blobdiff - wsockd/wsockd.c
remove old reference to the unsupported directory
[solanum.git] / wsockd / wsockd.c
index 5c5262cc78a4f306cc4468e69414844af4513974..66fa4efc234d3f1f7884e7e3d8773839847d352f 100644 (file)
@@ -1,8 +1,8 @@
 /*
- *  wsockd.c: charybdis websockets helper
+ *  wsockd.c: solanum websockets helper
  *  Copyright (C) 2007 Aaron Sethman <androsyn@ratbox.org>
  *  Copyright (C) 2007 ircd-ratbox development team
- *  Copyright (C) 2016 William Pitcock <nenolod@dereferenced.org>
+ *  Copyright (C) 2016 Ariadne Conill <ariadne@dereferenced.org>
  *
  *  This program is free software; you can redistribute it and/or modify
  *  it under the terms of the GNU General Public License as published by
  */
 
 #include "stdinc.h"
+#include "sha1.h"
 
 #define MAXPASSFD 4
 #ifndef READBUF_SIZE
 #define READBUF_SIZE 16384
 #endif
 
+#define WEBSOCKET_SERVER_KEY "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"
+#define WEBSOCKET_ANSWER_STRING_1 "HTTP/1.1 101 Switching Protocols\r\nAccess-Control-Allow-Origin: *\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: "
+#define WEBSOCKET_ANSWER_STRING_2 "\r\n\r\n"
+
 static void setup_signals(void);
 static pid_t ppid;
 
@@ -70,8 +75,12 @@ typedef struct _conn
 {
        rb_dlink_node node;
        mod_ctl_t *ctl;
+
        rawbuf_head_t *modbuf_out;
-       rawbuf_head_t *plainbuf_out;
+       rawbuf_head_t *modbuf_in;
+
+       buf_head_t plainbuf_out;
+       buf_head_t plainbuf_in;
 
        uint32_t id;
 
@@ -82,48 +91,101 @@ typedef struct _conn
        uint64_t plain_in;
        uint64_t plain_out;
        uint8_t flags;
-       void *stream;
+
+       char client_key[37];            /* maximum 36 bytes + nul */
 } conn_t;
 
+#define WEBSOCKET_OPCODE_TEXT_FRAME          1
+
+#define WEBSOCKET_MASK_LENGTH 4
+
+#define WEBSOCKET_MAX_UNEXTENDED_PAYLOAD_DATA_LENGTH 125
+
+typedef struct {
+       uint8_t opcode_rsv_fin; // opcode: 4, rsv1: 1, rsv2: 1, rsv3: 1, fin: 1
+       uint8_t payload_length_mask; // payload_length: 7, mask: 1
+} ws_frame_hdr_t;
+
+#define WEBSOCKET_FRAME_HDR_INIT ((ws_frame_hdr_t) { 0, 0 })
+
+typedef struct {
+       ws_frame_hdr_t header;
+       uint8_t payload_data[WEBSOCKET_MAX_UNEXTENDED_PAYLOAD_DATA_LENGTH];
+} ws_frame_payload_t;
+
+typedef struct {
+       ws_frame_hdr_t header;
+} ws_frame_t;
+
+typedef struct {
+       ws_frame_hdr_t header;
+       uint16_t payload_length_extended;
+} ws_frame_ext_t;
+
+#define WEBSOCKET_FRAME_EXT_INIT ((ws_frame_ext_t) { WEBSOCKET_FRAME_HDR_INIT, 0 })
+
+typedef struct {
+       ws_frame_hdr_t header;
+       uint64_t payload_length_extended;
+} ws_frame_ext2_t;
+
+static inline void
+ws_frame_set_opcode(ws_frame_hdr_t *header, int opcode)
+{
+       header->opcode_rsv_fin &= ~0xF;
+       header->opcode_rsv_fin |= opcode & 0xF;
+}
+
+static inline void
+ws_frame_set_fin(ws_frame_hdr_t *header, int fin)
+{
+       header->opcode_rsv_fin &= ~(0x1 << 7);
+       header->opcode_rsv_fin |= (fin << 7) & (0x1 << 7);
+}
+
+static void close_conn(conn_t * conn, int wait_plain, const char *fmt, ...);
+static void conn_mod_read_cb(rb_fde_t *fd, void *data);
+static void conn_plain_read_cb(rb_fde_t *fd, void *data);
+static void conn_plain_process_recvq(conn_t *conn);
+
 #define FLAG_CORK      0x01
 #define FLAG_DEAD      0x02
 #define FLAG_WSOCK     0x04
+#define FLAG_KEYED     0x08
 
 #define IsCork(x) ((x)->flags & FLAG_CORK)
 #define IsDead(x) ((x)->flags & FLAG_DEAD)
-#define IsWS(x)   ((x)->flags & FLAG_WSOCK)
+#define IsKeyed(x) ((x)->flags & FLAG_KEYED)
 
 #define SetCork(x) ((x)->flags |= FLAG_CORK)
 #define SetDead(x) ((x)->flags |= FLAG_DEAD)
 #define SetWS(x)   ((x)->flags |= FLAG_WSOCK)
+#define SetKeyed(x) ((x)->flags |= FLAG_KEYED)
 
 #define ClearCork(x) ((x)->flags &= ~FLAG_CORK)
-#define ClearDead(x) ((x)->flags &= ~FLAG_DEAD)
-#define ClearWS(x)   ((x)->flags &= ~FLAG_WSOCK)
 
 #define NO_WAIT 0x0
 #define WAIT_PLAIN 0x1
 
-#define HASH_WALK_SAFE(i, max, ptr, next, table) for(i = 0; i < max; i++) { RB_DLINK_FOREACH_SAFE(ptr, next, table[i].head)
-#define HASH_WALK_END }
 #define CONN_HASH_SIZE 2000
 #define connid_hash(x) (&connid_hash_table[(x % CONN_HASH_SIZE)])
 
+static const char *remote_closed = "Remote host closed the connection";
+
 static rb_dlink_list connid_hash_table[CONN_HASH_SIZE];
 static rb_dlink_list dead_list;
 
-#ifndef _WIN32
+static void conn_plain_read_shutdown_cb(rb_fde_t *fd, void *data);
+
 static void
 dummy_handler(int sig)
 {
        return;
 }
-#endif
 
 static void
 setup_signals()
 {
-#ifndef _WIN32
        struct sigaction act;
 
        act.sa_flags = 0;
@@ -146,38 +208,20 @@ setup_signals()
 
        act.sa_handler = dummy_handler;
        sigaction(SIGALRM, &act, 0);
-#endif
 }
 
 static int
 maxconn(void)
 {
-#if defined(RLIMIT_NOFILE) && defined(HAVE_SYS_RESOURCE_H)
        struct rlimit limit;
 
        if(!getrlimit(RLIMIT_NOFILE, &limit))
        {
                return limit.rlim_cur;
        }
-#endif /* RLIMIT_FD_MAX */
        return MAXCONNECTIONS;
 }
 
-static conn_t *
-conn_find_by_id(uint32_t id)
-{
-       rb_dlink_node *ptr;
-       conn_t *conn;
-
-       RB_DLINK_FOREACH(ptr, (connid_hash(id))->head)
-       {
-               conn = ptr->data;
-               if(conn->id == id && !IsDead(conn))
-                       return conn;
-       }
-       return NULL;
-}
-
 static void
 conn_add_id_hash(conn_t * conn, uint32_t id)
 {
@@ -188,8 +232,12 @@ conn_add_id_hash(conn_t * conn, uint32_t id)
 static void
 free_conn(conn_t * conn)
 {
+       rb_linebuf_donebuf(&conn->plainbuf_in);
+       rb_linebuf_donebuf(&conn->plainbuf_out);
+
+       rb_free_rawbuffer(conn->modbuf_in);
        rb_free_rawbuffer(conn->modbuf_out);
-       rb_free_rawbuffer(conn->plainbuf_out);
+
        rb_free(conn);
 }
 
@@ -208,19 +256,217 @@ clean_dead_conns(void *unused)
        dead_list.tail = dead_list.head = NULL;
 }
 
+static void
+conn_plain_write_sendq(rb_fde_t *fd, void *data)
+{
+       conn_t *conn = data;
+       int retlen;
+
+       if(IsDead(conn))
+               return;
+
+       while((retlen = rb_linebuf_flush(fd, &conn->plainbuf_out)) > 0)
+               conn->plain_out += retlen;
+
+       if(retlen == 0 || (retlen < 0 && !rb_ignore_errno(errno)))
+       {
+               close_conn(data, NO_WAIT, NULL);
+               return;
+       }
+
+       if(rb_linebuf_alloclen(&conn->plainbuf_out) > 0)
+               rb_setselect(conn->plain_fd, RB_SELECT_WRITE, conn_plain_write_sendq, conn);
+       else
+               rb_setselect(conn->plain_fd, RB_SELECT_WRITE, NULL, NULL);
+}
+
+static void
+conn_mod_write_sendq(rb_fde_t *fd, void *data)
+{
+       conn_t *conn = data;
+       const char *err;
+       int retlen;
+
+       if(IsDead(conn))
+               return;
+
+       while((retlen = rb_rawbuf_flush(conn->modbuf_out, fd)) > 0)
+               conn->mod_out += retlen;
+
+       if(retlen == 0 || (retlen < 0 && !rb_ignore_errno(errno)))
+       {
+               if(retlen == 0)
+                       close_conn(conn, WAIT_PLAIN, "%s", remote_closed);
+               err = strerror(errno);
+               close_conn(conn, WAIT_PLAIN, "Write error: %s", err);
+               return;
+       }
+
+       if(rb_rawbuf_length(conn->modbuf_out) > 0)
+               rb_setselect(conn->mod_fd, RB_SELECT_WRITE, conn_mod_write_sendq, conn);
+       else
+               rb_setselect(conn->mod_fd, RB_SELECT_WRITE, NULL, NULL);
+
+       if(IsCork(conn) && rb_rawbuf_length(conn->modbuf_out) == 0)
+       {
+               ClearCork(conn);
+               conn_plain_read_cb(conn->plain_fd, conn);
+       }
+}
+
+static void
+conn_mod_write(conn_t * conn, void *data, size_t len)
+{
+       if(IsDead(conn))        /* no point in queueing to a dead man */
+               return;
+       rb_rawbuf_append(conn->modbuf_out, data, len);
+}
+
+static void
+conn_mod_write_short_frame(conn_t * conn, void *data, int len)
+{
+       ws_frame_hdr_t hdr = WEBSOCKET_FRAME_HDR_INIT;
+
+       ws_frame_set_opcode(&hdr, WEBSOCKET_OPCODE_TEXT_FRAME);
+       ws_frame_set_fin(&hdr, 1);
+       hdr.payload_length_mask = (len + 2) & 0x7f;
+
+       conn_mod_write(conn, &hdr, sizeof(hdr));
+       conn_mod_write(conn, data, len);
+       conn_mod_write(conn, "\r\n", 2);
+}
+
+static void
+conn_mod_write_long_frame(conn_t * conn, void *data, int len)
+{
+       ws_frame_ext_t hdr = WEBSOCKET_FRAME_EXT_INIT;
+
+       ws_frame_set_opcode(&hdr.header, WEBSOCKET_OPCODE_TEXT_FRAME);
+       ws_frame_set_fin(&hdr.header, 1);
+       hdr.header.payload_length_mask = 126;
+       hdr.payload_length_extended = htons(len + 2);
+
+       conn_mod_write(conn, &hdr, sizeof(hdr));
+       conn_mod_write(conn, data, len);
+       conn_mod_write(conn, "\r\n", 2);
+}
+
+static void
+conn_mod_write_frame(conn_t *conn, void *data, int len)
+{
+       if(IsDead(conn))        /* no point in queueing to a dead man */
+               return;
+
+       if (len < 123)
+       {
+               conn_mod_write_short_frame(conn, data, len);
+               return;
+       }
+
+       conn_mod_write_long_frame(conn, data, len);
+}
+
+static void
+mod_write_ctl(rb_fde_t *F, void *data)
+{
+       mod_ctl_t *ctl = data;
+       mod_ctl_buf_t *ctl_buf;
+       rb_dlink_node *ptr, *next;
+       int retlen, x;
+
+       RB_DLINK_FOREACH_SAFE(ptr, next, ctl->writeq.head)
+       {
+               ctl_buf = ptr->data;
+               retlen = rb_send_fd_buf(ctl->F, ctl_buf->F, ctl_buf->nfds, ctl_buf->buf,
+                                       ctl_buf->buflen, ppid);
+               if(retlen > 0)
+               {
+                       rb_dlinkDelete(ptr, &ctl->writeq);
+                       for(x = 0; x < ctl_buf->nfds; x++)
+                               rb_close(ctl_buf->F[x]);
+                       rb_free(ctl_buf->buf);
+                       rb_free(ctl_buf);
+
+               }
+               if(retlen == 0 || (retlen < 0 && !rb_ignore_errno(errno)))
+                       exit(0);
+
+       }
+       if(rb_dlink_list_length(&ctl->writeq) > 0)
+               rb_setselect(ctl->F, RB_SELECT_WRITE, mod_write_ctl, ctl);
+}
+
+static void
+mod_cmd_write_queue(mod_ctl_t * ctl, const void *data, size_t len)
+{
+       mod_ctl_buf_t *ctl_buf;
+       ctl_buf = rb_malloc(sizeof(mod_ctl_buf_t));
+       ctl_buf->buf = rb_malloc(len);
+       ctl_buf->buflen = len;
+       memcpy(ctl_buf->buf, data, len);
+       ctl_buf->nfds = 0;
+       rb_dlinkAddTail(ctl_buf, &ctl_buf->node, &ctl->writeq);
+       mod_write_ctl(ctl->F, ctl);
+}
+
+static void
+close_conn(conn_t * conn, int wait_plain, const char *fmt, ...)
+{
+       va_list ap;
+       char reason[128];       /* must always be under 250 bytes */
+       uint8_t buf[256];
+       int len;
+       if(IsDead(conn))
+               return;
+
+       if (IsKeyed(conn))
+               conn_plain_process_recvq(conn);
+
+       rb_rawbuf_flush(conn->modbuf_out, conn->mod_fd);
+       rb_linebuf_flush(conn->plain_fd, &conn->plainbuf_out);
+       rb_close(conn->mod_fd);
+       SetDead(conn);
+
+       rb_dlinkDelete(&conn->node, connid_hash(conn->id));
+
+       if(!wait_plain || fmt == NULL)
+       {
+               rb_close(conn->plain_fd);
+               rb_dlinkAdd(conn, &conn->node, &dead_list);
+               return;
+       }
+
+       rb_setselect(conn->plain_fd, RB_SELECT_READ, conn_plain_read_shutdown_cb, conn);
+       rb_setselect(conn->plain_fd, RB_SELECT_WRITE, NULL, NULL);
+
+       va_start(ap, fmt);
+       vsnprintf(reason, sizeof(reason), fmt, ap);
+       va_end(ap);
+
+       buf[0] = 'D';
+       uint32_to_buf(&buf[1], conn->id);
+       rb_strlcpy((char *) &buf[5], reason, sizeof(buf) - 5);
+       len = (strlen(reason) + 1) + 5;
+       mod_cmd_write_queue(conn->ctl, buf, len);
+}
+
 static conn_t *
 make_conn(mod_ctl_t * ctl, rb_fde_t *mod_fd, rb_fde_t *plain_fd)
 {
        conn_t *conn = rb_malloc(sizeof(conn_t));
        conn->ctl = ctl;
-       conn->modbuf_out = rb_new_rawbuffer();
-       conn->plainbuf_out = rb_new_rawbuffer();
        conn->mod_fd = mod_fd;
        conn->plain_fd = plain_fd;
        conn->id = -1;
-       conn->stream = NULL;
        rb_set_nb(mod_fd);
        rb_set_nb(plain_fd);
+
+       rb_linebuf_newbuf(&conn->plainbuf_in);
+       rb_linebuf_newbuf(&conn->plainbuf_out);
+
+       conn->modbuf_in = rb_new_rawbuffer();
+       conn->modbuf_out = rb_new_rawbuffer();
+
        return conn;
 }
 
@@ -235,7 +481,361 @@ cleanup_bad_message(mod_ctl_t * ctl, mod_ctl_buf_t * ctlb)
 }
 
 static void
-wsock_process_accept(mod_ctl_t * ctl, mod_ctl_buf_t * ctlb)
+ws_frame_unmask(char *msg, int length, uint8_t maskval[WEBSOCKET_MASK_LENGTH])
+{
+       int i;
+
+       for (i = 0; i < length; i++)
+               msg[i] = msg[i] ^ maskval[i % 4];
+}
+
+static void
+conn_mod_process_frame(conn_t *conn, ws_frame_hdr_t *hdr, int masked)
+{
+       char msg[WEBSOCKET_MAX_UNEXTENDED_PAYLOAD_DATA_LENGTH];
+       uint8_t maskval[WEBSOCKET_MASK_LENGTH];
+       int dolen;
+
+       /* if we're masked, we get to collect the masking key for this frame */
+       if (masked)
+       {
+               dolen = rb_rawbuf_get(conn->modbuf_in, maskval, sizeof(maskval));
+               if (!dolen)
+               {
+                       close_conn(conn, WAIT_PLAIN, "websocket error: fault unpacking unmask key");
+                       return;
+               }
+       }
+
+       dolen = rb_rawbuf_get(conn->modbuf_in, msg, hdr->payload_length_mask);
+       if (!dolen)
+       {
+               close_conn(conn, WAIT_PLAIN, "websocket error: fault unpacking message");
+               return;
+       }
+
+       if (masked)
+               ws_frame_unmask(msg, dolen, maskval);
+
+       rb_linebuf_parse(&conn->plainbuf_out, msg, dolen, 1);
+}
+
+static void
+conn_mod_process_large(conn_t *conn, ws_frame_hdr_t *hdr, int masked)
+{
+       char msg[READBUF_SIZE];
+       uint16_t msglen;
+       uint8_t maskval[WEBSOCKET_MASK_LENGTH];
+       int dolen;
+
+       memset(msg, 0, sizeof msg);
+
+       dolen = rb_rawbuf_get(conn->modbuf_in, &msglen, sizeof(msglen));
+       if (!dolen)
+       {
+               close_conn(conn, WAIT_PLAIN, "websocket error: fault unpacking message size");
+               return;
+       }
+
+       msglen = ntohs(msglen);
+
+       if (masked)
+       {
+               dolen = rb_rawbuf_get(conn->modbuf_in, maskval, sizeof(maskval));
+               if (!dolen)
+               {
+                       close_conn(conn, WAIT_PLAIN, "websocket error: fault unpacking unmask key");
+                       return;
+               }
+       }
+
+       dolen = rb_rawbuf_get(conn->modbuf_in, msg, msglen);
+       if (!dolen)
+       {
+               close_conn(conn, WAIT_PLAIN, "websocket error: fault unpacking message");
+               return;
+       }
+
+       if (masked)
+               ws_frame_unmask(msg, dolen, maskval);
+
+       rb_linebuf_parse(&conn->plainbuf_out, msg, dolen, 1);
+}
+
+static void
+conn_mod_process_huge(conn_t *conn, ws_frame_hdr_t *hdr, int masked)
+{
+       /* XXX implement me */
+}
+
+static void
+conn_mod_process(conn_t *conn)
+{
+       ws_frame_hdr_t hdr;
+
+       while (1)
+       {
+               int masked;
+               int dolen = rb_rawbuf_get(conn->modbuf_in, &hdr, sizeof(hdr));
+               if (dolen != sizeof(hdr))
+                       break;
+
+               masked = (hdr.payload_length_mask >> 7) == 1;
+
+               hdr.payload_length_mask &= 0x7f;
+               switch (hdr.payload_length_mask)
+               {
+               case 126:
+                       conn_mod_process_large(conn, &hdr, masked);
+                       break;
+               case 127:
+                       conn_mod_process_huge(conn, &hdr, masked);
+                       break;
+               default:
+                       conn_mod_process_frame(conn, &hdr, masked);
+                       break;
+               }
+       }
+
+       conn_plain_write_sendq(conn->plain_fd, conn);
+}
+
+static void
+conn_mod_handshake_process(conn_t *conn)
+{
+       char inbuf[READBUF_SIZE];
+
+       memset(inbuf, 0, sizeof inbuf);
+
+       while (1)
+       {
+               char *p = NULL;
+
+               int dolen = rb_rawbuf_get(conn->modbuf_in, inbuf, sizeof inbuf);
+               if (!dolen)
+                       break;
+
+               if ((p = rb_strcasestr(inbuf, "Sec-WebSocket-Key:")) != NULL)
+               {
+                       char *start, *end;
+
+                       start = p + strlen("Sec-WebSocket-Key:");
+
+                       for (; start < (inbuf + READBUF_SIZE) && *start; start++)
+                       {
+                               if (*start != ' ' && *start != '\t')
+                                       break;
+                       }
+
+                       for (end = start; end < (inbuf + READBUF_SIZE) && *end; end++)
+                       {
+                               if (*end == '\r' || *end == '\n')
+                               {
+                                       *end = '\0';
+                                       break;
+                               }
+                       }
+
+                       rb_strlcpy(conn->client_key, start, sizeof(conn->client_key));
+                       SetKeyed(conn);
+               }
+       }
+
+       if (IsKeyed(conn))
+       {
+               SHA1 sha1;
+               uint8_t digest[SHA1_DIGEST_LENGTH];
+               char *resp;
+
+               sha1_init(&sha1);
+               sha1_update(&sha1, (uint8_t *) conn->client_key, strlen(conn->client_key));
+               sha1_update(&sha1, (uint8_t *) WEBSOCKET_SERVER_KEY, strlen(WEBSOCKET_SERVER_KEY));
+               sha1_final(&sha1, digest);
+
+               resp = (char *) rb_base64_encode(digest, SHA1_DIGEST_LENGTH);
+
+               conn_mod_write(conn, WEBSOCKET_ANSWER_STRING_1, strlen(WEBSOCKET_ANSWER_STRING_1));
+               conn_mod_write(conn, resp, strlen(resp));
+               conn_mod_write(conn, WEBSOCKET_ANSWER_STRING_2, strlen(WEBSOCKET_ANSWER_STRING_2));
+
+               rb_free(resp);
+       }
+
+       conn_mod_write_sendq(conn->mod_fd, conn);
+}
+
+static void
+conn_mod_read_cb(rb_fde_t *fd, void *data)
+{
+       char inbuf[READBUF_SIZE];
+
+       memset(inbuf, 0, sizeof inbuf);
+
+       conn_t *conn = data;
+       int length = 0;
+       if (conn == NULL)
+               return;
+
+       if (IsDead(conn))
+               return;
+
+       while (1)
+       {
+               if (IsDead(conn))
+                       return;
+
+               length = rb_read(fd, inbuf, sizeof(inbuf));
+
+               if (length < 0)
+               {
+                       if (rb_ignore_errno(errno))
+                       {
+                               rb_setselect(fd, RB_SELECT_READ, conn_mod_read_cb, conn);
+                               conn_plain_write_sendq(conn->plain_fd, conn);
+                       }
+                       else
+                               close_conn(conn, NO_WAIT, "Connection closed");
+
+                       return;
+               }
+               else if (length == 0)
+               {
+                       close_conn(conn, NO_WAIT, "Connection closed");
+                       return;
+               }
+
+               rb_rawbuf_append(conn->modbuf_in, inbuf, length);
+               if (!IsKeyed(conn))
+                       conn_mod_handshake_process(conn);
+               else
+                       conn_mod_process(conn);
+
+               if ((size_t) length < sizeof(inbuf))
+               {
+                       rb_setselect(fd, RB_SELECT_READ, conn_mod_read_cb, conn);
+                       return;
+               }
+       }
+}
+
+static bool
+plain_check_cork(conn_t * conn)
+{
+       if(rb_rawbuf_length(conn->modbuf_out) >= 4096)
+       {
+               /* if we have over 4k pending outbound, don't read until
+                * we've cleared the queue */
+               SetCork(conn);
+               rb_setselect(conn->plain_fd, RB_SELECT_READ, NULL, NULL);
+               /* try to write */
+               if (IsKeyed(conn))
+                       conn_mod_write_sendq(conn->mod_fd, conn);
+               return true;
+       }
+
+       return false;
+}
+
+static void
+conn_plain_process_recvq(conn_t *conn)
+{
+       char inbuf[READBUF_SIZE];
+
+       memset(inbuf, 0, sizeof inbuf);
+
+       while (1)
+       {
+               int dolen = rb_linebuf_get(&conn->plainbuf_in, inbuf, sizeof inbuf, LINEBUF_COMPLETE, LINEBUF_PARSED);
+               if (!dolen)
+                       break;
+
+               conn_mod_write_frame(conn, inbuf, dolen);
+       }
+
+       if (IsKeyed(conn))
+               conn_mod_write_sendq(conn->mod_fd, conn);
+}
+
+static void
+conn_plain_read_cb(rb_fde_t *fd, void *data)
+{
+       char inbuf[READBUF_SIZE];
+
+       memset(inbuf, 0, sizeof inbuf);
+
+       conn_t *conn = data;
+       int length = 0;
+       if(conn == NULL)
+               return;
+
+       if(IsDead(conn))
+               return;
+
+       if(plain_check_cork(conn))
+               return;
+
+       while(1)
+       {
+               if(IsDead(conn))
+                       return;
+
+               length = rb_read(conn->plain_fd, inbuf, sizeof(inbuf));
+
+               if(length == 0 || (length < 0 && !rb_ignore_errno(errno)))
+               {
+                       close_conn(conn, NO_WAIT, NULL);
+                       return;
+               }
+
+               if(length < 0)
+               {
+                       rb_setselect(conn->plain_fd, RB_SELECT_READ, conn_plain_read_cb, conn);
+                       if (IsKeyed(conn))
+                               conn_plain_process_recvq(conn);
+                       return;
+               }
+               conn->plain_in += length;
+
+               (void) rb_linebuf_parse(&conn->plainbuf_in, inbuf, length, 0);
+
+               if(IsDead(conn))
+                       return;
+               if(plain_check_cork(conn))
+                       return;
+       }
+}
+
+static void
+conn_plain_read_shutdown_cb(rb_fde_t *fd, void *data)
+{
+       char inbuf[READBUF_SIZE];
+       conn_t *conn = data;
+       int length = 0;
+
+       if(conn == NULL)
+               return;
+
+       while(1)
+       {
+               length = rb_read(conn->plain_fd, inbuf, sizeof(inbuf));
+
+               if(length == 0 || (length < 0 && !rb_ignore_errno(errno)))
+               {
+                       rb_close(conn->plain_fd);
+                       rb_dlinkAdd(conn, &conn->node, &dead_list);
+                       return;
+               }
+
+               if(length < 0)
+               {
+                       rb_setselect(conn->plain_fd, RB_SELECT_READ, conn_plain_read_shutdown_cb, conn);
+                       return;
+               }
+       }
+}
+
+static void
+wsock_process(mod_ctl_t * ctl, mod_ctl_buf_t * ctlb)
 {
        conn_t *conn;
        uint32_t id;
@@ -252,7 +852,8 @@ wsock_process_accept(mod_ctl_t * ctl, mod_ctl_buf_t * ctlb)
        if(rb_get_type(conn->plain_fd) == RB_FD_UNKNOWN)
                rb_set_type(conn->plain_fd, RB_FD_SOCKET);
 
-       // XXX todo
+       conn_mod_read_cb(conn->mod_fd, conn);
+       conn_plain_read_cb(conn->plain_fd, conn);
 }
 
 static void
@@ -274,7 +875,7 @@ mod_process_cmd_recv(mod_ctl_t * ctl)
                                        cleanup_bad_message(ctl, ctl_buf);
                                        break;
                                }
-                               wsock_process_accept(ctl, ctl_buf);
+                               wsock_process(ctl, ctl_buf);
                                break;
                        }
                default:
@@ -326,36 +927,6 @@ mod_read_ctl(rb_fde_t *F, void *data)
        rb_setselect(ctl->F, RB_SELECT_READ, mod_read_ctl, ctl);
 }
 
-static void
-mod_write_ctl(rb_fde_t *F, void *data)
-{
-       mod_ctl_t *ctl = data;
-       mod_ctl_buf_t *ctl_buf;
-       rb_dlink_node *ptr, *next;
-       int retlen, x;
-
-       RB_DLINK_FOREACH_SAFE(ptr, next, ctl->writeq.head)
-       {
-               ctl_buf = ptr->data;
-               retlen = rb_send_fd_buf(ctl->F, ctl_buf->F, ctl_buf->nfds, ctl_buf->buf,
-                                       ctl_buf->buflen, ppid);
-               if(retlen > 0)
-               {
-                       rb_dlinkDelete(ptr, &ctl->writeq);
-                       for(x = 0; x < ctl_buf->nfds; x++)
-                               rb_close(ctl_buf->F[x]);
-                       rb_free(ctl_buf->buf);
-                       rb_free(ctl_buf);
-
-               }
-               if(retlen == 0 || (retlen < 0 && !rb_ignore_errno(errno)))
-                       exit(0);
-
-       }
-       if(rb_dlink_list_length(&ctl->writeq) > 0)
-               rb_setselect(ctl->F, RB_SELECT_WRITE, mod_write_ctl, ctl);
-}
-
 static void
 read_pipe_ctl(rb_fde_t *F, void *data)
 {
@@ -374,7 +945,7 @@ int
 main(int argc, char **argv)
 {
        const char *s_ctlfd, *s_pipe, *s_pid;
-       int ctlfd, pipefd, x, maxfd;
+       int ctlfd, pipefd, maxfd, x;
        maxfd = maxconn();
 
        s_ctlfd = getenv("CTL_FD");
@@ -384,7 +955,7 @@ main(int argc, char **argv)
        if(s_ctlfd == NULL || s_pipe == NULL || s_pid == NULL)
        {
                fprintf(stderr,
-                       "This is the charybdis wsockd for internal ircd use.\n");
+                       "This is the solanum wsockd for internal ircd use.\n");
                fprintf(stderr,
                        "You aren't supposed to run me directly. Exiting.\n");
                exit(1);
@@ -393,8 +964,7 @@ main(int argc, char **argv)
        ctlfd = atoi(s_ctlfd);
        pipefd = atoi(s_pipe);
        ppid = atoi(s_pid);
-       x = 0;
-#ifndef _WIN32
+
        for(x = 0; x < maxfd; x++)
        {
                if(x != ctlfd && x != pipefd && x > 2)
@@ -413,10 +983,11 @@ main(int argc, char **argv)
                if(x > 2)
                        close(x);
        }
-#endif
+
        setup_signals();
        rb_lib_init(NULL, NULL, NULL, 0, maxfd, 1024, 4096);
-       rb_init_rawbuffers(1024);
+       rb_linebuf_init(4096);
+       rb_init_rawbuffers(4096);
 
        mod_ctl = rb_malloc(sizeof(mod_ctl_t));
        mod_ctl->F = rb_open(ctlfd, RB_FD_SOCKET, "ircd control socket");