git.sur5r.net Git - bacula/bacula/blob - bacula/src/lib/bsock.c

   1 /*
   2    Bacula® - The Network Backup Solution
   3
   4    Copyright (C) 2007-2010 Free Software Foundation Europe e.V.
   5
   6    The main author of Bacula is Kern Sibbald, with contributions from
   7    many others, a complete list can be found in the file AUTHORS.
   8    This program is Free Software; you can redistribute it and/or
   9    modify it under the terms of version three of the GNU Affero General Public
  10    License as published by the Free Software Foundation and included
  11    in the file LICENSE.
  12
  13    This program is distributed in the hope that it will be useful, but
  14    WITHOUT ANY WARRANTY; without even the implied warranty of
  15    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  16    General Public License for more details.
  17
  18    You should have received a copy of the GNU Affero General Public License
  19    along with this program; if not, write to the Free Software
  20    Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
  21    02110-1301, USA.
  22
  23    Bacula® is a registered trademark of Kern Sibbald.
  24    The licensor of Bacula is the Free Software Foundation Europe
  25    (FSFE), Fiduciary Program, Sumatrastrasse 25, 8006 Zürich,
  26    Switzerland, email:ftf@fsfeurope.org.
  27 */
  28 /*
  29  * Network Utility Routines
  30  *
  31  *  by Kern Sibbald
  32  *
  33  */
  34
  35
  36 #include "bacula.h"
  37 #include "jcr.h"
  38 #include <netdb.h>
  39
  40 #ifndef ENODATA                    /* not defined on BSD systems */
  41 #define ENODATA EPIPE
  42 #endif
  43
  44 #ifdef HAVE_WIN32
  45 #define socketRead(fd, buf, len)  ::recv(fd, buf, len, 0)
  46 #define socketWrite(fd, buf, len) ::send(fd, buf, len, 0)
  47 #define socketClose(fd)           ::closesocket(fd)
  48 #else
  49 #define socketRead(fd, buf, len)  ::read(fd, buf, len)
  50 #define socketWrite(fd, buf, len) ::write(fd, buf, len)
  51 #define socketClose(fd)           ::close(fd)
  52 #endif
  53
  54 /*
  55  * This is a non-class BSOCK "constructor"  because we want to
  56  *   call the Bacula smartalloc routines instead of new.
  57  */
  58 BSOCK *new_bsock()
  59 {
  60    BSOCK *bsock = (BSOCK *)malloc(sizeof(BSOCK));
  61    bsock->init();
  62    return bsock;
  63 }
  64
  65 void BSOCK::init()
  66 {
  67    memset(this, 0, sizeof(BSOCK));
  68    m_blocking = 1;
  69    msg = get_pool_memory(PM_BSOCK);
  70    errmsg = get_pool_memory(PM_MESSAGE);
  71    /*
  72     * ****FIXME**** reduce this to a few hours once
  73     *   heartbeats are implemented
  74     */
  75    timeout = 60 * 60 * 6 * 24;   /* 6 days timeout */
  76 }
  77
  78 /*
  79  * This is our "class destructor" that ensures that we use
  80  *   smartalloc rather than the system free().
  81  */
  82 void BSOCK::free_bsock()
  83 {
  84    destroy();
  85 }
  86
  87 void BSOCK::free_tls()
  88 {
  89    free_tls_connection(this->tls);
  90    this->tls = NULL;
  91 }
  92
  93 /*
  94  * Try to connect to host for max_retry_time at retry_time intervals.
  95  *   Note, you must have called the constructor prior to calling
  96  *   this routine.
  97  */
  98 bool BSOCK::connect(JCR * jcr, int retry_interval, utime_t max_retry_time,
  99                     utime_t heart_beat,
 100                     const char *name, char *host, char *service, int port,
 101                     int verbose)
 102 {
 103    bool ok = false;
 104    int i;
 105    int fatal = 0;
 106    time_t begin_time = time(NULL);
 107    time_t now;
 108    btimer_t *tid = NULL;
 109
 110    /* Try to trap out of OS call when time expires */
 111    if (max_retry_time) {
 112       tid = start_thread_timer(jcr, pthread_self(), (uint32_t)max_retry_time);
 113    }
 114
 115    for (i = 0; !open(jcr, name, host, service, port, heart_beat, &fatal);
 116         i -= retry_interval) {
 117       berrno be;
 118       if (fatal || (jcr && job_canceled(jcr))) {
 119          goto bail_out;
 120       }
 121       Dmsg4(100, "Unable to connect to %s on %s:%d. ERR=%s\n",
 122             name, host, port, be.bstrerror());
 123       if (i < 0) {
 124          i = 60 * 5;               /* complain again in 5 minutes */
 125          if (verbose)
 126             Qmsg4(jcr, M_WARNING, 0, _(
 127                "Could not connect to %s on %s:%d. ERR=%s\n"
 128                "Retrying ...\n"), name, host, port, be.bstrerror());
 129       }
 130       bmicrosleep(retry_interval, 0);
 131       now = time(NULL);
 132       if (begin_time + max_retry_time <= now) {
 133          Qmsg4(jcr, M_FATAL, 0, _("Unable to connect to %s on %s:%d. ERR=%s\n"),
 134                name, host, port, be.bstrerror());
 135          goto bail_out;
 136       }
 137    }
 138    ok = true;
 139
 140 bail_out:
 141    if (tid) {
 142       stop_thread_timer(tid);
 143    }
 144    return ok;
 145 }
 146
 147
 148 /*
 149  * Finish initialization of the pocket structure.
 150  */
 151 void BSOCK::fin_init(JCR * jcr, int sockfd, const char *who, const char *host, int port,
 152             struct sockaddr *lclient_addr)
 153 {
 154    Dmsg3(100, "who=%s host=%s port=%d\n", who, host, port);
 155    m_fd = sockfd;
 156    set_who(bstrdup(who));
 157    set_host(bstrdup(host));
 158    set_port(port);
 159    memcpy(&client_addr, lclient_addr, sizeof(client_addr));
 160    set_jcr(jcr);
 161 }
 162
 163 /*
 164  * Copy the address from the configuration dlist that gets passed in
 165  */
 166 void BSOCK::set_source_address(dlist *src_addr_list)
 167 {
 168    IPADDR *addr = NULL;
 169
 170    // delete the object we already have, if it's allocated
 171    if (src_addr) {
 172      free( src_addr);
 173      src_addr = NULL;
 174    }
 175
 176    if (src_addr_list) {
 177      addr = (IPADDR*) src_addr_list->first();
 178      src_addr = New( IPADDR(*addr));
 179    }
 180 }
 181
 182
 183 /*
 184  * Open a TCP connection to the server
 185  * Returns NULL
 186  * Returns BSOCK * pointer on success
 187  *
 188  */
 189 bool BSOCK::open(JCR *jcr, const char *name, char *host, char *service,
 190             int port, utime_t heart_beat, int *fatal)
 191 {
 192    int sockfd = -1;
 193    dlist *addr_list;
 194    IPADDR *ipaddr;
 195    bool connected = false;
 196    int turnon = 1;
 197    const char *errstr;
 198    int save_errno = 0;
 199
 200    /*
 201     * Fill in the structure serv_addr with the address of
 202     * the server that we want to connect with.
 203     */
 204    if ((addr_list = bnet_host2ipaddrs(host, 0, &errstr)) == NULL) {
 205       /* Note errstr is not malloc'ed */
 206       Qmsg2(jcr, M_ERROR, 0, _("gethostbyname() for host \"%s\" failed: ERR=%s\n"),
 207             host, errstr);
 208       Dmsg2(100, "bnet_host2ipaddrs() for host %s failed: ERR=%s\n",
 209             host, errstr);
 210       *fatal = 1;
 211       return false;
 212    }
 213
 214    foreach_dlist(ipaddr, addr_list) {
 215       ipaddr->set_port_net(htons(port));
 216       char allbuf[256 * 10];
 217       char curbuf[256];
 218       Dmsg2(100, "Current %sAll %s\n",
 219                    ipaddr->build_address_str(curbuf, sizeof(curbuf)),
 220                    build_addresses_str(addr_list, allbuf, sizeof(allbuf)));
 221       /* Open a TCP socket */
 222       if ((sockfd = socket(ipaddr->get_family(), SOCK_STREAM, 0)) < 0) {
 223          berrno be;
 224          save_errno = errno;
 225          *fatal = 1;
 226          Pmsg3(000, _("Socket open error. proto=%d port=%d. ERR=%s\n"),
 227             ipaddr->get_family(), ipaddr->get_port_host_order(), be.bstrerror());
 228          continue;
 229       }
 230
 231       /* Bind to the source address if it is set */
 232       if (src_addr) {
 233          if (bind(sockfd, src_addr->get_sockaddr(), src_addr->get_sockaddr_len()) < 0) {
 234             berrno be;
 235             save_errno = errno;
 236             *fatal = 1;
 237             Pmsg2(000, _("Source address bind error. proto=%d. ERR=%s\n"),
 238                   src_addr->get_family(), be.bstrerror() );
 239             continue;
 240          }
 241       }
 242
 243       /*
 244        * Keep socket from timing out from inactivity
 245        */
 246       if (setsockopt(sockfd, SOL_SOCKET, SO_KEEPALIVE, (sockopt_val_t)&turnon, sizeof(turnon)) < 0) {
 247          berrno be;
 248          Qmsg1(jcr, M_WARNING, 0, _("Cannot set SO_KEEPALIVE on socket: %s\n"),
 249                be.bstrerror());
 250       }
 251 #if defined(TCP_KEEPIDLE)
 252       if (heart_beat) {
 253          int opt = heart_beat
 254          if (setsockopt(sockfd, IPPROTO_IP, TCP_KEEPIDLE, (sockopt_val_t)&opt, sizeof(opt)) < 0) {
 255             berrno be;
 256             Qmsg1(jcr, M_WARNING, 0, _("Cannot set SO_KEEPIDLE on socket: %s\n"),
 257                   be.bstrerror());
 258          }
 259       }
 260 #endif
 261
 262       /* connect to server */
 263       if (::connect(sockfd, ipaddr->get_sockaddr(), ipaddr->get_sockaddr_len()) < 0) {
 264          save_errno = errno;
 265          socketClose(sockfd);
 266          continue;
 267       }
 268       *fatal = 0;
 269       connected = true;
 270       break;
 271    }
 272
 273    if (!connected) {
 274       free_addresses(addr_list);
 275       errno = save_errno | b_errno_win32;
 276       return false;
 277    }
 278    /*
 279     * Keep socket from timing out from inactivity
 280     *   Do this a second time out of paranoia
 281     */
 282    if (setsockopt(sockfd, SOL_SOCKET, SO_KEEPALIVE, (sockopt_val_t)&turnon, sizeof(turnon)) < 0) {
 283       berrno be;
 284       Qmsg1(jcr, M_WARNING, 0, _("Cannot set SO_KEEPALIVE on socket: %s\n"),
 285             be.bstrerror());
 286    }
 287    fin_init(jcr, sockfd, name, host, port, ipaddr->get_sockaddr());
 288    free_addresses(addr_list);
 289    return true;
 290 }
 291
 292 /*
 293  * Force read/write to use locking
 294  */
 295 bool BSOCK::set_locking()
 296 {
 297    int stat;
 298    if (m_use_locking) {
 299       return true;                      /* already set */
 300    }
 301    if ((stat = pthread_mutex_init(&m_mutex, NULL)) != 0) {
 302       berrno be;
 303       Qmsg(m_jcr, M_FATAL, 0, _("Could not init bsock mutex. ERR=%s\n"),
 304          be.bstrerror(stat));
 305       return false;
 306    }
 307    m_use_locking = true;
 308    return true;
 309 }
 310
 311 void BSOCK::clear_locking()
 312 {
 313    if (!m_use_locking) {
 314       return;
 315    }
 316    m_use_locking = false;
 317    pthread_mutex_destroy(&m_mutex);
 318    return;
 319 }
 320
 321 /*
 322  * Send a message over the network. The send consists of
 323  * two network packets. The first is sends a 32 bit integer containing
 324  * the length of the data packet which follows.
 325  *
 326  * Returns: false on failure
 327  *          true  on success
 328  */
 329 bool BSOCK::send()
 330 {
 331    int32_t rc;
 332    int32_t pktsiz;
 333    int32_t *hdr;
 334    bool ok = true;
 335
 336    if (errors) {
 337       if (!m_suppress_error_msgs) {
 338          Qmsg4(m_jcr, M_ERROR, 0,  _("Socket has errors=%d on call to %s:%s:%d\n"),
 339              errors, m_who, m_host, m_port);
 340       }
 341       return false;
 342    }
 343    if (is_terminated()) {
 344       if (!m_suppress_error_msgs) {
 345          Qmsg4(m_jcr, M_ERROR, 0,  _("Socket is terminated=%d on call to %s:%s:%d\n"),
 346              is_terminated(), m_who, m_host, m_port);
 347       }
 348       return false;
 349    }
 350    if (msglen > 4000000) {
 351       if (!m_suppress_error_msgs) {
 352          Qmsg4(m_jcr, M_ERROR, 0,
 353             _("Socket has insane msglen=%d on call to %s:%s:%d\n"),
 354              msglen, m_who, m_host, m_port);
 355       }
 356       return false;
 357    }
 358
 359    if (m_use_locking) P(m_mutex);
 360    /* Compute total packet length */
 361    if (msglen <= 0) {
 362       pktsiz = sizeof(pktsiz);               /* signal, no data */
 363    } else {
 364       pktsiz = msglen + sizeof(pktsiz);      /* data */
 365    }
 366    /* Store packet length at head of message -- note, we
 367     *  have reserved an int32_t just before msg, so we can
 368     *  store there
 369     */
 370    hdr = (int32_t *)(msg - (int)sizeof(pktsiz));
 371    *hdr = htonl(msglen);                     /* store signal/length */
 372
 373    out_msg_no++;            /* increment message number */
 374
 375    /* send data packet */
 376    timer_start = watchdog_time;  /* start timer */
 377    clear_timed_out();
 378    /* Full I/O done in one write */
 379    rc = write_nbytes(this, (char *)hdr, pktsiz);
 380    timer_start = 0;         /* clear timer */
 381    if (rc != pktsiz) {
 382       errors++;
 383       if (errno == 0) {
 384          b_errno = EIO;
 385       } else {
 386          b_errno = errno;
 387       }
 388       if (rc < 0) {
 389          if (!m_suppress_error_msgs) {
 390             Qmsg5(m_jcr, M_ERROR, 0,
 391                   _("Write error sending %d bytes to %s:%s:%d: ERR=%s\n"),
 392                   msglen, m_who,
 393                   m_host, m_port, this->bstrerror());
 394          }
 395       } else {
 396          Qmsg5(m_jcr, M_ERROR, 0,
 397                _("Wrote %d bytes to %s:%s:%d, but only %d accepted.\n"),
 398                msglen, m_who, m_host, m_port, rc);
 399       }
 400       ok = false;
 401    }
 402    if (m_use_locking) V(m_mutex);
 403    return ok;
 404 }
 405
 406 /*
 407  * Format and send a message
 408  *  Returns: false on error
 409  *           true  on success
 410  */
 411 bool BSOCK::fsend(const char *fmt, ...)
 412 {
 413    va_list arg_ptr;
 414    int maxlen;
 415
 416    if (errors || is_terminated()) {
 417       return false;
 418    }
 419    /* This probably won't work, but we vsnprintf, then if we
 420     * get a negative length or a length greater than our buffer
 421     * (depending on which library is used), the printf was truncated, so
 422     * get a bigger buffer and try again.
 423     */
 424    for (;;) {
 425       maxlen = sizeof_pool_memory(msg) - 1;
 426       va_start(arg_ptr, fmt);
 427       msglen = bvsnprintf(msg, maxlen, fmt, arg_ptr);
 428       va_end(arg_ptr);
 429       if (msglen > 0 && msglen < (maxlen - 5)) {
 430          break;
 431       }
 432       msg = realloc_pool_memory(msg, maxlen + maxlen / 2);
 433    }
 434    return send();
 435 }
 436
 437 /*
 438  * Receive a message from the other end. Each message consists of
 439  * two packets. The first is a header that contains the size
 440  * of the data that follows in the second packet.
 441  * Returns number of bytes read (may return zero)
 442  * Returns -1 on signal (BNET_SIGNAL)
 443  * Returns -2 on hard end of file (BNET_HARDEOF)
 444  * Returns -3 on error  (BNET_ERROR)
 445  *
 446  *  Unfortunately, it is a bit complicated because we have these
 447  *    four return types:
 448  *    1. Normal data
 449  *    2. Signal including end of data stream
 450  *    3. Hard end of file
 451  *    4. Error
 452  *  Using is_bnet_stop() and is_bnet_error() you can figure this all out.
 453  */
 454 int32_t BSOCK::recv()
 455 {
 456    int32_t nbytes;
 457    int32_t pktsiz;
 458
 459    msg[0] = 0;
 460    msglen = 0;
 461    if (errors || is_terminated()) {
 462       return BNET_HARDEOF;
 463    }
 464
 465    if (m_use_locking) P(m_mutex);
 466    read_seqno++;            /* bump sequence number */
 467    timer_start = watchdog_time;  /* set start wait time */
 468    clear_timed_out();
 469    /* get data size -- in int32_t */
 470    if ((nbytes = read_nbytes(this, (char *)&pktsiz, sizeof(int32_t))) <= 0) {
 471       timer_start = 0;      /* clear timer */
 472       /* probably pipe broken because client died */
 473       if (errno == 0) {
 474          b_errno = ENODATA;
 475       } else {
 476          b_errno = errno;
 477       }
 478       errors++;
 479       nbytes = BNET_HARDEOF;        /* assume hard EOF received */
 480       goto get_out;
 481    }
 482    timer_start = 0;         /* clear timer */
 483    if (nbytes != sizeof(int32_t)) {
 484       errors++;
 485       b_errno = EIO;
 486       Qmsg5(m_jcr, M_ERROR, 0, _("Read expected %d got %d from %s:%s:%d\n"),
 487             sizeof(int32_t), nbytes, m_who, m_host, m_port);
 488       nbytes = BNET_ERROR;
 489       goto get_out;
 490    }
 491
 492    pktsiz = ntohl(pktsiz);         /* decode no. of bytes that follow */
 493
 494    if (pktsiz == 0) {              /* No data transferred */
 495       timer_start = 0;             /* clear timer */
 496       in_msg_no++;
 497       msglen = 0;
 498       nbytes = 0;                  /* zero bytes read */
 499       goto get_out;
 500    }
 501
 502    /* If signal or packet size too big */
 503    if (pktsiz < 0 || pktsiz > 1000000) {
 504       if (pktsiz > 0) {            /* if packet too big */
 505          Qmsg3(m_jcr, M_FATAL, 0,
 506                _("Packet size too big from \"%s:%s:%d. Terminating connection.\n"),
 507                m_who, m_host, m_port);
 508          pktsiz = BNET_TERMINATE;  /* hang up */
 509       }
 510       if (pktsiz == BNET_TERMINATE) {
 511          set_terminated();
 512       }
 513       timer_start = 0;                /* clear timer */
 514       b_errno = ENODATA;
 515       msglen = pktsiz;                /* signal code */
 516       nbytes =  BNET_SIGNAL;          /* signal */
 517       goto get_out;
 518    }
 519
 520    /* Make sure the buffer is big enough + one byte for EOS */
 521    if (pktsiz >= (int32_t) sizeof_pool_memory(msg)) {
 522       msg = realloc_pool_memory(msg, pktsiz + 100);
 523    }
 524
 525    timer_start = watchdog_time;  /* set start wait time */
 526    clear_timed_out();
 527    /* now read the actual data */
 528    if ((nbytes = read_nbytes(this, msg, pktsiz)) <= 0) {
 529       timer_start = 0;      /* clear timer */
 530       if (errno == 0) {
 531          b_errno = ENODATA;
 532       } else {
 533          b_errno = errno;
 534       }
 535       errors++;
 536       Qmsg4(m_jcr, M_ERROR, 0, _("Read error from %s:%s:%d: ERR=%s\n"),
 537             m_who, m_host, m_port, this->bstrerror());
 538       nbytes = BNET_ERROR;
 539       goto get_out;
 540    }
 541    timer_start = 0;         /* clear timer */
 542    in_msg_no++;
 543    msglen = nbytes;
 544    if (nbytes != pktsiz) {
 545       b_errno = EIO;
 546       errors++;
 547       Qmsg5(m_jcr, M_ERROR, 0, _("Read expected %d got %d from %s:%s:%d\n"),
 548             pktsiz, nbytes, m_who, m_host, m_port);
 549       nbytes = BNET_ERROR;
 550       goto get_out;
 551    }
 552    /* always add a zero by to properly terminate any
 553     * string that was send to us. Note, we ensured above that the
 554     * buffer is at least one byte longer than the message length.
 555     */
 556    msg[nbytes] = 0; /* terminate in case it is a string */
 557    sm_check(__FILE__, __LINE__, false);
 558
 559 get_out:
 560    if (m_use_locking) V(m_mutex);
 561    return nbytes;                  /* return actual length of message */
 562 }
 563
 564
 565 /*
 566  * Send a signal
 567  */
 568 bool BSOCK::signal(int signal)
 569 {
 570    msglen = signal;
 571    if (signal == BNET_TERMINATE) {
 572       m_suppress_error_msgs = true;
 573    }
 574    return send();
 575 }
 576
 577 /*
 578  * Despool spooled attributes
 579  */
 580 bool BSOCK::despool(void update_attr_spool_size(ssize_t size), ssize_t tsize)
 581 {
 582    int32_t pktsiz;
 583    size_t nbytes;
 584    ssize_t last = 0, size = 0;
 585    int count = 0;
 586    JCR *jcr = get_jcr();
 587
 588    rewind(m_spool_fd);
 589
 590 #if defined(HAVE_POSIX_FADVISE) && defined(POSIX_FADV_WILLNEED)
 591    posix_fadvise(fileno(m_spool_fd), 0, 0, POSIX_FADV_WILLNEED);
 592 #endif
 593
 594    while (fread((char *)&pktsiz, 1, sizeof(int32_t), m_spool_fd) ==
 595           sizeof(int32_t)) {
 596       size += sizeof(int32_t);
 597       msglen = ntohl(pktsiz);
 598       if (msglen > 0) {
 599          if (msglen > (int32_t)sizeof_pool_memory(msg)) {
 600             msg = realloc_pool_memory(msg, msglen + 1);
 601          }
 602          nbytes = fread(msg, 1, msglen, m_spool_fd);
 603          if (nbytes != (size_t)msglen) {
 604             berrno be;
 605             Dmsg2(400, "nbytes=%d msglen=%d\n", nbytes, msglen);
 606             Qmsg1(get_jcr(), M_FATAL, 0, _("fread attr spool error. ERR=%s\n"),
 607                   be.bstrerror());
 608             update_attr_spool_size(tsize - last);
 609             return false;
 610          }
 611          size += nbytes;
 612          if ((++count & 0x3F) == 0) {
 613             update_attr_spool_size(size - last);
 614             last = size;
 615          }
 616       }
 617       send();
 618       if (jcr && job_canceled(jcr)) {
 619          return false;
 620       }
 621    }
 622    update_attr_spool_size(tsize - last);
 623    if (ferror(m_spool_fd)) {
 624       Qmsg(jcr, M_FATAL, 0, _("fread attr spool I/O error.\n"));
 625       return false;
 626    }
 627    return true;
 628 }
 629
 630 /*
 631  * Return the string for the error that occurred
 632  * on the socket. Only the first error is retained.
 633  */
 634 const char *BSOCK::bstrerror()
 635 {
 636    berrno be;
 637    if (errmsg == NULL) {
 638       errmsg = get_pool_memory(PM_MESSAGE);
 639    }
 640    pm_strcpy(errmsg, be.bstrerror(b_errno));
 641    return errmsg;
 642 }
 643
 644 int BSOCK::get_peer(char *buf, socklen_t buflen)
 645 {
 646 #if !defined(HAVE_WIN32)
 647     if (peer_addr.sin_family == 0) {
 648         socklen_t salen = sizeof(peer_addr);
 649         int rval = (getpeername)(m_fd, (struct sockaddr *)&peer_addr, &salen);
 650         if (rval < 0) return rval;
 651     }
 652     if (!inet_ntop(peer_addr.sin_family, &peer_addr.sin_addr, buf, buflen))
 653         return -1;
 654
 655     return 0;
 656 #else
 657     return -1;
 658 #endif
 659 }
 660
 661 /*
 662  * Set the network buffer size, suggested size is in size.
 663  *  Actual size obtained is returned in bs->msglen
 664  *
 665  *  Returns: false on failure
 666  *           true  on success
 667  */
 668 bool BSOCK::set_buffer_size(uint32_t size, int rw)
 669 {
 670    uint32_t dbuf_size, start_size;
 671
 672 #if defined(IP_TOS) && defined(IPTOS_THROUGHPUT)
 673    int opt;
 674    opt = IPTOS_THROUGHPUT;
 675    setsockopt(fd, IPPROTO_IP, IP_TOS, (sockopt_val_t)&opt, sizeof(opt));
 676 #endif
 677
 678    if (size != 0) {
 679       dbuf_size = size;
 680    } else {
 681       dbuf_size = DEFAULT_NETWORK_BUFFER_SIZE;
 682    }
 683    start_size = dbuf_size;
 684    if ((msg = realloc_pool_memory(msg, dbuf_size + 100)) == NULL) {
 685       Qmsg0(get_jcr(), M_FATAL, 0, _("Could not malloc BSOCK data buffer\n"));
 686       return false;
 687    }
 688
 689    /*
 690     * If user has not set the size, use the OS default -- i.e. do not
 691     *   try to set it.  This allows sys admins to set the size they
 692     *   want in the OS, and Bacula will comply. See bug #1493
 693     */
 694    if (size == 0) {
 695       msglen = dbuf_size;
 696       return true;
 697    }
 698
 699    if (rw & BNET_SETBUF_READ) {
 700       while ((dbuf_size > TAPE_BSIZE) && (setsockopt(m_fd, SOL_SOCKET,
 701               SO_RCVBUF, (sockopt_val_t) & dbuf_size, sizeof(dbuf_size)) < 0)) {
 702          berrno be;
 703          Qmsg1(get_jcr(), M_ERROR, 0, _("sockopt error: %s\n"), be.bstrerror());
 704          dbuf_size -= TAPE_BSIZE;
 705       }
 706       Dmsg1(200, "set network buffer size=%d\n", dbuf_size);
 707       if (dbuf_size != start_size) {
 708          Qmsg1(get_jcr(), M_WARNING, 0,
 709                _("Warning network buffer = %d bytes not max size.\n"), dbuf_size);
 710       }
 711       if (dbuf_size % TAPE_BSIZE != 0) {
 712          Qmsg1(get_jcr(), M_ABORT, 0,
 713                _("Network buffer size %d not multiple of tape block size.\n"),
 714                dbuf_size);
 715       }
 716    }
 717    if (size != 0) {
 718       dbuf_size = size;
 719    } else {
 720       dbuf_size = DEFAULT_NETWORK_BUFFER_SIZE;
 721    }
 722    start_size = dbuf_size;
 723    if (rw & BNET_SETBUF_WRITE) {
 724       while ((dbuf_size > TAPE_BSIZE) && (setsockopt(m_fd, SOL_SOCKET,
 725               SO_SNDBUF, (sockopt_val_t) & dbuf_size, sizeof(dbuf_size)) < 0)) {
 726          berrno be;
 727          Qmsg1(get_jcr(), M_ERROR, 0, _("sockopt error: %s\n"), be.bstrerror());
 728          dbuf_size -= TAPE_BSIZE;
 729       }
 730       Dmsg1(900, "set network buffer size=%d\n", dbuf_size);
 731       if (dbuf_size != start_size) {
 732          Qmsg1(get_jcr(), M_WARNING, 0,
 733                _("Warning network buffer = %d bytes not max size.\n"), dbuf_size);
 734       }
 735       if (dbuf_size % TAPE_BSIZE != 0) {
 736          Qmsg1(get_jcr(), M_ABORT, 0,
 737                _("Network buffer size %d not multiple of tape block size.\n"),
 738                dbuf_size);
 739       }
 740    }
 741
 742    msglen = dbuf_size;
 743    return true;
 744 }
 745
 746 /*
 747  * Set socket non-blocking
 748  * Returns previous socket flag
 749  */
 750 int BSOCK::set_nonblocking()
 751 {
 752 #ifndef HAVE_WIN32
 753    int oflags;
 754
 755    /* Get current flags */
 756    if ((oflags = fcntl(m_fd, F_GETFL, 0)) < 0) {
 757       berrno be;
 758       Qmsg1(get_jcr(), M_ABORT, 0, _("fcntl F_GETFL error. ERR=%s\n"), be.bstrerror());
 759    }
 760
 761    /* Set O_NONBLOCK flag */
 762    if ((fcntl(m_fd, F_SETFL, oflags|O_NONBLOCK)) < 0) {
 763       berrno be;
 764       Qmsg1(get_jcr(), M_ABORT, 0, _("fcntl F_SETFL error. ERR=%s\n"), be.bstrerror());
 765    }
 766
 767    m_blocking = 0;
 768    return oflags;
 769 #else
 770    int flags;
 771    u_long ioctlArg = 1;
 772
 773    flags = m_blocking;
 774    ioctlsocket(m_fd, FIONBIO, &ioctlArg);
 775    m_blocking = 0;
 776
 777    return flags;
 778 #endif
 779 }
 780
 781 /*
 782  * Set socket blocking
 783  * Returns previous socket flags
 784  */
 785 int BSOCK::set_blocking()
 786 {
 787 #ifndef HAVE_WIN32
 788    int oflags;
 789    /* Get current flags */
 790    if ((oflags = fcntl(m_fd, F_GETFL, 0)) < 0) {
 791       berrno be;
 792       Qmsg1(get_jcr(), M_ABORT, 0, _("fcntl F_GETFL error. ERR=%s\n"), be.bstrerror());
 793    }
 794
 795    /* Set O_NONBLOCK flag */
 796    if ((fcntl(m_fd, F_SETFL, oflags & ~O_NONBLOCK)) < 0) {
 797       berrno be;
 798       Qmsg1(get_jcr(), M_ABORT, 0, _("fcntl F_SETFL error. ERR=%s\n"), be.bstrerror());
 799    }
 800
 801    m_blocking = 1;
 802    return oflags;
 803 #else
 804    int flags;
 805    u_long ioctlArg = 0;
 806
 807    flags = m_blocking;
 808    ioctlsocket(m_fd, FIONBIO, &ioctlArg);
 809    m_blocking = 1;
 810
 811    return flags;
 812 #endif
 813 }
 814
 815 /*
 816  * Restores socket flags
 817  */
 818 void BSOCK::restore_blocking (int flags)
 819 {
 820 #ifndef HAVE_WIN32
 821    if ((fcntl(m_fd, F_SETFL, flags)) < 0) {
 822       berrno be;
 823       Qmsg1(get_jcr(), M_ABORT, 0, _("fcntl F_SETFL error. ERR=%s\n"), be.bstrerror());
 824    }
 825
 826    m_blocking = (flags & O_NONBLOCK) ? true : false;
 827 #else
 828    u_long ioctlArg = flags;
 829
 830    ioctlsocket(m_fd, FIONBIO, &ioctlArg);
 831    m_blocking = 1;
 832 #endif
 833 }
 834
 835 /*
 836  * Wait for a specified time for data to appear on
 837  * the BSOCK connection.
 838  *
 839  *   Returns: 1 if data available
 840  *            0 if timeout
 841  *           -1 if error
 842  */
 843 int BSOCK::wait_data(int sec, int usec)
 844 {
 845    fd_set fdset;
 846    struct timeval tv;
 847
 848    FD_ZERO(&fdset);
 849    FD_SET((unsigned)m_fd, &fdset);
 850    for (;;) {
 851       tv.tv_sec = sec;
 852       tv.tv_usec = usec;
 853       switch (select(m_fd + 1, &fdset, NULL, NULL, &tv)) {
 854       case 0:                      /* timeout */
 855          b_errno = 0;
 856          return 0;
 857       case -1:
 858          b_errno = errno;
 859          if (errno == EINTR) {
 860             continue;
 861          }
 862          return -1;                /* error return */
 863       default:
 864          b_errno = 0;
 865          return 1;
 866       }
 867    }
 868 }
 869
 870 /*
 871  * As above, but returns on interrupt
 872  */
 873 int BSOCK::wait_data_intr(int sec, int usec)
 874 {
 875    fd_set fdset;
 876    struct timeval tv;
 877
 878    if (this == NULL) {
 879       return -1;
 880    }
 881    FD_ZERO(&fdset);
 882    FD_SET((unsigned)m_fd, &fdset);
 883    tv.tv_sec = sec;
 884    tv.tv_usec = usec;
 885    switch (select(m_fd + 1, &fdset, NULL, NULL, &tv)) {
 886    case 0:                      /* timeout */
 887       b_errno = 0;
 888       return 0;
 889    case -1:
 890       b_errno = errno;
 891       return -1;                /* error return */
 892    default:
 893       b_errno = 0;
 894       break;
 895    }
 896    return 1;
 897 }
 898
 899 /*
 900  * Note, this routine closes and destroys all the sockets
 901  *  that are open including the duped ones.
 902  */
 903 #ifndef SHUT_RDWR
 904 #define SHUT_RDWR 2
 905 #endif
 906
 907 void BSOCK::close()
 908 {
 909    BSOCK *bsock = this;
 910    BSOCK *next;
 911
 912    if (!m_duped) {
 913       clear_locking();
 914    }
 915    for (; bsock; bsock = next) {
 916       next = bsock->m_next;           /* get possible pointer to next before destoryed */
 917       if (!bsock->m_duped) {
 918          /* Shutdown tls cleanly. */
 919          if (bsock->tls) {
 920             tls_bsock_shutdown(bsock);
 921             free_tls_connection(bsock->tls);
 922             bsock->tls = NULL;
 923          }
 924          if (bsock->is_timed_out()) {
 925             shutdown(bsock->m_fd, SHUT_RDWR);   /* discard any pending I/O */
 926          }
 927          socketClose(bsock->m_fd);      /* normal close */
 928       }
 929       bsock->destroy();
 930    }
 931    return;
 932 }
 933
 934 void BSOCK::destroy()
 935 {
 936    if (msg) {
 937       free_pool_memory(msg);
 938       msg = NULL;
 939    } else {
 940       ASSERT(1 == 0);              /* double close */
 941    }
 942    if (errmsg) {
 943       free_pool_memory(errmsg);
 944       errmsg = NULL;
 945    }
 946    if (m_who) {
 947       free(m_who);
 948       m_who = NULL;
 949    }
 950    if (m_host) {
 951       free(m_host);
 952       m_host = NULL;
 953    }
 954    if (src_addr) {
 955       free(src_addr);
 956       src_addr = NULL;
 957    }
 958    free(this);
 959 }
 960
 961 /* Commands sent to Director */
 962 static char hello[]    = "Hello %s calling\n";
 963
 964 /* Response from Director */
 965 static char OKhello[]   = "1000 OK:";
 966
 967 /*
 968  * Authenticate Director
 969  */
 970 bool BSOCK::authenticate_director(const char *name, const char *password,
 971                TLS_CONTEXT *tls_ctx, char *msg, int msglen)
 972 {
 973    int tls_local_need = BNET_TLS_NONE;
 974    int tls_remote_need = BNET_TLS_NONE;
 975    int compatible = true;
 976    char bashed_name[MAX_NAME_LENGTH];
 977    BSOCK *dir = this;        /* for readability */
 978
 979    msg[0] = 0;
 980    /*
 981     * Send my name to the Director then do authentication
 982     */
 983
 984    /* Timeout Hello after 15 secs */
 985    dir->start_timer(15);
 986    dir->fsend(hello, bashed_name);
 987
 988    if (get_tls_enable(tls_ctx)) {
 989       tls_local_need = get_tls_enable(tls_ctx) ? BNET_TLS_REQUIRED : BNET_TLS_OK;
 990    }
 991
 992    /* respond to Dir challenge */
 993    if (!cram_md5_respond(dir, password, &tls_remote_need, &compatible) ||
 994        /* Now challenge dir */
 995        !cram_md5_challenge(dir, password, tls_local_need, compatible)) {
 996       bsnprintf(msg, msglen, _("Director authorization problem at \"%s:%d\"\n"),
 997          dir->host(), dir->port());
 998       goto bail_out;
 999    }
1000
1001    /* Verify that the remote host is willing to meet our TLS requirements */
1002    if (tls_remote_need < tls_local_need && tls_local_need != BNET_TLS_OK && tls_remote_need != BNET_TLS_OK) {
1003       bsnprintf(msg, msglen, _("Authorization problem:"
1004              " Remote server at \"%s:%d\" did not advertise required TLS support.\n"),
1005              dir->host(), dir->port());
1006       goto bail_out;
1007    }
1008
1009    /* Verify that we are willing to meet the remote host's requirements */
1010    if (tls_remote_need > tls_local_need && tls_local_need != BNET_TLS_OK && tls_remote_need != BNET_TLS_OK) {
1011       bsnprintf(msg, msglen, _("Authorization problem with Director at \"%s:%d\":"
1012                      " Remote server requires TLS.\n"),
1013                      dir->host(), dir->port());
1014
1015       goto bail_out;
1016    }
1017
1018    /* Is TLS Enabled? */
1019    if (have_tls) {
1020       if (tls_local_need >= BNET_TLS_OK && tls_remote_need >= BNET_TLS_OK) {
1021          /* Engage TLS! Full Speed Ahead! */
1022          if (!bnet_tls_client(tls_ctx, dir, NULL)) {
1023             bsnprintf(msg, msglen, _("TLS negotiation failed with Director at \"%s:%d\"\n"),
1024                dir->host(), dir->port());
1025             goto bail_out;
1026          }
1027       }
1028    }
1029
1030    Dmsg1(6, ">dird: %s", dir->msg);
1031    if (dir->recv() <= 0) {
1032       dir->stop_timer();
1033       bsnprintf(msg, msglen, _("Bad response to Hello command: ERR=%s\n"
1034                       "The Director at \"%s:%d\" is probably not running.\n"),
1035                     dir->bstrerror(), dir->host(), dir->port());
1036       return false;
1037    }
1038
1039   dir->stop_timer();
1040    Dmsg1(10, "<dird: %s", dir->msg);
1041    if (strncmp(dir->msg, OKhello, sizeof(OKhello)-1) != 0) {
1042       bsnprintf(msg, msglen, _("Director at \"%s:%d\" rejected Hello command\n"),
1043          dir->host(), dir->port());
1044       return false;
1045    } else {
1046       bsnprintf(msg, msglen, "%s", dir->msg);
1047    }
1048    return true;
1049
1050 bail_out:
1051    dir->stop_timer();
1052    bsnprintf(msg, msglen, _("Authorization problem with Director at \"%s:%d\"\n"
1053              "Most likely the passwords do not agree.\n"
1054              "If you are using TLS, there may have been a certificate validation error during the TLS handshake.\n"
1055              "Please see http://www.bacula.org/en/rel-manual/Bacula_Freque_Asked_Questi.html#SECTION003760000000000000000 for help.\n"),
1056              dir->host(), dir->port());
1057    return false;
1058 }