Implement the common case of ATTACHSTREAM.

svn:r3751

doc/TODO

@@ -72,13 +72,15 @@ N  . Implement pending controller features.
     - EXTENDCIRCUIT
 R     - revised circ selection stuff.
       - Implement controller interface.
-    - ATTACHSTREAM
-      - Make streams have an 'unattached and not-automatically-attachable'
+    . ATTACHSTREAM
+      o Make streams have an 'unattached and not-automatically-attachable'
         state. ("Controller managed.")
-      - Add support to put new streams into this state rather than try to
+      o Add support to put new streams into this state rather than try to
         attach them automatically.  ("Hidden" config option.)
+      o Implement 'attach stream X to circuit Y' logic.
       - Time out never-attached streams.
-      - Implement 'attach stream X to circuit Y' logic.
+      - If we never get a CONNECTED back, we should put the stream back in
+        CONTROLLER_WAIT, not in CIRCUIT_WAIT.
     - Tests for new controller features
 R . HTTPS proxy for OR CONNECT stuff. (For outgoing SSL connections to
     other ORs.)

doc/control-spec.txt

@@ -100,6 +100,10 @@ the message.
                [The server didn't have enough of a given resource to
                 fulfill a given request.]
 
+        0x000A No such stream
+
+        0x000B No such circuit.
+
   The rest of the body should be a human-readable description of the error.
 
   In general, new error codes should only be added when they don't fall under
@@ -182,7 +186,7 @@ the message.
 
                 Status [1 octet]
                    (Sent connect=0,sent resolve=1,succeeded=2,failed=3,
-                    closed=4, new=5)
+                    closed=4, new connection=5, new resolve request=6)
                 Stream ID [4 octets]
                    (Must be unique to Tor process/time)
                 Target (NUL-terminated address-port string]
@@ -351,6 +355,9 @@ the message.
   associated with the specified circuit.  Each stream may be associated with
   at most one circuit, and multiple streams may share the same circuit.
 
+  If the circuit ID is 0, responsibility for attaching the given stream is
+  returned to Tor.
+
 3.16 POSTDESCRIPTOR (Type 0x000F)
 
   [Proposal; not finalized]

src/or/circuitlist.c

@@ -183,6 +183,22 @@ circuit_free_cpath_node(crypt_path_t *victim) {
   tor_free(victim);
 }
 
