connection_edge.c

/* Copyright 2001,2002,2003 Roger Dingledine, Matej Pfajfar. */
/* See LICENSE for licensing information */
/* $Id$ */

#include "or.h"
#include "tree.h"

extern or_options_t options; /* command-line and config-file options */
extern char *conn_state_to_string[][_CONN_TYPE_MAX+1];

static int connection_ap_handshake_process_socks(connection_t *conn);

static int connection_exit_begin_conn(cell_t *cell, circuit_t *circ);
static void connection_edge_consider_sending_sendme(connection_t *conn);

static uint32_t client_dns_lookup_entry(const char *address);
static void client_dns_set_entry(const char *address, uint32_t val);

void relay_header_pack(char *dest, const relay_header_t *src) {
  *(uint8_t*)(dest) = src->command;

  set_uint16(dest+1, htons(src->recognized));
  set_uint16(dest+3, htons(src->stream_id));
  memcpy(dest+5, src->integrity, 4);
  set_uint16(dest+9, htons(src->length));
}

void relay_header_unpack(relay_header_t *dest, const char *src) {
  dest->command = *(uint8_t*)(src);

  dest->recognized = ntohs(get_uint16(src+1));
  dest->stream_id = ntohs(get_uint16(src+3));
  memcpy(dest->integrity, src+5, 4);
  dest->length = ntohs(get_uint16(src+9));
}

/* mark and return -1 if there was an unexpected error with the conn,
 * else return 0.
 */
int connection_edge_process_inbuf(connection_t *conn) {

  assert(conn);
  assert(conn->type == CONN_TYPE_AP || conn->type == CONN_TYPE_EXIT);

  if(conn->inbuf_reached_eof) {
#ifdef HALF_OPEN
    /* eof reached; we're done reading, but we might want to write more. */
    conn->done_receiving = 1;
    shutdown(conn->s, 0); /* XXX check return, refactor NM */
    if (conn->done_sending) {
      connection_mark_for_close(conn, END_STREAM_REASON_DONE);
    } else {
      connection_edge_send_command(conn, circuit_get_by_conn(conn), RELAY_COMMAND_END,
                                   NULL, 0, conn->cpath_layer);
    }
    return 0;
#else
    /* eof reached, kill it. */
    log_fn(LOG_INFO,"conn (fd %d) reached eof. Closing.", conn->s);
    connection_mark_for_close(conn, END_STREAM_REASON_DONE);
    conn->hold_open_until_flushed = 1; /* just because we shouldn't read
                                          doesn't mean we shouldn't write */
    return 0;
#endif
  }

  switch(conn->state) {
    case AP_CONN_STATE_SOCKS_WAIT:
      if(connection_ap_handshake_process_socks(conn) < 0) {
        connection_mark_for_close(conn, END_STREAM_REASON_MISC);
        conn->hold_open_until_flushed = 1;
        return -1;
      }
      return 0;
    case AP_CONN_STATE_OPEN:
    case EXIT_CONN_STATE_OPEN:
      if(conn->package_window <= 0) {
        /* XXX this is still getting called rarely :( */
        log_fn(LOG_WARN,"called with package_window %d. Tell Roger.", conn->package_window);
        return 0;
      }
      if(connection_edge_package_raw_inbuf(conn) < 0) {
        connection_mark_for_close(conn, END_STREAM_REASON_MISC);
        return -1;
      }
      return 0;
    case EXIT_CONN_STATE_CONNECTING:
    case AP_CONN_STATE_RENDDESC_WAIT:
    case AP_CONN_STATE_CIRCUIT_WAIT:
    case AP_CONN_STATE_CONNECT_WAIT:
      log_fn(LOG_INFO,"data from edge while in '%s' state. Leaving it on buffer.",
                      conn_state_to_string[conn->type][conn->state]);
      return 0;
  }
  log_fn(LOG_WARN,"Got unexpected state %d. Closing.",conn->state);
  connection_mark_for_close(conn, END_STREAM_REASON_MISC);
  return -1;
}

int connection_edge_destroy(uint16_t circ_id, connection_t *conn) {
  assert(conn->type == CONN_TYPE_AP || conn->type == CONN_TYPE_EXIT);

  if(conn->marked_for_close)
    return 0; /* already marked; probably got an 'end' */
  log_fn(LOG_INFO,"CircID %d: At an edge. Marking connection for close.",
         circ_id);
  conn->has_sent_end = 1; /* we're closing the circuit, nothing to send to */
  connection_mark_for_close(conn, END_STREAM_REASON_DESTROY);
  conn->hold_open_until_flushed = 1;
  return 0;
}

