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

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