+/** DOCDOC **/
+circuit_t *
+circuit_get_by_global_id(uint32_t id)
+{
+  circuit_t *circ;
+  for (circ=global_circuitlist;circ;circ = circ->next) {
+    if (circ->global_identifier == id) {
+      if (circ->marked_for_close)
+        return NULL;
+      else
+        return circ;
+    }
+  }
+  return NULL;
+}
+
 /** Return a circ such that:
  *  - circ-\>n_circ_id or circ-\>p_circ_id is equal to <b>circ_id</b>, and
  *  - circ is attached to <b>conn</b>, either as p_conn, n-conn, or

src/or/circuituse.c

@@ -980,6 +980,36 @@ consider_recording_trackhost(connection_t *conn, circuit_t *circ) {
                       time(NULL) + options->TrackHostExitsExpire);
 }
 
+/* DOCDOC. Return as for connection_ap_handshake_attach_chosen_circuit. */
+int
+connection_ap_handshake_attach_chosen_circuit(connection_t *conn,
+                                              circuit_t *circ)
+{
+  tor_assert(conn);
+  tor_assert(conn->type == CONN_TYPE_AP);
+  tor_assert(conn->state == AP_CONN_STATE_CIRCUIT_WAIT ||
+             conn->state == AP_CONN_STATE_CONTROLLER_WAIT);
+  tor_assert(conn->socks_request);
+  tor_assert(circ);
+
+  conn->state = AP_CONN_STATE_CIRCUIT_WAIT;
+
+  if (!circ->timestamp_dirty)
+    circ->timestamp_dirty = time(NULL);
+
+  link_apconn_to_circ(conn, circ);
+  tor_assert(conn->socks_request);
+  if (conn->socks_request->command == SOCKS_COMMAND_CONNECT) {
+    consider_recording_trackhost(conn, circ);
+    connection_ap_handshake_send_begin(conn, circ);
+  } else {
+    connection_ap_handshake_send_resolve(conn, circ);
+  }
+
+  return 1;
+}
+
+
 /** 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);
@@ -1024,27 +1054,14 @@ int connection_ap_handshake_attach_circuit(connection_t *conn) {
     if (retval < 1)
       return retval;
 
-    /* We have found a suitable circuit for our conn. Hurray. */
-    tor_assert(circ);
-
-    log_fn(LOG_DEBUG,"Attaching apconn to general circ %d (stream %d sec old).",
+    log_fn(LOG_DEBUG,"Attaching apconn to circ %d (stream %d sec old).",
            circ->n_circ_id, conn_age);
     /* 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);
+    /* We have found a suitable circuit for our conn. Hurray. */
+    return connection_ap_handshake_attach_chosen_circuit(conn, circ);
 
-    link_apconn_to_circ(conn, circ);
-    tor_assert(conn->socks_request);
-    if (conn->socks_request->command == SOCKS_COMMAND_CONNECT) {
-      consider_recording_trackhost(conn, circ);
-      connection_ap_handshake_send_begin(conn, circ);
-    } else {
-      connection_ap_handshake_send_resolve(conn, circ);
-    }
-
-    return 1;
   } else { /* we're a rendezvous conn */
     circuit_t *rendcirc=NULL, *introcirc=NULL;
 

src/or/config.c

@@ -174,6 +174,7 @@ static config_var_t config_vars[] = {
   VAR("SysLog",              LINELIST_S, OldLogOptions,      NULL),
   OBSOLETE("TrafficShaping"),
   VAR("User",                STRING,   User,                 NULL),
+  VAR("__ManageConnections", BOOL,     ManageConnections,    "1"),
   { NULL, CONFIG_TYPE_OBSOLETE, 0, NULL }
 };
 #undef VAR
@@ -1155,6 +1156,9 @@ config_dump_options(or_options_t *options, int minimal)
     if (config_vars[i].type == CONFIG_TYPE_OBSOLETE ||
         config_vars[i].type == CONFIG_TYPE_LINELIST_S)
       continue;
+    /* Don't save 'hidden' control variables. */
+    if (!strcmpstart(config_vars[i].name, "__"))
+      continue;
     if (minimal && option_is_same(options, defaults, config_vars[i].name))
       continue;
     line = config_get_assigned_option(options, config_vars[i].name);

src/or/connection.c

@@ -1243,6 +1243,30 @@ connection_t *connection_get_by_identity_digest(const char *digest, int type)
   return best;
 }
 
+
+/** Return the connection with id <b>id</b> if it is not already
+ * marked for close.
+ */
+connection_t *
+connection_get_by_global_id(uint32_t id) {
+  int i, n;
+  connection_t *conn;
+  connection_t **carray;
+
+  get_connection_array(&carray,&n);
+  for (i=0;i<n;i++) {
+    conn = carray[i];
+    if (conn->global_identifier == id) {
+      if (!conn->marked_for_close)
+        return conn;
+      else
+        return NULL;
+    }
+  }
+  return NULL;
+}
+
+
 /** Return a connection of type <b>type</b> that is not marked for
  * close.
  */

src/or/connection_edge.c

@@ -839,16 +839,22 @@ static int connection_ap_handshake_process_socks(connection_t *conn) {
         return 0;
       }
       rep_hist_note_used_resolve(time(NULL)); /* help predict this next time */
