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