static char *connection_edge_end_reason(char *payload, uint16_t length) {
  if(length < 1) {
    log_fn(LOG_WARN,"End cell arrived with length 0. Should be at least 1.");
    return "MALFORMED";
  }
  if(*payload < _MIN_END_STREAM_REASON || *payload > _MAX_END_STREAM_REASON) {
    log_fn(LOG_WARN,"Reason for ending (%d) not recognized.",*payload);
    return "MALFORMED";
  }
  switch(*payload) {
    case END_STREAM_REASON_MISC:           return "misc error";
    case END_STREAM_REASON_RESOLVEFAILED:  return "resolve failed";
    case END_STREAM_REASON_CONNECTFAILED:  return "connect failed";
    case END_STREAM_REASON_EXITPOLICY:     return "exit policy failed";
    case END_STREAM_REASON_DESTROY:        return "destroyed";
    case END_STREAM_REASON_DONE:           return "closed normally";
    case END_STREAM_REASON_TIMEOUT:        return "gave up (timeout)";
  }
  assert(0);
  return "";
}

int connection_edge_end(connection_t *conn, char reason, crypt_path_t *cpath_layer) {
  char payload[5];
  int payload_len=1;
  circuit_t *circ;

  if(conn->has_sent_end) {
    log_fn(LOG_WARN,"It appears I've already sent the end. Are you calling me twice?");
    return -1;
  }

  payload[0] = reason;
  if(reason == END_STREAM_REASON_EXITPOLICY) {
    /* this is safe even for rend circs, because they never fail
     * because of exitpolicy */
    set_uint32(payload+1, htonl(conn->addr));
    payload_len += 4;
  }

  circ = circuit_get_by_conn(conn);
  if(circ && !circ->marked_for_close) {
    log_fn(LOG_DEBUG,"Marking conn (fd %d) and sending end.",conn->s);
    connection_edge_send_command(conn, circ, RELAY_COMMAND_END,
                                 payload, payload_len, cpath_layer);
  } else {
    log_fn(LOG_DEBUG,"Marking conn (fd %d); no circ to send end.",conn->s);
  }

  conn->has_sent_end = 1;
  return 0;
}

/* Make a relay cell out of 'relay_command' and 'payload', and
 * send it onto the open circuit 'circ'. If it's a control cell,
 * set fromconn to NULL, else it's the stream that's sending the
 * relay cell. Use cpath_layer NULL if you're responding to the OP;
 * If it's an outgoing cell it must specify the destination hop.
 *
 * If you can't send the cell, mark the circuit for close and
 * return -1. Else return 0.
 */
int connection_edge_send_command(connection_t *fromconn, circuit_t *circ,
                                 int relay_command, const char *payload,
                                 int payload_len, crypt_path_t *cpath_layer) {
  cell_t cell;
  relay_header_t rh;
  int cell_direction;

  if(!circ) {
    log_fn(LOG_WARN,"no circ. Closing conn.");
    assert(fromconn);
    connection_mark_for_close(fromconn, 0);
    return -1;
  }

  memset(&cell, 0, sizeof(cell_t));
  cell.command = CELL_RELAY;
  if(cpath_layer) {
    cell.circ_id = circ->n_circ_id;
    cell_direction = CELL_DIRECTION_OUT;
  } else {
    cell.circ_id = circ->p_circ_id;
    cell_direction = CELL_DIRECTION_IN;
  }

  memset(&rh, 0, sizeof(rh));
  rh.command = relay_command;
  if(fromconn)
    rh.stream_id = fromconn->stream_id; /* else it's 0 */
  rh.length = payload_len;
  relay_header_pack(cell.payload, &rh);
  if(payload_len)
    memcpy(cell.payload+RELAY_HEADER_SIZE, payload, payload_len);

  log_fn(LOG_DEBUG,"delivering %d cell %s.", relay_command,
         cell_direction == CELL_DIRECTION_OUT ? "forward" : "backward");

  if(circuit_package_relay_cell(&cell, circ, cell_direction, cpath_layer) < 0) {
    log_fn(LOG_WARN,"circuit_package_relay_cell failed. Closing.");
    circuit_mark_for_close(circ);
    return -1;
  }
  return 0;
}

