servlink/io.c

   1 /************************************************************************
   2  *   IRC - Internet Relay Chat, servlink/io.c
   3  *
   4  *   This program is free software; you can redistribute it and/or modify
   5  *   it under the terms of the GNU General Public License as published by
   6  *   the Free Software Foundation; either version 1, or (at your option)
   7  *   any later version.
   8  *
   9  *   This program is distributed in the hope that it will be useful,
  10  *   but WITHOUT ANY WARRANTY; without even the implied warranty of
  11  *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12  *   GNU General Public License for more details.
  13  *
  14  *   You should have received a copy of the GNU General Public License
  15  *   along with this program; if not, write to the Free Software
  16  *   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
  17  *
  18  *   $Id: io.c 22629 2006-05-23 17:47:03Z androsyn $
  19  */
  20
  21 #include "setup.h"
  22
  23 #include <sys/types.h>
  24 #include <sys/socket.h>
  25
  26 #include <assert.h>
  27 #include <errno.h>
  28 #include <string.h>
  29 #include <stdlib.h>
  30 #include <stdarg.h>
  31 #include <stdio.h>
  32 #include <unistd.h>
  33
  34 #ifdef HAVE_LIBZ
  35 #include <zlib.h>
  36 #endif
  37
  38 #include "stdinc.h"
  39 #include "servlink.h"
  40 #include "io.h"
  41 #include "control.h"
  42
  43 static int check_error(int, int, int);
  44
  45 static const char *
  46 fd_name(int fd)
  47 {
  48         if(fd == CONTROL.fd)
  49                 return "control";
  50         if(fd == LOCAL.fd)
  51                 return "data";
  52         if(fd == REMOTE.fd)
  53                 return "network";
  54
  55         /* uh oh... */
  56         return "unknown";
  57 }
  58
  59 #if defined( HAVE_LIBZ )
  60 static unsigned char tmp_buf[BUFLEN];
  61 static unsigned char tmp2_buf[BUFLEN];
  62 #endif
  63
  64 static unsigned char ctrl_buf[256] = "";
  65 static unsigned int ctrl_len = 0;
  66 static unsigned int ctrl_ofs = 0;
  67
  68 void
  69 io_loop(int nfds)
  70 {
  71         fd_set rfds;
  72         fd_set wfds;
  73         int i, ret;
  74
  75         /* loop forever */
  76         for (;;)
  77         {
  78                 FD_ZERO(&rfds);
  79                 FD_ZERO(&wfds);
  80
  81                 for (i = 0; i < 3; i++)
  82                 {
  83                         if(fds[i].read_cb)
  84                                 FD_SET(fds[i].fd, &rfds);
  85                         if(fds[i].write_cb)
  86                                 FD_SET(fds[i].fd, &wfds);
  87                 }
  88
  89                 /* we have <3 fds ever, so I don't think select is too painful */
  90                 ret = select(nfds, &rfds, &wfds, NULL, NULL);
  91
  92                 if(ret < 0)
  93                 {
  94                         check_error(ret, IO_SELECT, -1);        /* exit on fatal errors */
  95                 }
  96                 else if(ret > 0)
  97                 {
  98                         /* call any callbacks */
  99                         for (i = 0; i < 3; i++)
 100                         {
 101                                 if(FD_ISSET(fds[i].fd, &rfds) && fds[i].read_cb)
 102                                         (*fds[i].read_cb) ();
 103                                 if(FD_ISSET(fds[i].fd, &wfds) && fds[i].write_cb)
 104                                         (*fds[i].write_cb) ();
 105                         }
 106                 }
 107         }
 108 }
 109
 110 void
 111 send_data_blocking(int fd, unsigned char *data, int datalen)
 112 {
 113         int ret;
 114         fd_set wfds;
 115
 116         while (1)
 117         {
 118                 ret = write(fd, data, datalen);
 119
 120                 if(ret == datalen)
 121                         return;
 122                 else if(ret > 0)
 123                 {
 124                         data += ret;
 125                         datalen -= ret;
 126                 }
 127
 128                 ret = check_error(ret, IO_WRITE, fd);
 129
 130                 FD_ZERO(&wfds);
 131                 FD_SET(fd, &wfds);
 132
 133                 /* sleep until we can write to the fd */
 134                 while (1)
 135                 {
 136                         ret = select(fd + 1, NULL, &wfds, NULL, NULL);
 137
 138                         if(ret > 0)     /* break out so we can write */
 139                                 break;
 140
 141                         if(ret < 0)     /* error ? */
 142                                 check_error(ret, IO_SELECT, fd);        /* exit on fatal errors */
 143
 144                         /* loop on non-fatal errors */
 145                 }
 146         }
 147 }
 148
 149 /*
 150  * process_sendq:
 151  *
 152  * used before CMD_INIT to pass contents of SendQ from ircd
 153  * to servlink.  This data must _not_ be encrypted/compressed.
 154  */
 155 void
 156 process_sendq(struct ctrl_command *cmd)
 157 {
 158         send_data_blocking(REMOTE.fd, cmd->data, cmd->datalen);
 159 }
 160
 161 /*
 162  * process_recvq:
 163  *
 164  * used before CMD_INIT to pass contents of RecvQ from ircd
 165  * to servlink.  This data must be decrypted/decopmressed before
 166  * sending back to the ircd.
 167  */
 168 void
 169 process_recvq(struct ctrl_command *cmd)
 170 {
 171         int ret;
 172         unsigned char *buf;
 173         int blen;
 174         unsigned char *data = cmd->data;
 175         unsigned int datalen = cmd->datalen;
 176
 177         buf = data;
 178         blen = datalen;
 179         ret = -1;
 180         if(datalen > READLEN)
 181                 send_error("Error processing INJECT_RECVQ - buffer too long (%d > %d)",
 182                            datalen, READLEN);
 183
 184 #ifdef HAVE_LIBZ
 185         if(in_state.zip)
 186         {
 187                 /* decompress data */
 188                 in_state.zip_state.z_stream.next_in = buf;
 189                 in_state.zip_state.z_stream.avail_in = blen;
 190                 in_state.zip_state.z_stream.next_out = tmp2_buf;
 191                 in_state.zip_state.z_stream.avail_out = BUFLEN;
 192
 193                 buf = tmp2_buf;
 194                 while (in_state.zip_state.z_stream.avail_in)
 195                 {
 196                         if((ret = inflate(&in_state.zip_state.z_stream, Z_NO_FLUSH)) != Z_OK)
 197                         {
 198                                 if(!strncmp("ERROR ", (char *)in_state.zip_state.z_stream.next_in, 6))
 199                                         send_error("Received uncompressed ERROR");
 200                                 else
 201                                         send_error("Inflate failed: %s", zError(ret));
 202                         }
 203                         blen = BUFLEN - in_state.zip_state.z_stream.avail_out;
 204
 205                         if(in_state.zip_state.z_stream.avail_in)
 206                         {
 207                                 send_data_blocking(LOCAL.fd, buf, blen);
 208                                 blen = 0;
 209                                 in_state.zip_state.z_stream.next_out = buf;
 210                                 in_state.zip_state.z_stream.avail_out = BUFLEN;
 211                         }
 212                 }
 213
 214                 if(!blen)
 215                         return;
 216         }
 217 #endif
 218
 219         send_data_blocking(LOCAL.fd, buf, blen);
 220 }
 221
 222 void
 223 send_zipstats(struct ctrl_command *unused)
 224 {
 225 #ifdef HAVE_LIBZ
 226         int i = 0;
 227         int ret;
 228         uint32_t len;
 229         if(!in_state.active || !out_state.active)
 230                 send_error("Error processing CMD_ZIPSTATS - link is not active!");
 231         if(!in_state.zip || !out_state.zip)
 232                 send_error("Error processing CMD_ZIPSTATS - link is not compressed!");
 233
 234         ctrl_buf[i++] = RPL_ZIPSTATS;
 235         ctrl_buf[i++] = 0;
 236         ctrl_buf[i++] = 16;
 237
 238         len = (uint32_t) in_state.zip_state.z_stream.total_out;
 239         ctrl_buf[i++] = ((len >> 24) & 0xFF);
 240         ctrl_buf[i++] = ((len >> 16) & 0xFF);
 241         ctrl_buf[i++] = ((len >> 8) & 0xFF);
 242         ctrl_buf[i++] = ((len) & 0xFF);
 243
 244         len = (uint32_t) in_state.zip_state.z_stream.total_in;
 245         ctrl_buf[i++] = ((len >> 24) & 0xFF);
 246         ctrl_buf[i++] = ((len >> 16) & 0xFF);
 247         ctrl_buf[i++] = ((len >> 8) & 0xFF);
 248         ctrl_buf[i++] = ((len) & 0xFF);
 249
 250         len = (uint32_t) out_state.zip_state.z_stream.total_in;
 251         ctrl_buf[i++] = ((len >> 24) & 0xFF);
 252         ctrl_buf[i++] = ((len >> 16) & 0xFF);
 253         ctrl_buf[i++] = ((len >> 8) & 0xFF);
 254         ctrl_buf[i++] = ((len) & 0xFF);
 255
 256         len = (uint32_t) out_state.zip_state.z_stream.total_out;
 257         ctrl_buf[i++] = ((len >> 24) & 0xFF);
 258         ctrl_buf[i++] = ((len >> 16) & 0xFF);
 259         ctrl_buf[i++] = ((len >> 8) & 0xFF);
 260         ctrl_buf[i++] = ((len) & 0xFF);
 261
 262         in_state.zip_state.z_stream.total_in = 0;
 263         in_state.zip_state.z_stream.total_out = 0;
 264         out_state.zip_state.z_stream.total_in = 0;
 265         out_state.zip_state.z_stream.total_out = 0;
 266
 267         ret = check_error(write(CONTROL.fd, ctrl_buf, i), IO_WRITE, CONTROL.fd);
 268         if(ret < i)
 269         {
 270                 /* write incomplete, register write cb */
 271                 CONTROL.write_cb = write_ctrl;
 272                 /*  deregister read_cb */
 273                 CONTROL.read_cb = NULL;
 274                 ctrl_ofs = ret;
 275                 ctrl_len = i - ret;
 276                 return;
 277         }
 278 #else
 279         send_error("can't send_zipstats -- no zlib support!");
 280 #endif
 281 }
 282
 283 /* send_error
 284  *   - we ran into some problem, make a last ditch effort to
 285  *     flush the control fd sendq, then (blocking) send an
 286  *     error message over the control fd.
 287  */
 288 void
 289 send_error(const char *message, ...)
 290 {
 291         va_list args;
 292         static int sending_error = 0;
 293         struct linger linger_opt = { 1, 30 };   /* wait 30 seconds */
 294         int len;
 295
 296         if(sending_error)
 297                 exit(1);        /* we did _try_ */
 298
 299         sending_error = 1;
 300
 301         if(ctrl_len)            /* attempt to flush any data we have... */
 302         {
 303                 send_data_blocking(CONTROL.fd, (ctrl_buf + ctrl_ofs), ctrl_len);
 304         }
 305
 306         /* prepare the message, in in_buf, since we won't be using it again.. */
 307         in_state.buf[0] = RPL_ERROR;
 308         in_state.buf[1] = 0;
 309         in_state.buf[2] = 0;
 310
 311         va_start(args, message);
 312         len = vsprintf((char *) in_state.buf + 3, message, args);
 313         va_end(args);
 314
 315         in_state.buf[3 + len++] = '\0';
 316         in_state.buf[1] = len >> 8;
 317         in_state.buf[2] = len & 0xFF;
 318         len += 3;
 319
 320         send_data_blocking(CONTROL.fd, in_state.buf, len);
 321
 322         /* XXX - is this portable?
 323          *       this obviously will fail on a non socket.. */
 324         setsockopt(CONTROL.fd, SOL_SOCKET, SO_LINGER, &linger_opt, sizeof(struct linger));
 325
 326         /* well, we've tried... */
 327         exit(1);                /* now abort */
 328 }
 329
 330 /* read_ctrl
 331  *      called when a command is waiting on the control pipe
 332  */
 333 void
 334 read_ctrl(void)
 335 {
 336         int ret;
 337         unsigned char tmp[2];
 338         unsigned char *len;
 339         struct command_def *cdef;
 340         static struct ctrl_command cmd = { 0, 0, 0, 0, NULL };
 341
 342         if(cmd.command == 0)    /* we don't have a command yet */
 343         {
 344                 cmd.gotdatalen = 0;
 345                 cmd.datalen = 0;
 346                 cmd.readdata = 0;
 347                 cmd.data = NULL;
 348
 349                 /* read the command */
 350                 if(!(ret = check_error(read(CONTROL.fd, tmp, 1), IO_READ, CONTROL.fd)))
 351                         return;
 352
 353                 cmd.command = tmp[0];
 354         }
 355
 356         for (cdef = command_table; cdef->commandid; cdef++)
 357         {
 358                 if((int)cdef->commandid == cmd.command)
 359                         break;
 360         }
 361
 362         if(!cdef->commandid)
 363         {
 364                 send_error("Unsupported command (servlink/ircd out of sync?): %d", cmd.command);
 365                 /* NOTREACHED */
 366         }
 367
 368         /* read datalen for commands including data */
 369         if(cdef->flags & COMMAND_FLAG_DATA)
 370         {
 371                 if(cmd.gotdatalen < 2)
 372                 {
 373                         len = tmp;
 374                         if(!(ret = check_error(read(CONTROL.fd, len,
 375                                                     (2 - cmd.gotdatalen)), IO_READ, CONTROL.fd)))
 376                                 return;
 377
 378                         if(cmd.gotdatalen == 0)
 379                         {
 380                                 cmd.datalen = len[0] << 8;
 381                                 cmd.gotdatalen++;
 382                                 ret--;
 383                                 len++;
 384                         }
 385                         if(ret && (cmd.gotdatalen == 1))
 386                         {
 387                                 cmd.datalen |= len[0];
 388                                 cmd.gotdatalen++;
 389                                 if(cmd.datalen > 0)
 390                                         cmd.data = calloc(cmd.datalen, 1);
 391                         }
 392                 }
 393         }
 394
 395         if(cmd.readdata < cmd.datalen)  /* try to get any remaining data */
 396         {
 397                 if(!(ret = check_error(read(CONTROL.fd,
 398                                             (cmd.data + cmd.readdata),
 399                                             cmd.datalen - cmd.readdata), IO_READ, CONTROL.fd)))
 400                         return;
 401
 402                 cmd.readdata += ret;
 403                 if(cmd.readdata < cmd.datalen)
 404                         return;
 405         }
 406
 407         /* we now have the command and any data */
 408         (*cdef->handler) (&cmd);
 409
 410         if(cmd.datalen > 0)
 411                 free(cmd.data);
 412         cmd.command = 0;
 413 }
 414
 415 void
 416 write_ctrl(void)
 417 {
 418         int ret;
 419
 420         assert(ctrl_len);
 421
 422         if(!(ret = check_error(write(CONTROL.fd, (ctrl_buf + ctrl_ofs),
 423                                      ctrl_len), IO_WRITE, CONTROL.fd)))
 424                 return;         /* no data waiting */
 425
 426         ctrl_len -= ret;
 427
 428         if(!ctrl_len)
 429         {
 430                 /* write completed, de-register write cb */
 431                 CONTROL.write_cb = NULL;
 432                 /* reregister read_cb */
 433                 CONTROL.read_cb = read_ctrl;
 434                 ctrl_ofs = 0;
 435         }
 436         else
 437                 ctrl_ofs += ret;
 438 }
 439
 440 void
 441 read_data(void)
 442 {
 443         int ret, ret2;
 444         unsigned char *buf = out_state.buf;
 445         int blen;
 446         ret2 = -1;
 447         assert(!out_state.len);
 448
 449 #if defined(HAVE_LIBZ)
 450         if(out_state.zip || out_state.crypt)
 451                 buf = tmp_buf;
 452 #endif
 453
 454         while ((ret = check_error(read(LOCAL.fd, buf, READLEN), IO_READ, LOCAL.fd)))
 455         {
 456                 blen = ret;
 457 #ifdef HAVE_LIBZ
 458                 if(out_state.zip)
 459                 {
 460                         out_state.zip_state.z_stream.next_in = buf;
 461                         out_state.zip_state.z_stream.avail_in = ret;
 462
 463                         buf = out_state.buf;
 464                         out_state.zip_state.z_stream.next_out = buf;
 465                         out_state.zip_state.z_stream.avail_out = BUFLEN;
 466                         if(!(ret2 = deflate(&out_state.zip_state.z_stream,
 467                                             Z_PARTIAL_FLUSH)) == Z_OK)
 468                                 send_error("error compressing outgoing data - deflate returned: %s",
 469                                            zError(ret2));
 470
 471                         if(!out_state.zip_state.z_stream.avail_out)
 472                                 send_error("error compressing outgoing data - avail_out == 0");
 473                         if(out_state.zip_state.z_stream.avail_in)
 474                                 send_error("error compressing outgoing data - avail_in != 0");
 475
 476                         blen = BUFLEN - out_state.zip_state.z_stream.avail_out;
 477                 }
 478 #endif
 479
 480
 481                 ret = check_error(write(REMOTE.fd, out_state.buf, blen), IO_WRITE, REMOTE.fd);
 482                 if(ret < blen)
 483                 {
 484                         /* write incomplete, register write cb */
 485                         REMOTE.write_cb = write_net;
 486                         /*  deregister read_cb */
 487                         LOCAL.read_cb = NULL;
 488                         out_state.ofs = ret;
 489                         out_state.len = blen - ret;
 490                         return;
 491                 }
 492 #if defined(HAVE_LIBZ)
 493                 if(out_state.zip)
 494                         buf = tmp_buf;
 495 #endif
 496         }
 497
 498 }
 499
 500 void
 501 write_net(void)
 502 {
 503         int ret;
 504
 505         assert(out_state.len);
 506
 507         if(!(ret = check_error(write(REMOTE.fd,
 508                                      (out_state.buf + out_state.ofs),
 509                                      out_state.len), IO_WRITE, REMOTE.fd)))
 510                 return;         /* no data waiting */
 511
 512         out_state.len -= ret;
 513
 514         if(!out_state.len)
 515         {
 516                 /* write completed, de-register write cb */
 517                 REMOTE.write_cb = NULL;
 518                 /* reregister read_cb */
 519                 LOCAL.read_cb = read_data;
 520                 out_state.ofs = 0;
 521         }
 522         else
 523                 out_state.ofs += ret;
 524 }
 525
 526 void
 527 read_net(void)
 528 {
 529         int ret;
 530         int ret2;
 531         unsigned char *buf = in_state.buf;
 532         int blen;
 533         ret2 = -1;
 534         assert(!in_state.len);
 535
 536 #if defined(HAVE_LIBZ)
 537         if(in_state.zip)
 538                 buf = tmp_buf;
 539 #endif
 540
 541         while ((ret = check_error(read(REMOTE.fd, buf, READLEN), IO_READ, REMOTE.fd)))
 542         {
 543                 blen = ret;
 544 #ifdef HAVE_LIBZ
 545                 if(in_state.zip)
 546                 {
 547                         /* decompress data */
 548                         in_state.zip_state.z_stream.next_in = buf;
 549                         in_state.zip_state.z_stream.avail_in = ret;
 550                         in_state.zip_state.z_stream.next_out = in_state.buf;
 551                         in_state.zip_state.z_stream.avail_out = BUFLEN;
 552
 553                         while (in_state.zip_state.z_stream.avail_in)
 554                         {
 555                                 if((ret2 = inflate(&in_state.zip_state.z_stream,
 556                                                    Z_NO_FLUSH)) != Z_OK)
 557                                 {
 558                                         if(!strncmp("ERROR ", (char *)buf, 6))
 559                                                 send_error("Received uncompressed ERROR");
 560                                         send_error("Inflate failed: %s", zError(ret2));
 561                                 }
 562                                 blen = BUFLEN - in_state.zip_state.z_stream.avail_out;
 563
 564                                 if(in_state.zip_state.z_stream.avail_in)
 565                                 {
 566                                         if(blen)
 567                                         {
 568                                                 send_data_blocking(LOCAL.fd, in_state.buf, blen);
 569                                                 blen = 0;
 570                                         }
 571
 572                                         in_state.zip_state.z_stream.next_out = in_state.buf;
 573                                         in_state.zip_state.z_stream.avail_out = BUFLEN;
 574                                 }
 575                         }
 576
 577                         if(!blen)
 578                                 return; /* that didn't generate any decompressed input.. */
 579                 }
 580 #endif
 581
 582                 ret = check_error(write(LOCAL.fd, in_state.buf, blen), IO_WRITE, LOCAL.fd);
 583
 584                 if(ret < blen)
 585                 {
 586                         in_state.ofs = ret;
 587                         in_state.len = blen - ret;
 588                         /* write incomplete, register write cb */
 589                         LOCAL.write_cb = write_data;
 590                         /* deregister read_cb */
 591                         REMOTE.read_cb = NULL;
 592                         return;
 593                 }
 594 #if defined(HAVE_LIBZ)
 595                 if(in_state.zip)
 596                         buf = tmp_buf;
 597 #endif
 598         }
 599 }
 600
 601 void
 602 write_data(void)
 603 {
 604         int ret;
 605
 606         assert(in_state.len);
 607
 608         if(!(ret = check_error(write(LOCAL.fd,
 609                                      (in_state.buf + in_state.ofs),
 610                                      in_state.len), IO_WRITE, LOCAL.fd)))
 611                 return;
 612
 613         in_state.len -= ret;
 614
 615         if(!in_state.len)
 616         {
 617                 /* write completed, de-register write cb */
 618                 LOCAL.write_cb = NULL;
 619                 /* reregister read_cb */
 620                 REMOTE.read_cb = read_net;
 621                 in_state.ofs = 0;
 622         }
 623         else
 624                 in_state.ofs += ret;
 625 }
 626
 627 int
 628 check_error(int ret, int io, int fd)
 629 {
 630         if(ret > 0)             /* no error */
 631                 return ret;
 632         if(ret == 0)            /* EOF */
 633         {
 634                 send_error("%s failed on %s: EOF", IO_TYPE(io), FD_NAME(fd));
 635                 exit(1);        /* NOTREACHED */
 636         }
 637
 638         /* ret == -1.. */
 639         switch (errno)
 640         {
 641         case EINPROGRESS:
 642         case EWOULDBLOCK:
 643 #if EAGAIN != EWOULDBLOCK
 644         case EAGAIN:
 645 #endif
 646         case EALREADY:
 647         case EINTR:
 648 #ifdef ERESTART
 649         case ERESTART:
 650 #endif
 651                 /* non-fatal error, 0 bytes read */
 652                 return 0;
 653         }
 654
 655         /* fatal error */
 656         send_error("%s failed on %s: %s", IO_TYPE(io), FD_NAME(fd), strerror(errno));
 657         exit(1);                /* NOTREACHED */
 658 }