+            control_event_stream_status(conn, STREAM_EVENT_NEW_RESOLVE);
     } else { /* socks->command == SOCKS_COMMAND_CONNECT */
       if (socks->port == 0) {
         log_fn(LOG_NOTICE,"Application asked to connect to port 0. Refusing.");
         return -1;
       }
       rep_hist_note_used_port(socks->port, time(NULL)); /* help predict this next time */
+      control_event_stream_status(conn, STREAM_EVENT_NEW);
+    }
+    if (get_options()->ManageConnections) {
+      conn->state = AP_CONN_STATE_CIRCUIT_WAIT;
+      return connection_ap_handshake_attach_circuit(conn);
+    } else {
+      conn->state = AP_CONN_STATE_CONTROLLER_WAIT;
+      return 0;
     }
-    control_event_stream_status(conn, STREAM_EVENT_NEW);
-    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;

src/or/control.c

@@ -61,7 +61,8 @@ const char control_c_id[] = "$Id$";
 #define ERR_UNAUTHORIZED            0x0007
 #define ERR_REJECTED_AUTHENTICATION 0x0008
 #define ERR_RESOURCE_EXHAUSETED     0x0009
-#define ERR_TOO_LONG                0x000A
+#define ERR_NO_STREAM               0x000A
+#define ERR_NO_CIRC                 0x000B
 
 /* Recognized asynchronous event types. */
 #define _EVENT_MIN            0x0001
@@ -613,12 +614,6 @@ handle_control_getinfo(connection_t *conn, uint32_t len, const char *body)
     }
   });
 
-  if (msg_len > 65535) {
-    /* XXXX What if it *is* this long? */
-    send_control_error(conn, ERR_TOO_LONG, body);
-    goto done;
-  }
-
   msg = smartlist_join_strings2(answers, "\0", 1, 1, &msg_len);
   send_control_message(conn, CONTROL_CMD_INFOVALUE,
                        msg_len, msg_len?msg:NULL);