int connection_edge_process_relay_cell_not_open(
    relay_header_t *rh, cell_t *cell, circuit_t *circ,
    connection_t *conn, crypt_path_t *layer_hint) {
  uint32_t addr;

  if(rh->command == RELAY_COMMAND_END) {
    log_fn(LOG_INFO,"Edge got end (%s) before we're connected. Marking for close.",
      connection_edge_end_reason(cell->payload+RELAY_HEADER_SIZE, rh->length));
    if(CIRCUIT_IS_ORIGIN(circ))
      circuit_log_path(LOG_INFO,circ);
    conn->has_sent_end = 1; /* we just got an 'end', don't need to send one */
    connection_mark_for_close(conn, 0);
    /* XXX This is where we should check if reason is EXITPOLICY, and reattach */
    /* XXX here we should send a socks reject back if necessary, and hold
     * open til flushed */
    return 0;
  }
  if(conn->type == CONN_TYPE_AP && rh->command == RELAY_COMMAND_CONNECTED) {
    if(conn->state != AP_CONN_STATE_CONNECT_WAIT) {
      log_fn(LOG_WARN,"Got 'connected' while not in state connect_wait. Dropping.");
      return 0;
    }
//    log_fn(LOG_INFO,"Connected! Notifying application.");
    conn->state = AP_CONN_STATE_OPEN;
    if (rh->length >= 4) {
      addr = ntohl(get_uint32(cell->payload+RELAY_HEADER_SIZE));
      client_dns_set_entry(conn->socks_request->address, addr);
    }
    log_fn(LOG_INFO,"'connected' received after %d seconds.",
           (int)(time(NULL) - conn->timestamp_lastread));
    circuit_log_path(LOG_INFO,circ);
    connection_ap_handshake_socks_reply(conn, NULL, 0, 1);
    conn->socks_request->has_finished = 1;
    /* handle anything that might have queued */
    if (connection_edge_package_raw_inbuf(conn) < 0) {
      connection_mark_for_close(conn, END_STREAM_REASON_MISC);
      return 0;
    }
    return 0;
  } else {
    log_fn(LOG_WARN,"Got an unexpected relay command %d, in state %d (%s). Closing.",
           rh->command, conn->state, conn_state_to_string[conn->type][conn->state]);
    connection_mark_for_close(conn, END_STREAM_REASON_MISC);
    return -1;
  }
}

/* an incoming relay cell has arrived. return -1 if you want to tear down the
 * circuit, else 0. */
