Support closing safe OR connections.

Does not yet support flushing buffers before closing.

src/core/mainloop/connection.c

@@ -676,6 +676,10 @@ connection_free_minimal(connection_t *conn)
     buf_free(conn->outbuf);
   }
 
+  if (conn->safe_conn != NULL) {
+    safe_connection_unsubscribe_all(conn->safe_conn, conn->event_listener);
+  }
+
   event_listener_free(conn->event_listener);
   event_source_free(conn->event_source);
 
@@ -2142,7 +2146,6 @@ connection_connect_sockaddr,(connection_t *conn,
   log_fn(inprogress ? LOG_DEBUG : LOG_INFO, LD_NET,
          "Connection to socket %s (sock "TOR_SOCKET_T_FORMAT").",
          inprogress ? "in progress" : "established", s);
-  conn->s = s;
 
   if (!connection_uses_safe_conn(conn->type)) {
     conn->s = s;

src/core/or/connection_or.c

@@ -84,6 +84,7 @@
 
 event_label_t or_conn_link_protocol_version_ev = EVENT_LABEL_UNSET;
 event_label_t or_conn_open_ev = EVENT_LABEL_UNSET;
+event_label_t or_conn_closed_ev = EVENT_LABEL_UNSET;
 event_label_t or_conn_outgoing_packed_cell = EVENT_LABEL_UNSET;
 event_label_t or_conn_outgoing_fixed_cell = EVENT_LABEL_UNSET;
 event_label_t or_conn_outgoing_variable_cell = EVENT_LABEL_UNSET;
@@ -129,6 +130,7 @@ or_conn_register_events(event_registry_t *registry)
 {
   tor_assert(or_conn_link_protocol_version_ev == EVENT_LABEL_UNSET);
   tor_assert(or_conn_open_ev == EVENT_LABEL_UNSET);
+  tor_assert(or_conn_closed_ev == EVENT_LABEL_UNSET);
   tor_assert(or_conn_outgoing_packed_cell == EVENT_LABEL_UNSET);
   tor_assert(or_conn_outgoing_fixed_cell == EVENT_LABEL_UNSET);
   tor_assert(or_conn_outgoing_variable_cell == EVENT_LABEL_UNSET);
@@ -137,6 +139,8 @@ or_conn_register_events(event_registry_t *registry)
     event_registry_register_event(registry, "Decided protocol version");
   or_conn_open_ev = \
     event_registry_register_event(registry, "Connection is open");
+  or_conn_closed_ev = \
+    event_registry_register_event(registry, "Connection is closed");
   or_conn_outgoing_packed_cell = \
     event_registry_register_event(registry, "Outgoing packed cell");
   or_conn_outgoing_fixed_cell = \
@@ -1758,6 +1762,12 @@ connection_or_close_normally(or_connection_t *orconn, int flush)
   tor_assert(orconn);
   if (flush) connection_mark_and_flush_internal(TO_CONN(orconn));
   else connection_mark_for_close_internal(TO_CONN(orconn));
+
+  // TODO: support flush rather than closing immediately
+  event_data_t null_data = { .ptr = NULL };
+  event_source_publish(TO_CONN(orconn)->event_source,
+                       or_conn_closed_ev, null_data, NULL);
+
   if (orconn->chan) {
     chan = TLS_CHAN_TO_BASE(orconn->chan);
     /* Don't transition if we're already in closing, closed or error */
@@ -1779,6 +1789,12 @@ connection_or_close_for_error,(or_connection_t *orconn, int flush))
   tor_assert(orconn);
   if (flush) connection_mark_and_flush_internal(TO_CONN(orconn));
   else connection_mark_for_close_internal(TO_CONN(orconn));
+
+  // TODO: support flush rather than closing immediately
+  event_data_t null_data = { .ptr = NULL };
+  event_source_publish(TO_CONN(orconn)->event_source,
+                       or_conn_closed_ev, null_data, NULL);
+
   if (orconn->chan) {
     chan = TLS_CHAN_TO_BASE(orconn->chan);
     /* Don't transition if we're already in closing, closed or error */
@@ -3135,6 +3151,7 @@ connection_or_compute_authenticate_cell_body(or_connection_t *conn,
       log_fn(LOG_PROTOCOL_WARN, LD_OR, "Somebody asked us for an older TLS "
          "authentication method (AUTHTYPE_RSA_SHA256_TLSSECRET) "
          "which we don't support.");
+      goto err;
     }
   } else {
     char label[128];

src/core/or/connection_or.h

@@ -17,6 +17,7 @@
 
 extern event_label_t or_conn_link_protocol_version_ev;
 extern event_label_t or_conn_open_ev;
+extern event_label_t or_conn_closed_ev;
 extern event_label_t or_conn_outgoing_packed_cell;
 extern event_label_t or_conn_outgoing_fixed_cell;
 extern event_label_t or_conn_outgoing_variable_cell;

src/core/or/safe_connection.c

@@ -32,6 +32,9 @@ safe_or_conn_link_protocol_version_cb(event_label_t label, event_data_t data,
 static void
 safe_or_conn_open_cb(event_label_t label, event_data_t data, void *context);
 
+static void
+safe_or_conn_closed_cb(event_label_t label, event_data_t data, void *context);
+
 static tor_error_t
 safe_or_connection_update_state(safe_or_connection_t *safe_or_conn,
                                 or_conn_state_t new_state);
@@ -233,7 +236,7 @@ safe_connection_set_socket(safe_connection_t *safe_conn, tor_socket_t socket)
   tor_assert(!safe_conn->linked);
   tor_assert(SOCKET_OK(socket));
 
-  if (safe_conn->socket != TOR_INVALID_SOCKET) {
+  if (SOCKET_OK(safe_conn->socket)) {
     log_warn(LD_BUG, "We're overwriting a previous socket");
   }
   safe_conn->socket = socket;
@@ -245,6 +248,25 @@ safe_connection_set_socket(safe_connection_t *safe_conn, tor_socket_t socket)
   tor_mutex_release(&safe_conn->lock);
 }
 
+static void
+safe_connection_close_socket(safe_connection_t *safe_conn)
+{
+  tor_assert(safe_conn != NULL);
+  tor_mutex_acquire(&safe_conn->lock);
+
+  safe_connection_unregister_events(safe_conn);
+  event_listener_detach(safe_conn->event_listener);
+  // assume it's safe at this point we don't care about any more events
+  // TODO: improve this (possibly with something like a sentinel event)
+
+  if (SOCKET_OK(safe_conn->socket)) {
+    tor_close_socket(safe_conn->socket);
+    safe_conn->socket = TOR_INVALID_SOCKET;
+  }
+
+  tor_mutex_release(&safe_conn->lock);
+}
+
 static void
 safe_connection_read_cb(evutil_socket_t ev_sock, short fd, void *void_safe_conn)
 {
@@ -343,7 +365,10 @@ safe_connection_unregister_events(safe_connection_t *safe_conn)
   if (safe_conn->write_event != NULL) {
     tor_event_free(safe_conn->write_event);
   }
-  event_listener_detach(safe_conn->event_listener);
+
+  // we may still want to receive events, so we don't detach the
+  // event listener yet
+  // TODO: figure out a better way of handling this
 
   tor_mutex_release(&safe_conn->lock);
 }
@@ -360,6 +385,7 @@ safe_connection_register_events(safe_connection_t *safe_conn,
   // is either linked or has a socket, but not both (or neither)
 
   safe_connection_unregister_events(safe_conn);
+  event_listener_detach(safe_conn->event_listener);
 
   safe_conn->read_event = tor_event_new(event_base, safe_conn->socket,
                                         EV_READ|EV_PERSIST,
@@ -536,6 +562,9 @@ safe_or_connection_new(bool requires_buffers, bool is_outgoing,
   event_listener_set_callback(TO_SAFE_CONN(safe_or_conn)->event_listener,
                               or_conn_open_ev,
                               NULL, safe_or_conn_open_cb);
+  event_listener_set_callback(TO_SAFE_CONN(safe_or_conn)->event_listener,
+                              or_conn_closed_ev,
+                              NULL, safe_or_conn_closed_cb);
   event_listener_set_callback(TO_SAFE_CONN(safe_or_conn)->event_listener,
                               or_conn_outgoing_packed_cell,
                               NULL, safe_or_conn_outgoing_cell_cb);
@@ -553,6 +582,9 @@ safe_or_connection_new(bool requires_buffers, bool is_outgoing,
     event_source_subscribe(conn_event_source,
                            TO_SAFE_CONN(safe_or_conn)->event_listener,
                            or_conn_open_ev);
+    event_source_subscribe(conn_event_source,
+                           TO_SAFE_CONN(safe_or_conn)->event_listener,
+                           or_conn_closed_ev);
     event_source_subscribe(conn_event_source,
                            TO_SAFE_CONN(safe_or_conn)->event_listener,
                            or_conn_outgoing_packed_cell);
@@ -686,6 +718,27 @@ safe_or_conn_open_cb(event_label_t label, event_data_t data, void *context)
   tor_mutex_release(&TO_SAFE_CONN(safe_or_conn)->lock);
 }
 
+static void
+safe_or_conn_closed_cb(event_label_t label, event_data_t data, void *context)
+{
+  (void)data;
+
+  safe_or_connection_t *safe_or_conn = TO_SAFE_OR_CONN(context);
+  tor_assert(label == or_conn_closed_ev);
+  tor_assert(safe_or_conn != NULL);
+  tor_mutex_acquire(&TO_SAFE_CONN(safe_or_conn)->lock);
+
+  // TODO: we should support closing forcefully and closing gracefully
+  // with a CLOSING state (which only flushes remaining data)
+
+  if (safe_or_conn->state != SAFE_OR_CONN_STATE_CLOSED) {
+    // if we're already closed, then just ignore it
+    safe_or_connection_update_state(safe_or_conn, SAFE_OR_CONN_STATE_CLOSED);
+  }
+
+  tor_mutex_release(&TO_SAFE_CONN(safe_or_conn)->lock);
+}
+
 // TODO: we should get rid of this at some point
 void
 safe_or_connection_get_tls_desc(safe_or_connection_t *safe_or_conn,
@@ -926,6 +979,8 @@ safe_or_connection_update_state(safe_or_connection_t *safe_or_conn,
       log_warn(LD_OR, "Could not create a new tor TLS object");
       return E_ERROR;
     }
+    tor_tls_release_socket(safe_or_conn->tls);
+    // we want to have control over closing the socket
     if (safe_or_conn->remote_address_str != NULL) {
       tor_tls_set_logged_address(safe_or_conn->tls,
                                  safe_or_conn->remote_address_str);
@@ -993,6 +1048,11 @@ safe_or_connection_update_state(safe_or_connection_t *safe_or_conn,
                         TO_SAFE_CONN(safe_or_conn));
     event_source_publish(TO_SAFE_CONN(safe_or_conn)->event_source,
                          safe_or_conn_closed_ev, null_data, NULL);
+    if (safe_or_conn->tls != NULL) {
+      tor_tls_free(safe_or_conn->tls);
+      safe_or_conn->tls = NULL;
+    }
+    safe_connection_close_socket(TO_SAFE_CONN(safe_or_conn));
     break;
   default:
     log_warn(LD_OR, "Unexpected state");

src/core/or/scheduler_kist.c

@@ -234,6 +234,14 @@ update_socket_info_impl, (socket_table_ent_t *ent))
   const tor_socket_t sock = conn->safe_conn->socket;
   tor_mutex_release(&(conn->safe_conn->lock));
 
+  if (!SOCKET_OK(sock)) {
+    // the socket is closed, so we should not write anything
+    log_warn(LD_SCHED, "The socket for %p has been closed.", conn);
+    // TODO: this should be debug not warn
+    ent->limit = ent->cwnd = ent->unacked = ent->mss = ent->notsent = 0;
+    return;
+  }
+
   struct tcp_info tcp;
   socklen_t tcp_info_len = sizeof(tcp);
 
@@ -243,6 +251,15 @@ update_socket_info_impl, (socket_table_ent_t *ent))
 
   /* Gather information */
   if (getsockopt(sock, SOL_TCP, TCP_INFO, (void *)&(tcp), &tcp_info_len) < 0) {
+    if (errno == EBADF) {
+      // there is a race condition where the safe_conn could close the fd after
+      // we get it, in which case we should set a limit of 0
+      log_warn(LD_SCHED, "The socket for %p is not a valid file descriptor.",
+               conn);
+      // TODO: this should be debug not warn
+      ent->limit = ent->cwnd = ent->unacked = ent->mss = ent->notsent = 0;
+      return;
+    }
     if (errno == EINVAL) {
       /* Oops, this option is not provided by the kernel, we'll have to
        * disable KIST entirely. This can happen if tor was built on a machine
@@ -257,6 +274,15 @@ update_socket_info_impl, (socket_table_ent_t *ent))
     goto fallback;
   }
   if (ioctl(sock, SIOCOUTQNSD, &(ent->notsent)) < 0) {
+    if (errno == EBADF) {
+      // there is a race condition where the safe_conn could close the fd after
+      // we get it, in which case we should set a limit of 0
+      log_warn(LD_SCHED, "The socket for %p is not a valid file descriptor.",
+               conn);
+      // TODO: this should be debug not warn
+      ent->limit = ent->cwnd = ent->unacked = ent->mss = ent->notsent = 0;
+      return;
+    }
     if (errno == EINVAL) {
       log_notice(LD_SCHED, "Looks like our kernel doesn't have the support "
                            "for KIST anymore. We will fallback to the naive "