contrib/csup/proto.c

   1 /*-
   2  * Copyright (c) 2003-2006, Maxime Henrion <mux@FreeBSD.org>
   3  * All rights reserved.
   4  *
   5  * Redistribution and use in source and binary forms, with or without
   6  * modification, are permitted provided that the following conditions
   7  * are met:
   8  * 1. Redistributions of source code must retain the above copyright
   9  *    notice, this list of conditions and the following disclaimer.
  10  * 2. Redistributions in binary form must reproduce the above copyright
  11  *    notice, this list of conditions and the following disclaimer in the
  12  *    documentation and/or other materials provided with the distribution.
  13  *
  14  * THIS SOFTWARE IS PROVIDED BY THE AUTHOR AND CONTRIBUTORS ``AS IS'' AND
  15  * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
  16  * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
  17  * ARE DISCLAIMED.  IN NO EVENT SHALL THE AUTHOR OR CONTRIBUTORS BE LIABLE
  18  * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
  19  * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
  20  * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
  21  * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
  22  * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
  23  * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
  24  * SUCH DAMAGE.
  25  *
  26  * $FreeBSD$
  27  */
  28
  29 #include <sys/param.h>
  30 #include <sys/select.h>
  31 #include <sys/socket.h>
  32 #include <sys/types.h>
  33 #include <sys/stat.h>
  34
  35 #include <assert.h>
  36 #include <err.h>
  37 #include <errno.h>
  38 #include <limits.h>
  39 #include <netdb.h>
  40 #include <pthread.h>
  41 #include <signal.h>
  42 #include <stdarg.h>
  43 #include <stddef.h>
  44 #include <stdio.h>
  45 #include <stdlib.h>
  46 #include <string.h>
  47 #include <unistd.h>
  48
  49 #include "config.h"
  50 #include "detailer.h"
  51 #include "fattr.h"
  52 #include "fixups.h"
  53 #include "globtree.h"
  54 #include "keyword.h"
  55 #include "lister.h"
  56 #include "misc.h"
  57 #include "mux.h"
  58 #include "proto.h"
  59 #include "queue.h"
  60 #include "stream.h"
  61 #include "threads.h"
  62 #include "updater.h"
  63
  64 struct killer {
  65         pthread_t thread;
  66         sigset_t sigset;
  67         struct mux *mux;
  68         int killedby;
  69 };
  70
  71 static void              killer_start(struct killer *, struct mux *);
  72 static void             *killer_run(void *);
  73 static void              killer_stop(struct killer *);
  74
  75 static int               proto_waitconnect(int);
  76 static int               proto_greet(struct config *);
  77 static int               proto_negproto(struct config *);
  78 static int               proto_login(struct config *);
  79 static int               proto_fileattr(struct config *);
  80 static int               proto_xchgcoll(struct config *);
  81 static struct mux       *proto_mux(struct config *);
  82
  83 static int               proto_escape(struct stream *, const char *);
  84 static void              proto_unescape(char *);
  85
  86 static int
  87 proto_waitconnect(int s)
  88 {
  89         fd_set readfd;
  90         socklen_t len;
  91         int error, rv, soerror;
  92
  93         FD_ZERO(&readfd);
  94         FD_SET(s, &readfd);
  95
  96         do {
  97                 rv = select(s + 1, &readfd, NULL, NULL, NULL);
  98         } while (rv == -1 && errno == EINTR);
  99         if (rv == -1)
 100                 return (-1);
 101         /* Check that the connection was really successful. */
 102         len = sizeof(soerror);
 103         error = getsockopt(s, SOL_SOCKET, SO_ERROR, &soerror, &len);
 104         if (error) {
 105                 /* We have no choice but faking an error here. */
 106                 errno = ECONNREFUSED;
 107                 return (-1);
 108         }
 109         if (soerror) {
 110                 errno = soerror;
 111                 return (-1);
 112         }
 113         return (0);
 114 }
 115
 116 /* Connect to the CVSup server. */
 117 int
 118 proto_connect(struct config *config, int family, uint16_t port)
 119 {
 120         char addrbuf[NI_MAXHOST];
 121         /* Enough to hold sizeof("cvsup") or any port number. */
 122         char servname[8];
 123         struct addrinfo *res, *ai, hints;
 124         int error, opt, s;
 125
 126         s = -1;
 127         if (port != 0)
 128                 snprintf(servname, sizeof(servname), "%d", port);
 129         else {
 130                 strncpy(servname, "cvsup", sizeof(servname) - 1);
 131                 servname[sizeof(servname) - 1] = '\0';
 132         }
 133         memset(&hints, 0, sizeof(hints));
 134         hints.ai_family = family;
 135         hints.ai_socktype = SOCK_STREAM;
 136         error = getaddrinfo(config->host, servname, &hints, &res);
 137         /*
 138          * Try with the hardcoded port number for OSes that don't
 139          * have cvsup defined in the /etc/services file.
 140          */
 141         if (error == EAI_SERVICE) {
 142                 strncpy(servname, "5999", sizeof(servname) - 1);
 143                 servname[sizeof(servname) - 1] = '\0';
 144                 error = getaddrinfo(config->host, servname, &hints, &res);
 145         }
 146         if (error) {
 147                 lprintf(0, "Name lookup failure for \"%s\": %s\n", config->host,
 148                     gai_strerror(error));
 149                 return (STATUS_TRANSIENTFAILURE);
 150         }
 151         for (ai = res; ai != NULL; ai = ai->ai_next) {
 152                 s = socket(ai->ai_family, ai->ai_socktype, ai->ai_protocol);
 153                 if (s != -1) {
 154                         error = 0;
 155                         if (config->laddr != NULL) {
 156                                 opt = 1;
 157                                 (void)setsockopt(s, SOL_SOCKET, SO_REUSEADDR,
 158                                     &opt, sizeof(opt));
 159                                 error = bind(s, config->laddr,
 160                                     config->laddrlen);
 161                         }
 162                         if (!error) {
 163                                 error = connect(s, ai->ai_addr, ai->ai_addrlen);
 164                                 if (error && errno == EINTR)
 165                                         error = proto_waitconnect(s);
 166                         }
 167                         if (error)
 168                                 close(s);
 169                 }
 170                 (void)getnameinfo(ai->ai_addr, ai->ai_addrlen, addrbuf,
 171                     sizeof(addrbuf), NULL, 0, NI_NUMERICHOST);
 172                 if (s == -1 || error) {
 173                         lprintf(0, "Cannot connect to %s: %s\n", addrbuf,
 174                             strerror(errno));
 175                         continue;
 176                 }
 177                 lprintf(1, "Connected to %s\n", addrbuf);
 178                 freeaddrinfo(res);
 179                 config->socket = s;
 180                 return (STATUS_SUCCESS);
 181         }
 182         freeaddrinfo(res);
 183         return (STATUS_TRANSIENTFAILURE);
 184 }
 185
 186 /* Greet the server. */
 187 static int
 188 proto_greet(struct config *config)
 189 {
 190         char *line, *cmd, *msg, *swver;
 191         struct stream *s;
 192
 193         s = config->server;
 194         line = stream_getln(s, NULL);
 195         cmd = proto_get_ascii(&line);
 196         if (cmd == NULL)
 197                 goto bad;
 198         if (strcmp(cmd, "OK") == 0) {
 199                 (void)proto_get_ascii(&line);   /* major number */
 200                 (void)proto_get_ascii(&line);   /* minor number */
 201                 swver = proto_get_ascii(&line);
 202         } else if (strcmp(cmd, "!") == 0) {
 203                 msg = proto_get_rest(&line);
 204                 if (msg == NULL)
 205                         goto bad;
 206                 lprintf(-1, "Rejected by server: %s\n", msg);
 207                 return (STATUS_TRANSIENTFAILURE);
 208         } else
 209                 goto bad;
 210         lprintf(2, "Server software version: %s\n",
 211             swver != NULL ? swver : ".");
 212         return (STATUS_SUCCESS);
 213 bad:
 214         lprintf(-1, "Invalid greeting from server\n");
 215         return (STATUS_FAILURE);
 216 }
 217
 218 /* Negotiate protocol version with the server. */
 219 static int
 220 proto_negproto(struct config *config)
 221 {
 222         struct stream *s;
 223         char *cmd, *line, *msg;
 224         int error, maj, min;
 225
 226         s = config->server;
 227         proto_printf(s, "PROTO %d %d %s\n", PROTO_MAJ, PROTO_MIN, PROTO_SWVER);
 228         stream_flush(s);
 229         line = stream_getln(s, NULL);
 230         cmd = proto_get_ascii(&line);
 231         if (line == NULL)
 232                 goto bad;
 233         if (strcmp(cmd, "!") == 0) {
 234                 msg = proto_get_rest(&line);
 235                 lprintf(-1, "Protocol negotiation failed: %s\n", msg);
 236                 return (1);
 237         } else if (strcmp(cmd, "PROTO") != 0)
 238                 goto bad;
 239         error = proto_get_int(&line, &maj, 10);
 240         if (!error)
 241                 error = proto_get_int(&line, &min, 10);
 242         if (error)
 243                 goto bad;
 244         if (maj != PROTO_MAJ || min != PROTO_MIN) {
 245                 lprintf(-1, "Server protocol version %d.%d not supported "
 246                     "by client\n", maj, min);
 247                 return (STATUS_FAILURE);
 248         }
 249         return (STATUS_SUCCESS);
 250 bad:
 251         lprintf(-1, "Invalid PROTO command from server\n");
 252         return (STATUS_FAILURE);
 253 }
 254
 255 static int
 256 proto_login(struct config *config)
 257 {
 258         struct stream *s;
 259         char host[MAXHOSTNAMELEN];
 260         char *line, *cmd, *realm, *challenge, *msg;
 261
 262         s = config->server;
 263         gethostname(host, sizeof(host));
 264         host[sizeof(host) - 1] = '\0';
 265         proto_printf(s, "USER %s %s\n", getlogin(), host);
 266         stream_flush(s);
 267         line = stream_getln(s, NULL);
 268         cmd = proto_get_ascii(&line);
 269         realm = proto_get_ascii(&line);
 270         challenge = proto_get_ascii(&line);
 271         if (challenge == NULL || line != NULL)
 272                 goto bad;
 273         if (strcmp(realm, ".") != 0 || strcmp(challenge, ".") != 0) {
 274                 lprintf(-1, "Authentication required by the server and not "
 275                     "supported by client\n");
 276                 return (STATUS_FAILURE);
 277         }
 278         proto_printf(s, "AUTHMD5 . . .\n");
 279         stream_flush(s);
 280         line = stream_getln(s, NULL);
 281         cmd = proto_get_ascii(&line);
 282         if (strcmp(cmd, "OK") == 0)
 283                 return (STATUS_SUCCESS);
 284         if (strcmp(cmd, "!") == 0) {
 285                 msg = proto_get_rest(&line);
 286                 if (msg == NULL)
 287                         goto bad;
 288                 lprintf(-1, "Server error: %s\n", msg);
 289                 return (STATUS_FAILURE);
 290         }
 291 bad:
 292         lprintf(-1, "Invalid server reply to AUTHMD5\n");
 293         return (STATUS_FAILURE);
 294 }
 295
 296 /*
 297  * File attribute support negotiation.
 298  */
 299 static int
 300 proto_fileattr(struct config *config)
 301 {
 302         fattr_support_t support;
 303         struct stream *s;
 304         char *line, *cmd;
 305         int error, i, n, attr;
 306
 307         s = config->server;
 308         lprintf(2, "Negotiating file attribute support\n");
 309         proto_printf(s, "ATTR %d\n", FT_NUMBER);
 310         for (i = 0; i < FT_NUMBER; i++)
 311                 proto_printf(s, "%x\n", fattr_supported(i));
 312         proto_printf(s, ".\n");
 313         stream_flush(s);
 314         line = stream_getln(s, NULL);
 315         if (line == NULL)
 316                 goto bad;
 317         cmd = proto_get_ascii(&line);
 318         error = proto_get_int(&line, &n, 10);
 319         if (error || line != NULL || strcmp(cmd, "ATTR") != 0 || n > FT_NUMBER)
 320                 goto bad;
 321         for (i = 0; i < n; i++) {
 322                 line = stream_getln(s, NULL);
 323                 if (line == NULL)
 324                         goto bad;
 325                 error = proto_get_int(&line, &attr, 16);
 326                 if (error)
 327                         goto bad;
 328                 support[i] = fattr_supported(i) & attr;
 329         }
 330         for (i = n; i < FT_NUMBER; i++)
 331                 support[i] = 0;
 332         line = stream_getln(s, NULL);
 333         if (line == NULL || strcmp(line, ".") != 0)
 334                 goto bad;
 335         memcpy(config->fasupport, support, sizeof(config->fasupport));
 336         return (STATUS_SUCCESS);
 337 bad:
 338         lprintf(-1, "Protocol error negotiating attribute support\n");
 339         return (STATUS_FAILURE);
 340 }
 341
 342 /*
 343  * Exchange collection information.
 344  */
 345 static int
 346 proto_xchgcoll(struct config *config)
 347 {
 348         struct coll *coll;
 349         struct stream *s;
 350         struct globtree *diraccept, *dirrefuse;
 351         struct globtree *fileaccept, *filerefuse;
 352         char *line, *cmd, *collname, *pat;
 353         char *msg, *release, *ident, *rcskey, *prefix;
 354         size_t i, len;
 355         int error, flags, options;
 356
 357         s = config->server;
 358         lprintf(2, "Exchanging collection information\n");
 359         STAILQ_FOREACH(coll, &config->colls, co_next) {
 360                 proto_printf(s, "COLL %s %s %o %d\n", coll->co_name,
 361                     coll->co_release, coll->co_umask, coll->co_options);
 362                 for (i = 0; i < pattlist_size(coll->co_accepts); i++) {
 363                     proto_printf(s, "ACC %s\n",
 364                         pattlist_get(coll->co_accepts, i));
 365                 }
 366                 for (i = 0; i < pattlist_size(coll->co_refusals); i++) {
 367                     proto_printf(s, "REF %s\n",
 368                         pattlist_get(coll->co_refusals, i));
 369                 }
 370                 proto_printf(s, ".\n");
 371         }
 372         proto_printf(s, ".\n");
 373         stream_flush(s);
 374         STAILQ_FOREACH(coll, &config->colls, co_next) {
 375                 if (coll->co_options & CO_SKIP)
 376                         continue;
 377                 line = stream_getln(s, NULL);
 378                 if (line == NULL)
 379                         goto bad;
 380                 cmd = proto_get_ascii(&line);
 381                 collname = proto_get_ascii(&line);
 382                 release = proto_get_ascii(&line);
 383                 error = proto_get_int(&line, &options, 10);
 384                 if (error || line != NULL)
 385                         goto bad;
 386                 if (strcmp(cmd, "COLL") != 0 ||
 387                     strcmp(collname, coll->co_name) != 0 ||
 388                     strcmp(release, coll->co_release) != 0)
 389                         goto bad;
 390                 coll->co_options =
 391                     (coll->co_options | (options & CO_SERVMAYSET)) &
 392                     ~(~options & CO_SERVMAYCLEAR);
 393                 while ((line = stream_getln(s, NULL)) != NULL) {
 394                         if (strcmp(line, ".") == 0)
 395                                 break;
 396                         cmd = proto_get_ascii(&line);
 397                         if (cmd == NULL)
 398                                 goto bad;
 399                         if (strcmp(cmd, "!") == 0) {
 400                                 msg = proto_get_rest(&line);
 401                                 if (msg == NULL)
 402                                         goto bad;
 403                                 lprintf(-1, "Server message: %s\n", msg);
 404                         } else if (strcmp(cmd, "PRFX") == 0) {
 405                                 prefix = proto_get_ascii(&line);
 406                                 if (prefix == NULL || line != NULL)
 407                                         goto bad;
 408                                 coll->co_cvsroot = xstrdup(prefix);
 409                         } else if (strcmp(cmd, "KEYALIAS") == 0) {
 410                                 ident = proto_get_ascii(&line);
 411                                 rcskey = proto_get_ascii(&line);
 412                                 if (rcskey == NULL || line != NULL)
 413                                         goto bad;
 414                                 error = keyword_alias(coll->co_keyword, ident,
 415                                     rcskey);
 416                                 if (error)
 417                                         goto bad;
 418                         } else if (strcmp(cmd, "KEYON") == 0) {
 419                                 ident = proto_get_ascii(&line);
 420                                 if (ident == NULL || line != NULL)
 421                                         goto bad;
 422                                 error = keyword_enable(coll->co_keyword, ident);
 423                                 if (error)
 424                                         goto bad;
 425                         } else if (strcmp(cmd, "KEYOFF") == 0) {
 426                                 ident = proto_get_ascii(&line);
 427                                 if (ident == NULL || line != NULL)
 428                                         goto bad;
 429                                 error = keyword_disable(coll->co_keyword,
 430                                     ident);
 431                                 if (error)
 432                                         goto bad;
 433                         }
 434                 }
 435                 if (line == NULL)
 436                         goto bad;
 437                 keyword_prepare(coll->co_keyword);
 438
 439                 diraccept = globtree_true();
 440                 fileaccept = globtree_true();
 441                 dirrefuse = globtree_false();
 442                 filerefuse = globtree_false();
 443
 444                 if (pattlist_size(coll->co_accepts) > 0) {
 445                         globtree_free(diraccept);
 446                         globtree_free(fileaccept);
 447                         diraccept = globtree_false();
 448                         fileaccept = globtree_false();
 449                         flags = FNM_PATHNAME | FNM_LEADING_DIR |
 450                             FNM_PREFIX_DIRS;
 451                         for (i = 0; i < pattlist_size(coll->co_accepts); i++) {
 452                                 pat = pattlist_get(coll->co_accepts, i);
 453                                 diraccept = globtree_or(diraccept,
 454                                     globtree_match(pat, flags));
 455
 456                                 len = strlen(pat);
 457                                 if (coll->co_options & CO_CHECKOUTMODE &&
 458                                     (len == 0 || pat[len - 1] != '*')) {
 459                                         /* We must modify the pattern so that it
 460                                            refers to the RCS file, rather than
 461                                            the checked-out file. */
 462                                         xasprintf(&pat, "%s,v", pat);
 463                                         fileaccept = globtree_or(fileaccept,
 464                                             globtree_match(pat, flags));
 465                                         free(pat);
 466                                 } else {
 467                                         fileaccept = globtree_or(fileaccept,
 468                                             globtree_match(pat, flags));
 469                                 }
 470                         }
 471                 }
 472
 473                 for (i = 0; i < pattlist_size(coll->co_refusals); i++) {
 474                         pat = pattlist_get(coll->co_refusals, i);
 475                         dirrefuse = globtree_or(dirrefuse,
 476                             globtree_match(pat, 0));
 477                         len = strlen(pat);
 478                         if (coll->co_options & CO_CHECKOUTMODE &&
 479                             (len == 0 || pat[len - 1] != '*')) {
 480                                 /* We must modify the pattern so that it refers
 481                                    to the RCS file, rather than the checked-out
 482                                    file. */
 483                                 xasprintf(&pat, "%s,v", pat);
 484                                 filerefuse = globtree_or(filerefuse,
 485                                     globtree_match(pat, 0));
 486                                 free(pat);
 487                         } else {
 488                                 filerefuse = globtree_or(filerefuse,
 489                                     globtree_match(pat, 0));
 490                         }
 491                 }
 492
 493                 coll->co_dirfilter = globtree_and(diraccept,
 494                     globtree_not(dirrefuse));
 495                 coll->co_filefilter = globtree_and(fileaccept,
 496                     globtree_not(filerefuse));
 497
 498                 /* At this point we don't need the pattern lists anymore. */
 499                 pattlist_free(coll->co_accepts);
 500                 pattlist_free(coll->co_refusals);
 501                 coll->co_accepts = NULL;
 502                 coll->co_refusals = NULL;
 503
 504                 /* Set up a mask of file attributes that we don't want to sync
 505                    with the server. */
 506                 if (!(coll->co_options & CO_SETOWNER))
 507                         coll->co_attrignore |= FA_OWNER | FA_GROUP;
 508                 if (!(coll->co_options & CO_SETMODE))
 509                         coll->co_attrignore |= FA_MODE;
 510                 if (!(coll->co_options & CO_SETFLAGS))
 511                         coll->co_attrignore |= FA_FLAGS;
 512         }
 513         return (STATUS_SUCCESS);
 514 bad:
 515         lprintf(-1, "Protocol error during collection exchange\n");
 516         return (STATUS_FAILURE);
 517 }
 518
 519 static struct mux *
 520 proto_mux(struct config *config)
 521 {
 522         struct mux *m;
 523         struct stream *s, *wr;
 524         struct chan *chan0, *chan1;
 525         int id;
 526
 527         s = config->server;
 528         lprintf(2, "Establishing multiplexed-mode data connection\n");
 529         proto_printf(s, "MUX\n");
 530         stream_flush(s);
 531         m = mux_open(config->socket, &chan0);
 532         if (m == NULL) {
 533                 lprintf(-1, "Cannot open the multiplexer\n");
 534                 return (NULL);
 535         }
 536         id = chan_listen(m);
 537         if (id == -1) {
 538                 lprintf(-1, "ChannelMux.Listen failed: %s\n", strerror(errno));
 539                 mux_close(m);
 540                 return (NULL);
 541         }
 542         wr = stream_open(chan0, NULL, (stream_writefn_t *)chan_write, NULL);
 543         proto_printf(wr, "CHAN %d\n", id);
 544         stream_close(wr);
 545         chan1 = chan_accept(m, id);
 546         if (chan1 == NULL) {
 547                 lprintf(-1, "ChannelMux.Accept failed: %s\n", strerror(errno));
 548                 mux_close(m);
 549                 return (NULL);
 550         }
 551         config->chan0 = chan0;
 552         config->chan1 = chan1;
 553         return (m);
 554 }
 555
 556 /*
 557  * Initializes the connection to the CVSup server, that is handle
 558  * the protocol negotiation, logging in, exchanging file attributes
 559  * support and collections information, and finally run the update
 560  * session.
 561  */
 562 int
 563 proto_run(struct config *config)
 564 {
 565         struct thread_args lister_args;
 566         struct thread_args detailer_args;
 567         struct thread_args updater_args;
 568         struct thread_args *args;
 569         struct killer killer;
 570         struct threads *workers;
 571         struct mux *m;
 572         int i, status;
 573
 574         /*
 575          * We pass NULL for the close() function because we'll reuse
 576          * the socket after the stream is closed.
 577          */
 578         config->server = stream_open_fd(config->socket, stream_read_fd,
 579             stream_write_fd, NULL);
 580         status = proto_greet(config);
 581         if (status == STATUS_SUCCESS)
 582                 status = proto_negproto(config);
 583         if (status == STATUS_SUCCESS)
 584                 status = proto_login(config);
 585         if (status == STATUS_SUCCESS)
 586                 status = proto_fileattr(config);
 587         if (status == STATUS_SUCCESS)
 588                 status = proto_xchgcoll(config);
 589         if (status != STATUS_SUCCESS)
 590                 return (status);
 591
 592         /* Multi-threaded action starts here. */
 593         m = proto_mux(config);
 594         if (m == NULL)
 595                 return (STATUS_FAILURE);
 596
 597         stream_close(config->server);
 598         config->server = NULL;
 599         config->fixups = fixups_new();
 600         killer_start(&killer, m);
 601
 602         /* Start the worker threads. */
 603         workers = threads_new();
 604         args = &lister_args;
 605         args->config = config;
 606         args->status = -1;
 607         args->errmsg = NULL;
 608         args->rd = NULL;
 609         args->wr = stream_open(config->chan0,
 610             NULL, (stream_writefn_t *)chan_write, NULL);
 611         threads_create(workers, lister, args);
 612
 613         args = &detailer_args;
 614         args->config = config;
 615         args->status = -1;
 616         args->errmsg = NULL;
 617         args->rd = stream_open(config->chan0,
 618             (stream_readfn_t *)chan_read, NULL, NULL);
 619         args->wr = stream_open(config->chan1,
 620             NULL, (stream_writefn_t *)chan_write, NULL);
 621         threads_create(workers, detailer, args);
 622
 623         args = &updater_args;
 624         args->config = config;
 625         args->status = -1;
 626         args->errmsg = NULL;
 627         args->rd = stream_open(config->chan1,
 628             (stream_readfn_t *)chan_read, NULL, NULL);
 629         args->wr = NULL;
 630         threads_create(workers, updater, args);
 631
 632         lprintf(2, "Running\n");
 633         /* Wait for all the worker threads to finish. */
 634         status = STATUS_SUCCESS;
 635         for (i = 0; i < 3; i++) {
 636                 args = threads_wait(workers);
 637                 if (args->rd != NULL)
 638                         stream_close(args->rd);
 639                 if (args->wr != NULL)
 640                         stream_close(args->wr);
 641                 if (args->status != STATUS_SUCCESS) {
 642                         assert(args->errmsg != NULL);
 643                         if (status == STATUS_SUCCESS) {
 644                                 status = args->status;
 645                                 /* Shutdown the multiplexer to wake up all
 646                                    the other threads. */
 647                                 mux_shutdown(m, args->errmsg, status);
 648                         }
 649                         free(args->errmsg);
 650                 }
 651         }
 652         threads_free(workers);
 653         if (status == STATUS_SUCCESS) {
 654                 lprintf(2, "Shutting down connection to server\n");
 655                 chan_close(config->chan0);
 656                 chan_close(config->chan1);
 657                 chan_wait(config->chan0);
 658                 chan_wait(config->chan1);
 659                 mux_shutdown(m, NULL, STATUS_SUCCESS);
 660         }
 661         killer_stop(&killer);
 662         fixups_free(config->fixups);
 663         status = mux_close(m);
 664         if (status == STATUS_SUCCESS) {
 665                 lprintf(1, "Finished successfully\n");
 666         } else if (status == STATUS_INTERRUPTED) {
 667                 lprintf(-1, "Interrupted\n");
 668                 if (killer.killedby != -1)
 669                         kill(getpid(), killer.killedby);
 670         }
 671         return (status);
 672 }
 673
 674 /*
 675  * Write a string into the stream, escaping characters as needed.
 676  * Characters escaped:
 677  *
 678  * SPACE        -> "\_"
 679  * TAB          ->  "\t"
 680  * NEWLINE      -> "\n"
 681  * CR           -> "\r"
 682  * \            -> "\\"
 683  */
 684 static int
 685 proto_escape(struct stream *wr, const char *s)
 686 {
 687         size_t len;
 688         ssize_t n;
 689         char c;
 690
 691         /* Handle characters that need escaping. */
 692         do {
 693                 len = strcspn(s, " \t\r\n\\");
 694                 n = stream_write(wr, s, len);
 695                 if (n == -1)
 696                         return (-1);
 697                 c = s[len];
 698                 switch (c) {
 699                 case ' ':
 700                         n = stream_write(wr, "\\_", 2);
 701                         break;
 702                 case '\t':
 703                         n = stream_write(wr, "\\t", 2);
 704                         break;
 705                 case '\r':
 706                         n = stream_write(wr, "\\r", 2);
 707                         break;
 708                 case '\n':
 709                         n = stream_write(wr, "\\n", 2);
 710                         break;
 711                 case '\\':
 712                         n = stream_write(wr, "\\\\", 2);
 713                         break;
 714                 }
 715                 if (n == -1)
 716                         return (-1);
 717                 s += len + 1;
 718         } while (c != '\0');
 719         return (0);
 720 }
 721
 722 /*
 723  * A simple printf() implementation specifically tailored for csup.
 724  * List of the supported formats:
 725  *
 726  * %c           Print a char.
 727  * %d or %i     Print an int as decimal.
 728  * %x           Print an int as hexadecimal.
 729  * %o           Print an int as octal.
 730  * %t           Print a time_t as decimal.
 731  * %s           Print a char * escaping some characters as needed.
 732  * %S           Print a char * without escaping.
 733  * %f           Print an encoded struct fattr *.
 734  * %F           Print an encoded struct fattr *, specifying the supported
 735  *              attributes.
 736  */
 737 int
 738 proto_printf(struct stream *wr, const char *format, ...)
 739 {
 740         fattr_support_t *support;
 741         long long longval;
 742         struct fattr *fa;
 743         const char *fmt;
 744         va_list ap;
 745         char *cp, *s, *attr;
 746         ssize_t n;
 747         int rv, val, ignore;
 748         char c;
 749
 750         n = 0;
 751         rv = 0;
 752         fmt = format;
 753         va_start(ap, format);
 754         while ((cp = strchr(fmt, '%')) != NULL) {
 755                 if (cp > fmt) {
 756                         n = stream_write(wr, fmt, cp - fmt);
 757                         if (n == -1)
 758                                 return (-1);
 759                 }
 760                 if (*++cp == '\0')
 761                         goto done;
 762                 switch (*cp) {
 763                 case 'c':
 764                         c = va_arg(ap, int);
 765                         rv = stream_printf(wr, "%c", c);
 766                         break;
 767                 case 'd':
 768                 case 'i':
 769                         val = va_arg(ap, int);
 770                         rv = stream_printf(wr, "%d", val);
 771                         break;
 772                 case 'x':
 773                         val = va_arg(ap, int);
 774                         rv = stream_printf(wr, "%x", val);
 775                         break;
 776                 case 'o':
 777                         val = va_arg(ap, int);
 778                         rv = stream_printf(wr, "%o", val);
 779                         break;
 780                 case 'S':
 781                         s = va_arg(ap, char *);
 782                         assert(s != NULL);
 783                         rv = stream_printf(wr, "%s", s);
 784                         break;
 785                 case 's':
 786                         s = va_arg(ap, char *);
 787                         assert(s != NULL);
 788                         rv = proto_escape(wr, s);
 789                         break;
 790                 case 't':
 791                         longval = (long long)va_arg(ap, time_t);
 792                         rv = stream_printf(wr, "%lld", longval);
 793                         break;
 794                 case 'f':
 795                         fa = va_arg(ap, struct fattr *);
 796                         attr = fattr_encode(fa, NULL, 0);
 797                         rv = proto_escape(wr, attr);
 798                         free(attr);
 799                         break;
 800                 case 'F':
 801                         fa = va_arg(ap, struct fattr *);
 802                         support = va_arg(ap, fattr_support_t *);
 803                         ignore = va_arg(ap, int);
 804                         attr = fattr_encode(fa, *support, ignore);
 805                         rv = proto_escape(wr, attr);
 806                         free(attr);
 807                         break;
 808                 case '%':
 809                         n = stream_write(wr, "%", 1);
 810                         if (n == -1)
 811                                 return (-1);
 812                         break;
 813                 }
 814                 if (rv == -1)
 815                         return (-1);
 816                 fmt = cp + 1;
 817         }
 818         if (*fmt != '\0') {
 819                 rv = stream_printf(wr, "%s", fmt);
 820                 if (rv == -1)
 821                         return (-1);
 822         }
 823 done:
 824         va_end(ap);
 825         return (0);
 826 }
 827
 828 /*
 829  * Unescape the string, see proto_escape().
 830  */
 831 static void
 832 proto_unescape(char *s)
 833 {
 834         char *cp, *cp2;
 835
 836         cp = s;
 837         while ((cp = strchr(cp, '\\')) != NULL) {
 838                 switch (cp[1]) {
 839                 case '_':
 840                         *cp = ' ';
 841                         break;
 842                 case 't':
 843                         *cp = '\t';
 844                         break;
 845                 case 'r':
 846                         *cp = '\r';
 847                         break;
 848                 case 'n':
 849                         *cp = '\n';
 850                         break;
 851                 case '\\':
 852                         *cp = '\\';
 853                         break;
 854                 default:
 855                         *cp = *(cp + 1);
 856                 }
 857                 cp2 = ++cp;
 858                 while (*cp2 != '\0') {
 859                         *cp2 = *(cp2 + 1);
 860                         cp2++;
 861                 }
 862         }
 863 }
 864
 865 /*
 866  * Get an ascii token in the string.
 867  */
 868 char *
 869 proto_get_ascii(char **s)
 870 {
 871         char *ret;
 872
 873         ret = strsep(s, " ");
 874         if (ret == NULL)
 875                 return (NULL);
 876         /* Make sure we disallow 0-length fields. */
 877         if (*ret == '\0') {
 878                 *s = NULL;
 879                 return (NULL);
 880         }
 881         proto_unescape(ret);
 882         return (ret);
 883 }
 884
 885 /*
 886  * Get the rest of the string.
 887  */
 888 char *
 889 proto_get_rest(char **s)
 890 {
 891         char *ret;
 892
 893         if (s == NULL)
 894                 return (NULL);
 895         ret = *s;
 896         proto_unescape(ret);
 897         *s = NULL;
 898         return (ret);
 899 }
 900
 901 /*
 902  * Get an int token.
 903  */
 904 int
 905 proto_get_int(char **s, int *val, int base)
 906 {
 907         char *cp, *end;
 908         long longval;
 909
 910         cp = proto_get_ascii(s);
 911         if (cp == NULL)
 912                 return (-1);
 913         errno = 0;
 914         longval = strtol(cp, &end, base);
 915         if (errno || *end != '\0')
 916                 return (-1);
 917         if (longval > INT_MAX || longval < INT_MIN) {
 918                 errno = ERANGE;
 919                 return (-1);
 920         }
 921         *val = longval;
 922         return (0);
 923 }
 924
 925 /*
 926  * Get a time_t token.
 927  *
 928  * Ideally, we would use an intmax_t and strtoimax() here, but strtoll()
 929  * is more portable and 64bits should be enough for a timestamp.
 930  */
 931 int
 932 proto_get_time(char **s, time_t *val)
 933 {
 934         long long tmp;
 935         char *cp, *end;
 936
 937         cp = proto_get_ascii(s);
 938         if (cp == NULL)
 939                 return (-1);
 940         errno = 0;
 941         tmp = strtoll(cp, &end, 10);
 942         if (errno || *end != '\0')
 943                 return (-1);
 944         *val = (time_t)tmp;
 945         return (0);
 946 }
 947
 948 /* Start the killer thread.  It is used to protect against some signals
 949    during the multi-threaded run so that we can gracefully fail.  */
 950 static void
 951 killer_start(struct killer *k, struct mux *m)
 952 {
 953         int error;
 954
 955         k->mux = m;
 956         k->killedby = -1;
 957         sigemptyset(&k->sigset);
 958         sigaddset(&k->sigset, SIGINT);
 959         sigaddset(&k->sigset, SIGHUP);
 960         sigaddset(&k->sigset, SIGTERM);
 961         sigaddset(&k->sigset, SIGPIPE);
 962         pthread_sigmask(SIG_BLOCK, &k->sigset, NULL);
 963         error = pthread_create(&k->thread, NULL, killer_run, k);
 964         if (error)
 965                 err(1, "pthread_create");
 966 }
 967
 968 /* The main loop of the killer thread. */
 969 static void *
 970 killer_run(void *arg)
 971 {
 972         struct killer *k;
 973         int error, sig, old;
 974
 975         k = arg;
 976 again:
 977         error = sigwait(&k->sigset, &sig);
 978         assert(!error);
 979         if (sig == SIGINT || sig == SIGHUP || sig == SIGTERM) {
 980                 if (k->killedby == -1) {
 981                         k->killedby = sig;
 982                         /* Ensure we don't get canceled during the shutdown. */
 983                         pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &old);
 984                         mux_shutdown(k->mux, "Cleaning up ...",
 985                             STATUS_INTERRUPTED);
 986                         pthread_setcancelstate(old, NULL);
 987                 }
 988         }
 989         goto again;
 990 }
 991
 992 /* Stop the killer thread. */
 993 static void
 994 killer_stop(struct killer *k)
 995 {
 996         void *val;
 997         int error;
 998
 999         error = pthread_cancel(k->thread);
1000         assert(!error);
1001         pthread_join(k->thread, &val);
1002         assert(val == PTHREAD_CANCELED);
1003         pthread_sigmask(SIG_UNBLOCK, &k->sigset, NULL);
1004 }