@@ -642,7 +637,47 @@ handle_control_extendcircuit(connection_t *conn, uint32_t len,
 static int handle_control_attachstream(connection_t *conn, uint32_t len,
                                         const char *body)
 {
-  send_control_error(conn,ERR_UNRECOGNIZED_TYPE,"not yet implemented");
+  uint32_t conn_id;
+  uint32_t circ_id;
+  connection_t *ap_conn;
+  circuit_t *circ;
+
+  if (len < 8) {
+    send_control_error(conn, ERR_SYNTAX, "attachstream message too short");
+    return 0;
+  }
+
+  conn_id = ntohl(get_uint32(body));
+  circ_id = ntohl(get_uint32(body+4));
+
+  if (!(ap_conn = connection_get_by_global_id(conn_id))) {
+    send_control_error(conn, ERR_NO_STREAM,
+                       "No connection found with given ID");
+    return 0;
+  }
+  if (ap_conn->state != AP_CONN_STATE_CONTROLLER_WAIT) {
+    send_control_error(conn, ERR_NO_STREAM,
+                       "Connection was not managed by controller.");
+    return 0;
+  }
+
+  if (!circ_id) {
+    ap_conn->state = AP_CONN_STATE_CIRCUIT_WAIT;
+    if (connection_ap_handshake_attach_circuit(ap_conn)<0)
+      connection_mark_for_close(ap_conn);
+
+    return 0;
+  }
+
+
+  if (!(circ = circuit_get_by_global_id(circ_id))) {
+    send_control_error(conn, ERR_NO_CIRC, "No circuit found with given ID");
+    return 0;
+  }
+  if (connection_ap_handshake_attach_chosen_circuit(ap_conn, circ) != 1) {
+    send_control_error(conn, ERR_INTERNAL, "Unable to attach stream.");
+    return 0;
+  }
   return 0;
 }
 static int

src/or/or.h

@@ -256,15 +256,18 @@ typedef enum {
 /** State for a SOCKS connection: got a y.onion URL; waiting to receive
  * rendezvous rescriptor. */
 #define AP_CONN_STATE_RENDDESC_WAIT 6
+/** The controller will attach this connection to a circuit; it isn't our
+ * job to do so. */
+#define AP_CONN_STATE_CONTROLLER_WAIT 7
 /** State for a SOCKS connection: waiting for a completed circuit. */
-#define AP_CONN_STATE_CIRCUIT_WAIT 7
+#define AP_CONN_STATE_CIRCUIT_WAIT 8
 /** State for a SOCKS connection: sent BEGIN, waiting for CONNECTED. */
-#define AP_CONN_STATE_CONNECT_WAIT 8
+#define AP_CONN_STATE_CONNECT_WAIT 9
 /** State for a SOCKS connection: send RESOLVE, waiting for RESOLVED. */
-#define AP_CONN_STATE_RESOLVE_WAIT 9
+#define AP_CONN_STATE_RESOLVE_WAIT 10
 /** State for a SOCKS connection: ready to send and receive. */
-#define AP_CONN_STATE_OPEN 10
-#define _AP_CONN_STATE_MAX 10
+#define AP_CONN_STATE_OPEN 11
+#define _AP_CONN_STATE_MAX 11
 
 #define _DIR_CONN_STATE_MIN 1
 /** State for connection to directory server: waiting for connect(). */
@@ -1053,6 +1056,9 @@ typedef struct {
                                 * the control system. */
   int CookieAuthentication; /**< Boolean: do we enable cookie-based auth for
                              * the control system? */
+  int ManageConnections; /**< Boolean: Does Tor attach new connections to
+                          * circuits itself (1), or does it let the controller
+                          * deal? (0) */
 } or_options_t;
 
 #define MAX_SOCKS_REPLY_LEN 1024
@@ -1133,6 +1139,7 @@ void circuit_close_all_marked(void);
 circuit_t *circuit_new(uint16_t p_circ_id, connection_t *p_conn);
 circuit_t *circuit_get_by_circ_id_conn(uint16_t circ_id, connection_t *conn);
 circuit_t *circuit_get_by_conn(connection_t *conn);
+circuit_t *circuit_get_by_global_id(uint32_t id);
 circuit_t *circuit_get_by_rend_query_and_purpose(const char *rend_query, uint8_t purpose);
 circuit_t *circuit_get_next_by_pk_and_purpose(circuit_t *start,
                                          const char *digest, uint8_t purpose);
@@ -1172,6 +1179,8 @@ circuit_t *circuit_launch_by_nickname(uint8_t purpose, const char *exit_nickname
 circuit_t *circuit_launch_by_identity(uint8_t purpose, const char *exit_digest,
                                       int need_uptime, int need_capacity, int is_internal);
 void circuit_reset_failure_count(int timeout);
+int connection_ap_handshake_attach_chosen_circuit(connection_t *conn,
+                                                  circuit_t *circ);
 int connection_ap_handshake_attach_circuit(connection_t *conn);
 
 /********************************* command.c ***************************/
@@ -1260,6 +1269,7 @@ void connection_write_to_buf(const char *string, size_t len, connection_t *conn)
 
 connection_t *connection_exact_get_by_addr_port(uint32_t addr, uint16_t port);
 connection_t *connection_get_by_identity_digest(const char *digest, int type);
+connection_t *connection_get_by_global_id(uint32_t id);
 
 connection_t *connection_get_by_type(int type);
 connection_t *connection_get_by_type_state(int type, int state);
@@ -1365,7 +1375,8 @@ typedef enum stream_status_event_t {
   STREAM_EVENT_SUCCEEDED    = 2,
   STREAM_EVENT_FAILED       = 3,
   STREAM_EVENT_CLOSED       = 4,
-  STREAM_EVENT_NEW          = 5
+  STREAM_EVENT_NEW          = 5,
+  STREAM_EVENT_NEW_RESOLVE  = 6
 } stream_status_event_t;
 
 typedef enum or_conn_status_event_t {