int connection_edge_process_relay_cell(cell_t *cell, circuit_t *circ,
                                       connection_t *conn,
                                       crypt_path_t *layer_hint) {
  static int num_seen=0;
  uint32_t addr;
  relay_header_t rh;

  assert(cell && circ);

  relay_header_unpack(&rh, cell->payload);
//  log_fn(LOG_DEBUG,"command %d stream %d", rh.command, rh.stream_id);
  num_seen++;
  log_fn(LOG_DEBUG,"Now seen %d relay cells here.", num_seen);

  /* either conn is NULL, in which case we've got a control cell, or else
   * conn points to the recognized stream. */

  if(conn &&
     conn->state != AP_CONN_STATE_OPEN &&
     conn->state != EXIT_CONN_STATE_OPEN) {
    return connection_edge_process_relay_cell_not_open(
             &rh, cell, circ, conn, layer_hint);
  }

  switch(rh.command) {
    case RELAY_COMMAND_DROP:
      log_fn(LOG_INFO,"Got a relay-level padding cell. Dropping.");
      return 0;
    case RELAY_COMMAND_BEGIN:
      if (layer_hint &&
          circ->purpose != CIRCUIT_PURPOSE_S_REND_JOINED) {
        log_fn(LOG_WARN,"relay begin request unsupported at AP. Dropping.");
        return 0;
      }
      if(conn) {
        log_fn(LOG_WARN,"begin cell for known stream. Dropping.");
        return 0;
      }
      connection_exit_begin_conn(cell, circ);
      return 0;
    case RELAY_COMMAND_DATA:
      ++stats_n_data_cells_received;
      if((layer_hint && --layer_hint->deliver_window < 0) ||
         (!layer_hint && --circ->deliver_window < 0)) {
        log_fn(LOG_WARN,"(relay data) circ deliver_window below 0. Killing.");
        connection_mark_for_close(conn, END_STREAM_REASON_MISC);
        return -1;
      }
      log_fn(LOG_DEBUG,"circ deliver_window now %d.", layer_hint ?
             layer_hint->deliver_window : circ->deliver_window);

      circuit_consider_sending_sendme(circ, layer_hint);

      if(!conn) {
        log_fn(LOG_INFO,"data cell dropped, unknown stream.");
        return 0;
      }

      if(--conn->deliver_window < 0) { /* is it below 0 after decrement? */
        log_fn(LOG_WARN,"(relay data) conn deliver_window below 0. Killing.");
        return -1; /* somebody's breaking protocol. kill the whole circuit. */
      }

      stats_n_data_bytes_received += rh.length;
      connection_write_to_buf(cell->payload + RELAY_HEADER_SIZE,
                              rh.length, conn);
      connection_edge_consider_sending_sendme(conn);
      return 0;
    case RELAY_COMMAND_END:
      if(!conn) {
        log_fn(LOG_INFO,"end cell (%s) dropped, unknown stream.",
          connection_edge_end_reason(cell->payload+RELAY_HEADER_SIZE, rh.length));
        return 0;
      }
      if(rh.length >= 5 &&
         *(cell->payload+RELAY_HEADER_SIZE) == END_STREAM_REASON_EXITPOLICY) {

        /* XXX this will never be reached, since we're in 'connect_wait' state
         * but this is only reached when we're in 'open' state. Put it higher. */

        /* No need to close the connection. We'll hold it open while
         * we try a new exit node.
         * cell->payload+RELAY_HEADER_SIZE+1 holds the destination addr.
         */
        addr = ntohl(get_uint32(cell->payload+RELAY_HEADER_SIZE+1));
        client_dns_set_entry(conn->socks_request->address, addr);
        conn->state = AP_CONN_STATE_CIRCUIT_WAIT;
        if(connection_ap_handshake_attach_circuit(conn) >= 0)
          return 0;
        /* else, conn will get closed below */
      }
/* XXX add to this log_fn the exit node's nickname? */
      log_fn(LOG_INFO,"end cell (%s) for stream %d. Removing stream.",
        connection_edge_end_reason(cell->payload+RELAY_HEADER_SIZE, rh.length),
        conn->stream_id);

#ifdef HALF_OPEN
      conn->done_sending = 1;
      shutdown(conn->s, 1); /* XXX check return; refactor NM */
      if (conn->done_receiving) {
        /* We just *got* an end; no reason to send one. */
        conn->has_sent_end = 1;
        connection_mark_for_close(conn, 0);
        conn->hold_open_until_flushed = 1;
      }
#else
      /* We just *got* an end; no reason to send one. */
      conn->has_sent_end = 1;
      connection_mark_for_close(conn, 0);
      conn->hold_open_until_flushed = 1;
#endif
      return 0;
    case RELAY_COMMAND_EXTEND:
      if(conn) {
        log_fn(LOG_WARN,"'extend' for non-zero stream. Dropping.");
        return 0;
      }
      return circuit_extend(cell, circ);
    case RELAY_COMMAND_EXTENDED:
      if(!layer_hint) {
        log_fn(LOG_WARN,"'extended' unsupported at non-origin. Dropping.");
        return 0;
      }
      log_fn(LOG_DEBUG,"Got an extended cell! Yay.");
      if(circuit_finish_handshake(circ, cell->payload+RELAY_HEADER_SIZE) < 0) {
        log_fn(LOG_WARN,"circuit_finish_handshake failed.");
        return -1;
      }
      if (circuit_send_next_onion_skin(circ)<0) {
        log_fn(LOG_INFO,"circuit_send_next_onion_skin() failed.");
        return -1;
      }
      return 0;
    case RELAY_COMMAND_TRUNCATE:
      if(layer_hint) {
        log_fn(LOG_WARN,"'truncate' unsupported at origin. Dropping.");
        return 0;
      }
      if(circ->n_conn) {
        connection_send_destroy(circ->n_circ_id, circ->n_conn);
        circ->n_conn = NULL;
      }
      log_fn(LOG_DEBUG, "Processed 'truncate', replying.");
      connection_edge_send_command(NULL, circ, RELAY_COMMAND_TRUNCATED,
                                   NULL, 0, NULL);
      return 0;
    case RELAY_COMMAND_TRUNCATED:
      if(!layer_hint) {
        log_fn(LOG_WARN,"'truncated' unsupported at non-origin. Dropping.");
        return 0;
      }
      circuit_truncated(circ, layer_hint);
      return 0;
    case RELAY_COMMAND_CONNECTED:
      if(conn) {
        log_fn(LOG_WARN,"'connected' unsupported while open. Closing circ.");
        return -1;
      }
      log_fn(LOG_INFO,"'connected' received, no conn attached anymore. Ignoring.");
      return 0;
    case RELAY_COMMAND_SENDME:
      if(!conn) {
        if(layer_hint) {
          layer_hint->package_window += CIRCWINDOW_INCREMENT;
          log_fn(LOG_DEBUG,"circ-level sendme at origin, packagewindow %d.",
                 layer_hint->package_window);
          circuit_resume_edge_reading(circ, layer_hint);
        } else {
          circ->package_window += CIRCWINDOW_INCREMENT;
          log_fn(LOG_DEBUG,"circ-level sendme at non-origin, packagewindow %d.",
                 circ->package_window);
          circuit_resume_edge_reading(circ, layer_hint);
        }
        return 0;
      }
      conn->package_window += STREAMWINDOW_INCREMENT;
      log_fn(LOG_DEBUG,"stream-level sendme, packagewindow now %d.", conn->package_window);
      connection_start_reading(conn);
      connection_edge_package_raw_inbuf(conn); /* handle whatever might still be on the inbuf */
      return 0;
    case RELAY_COMMAND_ESTABLISH_INTRO:
    case RELAY_COMMAND_ESTABLISH_RENDEZVOUS:
    case RELAY_COMMAND_INTRODUCE1:
    case RELAY_COMMAND_INTRODUCE2:
    case RELAY_COMMAND_RENDEZVOUS1:
    case RELAY_COMMAND_RENDEZVOUS2:
    case RELAY_COMMAND_INTRO_ESTABLISHED:
    case RELAY_COMMAND_RENDEZVOUS_ESTABLISHED:
      rend_process_relay_cell(circ, rh.command, rh.length,
                              cell->payload+RELAY_HEADER_SIZE);
      return 0;
  }
  log_fn(LOG_WARN,"unknown relay command %d.",rh.command);
  return -1;
}

