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