UNIXworkcode

1 /* 2 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS HEADER. 3 * 4 * Copyright 2013 Olaf Wintermann. All rights reserved. 5 * 6 * Redistribution and use in source and binary forms, with or without 7 * modification, are permitted provided that the following conditions are met: 8 * 9 * 1. Redistributions of source code must retain the above copyright 10 * notice, this list of conditions and the following disclaimer. 11 * 12 * 2. Redistributions in binary form must reproduce the above copyright 13 * notice, this list of conditions and the following disclaimer in the 14 * documentation and/or other materials provided with the distribution. 15 * 16 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" 17 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE 18 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE 19 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE 20 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR 21 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF 22 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS 23 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN 24 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) 25 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE 26 * POSSIBILITY OF SUCH DAMAGE. 27 */ 28 29 #ifdef __gnu_linux__ 30 #define _GNU_SOURCE 31 #endif 32 33 #include <unistd.h> 34 #include <stdlib.h> 35 36 #ifdef XP_UNIX 37 #include <sys/uio.h> 38 #include <sys/uio.h> 39 #endif 40 41 #ifdef XP_WIN32 42 43 #endif 44 45 #if defined(LINUX) || defined(SOLARIS) 46 #include <sys/sendfile.h> 47 #define WS_SENDFILE 48 #elif defined(BSD) 49 #if defined(__NetBSD__) || defined(__OpenBSD__) 50 #define net_sys_sendfile net_fallback_sendfile 51 #else 52 #define WS_SENDFILE 53 #endif 54 #endif 55 56 #ifdef WS_SENDFILE 57 #define NET_SYS_SENDFILE net_sys_sendfile 58 #else 59 #define NET_SYS_SENDFILE net_fallback_sendfile 60 #endif 61 62 63 64 #include "../daemon/vfs.h" 65 #include "io.h" 66 #include "pool.h" 67 #include "../daemon/netsite.h" 68 #include "../daemon/event.h" 69 #include "cx/utils.h" 70 #include <cx/printf.h> 71 72 IOStream native_io_funcs = { 73 (io_write_f)net_sys_write, 74 (io_writev_f)net_sys_writev, 75 (io_read_f)net_sys_read, 76 (io_sendfile_f)NET_SYS_SENDFILE, 77 (io_close_f)net_sys_close, 78 NULL, 79 (io_setmode_f)net_sys_setmode, 80 (io_poll_f)net_sys_poll, 81 0, 82 0 83 }; 84 85 IOStream http_io_funcs = { 86 (io_write_f)net_http_write, 87 (io_writev_f)net_http_writev, 88 (io_read_f)net_http_read, 89 (io_sendfile_f)net_http_sendfile, 90 (io_close_f)net_http_close, 91 (io_finish_f)net_http_finish, 92 (io_setmode_f)net_http_setmode, 93 (io_poll_f)net_http_poll, 94 0, 95 IO_STREAM_TYPE_HTTP 96 }; 97 98 IOStream ssl_io_funcs = { 99 (io_write_f)net_ssl_write, 100 (io_writev_f)net_ssl_writev, 101 (io_read_f)net_ssl_read, 102 NULL, 103 (io_close_f)net_ssl_close, 104 (io_finish_f)net_ssl_finish, 105 (io_setmode_f)net_ssl_setmode, 106 (io_poll_f)net_ssl_poll, 107 0, 108 IO_STREAM_TYPE_SSL 109 }; 110 111 static int net_write_max_attempts = 16384; 112 113 void io_set_max_writes(int n) { 114 net_write_max_attempts = 1; 115 } 116 117 /* 118 * Sysstream implementation 119 */ 120 121 IOStream* Sysstream_new(pool_handle_t *pool, SYS_SOCKET fd) { 122 Sysstream *st = pool_malloc(pool, sizeof(Sysstream)); 123 st->st = native_io_funcs; 124 st->fd = fd; 125 return (IOStream*)st; 126 } 127 128 #ifdef XP_UNIX 129 ssize_t net_sys_write(Sysstream *st, const void *buf, size_t nbytes) { 130 ssize_t r = write(st->fd, buf, nbytes); 131 st->st.io_errno = errno; 132 return r; 133 } 134 135 ssize_t net_sys_writev(Sysstream *st, struct iovec *iovec, int iovcnt) { 136 ssize_t r = writev(st->fd, iovec, iovcnt); 137 st->st.io_errno = errno; 138 return r; 139 } 140 141 ssize_t net_sys_read(Sysstream *st, void *buf, size_t nbytes) { 142 ssize_t r = read(st->fd, buf, nbytes); 143 st->st.io_errno = errno; 144 return r; 145 } 146 147 #ifdef WS_SENDFILE 148 ssize_t net_sys_sendfile(Sysstream *st, sendfiledata *sfd) { 149 ssize_t ret = 0; 150 off_t fileoffset = sfd->offset; 151 if(sfd->fd->fd != -1) { 152 #ifdef BSD 153 struct iovec hdvec; 154 hdvec.iov_base = (void*)sfd->header; 155 hdvec.iov_len = sfd->hlen; 156 struct iovec trvec; 157 trvec.iov_base = (void*)sfd->trailer; 158 trvec.iov_len = sfd->tlen; 159 struct sf_hdtr hdtr; 160 hdtr.headers = &hdvec; 161 hdtr.hdr_cnt = sfd->hlen > 0 ? 1 : 0; 162 hdtr.trailers = &trvec; 163 hdtr.trl_cnt = sfd->tlen > 0 ? 1 : 0; 164 165 off_t len = sfd->len; 166 #ifdef OSX 167 ret = sendfile(sfd->fd->fd, st->fd, fileoffset, &len, &hdtr, 0); 168 #else // BSD 169 ret = sendfile( 170 sfd->fd->fd, 171 st->fd, 172 fileoffset, 173 sfd->len, 174 &hdtr, 175 NULL, 176 0); 177 #endif 178 if(ret == 0) { 179 ret = sfd->hlen + sfd->tlen + sfd->len; 180 } 181 182 #else // Solaris/Linux 183 if(sfd->header) { 184 ret += write(st->fd, sfd->header, sfd->hlen); 185 } 186 ret += sendfile(st->fd, sfd->fd->fd, &fileoffset, sfd->len); 187 if(sfd->trailer) { 188 ret += write(st->fd, sfd->trailer, sfd->tlen); 189 } 190 #endif 191 } else { 192 return net_fallback_sendfile((IOStream*)st, sfd); 193 } 194 st->st.io_errno = errno; 195 return ret; 196 } 197 #endif 198 199 void net_sys_close(Sysstream *st) { 200 system_close(st->fd); 201 } 202 203 void net_sys_setmode(Sysstream *st, int mode) { 204 int flags; 205 if (-1 == (flags = fcntl(st->fd, F_GETFL, 0))) { 206 flags = 0; 207 } 208 if(mode == IO_MODE_BLOCKING) { 209 if (fcntl(st->fd, F_SETFL, flags & ~O_NONBLOCK) != 0) { 210 perror("fcntl"); 211 // TODO: error 212 } 213 } else if(mode == IO_MODE_NONBLOCKING) { 214 if (fcntl(st->fd, F_SETFL, flags | O_NONBLOCK) != 0) { 215 perror("fcntl"); 216 // TODO: error 217 } 218 } 219 } 220 221 int net_sys_poll(Sysstream *st, EventHandler *ev, int events, Event *cb) { 222 switch(events) { 223 default: return -1; 224 case IO_POLL_NONE: return ev_remove_poll(ev, st->fd); 225 case IO_POLL_IN: return ev_pollin(ev, st->fd, cb); 226 case IO_POLL_OUT: return ev_pollout(ev, st->fd, cb); 227 case IO_POLL_IN | IO_POLL_OUT: return -1; // TODO: implement 228 } 229 } 230 231 #elif defined(XP_WIN32) 232 233 ssize_t net_sys_write(Sysstream *st, void *buf, size_t nbytes) { 234 int ret = send(st->fd, buf, nbytes, 0); 235 if(ret == SOCKET_ERROR) { 236 return IO_ERROR; 237 } 238 return ret; 239 } 240 241 ssize_t net_sys_writev(Sysstream *st, struct iovec *iovec, int iovcnt) { 242 // TODO 243 } 244 245 ssize_t net_sys_read(Sysstream *st, void *buf, size_t nbytes) { 246 int ret = recv(st->fd, buf, nbytes, 0); 247 if(ret == SOCKET_ERROR) { 248 return IO_ERROR; 249 } 250 return ret; 251 } 252 253 ssize_t net_sys_sendfile(Sysstream *st, sendfiledata *sfd) { 254 // TODO 255 } 256 257 void net_sys_close(Sysstream *st) { 258 closesocket(st->fd); 259 } 260 261 #endif 262 263 264 /* 265 * HttpStream implementation 266 */ 267 268 IOStream* httpstream_new(pool_handle_t *pool, IOStream *fd) { 269 HttpStream *st = pool_malloc(pool, sizeof(HttpStream)); 270 st->st = http_io_funcs; 271 st->fd = fd; 272 st->written = 0; 273 st->max_read = 0; 274 st->read = 0; 275 st->read_total = 0; 276 st->readbuf = NULL; 277 st->bufsize = 0; 278 st->buflen = NULL; 279 st->bufpos = NULL; 280 st->chunk_buf_pos = 0; 281 st->current_chunk_length = 0; 282 st->current_chunk_pos = 0; 283 st->current_trailer = 0; 284 st->write_chunk_buf_len = 0; 285 st->write_chunk_buf_pos = 0; 286 st->chunked_enc = WS_FALSE; 287 st->read_eof = WS_TRUE; 288 st->write_eof = WS_FALSE; 289 return (IOStream*)st; 290 } 291 292 int httpstream_enable_chunked_read(IOStream *st, char *buffer, size_t bufsize, int *cursize, int *pos) { 293 if(st->read != (io_read_f)net_http_read) { 294 log_ereport(LOG_FAILURE, "%s", "httpstream_enable_chunked_read: IOStream is not an HttpStream"); 295 return 1; 296 } 297 st->read = (io_read_f)net_http_read_chunked; 298 HttpStream *http = (HttpStream*)st; 299 http->max_read = 0; 300 http->read = 0; 301 http->readbuf = buffer; 302 http->bufsize = bufsize; 303 http->buflen = cursize; 304 http->bufpos = pos; 305 http->chunk_buf_pos = 0; 306 http->read_eof = WS_FALSE; 307 return 0; 308 } 309 310 int httpstream_enable_chunked_write(IOStream *st) { 311 if(st->type != IO_STREAM_TYPE_HTTP) { 312 log_ereport(LOG_FAILURE, "%s", "httpstream_enable_chunked_write: IOStream is not an HttpStream"); 313 return 1; 314 } 315 HttpStream *http = (HttpStream*)st; 316 http->chunked_enc = WS_TRUE; 317 return 0; 318 } 319 320 int httpstream_set_max_read(IOStream *st, int64_t maxread) { 321 if(st->write != (io_write_f)net_http_write) { 322 log_ereport(LOG_FAILURE, "%s", "httpstream_set_max_read: IOStream is not an HttpStream"); 323 return 1; 324 } 325 HttpStream *http = (HttpStream*)st; 326 http->max_read = maxread; 327 return 0; 328 } 329 330 WSBool httpstream_eof(IOStream *st) { 331 HttpStream *http = (HttpStream*)st; 332 return http->read_eof; 333 } 334 335 int64_t httpstream_written(IOStream *st) { 336 HttpStream *http = (HttpStream*)st; 337 return http->written; 338 } 339 340 /* 341 * iovec callback func 342 * returns number of payload bytes written (number of bytes returned back to the net_write caller) 343 */ 344 typedef ssize_t(*writeop_finish_func)(HttpStream *st, char *base, size_t len, size_t written, void *udata); 345 346 static ssize_t httpstream_finish_prev_header(HttpStream *st, char *base, size_t len, size_t written, void *udata) { 347 st->write_chunk_buf_pos += written; 348 if(st->write_chunk_buf_pos == st->write_chunk_buf_len) { 349 st->write_chunk_buf_len = 0; 350 st->write_chunk_buf_pos = 0; 351 } 352 return 0; 353 } 354 355 static ssize_t httpstream_finish_data(HttpStream *st, char *base, size_t len, size_t written, void *udata) { 356 st->current_chunk_pos += written; 357 if(st->current_chunk_pos == st->current_chunk_length) { 358 st->current_chunk_length = 0; 359 st->current_chunk_pos = 0; 360 st->current_trailer = 2; 361 } 362 return written; 363 } 364 365 static ssize_t httpstream_finish_new_header(HttpStream *st, char *base, size_t len, size_t written, void *udata) { 366 size_t *chunk_len = udata; 367 st->current_chunk_length = *chunk_len; 368 st->current_chunk_pos = 0; // new chunk started 369 if(written < len) { 370 st->write_chunk_buf_len = len-written; 371 st->write_chunk_buf_pos = 0; 372 memcpy(st->write_chunk_buf + st->write_chunk_buf_pos, base+written, st->write_chunk_buf_len); 373 } else { 374 st->write_chunk_buf_len = 0; 375 st->write_chunk_buf_pos = 0; 376 } 377 return 0; 378 } 379 380 static ssize_t httpstream_finish_trailer(HttpStream *st, char *base, size_t len, size_t written, void *udata) { 381 st->current_trailer -= written; 382 return 0; 383 } 384 385 ssize_t net_http_write(HttpStream *st, const void *buf, size_t nbytes) { 386 st->st.io_errno = 0; 387 if(st->write_eof) return 0; 388 IOStream *fd = st->fd; 389 if(!st->chunked_enc) { 390 ssize_t w = fd->write(fd, buf, nbytes); 391 st->written += w > 0 ? w : 0; 392 return w; 393 } else { 394 struct iovec io[8]; 395 writeop_finish_func io_finished[8]; 396 void *io_finished_udata[8]; 397 int iovec_len = 0; 398 399 char *str_crlf = "\r\n"; 400 401 size_t prev_chunk_len = st->current_chunk_length; 402 size_t new_chunk_len = 0; 403 404 // was the previous chunk header completely sent? 405 if(st->write_chunk_buf_len > 0) { 406 io[0].iov_base = &st->write_chunk_buf[st->write_chunk_buf_pos]; 407 io[0].iov_len = st->write_chunk_buf_len - st->write_chunk_buf_pos; 408 io_finished[0] = httpstream_finish_prev_header; 409 io_finished_udata[0] = &prev_chunk_len; 410 iovec_len++; 411 } 412 413 // was the previous chunk payload completely sent? 414 if(st->current_chunk_length != 0) { 415 size_t chunk_remaining = st->current_chunk_length - st->current_chunk_pos; 416 size_t prev_nbytes = chunk_remaining > nbytes ? nbytes : chunk_remaining; 417 io[iovec_len].iov_base = (char*)buf; 418 io[iovec_len].iov_len = prev_nbytes; 419 io_finished[iovec_len] = httpstream_finish_data; 420 buf = ((char*)buf) + prev_nbytes; 421 nbytes -= prev_nbytes; 422 iovec_len++; 423 424 io[iovec_len].iov_base = str_crlf; 425 io[iovec_len].iov_len = 2; 426 io_finished[iovec_len] = httpstream_finish_trailer; 427 iovec_len++; 428 } else if(st->current_trailer > 0) { 429 io[iovec_len].iov_base = str_crlf + 2 - st->current_trailer; 430 io[iovec_len].iov_len = st->current_trailer; 431 io_finished[iovec_len] = httpstream_finish_trailer; 432 iovec_len++; 433 } 434 435 // TODO: on some plattforms iov_len is smaller than size_t 436 // if nbytes > INT_MAX, it should be devided into multiple 437 // iovec entries 438 char chunk_len[16]; 439 if(nbytes > 0) { 440 new_chunk_len = nbytes; 441 io[iovec_len].iov_base = chunk_len; 442 io[iovec_len].iov_len = snprintf(chunk_len, 16, "%zx\r\n", nbytes); 443 io_finished[iovec_len] = httpstream_finish_new_header; 444 io_finished_udata[iovec_len] = &new_chunk_len; 445 iovec_len++; 446 447 io[iovec_len].iov_base = (char*)buf; 448 io[iovec_len].iov_len = nbytes; 449 io_finished[iovec_len] = httpstream_finish_data; 450 iovec_len++; 451 452 io[iovec_len].iov_base = str_crlf; 453 io[iovec_len].iov_len = 2; 454 io_finished[iovec_len] = httpstream_finish_trailer; 455 iovec_len++; 456 } 457 458 ssize_t wv = fd->writev(fd, io, iovec_len); 459 if(wv <= 0) { 460 st->st.io_errno = net_errno(st->fd); 461 return wv; 462 } 463 464 ssize_t ret_w = 0; 465 int i = 0; 466 while(wv > 0) { 467 char *base = io[i].iov_base; 468 size_t len = io[i].iov_len; 469 size_t wlen = wv > len ? len : wv; 470 ret_w += io_finished[i](st, base, len, wlen, io_finished_udata[i]); 471 wv -= wlen; 472 i++; 473 } 474 475 st->written += ret_w; 476 if(ret_w == 0) { 477 st->st.io_errno = EWOULDBLOCK; // not sure if this is really correct 478 //ret_w = -1; 479 } 480 return ret_w; 481 } 482 } 483 484 ssize_t net_http_writev(HttpStream *st, struct iovec *iovec, int iovcnt) { 485 if(st->write_eof) return 0; 486 IOStream *fd = st->fd; 487 if(st->chunked_enc) { 488 struct iovec *io = calloc(iovcnt + 1, sizeof(struct iovec)); 489 if(!io) { 490 return 0; 491 } 492 char chunk_len[16]; 493 io[0].iov_base = chunk_len; 494 size_t len = 0; 495 for(int i=0;i<iovcnt;i++) { 496 len += iovec[i].iov_len; 497 } 498 io[0].iov_len = snprintf(chunk_len, 16, "\r\n%zx\r\n", len); 499 memcpy(io + 1, iovec, iovcnt * sizeof(struct iovec)); 500 ssize_t r = fd->writev(fd, io, iovcnt + 1); 501 502 ssize_t ret = r - io[0].iov_len; 503 free(io); 504 st->written += ret; 505 return ret; 506 } else { 507 ssize_t w = fd->writev(fd, iovec, iovcnt); 508 st->written += w; 509 return w; 510 } 511 } 512 513 ssize_t net_http_read(HttpStream *st, void *buf, size_t nbytes) { 514 if(st->read >= st->max_read) { 515 st->read_eof = WS_TRUE; 516 return 0; 517 } 518 ssize_t r = st->fd->read(st->fd, buf, nbytes); 519 if(r < 0) { 520 st->st.io_errno = st->fd->io_errno; 521 } 522 st->read += r; 523 return r; 524 } 525 526 #define BUF_UNNEEDED_DIFF 64 527 /* 528 * read from st->chunk_buf first, read from st->fd if perform_io is true 529 */ 530 static ssize_t net_http_read_buffered(HttpStream *st, char *buf, size_t nbytes, WSBool read_data, WSBool *perform_io) { 531 ssize_t r = 0; 532 533 //memset(buf, 'x', nbytes); 534 //char *orig_buf = buf; 535 536 // copy available data from st->readbuf to buf 537 int pos = *st->bufpos; 538 size_t buf_available = *st->buflen - pos; 539 if(buf_available) { 540 size_t cplen = buf_available > nbytes ? nbytes : buf_available; 541 if(read_data) { 542 // if we read data (and not a chunk header), we limit the 543 // amount of bytes we copy 544 size_t chunk_available = st->max_read - st->read; 545 cplen = cplen > chunk_available ? chunk_available : cplen; 546 st->read += cplen; 547 } 548 memcpy(buf, st->readbuf + pos, cplen); 549 *st->bufpos += cplen; 550 r += cplen; 551 buf += cplen; 552 nbytes -= cplen; 553 } 554 555 // maybe perform IO and refill the read buffer 556 // if we read data (read_data == true), make sure not to perform IO, 557 // when a chunk is completed 558 // 559 // if we read a chunk header (read_data == false) it is very important 560 // to not perform IO, if we have previously copied data from readbuf 561 // this ensures we never override non-chunk-header data 562 if(*perform_io && ((read_data && nbytes > 0 && st->max_read - st->read) || (!read_data && r == 0))) { 563 if(*st->buflen - *st->bufpos > 0) { 564 printf("todo: fix, should not happen, remove later\n"); 565 } 566 // fill buffer again 567 ssize_t rlen = st->fd->read(st->fd, st->readbuf, st->bufsize); 568 *st->buflen = rlen; 569 *st->bufpos = 0; 570 *perform_io = WS_FALSE; 571 if(rlen < 0) { 572 st->st.io_errno = st->fd->io_errno; 573 } 574 575 if(rlen > 0) { 576 // call func again to get data from buffer (no IO will be performed) 577 r += net_http_read_buffered(st, buf, nbytes, read_data, perform_io); 578 } 579 } 580 581 return r; 582 } 583 584 585 /* 586 * parses a chunk header 587 * the chunk length is stored in chunklen 588 * return: 0 if the data is incomplete 589 * -1 if an error occured 590 * >0 chunk header length 591 */ 592 int http_stream_parse_chunk_header(char *str, int len, WSBool first, int64_t *chunklen) { 593 char *hdr_start = NULL; 594 char *hdr_end = NULL; 595 int i = 0; 596 if(first) { 597 hdr_start = str; 598 } else { 599 if(len < 3) { 600 return 0; 601 } 602 if(str[0] == '\r' && str[1] == '\n') { 603 hdr_start = str+2; 604 i = 2; 605 } else if(str[0] == '\n') { 606 hdr_start = str+1; 607 i = 1; 608 } else { 609 return -1; 610 } 611 } 612 613 for(;i<len;i++) { 614 char c = str[i]; 615 if(c == '\r' || c == '\n') { 616 hdr_end = str+i; 617 break; 618 } 619 } 620 if(!hdr_end || i == len) { 621 return 0; // incomplete 622 } 623 624 if(*hdr_end == '\r') { 625 // we also need '\n' 626 if(hdr_end[1] != '\n') { 627 return -1; 628 } 629 i++; // '\n' found 630 } 631 632 // check if the first character is a number 633 char f = hdr_start[0]; 634 if(!(isdigit(f) || (f >= 'A' && f <= 'F') || (f >= 'a' && f <= 'f'))) { 635 return -1; 636 } 637 638 // parse 639 char save_c = *hdr_end; 640 *hdr_end = '\0'; 641 char *end; 642 int64_t clen; 643 errno = 0; 644 clen = strtoll(hdr_start, &end, 16); 645 *hdr_end = save_c; 646 if(errno == 0 && end != hdr_end) { 647 return -1; 648 } 649 i++; 650 651 if(clen == 0) { 652 // chunk length of 0 indicates the end 653 // an additional \r\n is required (we also accept \n) 654 if(i >= len) { 655 return 0; 656 } 657 if(str[i] == '\n') { 658 i++; 659 } else if(str[i] == '\r') { 660 if(++i >= len) { 661 return 0; 662 } 663 if(str[i] == '\n') { 664 i++; 665 } else { 666 return -1; 667 } 668 } else { 669 return -1; 670 } 671 } 672 673 *chunklen = clen; 674 return i; 675 } 676 677 ssize_t net_http_read_chunked(HttpStream *st, void *buf, size_t nbytes) { 678 if(st->read_eof) { 679 return 0; 680 } 681 682 char *rbuf = buf; // buffer pos 683 size_t rd = 0; // number of bytes read 684 size_t rbuflen = nbytes; // number of bytes until end of buf 685 WSBool perform_io = WS_TRUE; // we do only 1 read before we abort 686 while(rd < nbytes && (perform_io || (st->max_read - st->read) > 0)) { 687 // how many bytes are available in the current chunk 688 size_t chunk_available = st->max_read - st->read; 689 if(chunk_available > 0) { 690 ssize_t r = net_http_read_buffered(st, rbuf, rbuflen, TRUE, &perform_io); 691 if(r == 0) { 692 break; 693 } 694 rd += r; 695 st->read_total += r; 696 rbuf += r; 697 rbuflen -= r; 698 } else { 699 int chunkbuf_avail = HTTP_STREAM_CBUF_SIZE - st->chunk_buf_pos; 700 if(chunkbuf_avail == 0) { 701 // for some reason HTTP_STREAM_CBUF_SIZE is not enough 702 // to store the chunk header 703 // this indicates that something has gone wrong (or this is an attack) 704 st->read_eof = WS_TRUE; 705 return -1; 706 } 707 // fill st->chunk_buf 708 ssize_t r = net_http_read_buffered(st, &st->chunk_buf[st->chunk_buf_pos], chunkbuf_avail, FALSE, &perform_io); 709 if(r == 0) { 710 break; 711 } 712 int chunkbuf_len = st->chunk_buf_pos + r; 713 int64_t chunklen; 714 int ret = http_stream_parse_chunk_header(st->chunk_buf, chunkbuf_len, st->read_total > 0 ? FALSE : TRUE, &chunklen); 715 if(ret == 0) { 716 // incomplete chunk header 717 st->chunk_buf_pos = chunkbuf_len; 718 } else if(ret < 0) { 719 // error 720 st->read_eof = WS_TRUE; 721 return -1; 722 } else if(ret > 0) { 723 st->max_read = chunklen; 724 st->read = 0; 725 int remaining_len = chunkbuf_len - ret; 726 if(remaining_len > 0) { 727 // we have read more into chunk_buf than the chunk_header 728 // it is save to just move bufpos back 729 *st->bufpos -= remaining_len; 730 } 731 //st->remaining_len = chunkbuf_len - ret; 732 st->chunk_buf_pos = 0; 733 734 if(chunklen == 0) { 735 st->read_eof = WS_TRUE; 736 break; 737 } 738 } 739 } 740 741 if(!perform_io && rd == 0) { 742 perform_io = WS_TRUE; 743 } 744 } 745 746 return rd; 747 } 748 749 ssize_t net_http_sendfile(HttpStream *st, sendfiledata *sfd) { 750 if(st->write_eof) return 0; 751 ssize_t ret = 0; 752 // TODO: support chunked transfer encoding 753 if(st->fd->sendfile) { 754 ret = st->fd->sendfile(st->fd, sfd); 755 } else { 756 ret = net_fallback_sendfile((IOStream*)st, sfd); 757 } 758 759 st->written += ret > 0 ? ret : 0; 760 761 return ret; 762 } 763 764 void net_http_close(HttpStream *st) { 765 st->fd->close(st->fd); 766 } 767 768 void net_http_finish(HttpStream *st) { 769 if(st->chunked_enc && !st->write_eof) { 770 st->fd->write(st->fd, "0\r\n\r\n", 5); 771 } 772 st->write_eof = WS_TRUE; 773 } 774 775 void net_http_setmode(HttpStream *st, int mode) { 776 st->fd->setmode(st->fd, mode); 777 } 778 779 int net_http_poll(HttpStream *st, EventHandler *ev, int events, Event *cb) { 780 return st->fd->poll(st->fd, ev, events, cb); 781 } 782 783 784 /* 785 * SSLStream implementation 786 */ 787 788 IOStream* sslstream_new(pool_handle_t *pool, SSL *ssl) { 789 SSLStream *st = pool_malloc(pool, sizeof(SSLStream)); 790 st->st = ssl_io_funcs; 791 st->ssl = ssl; 792 st->error = 0; 793 return (IOStream*)st; 794 } 795 796 ssize_t net_ssl_write(SSLStream *st, const void *buf, size_t nbytes) { 797 int ret = SSL_write(st->ssl, buf, nbytes); 798 if(ret <= 0) { 799 st->error = SSL_get_error(st->ssl, ret); 800 if(st->error == SSL_ERROR_WANT_WRITE || st->error == SSL_ERROR_WANT_READ) { 801 st->st.io_errno = EWOULDBLOCK; 802 } else { 803 st->st.io_errno = -1; 804 } 805 ret = -1; 806 } 807 return ret; 808 } 809 810 ssize_t net_ssl_writev(SSLStream *st, struct iovec *iovec, int iovcnt) { 811 ssize_t r = 0; 812 for(int i=0;i<iovcnt;i++) { 813 int ret = SSL_write(st->ssl, iovec[i].iov_base, iovec[i].iov_len); 814 if(ret <= 0) { 815 if(r == 0) { 816 st->error = SSL_get_error(st->ssl, ret); 817 if(st->error == SSL_ERROR_WANT_WRITE || st->error == SSL_ERROR_WANT_READ) { 818 st->st.io_errno = EWOULDBLOCK; 819 } else { 820 st->st.io_errno = -1; 821 } 822 } 823 break; 824 } 825 r += ret; 826 if(ret < iovec[i].iov_len) { 827 break; 828 } 829 } 830 return r == 0 ? -1 : r; 831 } 832 833 ssize_t net_ssl_read(SSLStream *st, void *buf, size_t nbytes) { 834 int ret = SSL_read(st->ssl, buf, nbytes); 835 if(ret <= 0) { 836 st->error = SSL_get_error(st->ssl, ret); 837 } 838 return ret; 839 } 840 841 void net_ssl_close(SSLStream *st) { 842 int ret = SSL_shutdown(st->ssl); 843 if(ret != 1) { 844 st->error = SSL_get_error(st->ssl, ret); 845 } 846 system_close(SSL_get_fd(st->ssl)); 847 } 848 849 void net_ssl_finish(SSLStream *st) { 850 851 } 852 853 void net_ssl_setmode(SSLStream *st, int mode) { 854 int flags; 855 if (-1 == (flags = fcntl(SSL_get_fd(st->ssl), F_GETFL, 0))) { 856 flags = 0; 857 } 858 if(mode == IO_MODE_BLOCKING) { 859 if (fcntl(SSL_get_fd(st->ssl), F_SETFL, flags & ~O_NONBLOCK) != 0) { 860 perror("fcntl"); 861 // TODO: error 862 } 863 } else if(mode == IO_MODE_NONBLOCKING) { 864 if (fcntl(SSL_get_fd(st->ssl), F_SETFL, flags | O_NONBLOCK) != 0) { 865 perror("fcntl"); 866 // TODO: error 867 } 868 } 869 } 870 871 int net_ssl_poll(SSLStream *st, EventHandler *ev, int events, Event *cb) { 872 int fd = SSL_get_fd(st->ssl); 873 switch(events) { 874 default: return -1; 875 case IO_POLL_NONE: return ev_remove_poll(ev, fd); 876 case IO_POLL_IN: return ev_pollin(ev, fd, cb); 877 case IO_POLL_OUT: return ev_pollout(ev, fd, cb); 878 case IO_POLL_IN | IO_POLL_OUT: return -1; // TODO: implement 879 } 880 } 881 882 /* -------------------- public nsapi network functions -------------------- */ 883 884 ssize_t net_read(SYS_NETFD fd, void *buf, size_t nbytes) { 885 ssize_t r = ((IOStream*)fd)->read(fd, buf, nbytes); 886 if(r == 0) { 887 return IO_EOF; 888 } else if(r < 0) { 889 ((IOStream*)fd)->io_errno = errno; 890 return IO_ERROR; 891 } 892 return r; 893 } 894 895 ssize_t net_write(SYS_NETFD fd, const void *buf, size_t nbytes) { 896 size_t w = 0; 897 size_t remaining = nbytes; 898 const char *cbuf = buf; 899 ssize_t r = 0; 900 int attempts = 0; 901 while(w < nbytes && attempts < net_write_max_attempts) { 902 r = ((IOStream*)fd)->write(fd, cbuf, remaining); 903 if(r <= 0) { 904 break; 905 } 906 w += r; 907 cbuf += r; 908 remaining -= r; 909 attempts++; 910 } 911 if(r < 0 && w == 0) { 912 return IO_ERROR; 913 } 914 return w; 915 } 916 917 ssize_t net_writev(SYS_NETFD fd, struct iovec *iovec, int iovcnt) { 918 ssize_t r = ((IOStream*)fd)->writev(fd, iovec, iovcnt); 919 if(r < 0) { 920 ((IOStream*)fd)->io_errno = errno; 921 return IO_ERROR; 922 } 923 return r; 924 } 925 926 ssize_t net_printf(SYS_NETFD fd, char *format, ...) { 927 va_list arg; 928 va_start(arg, format); 929 cxmutstr buf = cx_vasprintf_a(cxDefaultAllocator, format, arg); 930 ssize_t r = buf.length > 0 ? net_write(fd, buf.ptr, buf.length) : 0; 931 free(buf.ptr); 932 va_end(arg); 933 if(r < 0) { 934 ((IOStream*)fd)->io_errno = errno; 935 } 936 return r; 937 } 938 939 ssize_t net_sendfile(SYS_NETFD fd, sendfiledata *sfd) { 940 IOStream *out = fd; 941 if(out->sendfile && sfd->fd && sfd->fd->fd != -1) { 942 ssize_t r = out->sendfile(fd, sfd); 943 if(r < 0) { 944 out->io_errno = errno; 945 return IO_ERROR; 946 } 947 return r; 948 } else { 949 // stream/file does not support sendfile 950 // do regular copy 951 return net_fallback_sendfile(out, sfd); 952 } 953 } 954 955 // private 956 ssize_t net_fallback_sendfile(IOStream *fd, sendfiledata *sfd) { 957 char *buf = malloc(4096); 958 if(!buf) { 959 // TODO: out of memory error 960 return IO_ERROR; 961 } 962 char *header = (char*)sfd->header; 963 int hlen = sfd->hlen; 964 char *trailer = (char*)sfd->trailer; 965 int tlen = sfd->tlen; 966 if(header == NULL) { 967 hlen = 0; 968 } 969 if(trailer == NULL) { 970 tlen = 0; 971 } 972 973 ssize_t r; 974 while(hlen > 0) { 975 r = fd->write(fd, header, hlen); 976 header += r; 977 hlen -= r; 978 if(r <= 0) { 979 free(buf); 980 fd->io_errno = errno; 981 return IO_ERROR; 982 } 983 } 984 985 if(system_lseek(sfd->fd, sfd->offset, SEEK_SET) == -1) { 986 free(buf); 987 fd->io_errno = errno; 988 return IO_ERROR; 989 } 990 991 size_t length = sfd->len; 992 while(length > 0) { 993 // TODO: remove 994 if(length > sfd->len) { 995 log_ereport(LOG_WARN, "net_fallback_sendfile: length > sfd->len: %zu > %zu", length, sfd->len); 996 free(buf); 997 return IO_ERROR; 998 } 999 1000 if((r = system_fread(sfd->fd, buf, 4096)) <= 0) { 1001 break; 1002 } 1003 char *write_buf = buf; 1004 while(r > 0) { 1005 ssize_t w = fd->write(fd, write_buf, r); 1006 // TODO: remove 1007 if(w > r) { 1008 log_ereport(LOG_WARN, "net_fallback_sendfile: w > r, %zd > %zd", w, r); 1009 w = 0; 1010 } 1011 1012 if(w <= 0) { 1013 free(buf); 1014 fd->io_errno = errno; 1015 return IO_ERROR; 1016 } 1017 r -= w; 1018 length -= w; 1019 write_buf += w; 1020 } 1021 } 1022 free(buf); 1023 if(length > 0) { 1024 fd->io_errno = errno; 1025 return IO_ERROR; 1026 } 1027 1028 while(tlen > 0) { 1029 r = fd->write(fd, trailer, tlen); 1030 trailer += r; 1031 tlen -= r; 1032 if(r <= 0) { 1033 fd->io_errno = errno; 1034 return IO_ERROR; 1035 } 1036 } 1037 1038 return sfd->hlen + sfd->len + sfd->tlen; 1039 } 1040 1041 int net_flush(SYS_NETFD sd) { 1042 // TODO: implement 1043 return 0; 1044 } 1045 1046 void net_close(SYS_NETFD fd) { 1047 ((IOStream*)fd)->close(fd); 1048 } 1049 1050 int net_setnonblock(SYS_NETFD fd, int nonblock) { 1051 ((IOStream*)fd)->setmode( 1052 fd, 1053 nonblock ? IO_MODE_NONBLOCKING : IO_MODE_BLOCKING); 1054 return 0; 1055 } 1056 1057 int net_errno(SYS_NETFD fd) { 1058 return ((IOStream*)fd)->io_errno; 1059 } 1060 1061 // private 1062 void net_finish(SYS_NETFD fd) { 1063 ((IOStream*)fd)->finish(fd); 1064 } 1065