src/or/connection.c

   1 /* Copyright 2001,2002,2003 Roger Dingledine, Matej Pfajfar. */
   2 /* See LICENSE for licensing information */
   3 /* $Id$ */
   4
   5 #include "or.h"
   6
   7 /********* START VARIABLES **********/
   8
   9 extern or_options_t options; /* command-line and config-file options */
  10
  11 char *conn_type_to_string[] = {
  12   "",            /* 0 */
  13   "OP listener", /* 1 */
  14   "OP",          /* 2 */
  15   "OR listener", /* 3 */
  16   "OR",          /* 4 */
  17   "Exit",        /* 5 */
  18   "App listener",/* 6 */
  19   "App",         /* 7 */
  20   "Dir listener",/* 8 */
  21   "Dir",         /* 9 */
  22   "DNS worker",  /* 10 */
  23   "CPU worker",  /* 11 */
  24 };
  25
  26 char *conn_state_to_string[][_CONN_TYPE_MAX+1] = {
  27   { NULL }, /* no type associated with 0 */
  28   { NULL }, /* op listener, obsolete */
  29   { NULL }, /* op, obsolete */
  30   { "ready" }, /* or listener, 0 */
  31   { "",                         /* OR, 0 */
  32     "connect()ing",                 /* 1 */
  33     "handshaking",                  /* 2 */
  34     "open" },                       /* 3 */
  35   { "",                          /* exit, 0 */
  36     "waiting for dest info",           /* 1 */
  37     "connecting",                      /* 2 */
  38     "open",                            /* 3 */
  39     "resolve failed" },                /* 4 */
  40   { "ready" }, /* app listener, 0 */
  41   { "", /* 0 */
  42     "", /* 1 */
  43     "", /* 2 */
  44     "", /* 3 */
  45     "", /* 4 */
  46     "awaiting dest info",         /* app, 5 */
  47     "waiting for rendezvous desc",     /* 6 */
  48     "waiting for safe circuit",        /* 7 */
  49     "waiting for connected",           /* 8 */
  50     "open" },                          /* 9 */
  51   { "ready" }, /* dir listener, 0 */
  52   { "",                           /* dir, 0 */
  53     "connecting",                      /* 1 */
  54     "client sending",                  /* 2 */
  55     "client reading",                  /* 3 */
  56     "awaiting command",                /* 4 */
  57     "writing" },                       /* 5 */
  58   { "",                    /* dns worker, 0 */
  59     "idle",                            /* 1 */
  60     "busy" },                          /* 2 */
  61   { "",                    /* cpu worker, 0 */
  62     "idle",                            /* 1 */
  63     "busy with onion",                 /* 2 */
  64     "busy with handshake" },           /* 3 */
  65 };
  66
  67 /********* END VARIABLES ************/
  68
  69 static int connection_init_accepted_conn(connection_t *conn);
  70 static int connection_handle_listener_read(connection_t *conn, int new_type);
  71 static int connection_receiver_bucket_should_increase(connection_t *conn);
  72
  73 /**************************************************************/
  74
  75 connection_t *connection_new(int type) {
  76   connection_t *conn;
  77   time_t now = time(NULL);
  78
  79   conn = tor_malloc_zero(sizeof(connection_t));
  80   conn->magic = CONNECTION_MAGIC;
  81   conn->s = -1; /* give it a default of 'not used' */
  82
  83   conn->type = type;
  84   if(!connection_is_listener(conn)) { /* listeners never use their buf */
  85     conn->inbuf = buf_new();
  86     conn->outbuf = buf_new();
  87   }
  88   if (type == CONN_TYPE_AP) {
  89     conn->socks_request = tor_malloc_zero(sizeof(socks_request_t));
  90   }
  91
  92   conn->next_circ_id = crypto_pseudo_rand_int(1<<15);
  93
  94   conn->timestamp_created = now;
  95   conn->timestamp_lastread = now;
  96   conn->timestamp_lastwritten = now;
  97
  98   return conn;
  99 }
 100
 101 void connection_free(connection_t *conn) {
 102   assert(conn);
 103   assert(conn->magic == CONNECTION_MAGIC);
 104
 105   if(!connection_is_listener(conn)) {
 106     buf_free(conn->inbuf);
 107     buf_free(conn->outbuf);
 108   }
 109   tor_free(conn->address);
 110
 111   if(connection_speaks_cells(conn)) {
 112     directory_set_dirty();
 113     if (conn->tls)
 114       tor_tls_free(conn->tls);
 115   }
 116
 117   if (conn->onion_pkey)
 118     crypto_free_pk_env(conn->onion_pkey);
 119   if (conn->identity_pkey)
 120     crypto_free_pk_env(conn->identity_pkey);
 121   tor_free(conn->nickname);
 122   tor_free(conn->socks_request);
 123
 124   if(conn->s >= 0) {
 125     log_fn(LOG_INFO,"closing fd %d.",conn->s);
 126     close(conn->s);
 127   }
 128   memset(conn, 0xAA, sizeof(connection_t)); /* poison memory */
 129   free(conn);
 130 }
 131
 132 void connection_free_all(void) {
 133   int i, n;
 134   connection_t **carray;
 135
 136   get_connection_array(&carray,&n);
 137   for(i=0;i<n;i++)
 138     connection_free(carray[i]);
 139 }
 140
 141 /* Close the underlying socket for conn, so we don't try to flush it.
 142  * Must be used in conjunction with (right before) connection_mark_for_close
 143  */
 144 void connection_close_immediate(connection_t *conn)
 145 {
 146   assert_connection_ok(conn,0);
 147   if (conn->s < 0) {
 148     log_fn(LOG_WARN,"Attempt to close already-closed connection.");
 149     return;
 150   }
 151   if (conn->outbuf_flushlen) {
 152     log_fn(LOG_INFO,"Closing connection (fd %d, type %s, state %d) with data on outbuf.",
 153            conn->s, CONN_TYPE_TO_STRING(conn->type), conn->state);
 154   }
 155   close(conn->s);
 156   conn->s = -1;
 157   if(!connection_is_listener(conn)) {
 158     buf_clear(conn->outbuf);
 159     conn->outbuf_flushlen = 0;
 160   }
 161 }
 162
 163 int
 164 _connection_mark_for_close(connection_t *conn, char reason)
 165 {
 166   int retval = 0;
 167   assert_connection_ok(conn,0);
 168
 169   if (conn->marked_for_close) {
 170     log(LOG_WARN, "Double mark-for-close on connection.");
 171     return -1;
 172   }
 173
 174   switch (conn->type)
 175     {
 176     case CONN_TYPE_OR_LISTENER:
 177     case CONN_TYPE_AP_LISTENER:
 178     case CONN_TYPE_DIR_LISTENER:
 179     case CONN_TYPE_CPUWORKER:
 180       /* No special processing needed. */
 181       break;
 182     case CONN_TYPE_DIR:
 183       if(conn->purpose == DIR_PURPOSE_FETCH_RENDDESC)
 184         rend_client_desc_fetched(conn->rend_query, 0);
 185       break;
 186     case CONN_TYPE_OR:
 187       /* Remember why we're closing this connection. */
 188       if (conn->state != OR_CONN_STATE_OPEN) {
 189         /* XXX Nick: this still isn't right, because it might be
 190          * dying even though we didn't initiate the connect. Can
 191          * you look at this more? -RD */
 192         if(conn->nickname)
 193           rep_hist_note_connect_failed(conn->nickname, time(NULL));
 194       } else if (reason == CLOSE_REASON_UNUSED_OR_CONN) {
 195         rep_hist_note_disconnect(conn->nickname, time(NULL));
 196       } else {
 197         rep_hist_note_connection_died(conn->nickname, time(NULL));
 198       }
 199       /* No special processing needed. */
 200       break;
 201     case CONN_TYPE_AP:
 202       if (conn->socks_request->has_finished == 0) {
 203         log_fn(LOG_INFO,"Cleaning up AP -- sending socks reject.");
 204         connection_ap_handshake_socks_reply(conn, NULL, 0, 0);
 205         conn->socks_request->has_finished = 1;
 206         conn->hold_open_until_flushed = 1;
 207       }
 208       /* fall through, to do things for both ap and exit */
 209     case CONN_TYPE_EXIT:
 210       if (conn->state == EXIT_CONN_STATE_RESOLVING)
 211         connection_dns_remove(conn);
 212       if (!conn->has_sent_end && reason &&
 213           connection_edge_end(conn, reason, conn->cpath_layer) < 0)
 214         retval = -1;
 215       break;
 216     case CONN_TYPE_DNSWORKER:
 217       if (conn->state == DNSWORKER_STATE_BUSY) {
 218         dns_cancel_pending_resolve(conn->address);
 219       }
 220       break;
 221     default:
 222       log(LOG_ERR, "Unknown connection type %d", conn->type);
 223       ;
 224     }
 225   conn->marked_for_close = 1;
 226
 227   /* in case we're going to be held-open-til-flushed, reset
 228    * the number of seconds since last successful write, so
 229    * we get our whole 15 seconds */
 230   conn->timestamp_lastwritten = time(NULL);
 231
 232   return retval;
 233 }
 234
 235 void connection_expire_held_open(void)
 236 {
 237   connection_t **carray, *conn;
 238   int n, i;
 239   time_t now;
 240
 241   now = time(NULL);
 242
 243   get_connection_array(&carray, &n);
 244   for (i = 0; i < n; ++i) {
 245     conn = carray[i];
 246     /* If we've been holding the connection open, but we haven't written
 247      * for 15 seconds...
 248      */
 249     if (conn->hold_open_until_flushed) {
 250       assert(conn->marked_for_close);
 251       if (now - conn->timestamp_lastwritten >= 15) {
 252         log_fn(LOG_WARN,"Giving up on marked_for_close conn that's been flushing for 15s (fd %d, type %s, state %d).",
 253                conn->s, CONN_TYPE_TO_STRING(conn->type), conn->state);
 254         conn->hold_open_until_flushed = 0;
 255       }
 256     }
 257   }
 258 }
 259
 260 int connection_create_listener(char *bindaddress, uint16_t bindport, int type) {
 261   struct sockaddr_in bindaddr; /* where to bind */
 262   struct hostent *rent;
 263   connection_t *conn;
 264   int s; /* the socket we're going to make */
 265   int one=1;
 266
 267   memset(&bindaddr,0,sizeof(struct sockaddr_in));
 268   bindaddr.sin_family = AF_INET;
 269   bindaddr.sin_port = htons(bindport);
 270   rent = gethostbyname(bindaddress);
 271   if (!rent) {
 272     log_fn(LOG_WARN,"Can't resolve BindAddress %s",bindaddress);
 273     return -1;
 274   }
 275   if(rent->h_length != 4)
 276     return -1; /* XXX complain */
 277   memcpy(&(bindaddr.sin_addr.s_addr),rent->h_addr,rent->h_length);
 278
 279   s = socket(PF_INET,SOCK_STREAM,IPPROTO_TCP);
 280   if (s < 0) {
 281     log_fn(LOG_WARN,"Socket creation failed.");
 282     return -1;
 283   }
 284
 285   setsockopt(s, SOL_SOCKET, SO_REUSEADDR, (void*) &one, sizeof(one));
 286
 287   if(bind(s,(struct sockaddr *)&bindaddr,sizeof(bindaddr)) < 0) {
 288     log_fn(LOG_WARN,"Could not bind to port %u: %s",bindport,strerror(errno));
 289     return -1;
 290   }
 291
 292   if(listen(s,SOMAXCONN) < 0) {
 293     log_fn(LOG_WARN,"Could not listen on port %u: %s",bindport,strerror(errno));
 294     return -1;
 295   }
 296
 297   set_socket_nonblocking(s);
 298
 299   conn = connection_new(type);
 300   conn->s = s;
 301
 302   if(connection_add(conn) < 0) { /* no space, forget it */
 303     log_fn(LOG_WARN,"connection_add failed. Giving up.");
 304     connection_free(conn);
 305     return -1;
 306   }
 307
 308   log_fn(LOG_DEBUG,"%s listening on port %u.",conn_type_to_string[type], bindport);
 309
 310   conn->state = LISTENER_STATE_READY;
 311   connection_start_reading(conn);
 312
 313   return 0;
 314 }
 315
 316 static int connection_handle_listener_read(connection_t *conn, int new_type) {
 317   int news; /* the new socket */
 318   connection_t *newconn;
 319   struct sockaddr_in remote; /* information about the remote peer when connecting to other routers */
 320   int remotelen = sizeof(struct sockaddr_in); /* length of the remote address */
 321 #ifdef MS_WINDOWS
 322   int e;
 323 #endif
 324
 325   news = accept(conn->s,(struct sockaddr *)&remote,&remotelen);
 326   if (news == -1) { /* accept() error */
 327     if(ERRNO_EAGAIN(errno)) {
 328 #ifdef MS_WINDOWS
 329       e = correct_socket_errno(conn->s);
 330       if (ERRNO_EAGAIN(e))
 331         return 0;
 332 #else
 333       return 0; /* he hung up before we could accept(). that's fine. */
 334 #endif
 335     }
 336     /* else there was a real error. */
 337     log_fn(LOG_WARN,"accept() failed. Closing listener.");
 338     connection_mark_for_close(conn,0);
 339     return -1;
 340   }
 341   log(LOG_INFO,"Connection accepted on socket %d (child of fd %d).",news, conn->s);
 342
 343   set_socket_nonblocking(news);
 344
 345   newconn = connection_new(new_type);
 346   newconn->s = news;
 347
 348   newconn->address = tor_strdup(inet_ntoa(remote.sin_addr)); /* remember the remote address */
 349   newconn->addr = ntohl(remote.sin_addr.s_addr);
 350   newconn->port = ntohs(remote.sin_port);
 351
 352   if(connection_add(newconn) < 0) { /* no space, forget it */
 353     connection_free(newconn);
 354     return 0; /* no need to tear down the parent */
 355   }
 356
 357   if(connection_init_accepted_conn(newconn) < 0) {
 358     connection_mark_for_close(newconn,0);
 359     return 0;
 360   }
 361   return 0;
 362 }
 363
 364 static int connection_init_accepted_conn(connection_t *conn) {
 365
 366   connection_start_reading(conn);
 367
 368   switch(conn->type) {
 369     case CONN_TYPE_OR:
 370       return connection_tls_start_handshake(conn, 1);
 371     case CONN_TYPE_AP:
 372       conn->state = AP_CONN_STATE_SOCKS_WAIT;
 373       break;
 374     case CONN_TYPE_DIR:
 375       conn->purpose = DIR_PURPOSE_SERVER;
 376       conn->state = DIR_CONN_STATE_SERVER_COMMAND_WAIT;
 377       break;
 378   }
 379   return 0;
 380 }
 381
 382 /* take conn, make a nonblocking socket; try to connect to
 383  * addr:port (they arrive in *host order*). If fail, return -1. Else
 384  * assign s to conn->s: if connected return 1, if eagain return 0.
 385  * address is used to make the logs useful.
 386  */
 387 int connection_connect(connection_t *conn, char *address, uint32_t addr, uint16_t port) {
 388   int s;
 389   struct sockaddr_in dest_addr;
 390
 391   s=socket(PF_INET,SOCK_STREAM,IPPROTO_TCP);
 392   if (s < 0) {
 393     log_fn(LOG_WARN,"Error creating network socket.");
 394     return -1;
 395   }
 396   set_socket_nonblocking(s);
 397
 398   memset(&dest_addr,0,sizeof(dest_addr));
 399   dest_addr.sin_family = AF_INET;
 400   dest_addr.sin_port = htons(port);
 401   dest_addr.sin_addr.s_addr = htonl(addr);
 402
 403   log_fn(LOG_DEBUG,"Connecting to %s:%u.",address,port);
 404
 405   if(connect(s,(struct sockaddr *)&dest_addr,sizeof(dest_addr)) < 0) {
 406     if(!ERRNO_CONN_EINPROGRESS(errno)) {
 407       /* yuck. kill it. */
 408       log_fn(LOG_INFO,"Connect() to %s:%u failed: %s",address,port,strerror(errno));
 409       close(s);
 410       return -1;
 411     } else {
 412       /* it's in progress. set state appropriately and return. */
 413       conn->s = s;
 414       log_fn(LOG_DEBUG,"connect in progress, socket %d.",s);
 415       return 0;
 416     }
 417   }
 418
 419   /* it succeeded. we're connected. */
 420   log_fn(LOG_INFO,"Connection to %s:%u established.",address,port);
 421   conn->s = s;
 422   return 1;
 423 }
 424
 425 static void listener_close_if_present(int type) {
 426   connection_t *conn;
 427   assert(type == CONN_TYPE_OR_LISTENER ||
 428          type == CONN_TYPE_AP_LISTENER ||
 429          type == CONN_TYPE_DIR_LISTENER);
 430   conn = connection_get_by_type(type);
 431   if (conn) {
 432     connection_close_immediate(conn);
 433     connection_mark_for_close(conn,0);
 434   }
 435 }
 436
 437 /* start all connections that should be up but aren't */
 438 int retry_all_connections(void) {
 439
 440   if(options.ORPort) {
 441     router_retry_connections();
 442   }
 443
 444   if(options.ORPort) {
 445     listener_close_if_present(CONN_TYPE_OR_LISTENER);
 446     if(connection_create_listener(options.ORBindAddress,
 447                                   (uint16_t) options.ORPort,
 448                                   CONN_TYPE_OR_LISTENER) < 0)
 449       return -1;
 450   }
 451
 452   if(options.DirPort) {
 453     listener_close_if_present(CONN_TYPE_DIR_LISTENER);
 454     if(connection_create_listener(options.DirBindAddress,
 455                                   (uint16_t) options.DirPort,
 456                                   CONN_TYPE_DIR_LISTENER) < 0)
 457       return -1;
 458   }
 459
 460   if(options.SocksPort) {
 461     listener_close_if_present(CONN_TYPE_AP_LISTENER);
 462     if(connection_create_listener(options.SocksBindAddress,
 463                                   (uint16_t) options.SocksPort,
 464                                   CONN_TYPE_AP_LISTENER) < 0)
 465       return -1;
 466   }
 467
 468   return 0;
 469 }
 470
 471 extern int global_read_bucket;
 472
 473 /* how many bytes at most can we read onto this connection? */
 474 int connection_bucket_read_limit(connection_t *conn) {
 475   int at_most;
 476
 477   if(options.LinkPadding) {
 478     at_most = global_read_bucket;
 479   } else {
 480     /* do a rudimentary round-robin so one circuit can't hog a connection */
 481     if(connection_speaks_cells(conn)) {
 482       at_most = 32*(CELL_NETWORK_SIZE);
 483     } else {
 484       at_most = 32*(RELAY_PAYLOAD_SIZE);
 485     }
 486
 487     if(at_most > global_read_bucket)
 488       at_most = global_read_bucket;
 489   }
 490
 491   if(connection_speaks_cells(conn) && conn->state == OR_CONN_STATE_OPEN)
 492     if(at_most > conn->receiver_bucket)
 493       at_most = conn->receiver_bucket;
 494
 495   return at_most;
 496 }
 497
 498 /* we just read num_read onto conn. Decrement buckets appropriately. */
 499 void connection_bucket_decrement(connection_t *conn, int num_read) {
 500   global_read_bucket -= num_read; assert(global_read_bucket >= 0);
 501   if(connection_speaks_cells(conn) && conn->state == OR_CONN_STATE_OPEN) {
 502     conn->receiver_bucket -= num_read; assert(conn->receiver_bucket >= 0);
 503   }
 504   if(global_read_bucket == 0) {
 505     log_fn(LOG_DEBUG,"global bucket exhausted. Pausing.");
 506     conn->wants_to_read = 1;
 507     connection_stop_reading(conn);
 508     return;
 509   }
 510   if(connection_speaks_cells(conn) &&
 511      conn->state == OR_CONN_STATE_OPEN &&
 512      conn->receiver_bucket == 0) {
 513       log_fn(LOG_DEBUG,"receiver bucket exhausted. Pausing.");
 514       conn->wants_to_read = 1;
 515       connection_stop_reading(conn);
 516   }
 517 }
 518
 519 /* keep a timeval to know when time has passed enough to refill buckets */
 520 static struct timeval current_time;
 521
 522 void connection_bucket_init(void) {
 523   tor_gettimeofday(&current_time);
 524   global_read_bucket = options.BandwidthBurst; /* start it at max traffic */
 525 }
 526
 527 /* some time has passed; increment buckets appropriately. */
 528 void connection_bucket_refill(struct timeval *now) {
 529   int i, n;
 530   connection_t *conn;
 531   connection_t **carray;
 532
 533   if(now->tv_sec <= current_time.tv_sec)
 534     return; /* wait until the second has rolled over */
 535
 536   current_time.tv_sec = now->tv_sec; /* update current_time */
 537   /* (ignore usecs for now) */
 538
 539   /* refill the global bucket */
 540   if(global_read_bucket < options.BandwidthBurst) {
 541     global_read_bucket += options.BandwidthRate;
 542     log_fn(LOG_DEBUG,"global_read_bucket now %d.", global_read_bucket);
 543   }
 544
 545   /* refill the per-connection buckets */
 546   get_connection_array(&carray,&n);
 547   for(i=0;i<n;i++) {
 548     conn = carray[i];
 549
 550     if(connection_receiver_bucket_should_increase(conn)) {
 551       conn->receiver_bucket += conn->bandwidth;
 552       //log_fn(LOG_DEBUG,"Receiver bucket %d now %d.", i, conn->receiver_bucket);
 553     }
 554
 555     if(conn->wants_to_read == 1 /* it's marked to turn reading back on now */
 556        && global_read_bucket > 0 /* and we're allowed to read */
 557        && (!connection_speaks_cells(conn) ||
 558            conn->state != OR_CONN_STATE_OPEN ||
 559            conn->receiver_bucket > 0)) {
 560       /* and either a non-cell conn or a cell conn with non-empty bucket */
 561       log_fn(LOG_DEBUG,"waking up conn (fd %d)",conn->s);
 562       conn->wants_to_read = 0;
 563       connection_start_reading(conn);
 564       if(conn->wants_to_write == 1) {
 565         conn->wants_to_write = 0;
 566         connection_start_writing(conn);
 567       }
 568     }
 569   }
 570 }
 571
 572 static int connection_receiver_bucket_should_increase(connection_t *conn) {
 573   assert(conn);
 574
 575   if(!connection_speaks_cells(conn))
 576     return 0; /* edge connections don't use receiver_buckets */
 577   if(conn->state != OR_CONN_STATE_OPEN)
 578     return 0; /* only open connections play the rate limiting game */
 579
 580   assert(conn->bandwidth > 0);
 581   if(conn->receiver_bucket > 9*conn->bandwidth)
 582     return 0;
 583
 584   return 1;
 585 }
 586
 587 int connection_handle_read(connection_t *conn) {
 588
 589   conn->timestamp_lastread = time(NULL);
 590
 591   switch(conn->type) {
 592     case CONN_TYPE_OR_LISTENER:
 593       return connection_handle_listener_read(conn, CONN_TYPE_OR);
 594     case CONN_TYPE_AP_LISTENER:
 595       return connection_handle_listener_read(conn, CONN_TYPE_AP);
 596     case CONN_TYPE_DIR_LISTENER:
 597       return connection_handle_listener_read(conn, CONN_TYPE_DIR);
 598   }
 599
 600   if(connection_read_to_buf(conn) < 0) {
 601     if(conn->type == CONN_TYPE_DIR &&
 602        conn->state == DIR_CONN_STATE_CONNECTING) {
 603        /* it's a directory server and connecting failed: forget about this router */
 604        /* XXX I suspect pollerr may make Windows not get to this point. :( */
 605        router_mark_as_down(conn->nickname);
 606     }
 607     /* There's a read error; kill the connection.*/
 608     connection_close_immediate(conn); /* Don't flush; connection is dead. */
 609     connection_mark_for_close(conn, END_STREAM_REASON_MISC);
 610     return -1;
 611   }
 612   if(connection_process_inbuf(conn) < 0) {
 613 //    log_fn(LOG_DEBUG,"connection_process_inbuf returned -1.");
 614     return -1;
 615   }
 616   return 0;
 617 }
 618
 619 /* return -1 if we want to break conn, else return 0 */
 620 int connection_read_to_buf(connection_t *conn) {
 621   int result;
 622   int at_most;
 623
 624   /* how many bytes are we allowed to read? */
 625   at_most = connection_bucket_read_limit(conn);
 626
 627   if(connection_speaks_cells(conn) && conn->state != OR_CONN_STATE_CONNECTING) {
 628     if(conn->state == OR_CONN_STATE_HANDSHAKING) {
 629       /* continue handshaking even if global token bucket is empty */
 630       return connection_tls_continue_handshake(conn);
 631     }
 632
 633     /* else open, or closing */
 634     result = read_to_buf_tls(conn->tls, at_most, conn->inbuf);
 635
 636     switch(result) {
 637       case TOR_TLS_ERROR:
 638       case TOR_TLS_CLOSE:
 639         log_fn(LOG_INFO,"tls error. breaking.");
 640         return -1; /* XXX deal with close better */
 641       case TOR_TLS_WANTWRITE:
 642         connection_start_writing(conn);
 643         return 0;
 644       case TOR_TLS_WANTREAD: /* we're already reading */
 645       case TOR_TLS_DONE: /* no data read, so nothing to process */
 646         result = 0;
 647         break; /* so we call bucket_decrement below */
 648     }
 649   } else {
 650     result = read_to_buf(conn->s, at_most, conn->inbuf,
 651                          &conn->inbuf_reached_eof);
 652
 653 //  log(LOG_DEBUG,"connection_read_to_buf(): read_to_buf returned %d.",read_result);
 654
 655     if(result < 0)
 656       return -1;
 657   }
 658
 659   connection_bucket_decrement(conn, result);
 660   return 0;
 661 }
 662
 663 int connection_fetch_from_buf(char *string, int len, connection_t *conn) {
 664   return fetch_from_buf(string, len, conn->inbuf);
 665 }
 666
 667 int connection_find_on_inbuf(char *string, int len, connection_t *conn) {
 668   return find_on_inbuf(string, len, conn->inbuf);
 669 }
 670
 671 int connection_wants_to_flush(connection_t *conn) {
 672   return conn->outbuf_flushlen;
 673 }
 674
 675 int connection_outbuf_too_full(connection_t *conn) {
 676   return (conn->outbuf_flushlen > 10*CELL_PAYLOAD_SIZE);
 677 }
 678
 679 /* return -1 if you want to break the conn, else return 0 */
 680 int connection_handle_write(connection_t *conn) {
 681
 682   assert(!connection_is_listener(conn));
 683
 684   conn->timestamp_lastwritten = time(NULL);
 685
 686   if (connection_speaks_cells(conn) &&
 687       conn->state != OR_CONN_STATE_CONNECTING) {
 688     if (conn->state == OR_CONN_STATE_HANDSHAKING) {
 689       connection_stop_writing(conn);
 690       if(connection_tls_continue_handshake(conn) < 0) {
 691         connection_close_immediate(conn); /* Don't flush; connection is dead. */
 692         connection_mark_for_close(conn, 0);
 693         return -1;
 694       }
 695       return 0;
 696     }
 697
 698     /* else open, or closing */
 699     switch(flush_buf_tls(conn->tls, conn->outbuf, &conn->outbuf_flushlen)) {
 700       case TOR_TLS_ERROR:
 701       case TOR_TLS_CLOSE:
 702         log_fn(LOG_INFO,"tls error. breaking.");
 703         connection_close_immediate(conn); /* Don't flush; connection is dead. */
 704         connection_mark_for_close(conn, 0);
 705         return -1; /* XXX deal with close better */
 706       case TOR_TLS_WANTWRITE:
 707         log_fn(LOG_DEBUG,"wanted write.");
 708         /* we're already writing */
 709         return 0;
 710       case TOR_TLS_WANTREAD:
 711         /* Make sure to avoid a loop if the receive buckets are empty. */
 712         log_fn(LOG_DEBUG,"wanted read.");
 713         if(!connection_is_reading(conn)) {
 714           connection_stop_writing(conn);
 715           conn->wants_to_write = 1;
 716           /* we'll start reading again when the next second arrives,
 717            * and then also start writing again.
 718            */
 719         }
 720         /* else no problem, we're already reading */
 721         return 0;
 722       /* case TOR_TLS_DONE:
 723        * for TOR_TLS_DONE, fall through to check if the flushlen
 724        * is empty, so we can stop writing.
 725        */
 726     }
 727   } else {
 728     if (flush_buf(conn->s, conn->outbuf, &conn->outbuf_flushlen) < 0) {
 729       connection_close_immediate(conn); /* Don't flush; connection is dead. */
 730       connection_mark_for_close(conn, END_STREAM_REASON_MISC);
 731       return -1;
 732     }
 733     /* conns in CONNECTING state will fall through... */
 734   }
 735
 736   if(!connection_wants_to_flush(conn)) /* it's done flushing */
 737     if(connection_finished_flushing(conn) < 0) /* ...and get handled here. */
 738       return -1;
 739
 740   return 0;
 741 }
 742
 743 void connection_write_to_buf(const char *string, int len, connection_t *conn) {
 744
 745   if(!len || conn->marked_for_close)
 746     return;
 747
 748   if(write_to_buf(string, len, conn->outbuf) < 0) {
 749     log_fn(LOG_WARN,"write_to_buf failed. Closing connection (fd %d).", conn->s);
 750     connection_mark_for_close(conn,0);
 751     return;
 752   }
 753
 754   connection_start_writing(conn);
 755 #define MIN_TLS_FLUSHLEN 15872
 756 /* openssl tls record size is 16383, this is close. The goal here is to
 757  * push data out as soon as we know there's enough for a tls record, so
 758  * during periods of high load we won't read the entire megabyte from
 759  * input before pushing any data out. */
 760   if(connection_speaks_cells(conn) &&
 761      conn->outbuf_flushlen < MIN_TLS_FLUSHLEN &&
 762      conn->outbuf_flushlen+len >= MIN_TLS_FLUSHLEN) {
 763     len -= (MIN_TLS_FLUSHLEN - conn->outbuf_flushlen);
 764     conn->outbuf_flushlen = MIN_TLS_FLUSHLEN;
 765     if(connection_handle_write(conn) < 0) {
 766       log_fn(LOG_WARN,"flushing failed.");
 767       connection_mark_for_close(conn,0);
 768     }
 769   }
 770   if(len > 0) { /* if there's any left over */
 771     conn->outbuf_flushlen += len;
 772     connection_start_writing(conn);
 773     /* because connection_handle_write() above might have stopped writing */
 774   }
 775 }
 776
 777 connection_t *connection_exact_get_by_addr_port(uint32_t addr, uint16_t port) {
 778   int i, n;
 779   connection_t *conn;
 780   connection_t **carray;
 781
 782   get_connection_array(&carray,&n);
 783   for(i=0;i<n;i++) {
 784     conn = carray[i];
 785     if(conn->addr == addr && conn->port == port && !conn->marked_for_close)
 786       return conn;
 787   }
 788   return NULL;
 789 }
 790
 791 connection_t *connection_twin_get_by_addr_port(uint32_t addr, uint16_t port) {
 792   /* Find a connection to the router described by addr and port,
 793    *   or alternately any router which knows its key.
 794    * This connection *must* be in 'open' state.
 795    * If not, return NULL.
 796    */
 797   int i, n;
 798   connection_t *conn;
 799   routerinfo_t *router;
 800   connection_t **carray;
 801
 802   /* first check if it's there exactly */
 803   conn = connection_exact_get_by_addr_port(addr,port);
 804   if(conn && connection_state_is_open(conn)) {
 805     log(LOG_DEBUG,"connection_twin_get_by_addr_port(): Found exact match.");
 806     return conn;
 807   }
 808
 809   /* now check if any of the other open connections are a twin for this one */
 810
 811   router = router_get_by_addr_port(addr,port);
 812   if(!router)
 813     return NULL;
 814
 815   get_connection_array(&carray,&n);
 816   for(i=0;i<n;i++) {
 817     conn = carray[i];
 818     assert(conn);
 819     if(connection_state_is_open(conn) &&
 820        !crypto_pk_cmp_keys(conn->onion_pkey, router->onion_pkey)) {
 821       log(LOG_DEBUG,"connection_twin_get_by_addr_port(): Found twin (%s).",conn->address);
 822       return conn;
 823     }
 824   }
 825   return NULL;
 826 }
 827
 828 connection_t *connection_get_by_type(int type) {
 829   int i, n;
 830   connection_t *conn;
 831   connection_t **carray;
 832
 833   get_connection_array(&carray,&n);
 834   for(i=0;i<n;i++) {
 835     conn = carray[i];
 836     if(conn->type == type && !conn->marked_for_close)
 837       return conn;
 838   }
 839   return NULL;
 840 }
 841
 842 connection_t *connection_get_by_type_state(int type, int state) {
 843   int i, n;
 844   connection_t *conn;
 845   connection_t **carray;
 846
 847   get_connection_array(&carray,&n);
 848   for(i=0;i<n;i++) {
 849     conn = carray[i];
 850     if(conn->type == type && conn->state == state && !conn->marked_for_close)
 851       return conn;
 852   }
 853   return NULL;
 854 }
 855
 856 connection_t *connection_get_by_type_state_lastwritten(int type, int state) {
 857   int i, n;
 858   connection_t *conn, *best=NULL;
 859   connection_t **carray;
 860
 861   get_connection_array(&carray,&n);
 862   for(i=0;i<n;i++) {
 863     conn = carray[i];
 864     if(conn->type == type && conn->state == state && !conn->marked_for_close)
 865       if(!best || conn->timestamp_lastwritten < best->timestamp_lastwritten)
 866         best = conn;
 867   }
 868   return best;
 869 }
 870
 871 connection_t *connection_get_by_type_rendquery(int type, char *rendquery) {
 872   int i, n;
 873   connection_t *conn;
 874   connection_t **carray;
 875
 876   get_connection_array(&carray,&n);
 877   for(i=0;i<n;i++) {
 878     conn = carray[i];
 879     if(conn->type == type &&
 880        !conn->marked_for_close &&
 881        !rend_cmp_service_ids(rendquery, conn->rend_query))
 882       return conn;
 883   }
 884   return NULL;
 885 }
 886
 887 int connection_is_listener(connection_t *conn) {
 888   if(conn->type == CONN_TYPE_OR_LISTENER ||
 889      conn->type == CONN_TYPE_AP_LISTENER ||
 890      conn->type == CONN_TYPE_DIR_LISTENER)
 891     return 1;
 892   return 0;
 893 }
 894
 895 int connection_state_is_open(connection_t *conn) {
 896   assert(conn);
 897
 898   if(conn->marked_for_close)
 899     return 0;
 900
 901   if((conn->type == CONN_TYPE_OR && conn->state == OR_CONN_STATE_OPEN) ||
 902      (conn->type == CONN_TYPE_AP && conn->state == AP_CONN_STATE_OPEN) ||
 903      (conn->type == CONN_TYPE_EXIT && conn->state == EXIT_CONN_STATE_OPEN))
 904     return 1;
 905
 906   return 0;
 907 }
 908
 909 int connection_send_destroy(uint16_t circ_id, connection_t *conn) {
 910   cell_t cell;
 911
 912   assert(conn);
 913   assert(connection_speaks_cells(conn));
 914
 915   memset(&cell, 0, sizeof(cell_t));
 916   cell.circ_id = circ_id;
 917   cell.command = CELL_DESTROY;
 918   log_fn(LOG_INFO,"Sending destroy (circID %d).", circ_id);
 919   connection_or_write_cell_to_buf(&cell, conn);
 920   return 0;
 921 }
 922
 923 int connection_process_inbuf(connection_t *conn) {
 924
 925   assert(conn);
 926
 927   switch(conn->type) {
 928     case CONN_TYPE_OR:
 929       return connection_or_process_inbuf(conn);
 930     case CONN_TYPE_EXIT:
 931     case CONN_TYPE_AP:
 932       return connection_edge_process_inbuf(conn);
 933     case CONN_TYPE_DIR:
 934       return connection_dir_process_inbuf(conn);
 935     case CONN_TYPE_DNSWORKER:
 936       return connection_dns_process_inbuf(conn);
 937     case CONN_TYPE_CPUWORKER:
 938       return connection_cpu_process_inbuf(conn);
 939     default:
 940       log_fn(LOG_WARN,"got unexpected conn->type %d.", conn->type);
 941       return -1;
 942   }
 943 }
 944
 945 int connection_finished_flushing(connection_t *conn) {
 946
 947   assert(conn);
 948
 949 //  log_fn(LOG_DEBUG,"entered. Socket %u.", conn->s);
 950
 951   switch(conn->type) {
 952     case CONN_TYPE_OR:
 953       return connection_or_finished_flushing(conn);
 954     case CONN_TYPE_AP:
 955     case CONN_TYPE_EXIT:
 956       return connection_edge_finished_flushing(conn);
 957     case CONN_TYPE_DIR:
 958       return connection_dir_finished_flushing(conn);
 959     case CONN_TYPE_DNSWORKER:
 960       return connection_dns_finished_flushing(conn);
 961     case CONN_TYPE_CPUWORKER:
 962       return connection_cpu_finished_flushing(conn);
 963     default:
 964       log_fn(LOG_WARN,"got unexpected conn->type %d.", conn->type);
 965       return -1;
 966   }
 967 }
 968
 969 void assert_connection_ok(connection_t *conn, time_t now)
 970 {
 971   assert(conn);
 972   assert(conn->magic == CONNECTION_MAGIC);
 973   assert(conn->type >= _CONN_TYPE_MIN);
 974   assert(conn->type <= _CONN_TYPE_MAX);
 975
 976   if(conn->outbuf_flushlen > 0) {
 977     assert(connection_is_writing(conn) || conn->wants_to_write);
 978   }
 979
 980   if(conn->hold_open_until_flushed)
 981     assert(conn->marked_for_close);
 982
 983   /* XXX check: wants_to_read, wants_to_write, s, poll_index,
 984    * marked_for_close. */
 985
 986   /* buffers */
 987   if (!connection_is_listener(conn)) {
 988     assert_buf_ok(conn->inbuf);
 989     assert_buf_ok(conn->outbuf);
 990   }
 991
 992 #if 0 /* computers often go back in time; no way to know */
 993   assert(!now || conn->timestamp_lastread <= now);
 994   assert(!now || conn->timestamp_lastwritten <= now);
 995   assert(conn->timestamp_created <= conn->timestamp_lastread);
 996   assert(conn->timestamp_created <= conn->timestamp_lastwritten);
 997 #endif
 998
 999   /* XXX Fix this; no longer so.*/
1000 #if 0
1001   if(conn->type != CONN_TYPE_OR && conn->type != CONN_TYPE_DIR)
1002     assert(!conn->pkey);
1003   /* pkey is set if we're a dir client, or if we're an OR in state OPEN
1004    * connected to another OR.
1005    */
1006 #endif
1007
1008   if (conn->type != CONN_TYPE_OR) {
1009     assert(!conn->tls);
1010   } else {
1011     if(conn->state == OR_CONN_STATE_OPEN) {
1012       /* assert(conn->bandwidth > 0); */
1013       /* the above isn't necessarily true: if we just did a TLS
1014        * handshake but we didn't recognize the other peer, or it
1015        * gave a bad cert/etc, then we won't have assigned bandwidth,
1016        * yet it will be open. -RD
1017        */
1018       assert(conn->receiver_bucket >= 0);
1019     }
1020     assert(conn->addr && conn->port);
1021     assert(conn->address);
1022     if (conn->state != OR_CONN_STATE_CONNECTING)
1023       assert(conn->tls);
1024   }
1025
1026   if (conn->type != CONN_TYPE_EXIT && conn->type != CONN_TYPE_AP) {
1027     assert(!conn->stream_id);
1028     assert(!conn->next_stream);
1029     assert(!conn->cpath_layer);
1030     assert(!conn->package_window);
1031     assert(!conn->deliver_window);
1032     assert(!conn->done_sending);
1033     assert(!conn->done_receiving);
1034   } else {
1035     /* XXX unchecked: package window, deliver window. */
1036   }
1037   if (conn->type == CONN_TYPE_AP) {
1038     assert(conn->socks_request);
1039     if (conn->state == AP_CONN_STATE_OPEN) {
1040       assert(conn->socks_request->has_finished);
1041       assert(conn->cpath_layer);
1042       assert_cpath_layer_ok(conn->cpath_layer);
1043     }
1044   } else {
1045     assert(!conn->socks_request);
1046   }
1047   if(conn->type != CONN_TYPE_DIR) {
1048     assert(!conn->purpose); /* only used for dir types currently */
1049   }
1050
1051   switch(conn->type)
1052     {
1053     case CONN_TYPE_OR_LISTENER:
1054     case CONN_TYPE_AP_LISTENER:
1055     case CONN_TYPE_DIR_LISTENER:
1056       assert(conn->state == LISTENER_STATE_READY);
1057       break;
1058     case CONN_TYPE_OR:
1059       assert(conn->state >= _OR_CONN_STATE_MIN &&
1060              conn->state <= _OR_CONN_STATE_MAX);
1061       break;
1062     case CONN_TYPE_EXIT:
1063       assert(conn->state >= _EXIT_CONN_STATE_MIN &&
1064              conn->state <= _EXIT_CONN_STATE_MAX);
1065       break;
1066     case CONN_TYPE_AP:
1067       assert(conn->state >= _AP_CONN_STATE_MIN &&
1068              conn->state <= _AP_CONN_STATE_MAX);
1069       assert(conn->socks_request);
1070       break;
1071     case CONN_TYPE_DIR:
1072       assert(conn->state >= _DIR_CONN_STATE_MIN &&
1073              conn->state <= _DIR_CONN_STATE_MAX);
1074       assert(conn->purpose >= _DIR_PURPOSE_MIN &&
1075              conn->purpose <= _DIR_PURPOSE_MAX);
1076       break;
1077     case CONN_TYPE_DNSWORKER:
1078       assert(conn->state == DNSWORKER_STATE_IDLE ||
1079              conn->state == DNSWORKER_STATE_BUSY);
1080       break;
1081     case CONN_TYPE_CPUWORKER:
1082       assert(conn->state >= _CPUWORKER_STATE_MIN &&
1083              conn->state <= _CPUWORKER_STATE_MAX);
1084       break;
1085     default:
1086       assert(0);
1087   }
1088 }
1089
1090 /*
1091   Local Variables:
1092   mode:c
1093   indent-tabs-mode:nil
1094   c-basic-offset:2
1095   End:
1096 */