int connection_edge_finished_flushing(connection_t *conn) {
  unsigned char connected_payload[4];
  int e, len=sizeof(e);

  assert(conn);
  assert(conn->type == CONN_TYPE_AP || conn->type == CONN_TYPE_EXIT);

  switch(conn->state) {
    case EXIT_CONN_STATE_CONNECTING:
      if (getsockopt(conn->s, SOL_SOCKET, SO_ERROR, (void*)&e, &len) < 0)  { /* not yet */
        if(!ERRNO_CONN_EINPROGRESS(errno)) {
          /* yuck. kill it. */
          log_fn(LOG_DEBUG,"in-progress exit connect failed. Removing.");
          connection_mark_for_close(conn, END_STREAM_REASON_CONNECTFAILED);
          return -1;
        } else {
          log_fn(LOG_DEBUG,"in-progress exit connect still waiting.");
          return 0; /* no change, see if next time is better */
        }
      }
      /* the connect has finished. */

      log_fn(LOG_INFO,"Exit connection to %s:%u established.",
          conn->address,conn->port);

      conn->state = EXIT_CONN_STATE_OPEN;
      connection_watch_events(conn, POLLIN); /* stop writing, continue reading */
      if(connection_wants_to_flush(conn)) /* in case there are any queued relay cells */
        connection_start_writing(conn);
      /* deliver a 'connected' relay cell back through the circuit. */
      if(connection_edge_is_rendezvous_stream(conn)) {
        if(connection_edge_send_command(conn, circuit_get_by_conn(conn),
           RELAY_COMMAND_CONNECTED, NULL, 0, conn->cpath_layer) < 0)
          return 0; /* circuit is closed, don't continue */
      } else {
        *(uint32_t*)connected_payload = htonl(conn->addr);
        if(connection_edge_send_command(conn, circuit_get_by_conn(conn),
           RELAY_COMMAND_CONNECTED, connected_payload, 4, conn->cpath_layer) < 0)
          return 0; /* circuit is closed, don't continue */
      }
      assert(conn->package_window > 0);
      return connection_edge_process_inbuf(conn); /* in case the server has written anything */
    case AP_CONN_STATE_OPEN:
    case EXIT_CONN_STATE_OPEN:
      connection_stop_writing(conn);
      connection_edge_consider_sending_sendme(conn);
      return 0;
    case AP_CONN_STATE_SOCKS_WAIT:
    case AP_CONN_STATE_RENDDESC_WAIT:
    case AP_CONN_STATE_CIRCUIT_WAIT:
    case AP_CONN_STATE_CONNECT_WAIT:
      connection_stop_writing(conn);
      return 0;
    default:
      log_fn(LOG_WARN,"BUG: called in unexpected state %d.", conn->state);
      return -1;
  }
  return 0;
}

uint64_t stats_n_data_cells_packaged = 0;
uint64_t stats_n_data_bytes_packaged = 0;
uint64_t stats_n_data_cells_received = 0;
uint64_t stats_n_data_bytes_received = 0;

int connection_edge_package_raw_inbuf(connection_t *conn) {
  int amount_to_process, length;
  char payload[CELL_PAYLOAD_SIZE];
  circuit_t *circ;

  assert(conn);
  assert(!connection_speaks_cells(conn));

repeat_connection_edge_package_raw_inbuf:

  circ = circuit_get_by_conn(conn);
  if(!circ) {
    log_fn(LOG_INFO,"conn has no circuits! Closing.");
    return -1;
  }

  if(circuit_consider_stop_edge_reading(circ, conn->cpath_layer))
    return 0;

  if(conn->package_window <= 0) {
    log_fn(LOG_WARN,"called with package_window %d. Tell Roger.", conn->package_window);
    connection_stop_reading(conn);
    return 0;
  }

  amount_to_process = buf_datalen(conn->inbuf);

  if(!amount_to_process)
    return 0;

  if(amount_to_process > RELAY_PAYLOAD_SIZE) {
    length = RELAY_PAYLOAD_SIZE;
  } else {
    length = amount_to_process;
  }
  stats_n_data_bytes_packaged += length;
  stats_n_data_cells_packaged += 1;

  connection_fetch_from_buf(payload, length, conn);

  log_fn(LOG_DEBUG,"(%d) Packaging %d bytes (%d waiting).", conn->s, length,
         (int)buf_datalen(conn->inbuf));

  if(connection_edge_send_command(conn, circ, RELAY_COMMAND_DATA,
                               payload, length, conn->cpath_layer) < 0)
    return 0; /* circuit is closed, don't continue */

  if(!conn->cpath_layer) { /* non-rendezvous exit */
    assert(circ->package_window > 0);
    circ->package_window--;
  } else { /* we're an AP, or an exit on a rendezvous circ */
    assert(conn->cpath_layer->package_window > 0);
    conn->cpath_layer->package_window--;
  }

  if(--conn->package_window <= 0) { /* is it 0 after decrement? */
    connection_stop_reading(conn);
    log_fn(LOG_DEBUG,"conn->package_window reached 0.");
    circuit_consider_stop_edge_reading(circ, conn->cpath_layer);
    return 0; /* don't process the inbuf any more */
  }
  log_fn(LOG_DEBUG,"conn->package_window is now %d",conn->package_window);

  /* handle more if there's more, or return 0 if there isn't */
  goto repeat_connection_edge_package_raw_inbuf;
}

#define MAX_STREAM_RETRIES 4
void connection_ap_expire_beginning(void) {
  connection_t **carray;
  connection_t *conn;
  circuit_t *circ;
  int n, i;
  time_t now = time(NULL);

  get_connection_array(&carray, &n);

  for (i = 0; i < n; ++i) {
    conn = carray[i];
    if (conn->type != CONN_TYPE_AP ||
        conn->state != AP_CONN_STATE_CONNECT_WAIT)
      continue;
    if (now - conn->timestamp_lastread < 15)
      continue;
    conn->num_retries++;
    circ = circuit_get_by_conn(conn);
    if(circ->purpose == CIRCUIT_PURPOSE_C_REND_JOINED) {
      if (now - conn->timestamp_lastread > 45) {
        log_fn(LOG_WARN,"Rend stream is %d seconds late. Giving up.",
               (int)(now - conn->timestamp_lastread));
        connection_mark_for_close(conn,END_STREAM_REASON_TIMEOUT);
      }
      continue;
    }
    assert(circ->purpose == CIRCUIT_PURPOSE_C_GENERAL);
    if(conn->num_retries >= MAX_STREAM_RETRIES) {
      log_fn(LOG_WARN,"Stream is %d seconds late. Giving up.",
             15*conn->num_retries);
      circuit_log_path(LOG_WARN, circ);
      connection_mark_for_close(conn,END_STREAM_REASON_TIMEOUT);
    } else {
      log_fn(LOG_WARN,"Stream is %d seconds late. Retrying.",
             (int)(now - conn->timestamp_lastread));
      circuit_log_path(LOG_WARN, circ);
      /* send an end down the circuit */
      connection_edge_end(conn, END_STREAM_REASON_TIMEOUT, conn->cpath_layer);
      /* un-mark it as ending, since we're going to reuse it */
      conn->has_sent_end = 0;
      /* move it back into 'pending' state. */
      conn->state = AP_CONN_STATE_CIRCUIT_WAIT;
      circuit_detach_stream(circ, conn);
      /* kludge to make us not try this circuit again, yet to allow
       * current streams on it to survive if they can: make it
       * unattractive to use for new streams */
      assert(circ->timestamp_dirty);
      circ->timestamp_dirty -= options.NewCircuitPeriod;
      /* give our stream another 15 seconds to try */
      conn->timestamp_lastread += 15;
      /* attaching to a dirty circuit is fine */
      if(connection_ap_handshake_attach_circuit(conn)<0) {
        /* it will never work */
        /* Don't need to send end -- we're not connected */
        connection_mark_for_close(conn, 0);
      }
    } /* end if max_retries */
  } /* end for */
}

/* Tell any APs that are waiting for a new circuit that one is available */
void connection_ap_attach_pending(void)
{
  connection_t **carray;
  connection_t *conn;
  int n, i;

  get_connection_array(&carray, &n);

  for (i = 0; i < n; ++i) {
    conn = carray[i];
    if (conn->type != CONN_TYPE_AP ||
        conn->state != AP_CONN_STATE_CIRCUIT_WAIT)
      continue;
    if(connection_ap_handshake_attach_circuit(conn) < 0) {
      /* -1 means it will never work */
      /* Don't send end; there is no 'other side' yet */
      connection_mark_for_close(conn,0);
    }
  }
}

static void connection_edge_consider_sending_sendme(connection_t *conn) {
  circuit_t *circ;

  if(connection_outbuf_too_full(conn))
    return;

  circ = circuit_get_by_conn(conn);
  if(!circ) {
    /* this can legitimately happen if the destroy has already
     * arrived and torn down the circuit */
    log_fn(LOG_INFO,"No circuit associated with conn. Skipping.");
    return;
  }

  while(conn->deliver_window < STREAMWINDOW_START - STREAMWINDOW_INCREMENT) {
    log_fn(LOG_DEBUG,"Outbuf %d, Queueing stream sendme.", conn->outbuf_flushlen);
    conn->deliver_window += STREAMWINDOW_INCREMENT;
    if(connection_edge_send_command(conn, circ, RELAY_COMMAND_SENDME,
                                    NULL, 0, conn->cpath_layer) < 0) {
      log_fn(LOG_WARN,"connection_edge_send_command failed. Returning.");
      return; /* the circuit's closed, don't continue */
    }
  }
}

/* return -1 if an unexpected error with conn, else 0. */
static int connection_ap_handshake_process_socks(connection_t *conn) {
  socks_request_t *socks;
  int sockshere;

  assert(conn);
  assert(conn->type == CONN_TYPE_AP);
  assert(conn->state == AP_CONN_STATE_SOCKS_WAIT);
  assert(conn->socks_request);
  socks = conn->socks_request;

  log_fn(LOG_DEBUG,"entered.");

  sockshere = fetch_from_buf_socks(conn->inbuf, socks);
  if(sockshere == -1 || sockshere == 0) {
    if(socks->replylen) { /* we should send reply back */
      log_fn(LOG_DEBUG,"reply is already set for us. Using it.");
      connection_ap_handshake_socks_reply(conn, socks->reply, socks->replylen, 0);
    } else if(sockshere == -1) { /* send normal reject */
      log_fn(LOG_WARN,"Fetching socks handshake failed. Closing.");
      connection_ap_handshake_socks_reply(conn, NULL, 0, 0);
    } else {
      log_fn(LOG_DEBUG,"socks handshake not all here yet.");
    }
    if (sockshere == -1)
      conn->socks_request->has_finished = 1;
    return sockshere;
  } /* else socks handshake is done, continue processing */

  /* this call _modifies_ socks->address iff it's a hidden-service request */
  if (rend_parse_rendezvous_address(socks->address) < 0) {
    /* normal request */
    conn->state = AP_CONN_STATE_CIRCUIT_WAIT;
    return connection_ap_handshake_attach_circuit(conn);
  } else {
    /* it's a hidden-service request */
    rend_cache_entry_t *entry;

    strcpy(conn->rend_query, socks->address); /* this strcpy is safe -RD */
    log_fn(LOG_INFO,"Got a hidden service request for ID '%s'", conn->rend_query);
    /* see if we already have it cached */
    if (rend_cache_lookup_entry(conn->rend_query, &entry) == 1 &&
#define NUM_SECONDS_BEFORE_REFETCH (60*15)
      entry->received + NUM_SECONDS_BEFORE_REFETCH < time(NULL)) {
      conn->state = AP_CONN_STATE_CIRCUIT_WAIT;
      return connection_ap_handshake_attach_circuit(conn);
    } else {
      conn->state = AP_CONN_STATE_RENDDESC_WAIT;
      if(!connection_get_by_type_rendquery(CONN_TYPE_DIR, conn->rend_query)) {
        /* not one already; initiate a dir rend desc lookup */
        directory_initiate_command(router_pick_directory_server(),
                                   DIR_PURPOSE_FETCH_RENDDESC,
                                   conn->rend_query, strlen(conn->rend_query));
      }
      return 0;
    }
  }
  return 0;
}

/* Find an open circ that we're happy with: return 1. if there isn't
 * one, and there isn't one on the way, launch one and return 0. if it
 * will never work, return -1.
 * write the found or in-progress or launched circ into *circp.
 */
static int
circuit_get_open_circ_or_launch(connection_t *conn,
                                uint8_t desired_circuit_purpose,
                                circuit_t **circp) {
  circuit_t *circ;
  uint32_t addr;

  assert(conn);
  assert(circp);
  assert(conn->state == AP_CONN_STATE_CIRCUIT_WAIT);

  circ = circuit_get_best(conn, 1, desired_circuit_purpose);

  if(circ) {
    *circp = circ;
    return 1; /* we're happy */
  }

  if(!connection_edge_is_rendezvous_stream(conn)) { /* general purpose circ */
    addr = client_dns_lookup_entry(conn->socks_request->address);
    if(router_exit_policy_all_routers_reject(addr, conn->socks_request->port)) {
      log_fn(LOG_WARN,"No Tor server exists that allows exit to %s:%d. Rejecting.",
             conn->socks_request->address, conn->socks_request->port);
      return -1;
    }
  }

  /* is one already on the way? */
  circ = circuit_get_best(conn, 0, desired_circuit_purpose);
  if(!circ) {
    char *exitname=NULL;
    uint8_t new_circ_purpose;

    if(desired_circuit_purpose == CIRCUIT_PURPOSE_C_INTRODUCING) {
      /* need to pick an intro point */
      exitname = rend_client_get_random_intro(conn->rend_query);
      if(!exitname) {
        log_fn(LOG_WARN,"Couldn't get an intro point for '%s'. Closing conn.",
               conn->rend_query);
        return -1;
      }
      if(!router_get_by_nickname(exitname)) {
        log_fn(LOG_WARN,"Advertised intro point '%s' is not known. Closing.", exitname);
        return -1;
      }
      log_fn(LOG_INFO,"Chose %s as intro point for %s.", exitname, conn->rend_query);
    }

    if(desired_circuit_purpose == CIRCUIT_PURPOSE_C_REND_JOINED)
      new_circ_purpose = CIRCUIT_PURPOSE_C_ESTABLISH_REND;
    else
      new_circ_purpose = desired_circuit_purpose;

    circ = circuit_launch_new(new_circ_purpose, exitname);
    tor_free(exitname);

    if(circ &&
       (desired_circuit_purpose != CIRCUIT_PURPOSE_C_GENERAL)) {
      /* then write the service_id into circ */
      strcpy(circ->rend_query, conn->rend_query);
    }
  }
  if(!circ)
    log_fn(LOG_INFO,"No safe circuit (purpose %d) ready for edge connection; delaying.",
           desired_circuit_purpose);
  *circp = circ;
  return 0;
}

void link_apconn_to_circ(connection_t *apconn, circuit_t *circ) {
  /* add it into the linked list of streams on this circuit */
  log_fn(LOG_DEBUG,"attaching new conn to circ. n_circ_id %d.", circ->n_circ_id);
  apconn->next_stream = circ->p_streams;
  /* assert_connection_ok(conn, time(NULL)); */
  circ->p_streams = apconn;

  assert(CIRCUIT_IS_ORIGIN(circ) && circ->cpath && circ->cpath->prev);
  assert(circ->cpath->prev->state == CPATH_STATE_OPEN);
  apconn->cpath_layer = circ->cpath->prev;
}

/* Try to find a safe live circuit for CONN_TYPE_AP connection conn. If
 * we don't find one: if conn cannot be handled by any known nodes,
 * warn and return -1 (conn needs to die);
 * else launch new circuit (if necessary) and return 0.
 * Otherwise, associate conn with a safe live circuit, do the
 * right next step, and return 1.
 */
int connection_ap_handshake_attach_circuit(connection_t *conn) {
  int retval;

  assert(conn);
  assert(conn->type == CONN_TYPE_AP);
  assert(conn->state == AP_CONN_STATE_CIRCUIT_WAIT);
  assert(conn->socks_request);

  if(!connection_edge_is_rendezvous_stream(conn)) { /* we're a general conn */
    circuit_t *circ=NULL;

    /* find the circuit that we should use, if there is one. */
    retval = circuit_get_open_circ_or_launch(conn, CIRCUIT_PURPOSE_C_GENERAL, &circ);
    if(retval < 1)
      return retval;

    /* We have found a suitable circuit for our conn. Hurray. */

    /* here, print the circ's path. so people can figure out which circs are sucking. */
    circuit_log_path(LOG_INFO,circ);

    if(!circ->timestamp_dirty)
      circ->timestamp_dirty = time(NULL);

    link_apconn_to_circ(conn, circ);
    connection_ap_handshake_send_begin(conn, circ);

    return 1;

  } else { /* we're a rendezvous conn */
    circuit_t *rendcirc=NULL, *introcirc=NULL;

    /* before anything else, see if we've already been attached
     * to a rendezvous circuit */
    if(conn->cpath_layer) {
      return 1;
    }

    /* else, start by finding a rendezvous circuit for us */

    retval = circuit_get_open_circ_or_launch(conn, CIRCUIT_PURPOSE_C_REND_JOINED, &rendcirc);
    if(retval < 0) return -1; /* failed */

    if(retval > 0) {
      /* one is already established, attach */
      log_fn(LOG_INFO,"rend joined circ already here. reusing.");
      link_apconn_to_circ(conn, rendcirc);
      if(connection_ap_handshake_send_begin(conn, rendcirc) < 0)
        return 0; /* already marked, let them fade away */
      return 1;
    }

    if(rendcirc &&
       rendcirc->purpose == CIRCUIT_PURPOSE_C_REND_READY &&
       rendcirc->build_state->pending_final_cpath) {
      log_fn(LOG_INFO,"pending-join circ already here. reusing.");
      link_apconn_to_circ(conn, rendcirc);
      /* don't send the begin, because we're still waiting for contact from bob */
      return 1;
    }

    /* it's on its way. find an intro circ. */
    retval = circuit_get_open_circ_or_launch(conn, CIRCUIT_PURPOSE_C_INTRODUCING, &introcirc);
    if(retval < 0) return -1; /* failed */

    if(retval > 0) {
      log_fn(LOG_INFO,"Intro circ is ready for us");
      if(rendcirc &&
         rendcirc->purpose == CIRCUIT_PURPOSE_C_REND_READY) {
        /* then we know !pending_final_cpath, from above */
        log_fn(LOG_INFO,"intro and rend circs are both ready. introducing.");
        /* this call marks introcirc for close */
        if(rend_client_send_introduction(introcirc, rendcirc) < 0) {
          return -1;
        }
        /* now attach conn to rendcirc */
        link_apconn_to_circ(conn, rendcirc);