control.c

/* Copyright (c) 2004-2006, Roger Dingledine, Nick Mathewson.
 * Copyright (c) 2007-2018, The Tor Project, Inc. */
/* See LICENSE for licensing information */

/**
 * \file control.c
 * \brief Implementation for Tor's control-socket interface.
 *
 * A "controller" is an external program that monitors and controls a Tor
 * instance via a text-based protocol. It connects to Tor via a connection
 * to a local socket.
 *
 * The protocol is line-driven.  The controller sends commands terminated by a
 * CRLF.  Tor sends lines that are either <em>replies</em> to what the
 * controller has said, or <em>events</em> that Tor sends to the controller
 * asynchronously based on occurrences in the Tor network model.
 *
 * See the control-spec.txt file in the torspec.git repository for full
 * details on protocol.
 *
 * This module generally has two kinds of entry points: those based on having
 * received a command on a controller socket, which are handled in
 * connection_control_process_inbuf(), and dispatched to individual functions
 * with names like control_handle_COMMANDNAME(); and those based on events
 * that occur elsewhere in Tor, which are handled by functions with names like
 * control_event_EVENTTYPE().
 *
 * Controller events are not sent immediately; rather, they are inserted into
 * the queued_control_events array, and flushed later from
 * flush_queued_events_cb().  Doing this simplifies our callgraph greatly,
 * by limiting the number of places in Tor that can call back into the network
 * stack.
 **/

#define CONTROL_PRIVATE

#include "core/or/or.h"
#include "app/config/config.h"
#include "app/config/confparse.h"
#include "app/main/main.h"
#include "core/mainloop/connection.h"
#include "core/mainloop/mainloop.h"
#include "core/or/channel.h"
#include "core/or/channeltls.h"
#include "core/or/circuitbuild.h"
#include "core/or/circuitlist.h"
#include "core/or/circuitstats.h"
#include "core/or/circuituse.h"
#include "core/or/command.h"
#include "core/or/connection_edge.h"
#include "core/or/connection_or.h"
#include "core/or/policies.h"
#include "core/or/reasons.h"
#include "core/or/versions.h"
#include "core/proto/proto_control0.h"
#include "core/proto/proto_http.h"
#include "feature/client/addressmap.h"
#include "feature/client/bridges.h"
#include "feature/client/dnsserv.h"
#include "feature/client/entrynodes.h"
#include "feature/control/control.h"
#include "feature/control/fmt_serverstatus.h"
#include "feature/control/getinfo_geoip.h"
#include "feature/dircache/dirserv.h"
#include "feature/dirclient/dirclient.h"
#include "feature/dirclient/dlstatus.h"
#include "feature/dircommon/directory.h"
#include "feature/hibernate/hibernate.h"
#include "feature/hs/hs_cache.h"
#include "feature/hs/hs_common.h"
#include "feature/hs/hs_control.h"
#include "feature/hs_common/shared_random_client.h"
#include "feature/nodelist/authcert.h"
#include "feature/nodelist/dirlist.h"
#include "feature/nodelist/microdesc.h"
#include "feature/nodelist/networkstatus.h"
#include "feature/nodelist/nodelist.h"
#include "feature/nodelist/routerinfo.h"
#include "feature/nodelist/routerlist.h"
#include "feature/relay/router.h"
#include "feature/relay/routermode.h"
#include "feature/relay/selftest.h"
#include "feature/rend/rendclient.h"
#include "feature/rend/rendcommon.h"
#include "feature/rend/rendparse.h"
#include "feature/rend/rendservice.h"
#include "feature/stats/geoip_stats.h"
#include "feature/stats/predict_ports.h"
#include "lib/container/buffers.h"
#include "lib/crypt_ops/crypto_rand.h"
#include "lib/crypt_ops/crypto_util.h"
#include "lib/encoding/confline.h"
#include "lib/evloop/compat_libevent.h"

#include "feature/dircache/cached_dir_st.h"
#include "feature/control/control_connection_st.h"
#include "core/or/cpath_build_state_st.h"
#include "core/or/entry_connection_st.h"
#include "feature/nodelist/extrainfo_st.h"
#include "feature/nodelist/networkstatus_st.h"
#include "feature/nodelist/node_st.h"
#include "core/or/or_connection_st.h"
#include "core/or/or_circuit_st.h"
#include "core/or/origin_circuit_st.h"
#include "feature/nodelist/microdesc_st.h"
#include "feature/rend/rend_authorized_client_st.h"
#include "feature/rend/rend_encoded_v2_service_descriptor_st.h"
#include "feature/rend/rend_service_descriptor_st.h"
#include "feature/nodelist/routerinfo_st.h"
#include "feature/nodelist/routerlist_st.h"
#include "core/or/socks_request_st.h"

#ifdef HAVE_UNISTD_H
#include <unistd.h>
#endif
#ifdef HAVE_SYS_STAT_H
#include <sys/stat.h>
#endif

#ifndef _WIN32
#include <pwd.h>
#include <sys/resource.h>
#endif

#include "lib/crypt_ops/crypto_s2k.h"
#include "lib/evloop/procmon.h"
#include "lib/evloop/compat_libevent.h"

/** Yield true iff <b>s</b> is the state of a control_connection_t that has
 * finished authentication and is accepting commands. */
#define STATE_IS_OPEN(s) ((s) == CONTROL_CONN_STATE_OPEN)

/** Bitfield: The bit 1&lt;&lt;e is set if <b>any</b> open control
 * connection is interested in events of type <b>e</b>.  We use this
 * so that we can decide to skip generating event messages that nobody
 * has interest in without having to walk over the global connection
 * list to find out.
 **/
typedef uint64_t event_mask_t;

/** An event mask of all the events that any controller is interested in
 * receiving. */
static event_mask_t global_event_mask = 0;

/** True iff we have disabled log messages from being sent to the controller */
static int disable_log_messages = 0;

/** Macro: true if any control connection is interested in events of type
 * <b>e</b>. */
#define EVENT_IS_INTERESTING(e) \
  (!! (global_event_mask & EVENT_MASK_(e)))

/** Macro: true if any event from the bitfield 'e' is interesting. */
#define ANY_EVENT_IS_INTERESTING(e) \
  (!! (global_event_mask & (e)))

/** If we're using cookie-type authentication, how long should our cookies be?
 */
#define AUTHENTICATION_COOKIE_LEN 32

/** If true, we've set authentication_cookie to a secret code and
 * stored it to disk. */
static int authentication_cookie_is_set = 0;
/** If authentication_cookie_is_set, a secret cookie that we've stored to disk
 * and which we're using to authenticate controllers.  (If the controller can
 * read it off disk, it has permission to connect.) */
static uint8_t *authentication_cookie = NULL;

#define SAFECOOKIE_SERVER_TO_CONTROLLER_CONSTANT \
  "Tor safe cookie authentication server-to-controller hash"
#define SAFECOOKIE_CONTROLLER_TO_SERVER_CONSTANT \
  "Tor safe cookie authentication controller-to-server hash"
#define SAFECOOKIE_SERVER_NONCE_LEN DIGEST256_LEN

/** The list of onion services that have been added via ADD_ONION that do not
 * belong to any particular control connection.
 */
static smartlist_t *detached_onion_services = NULL;

/** A sufficiently large size to record the last bootstrap phase string. */
#define BOOTSTRAP_MSG_LEN 1024

/** What was the last bootstrap phase message we sent? We keep track
 * of this so we can respond to getinfo status/bootstrap-phase queries. */
static char last_sent_bootstrap_message[BOOTSTRAP_MSG_LEN];

static void connection_printf_to_buf(control_connection_t *conn,
                                     const char *format, ...)
  CHECK_PRINTF(2,3);
static void send_control_event_impl(uint16_t event,
                                    const char *format, va_list ap)
  CHECK_PRINTF(2,0);
static int control_event_status(int type, int severity, const char *format,
                                va_list args)
  CHECK_PRINTF(3,0);

static void send_control_done(control_connection_t *conn);
static void send_control_event(uint16_t event,
                               const char *format, ...)
  CHECK_PRINTF(2,3);
static int handle_control_setconf(control_connection_t *conn, uint32_t len,
                                  char *body);
static int handle_control_resetconf(control_connection_t *conn, uint32_t len,
                                    char *body);
static int handle_control_getconf(control_connection_t *conn, uint32_t len,
                                  const char *body);
static int handle_control_loadconf(control_connection_t *conn, uint32_t len,
                                  const char *body);
static int handle_control_setevents(control_connection_t *conn, uint32_t len,
                                    const char *body);
static int handle_control_authenticate(control_connection_t *conn,
                                       uint32_t len,
                                       const char *body);
static int handle_control_signal(control_connection_t *conn, uint32_t len,
                                 const char *body);
static int handle_control_mapaddress(control_connection_t *conn, uint32_t len,
                                     const char *body);
static char *list_getinfo_options(void);
static int handle_control_getinfo(control_connection_t *conn, uint32_t len,
                                  const char *body);
static int handle_control_extendcircuit(control_connection_t *conn,
                                        uint32_t len,
                                        const char *body);
static int handle_control_setcircuitpurpose(control_connection_t *conn,
                                            uint32_t len, const char *body);
static int handle_control_attachstream(control_connection_t *conn,
                                       uint32_t len,
                                        const char *body);
static int handle_control_postdescriptor(control_connection_t *conn,
                                         uint32_t len,
                                         const char *body);
static int handle_control_redirectstream(control_connection_t *conn,
                                         uint32_t len,
                                         const char *body);
static int handle_control_closestream(control_connection_t *conn, uint32_t len,
                                      const char *body);
static int handle_control_closecircuit(control_connection_t *conn,
                                       uint32_t len,
                                       const char *body);
static int handle_control_resolve(control_connection_t *conn, uint32_t len,
                                  const char *body);
static int handle_control_usefeature(control_connection_t *conn,
                                     uint32_t len,
                                     const char *body);
static int handle_control_hsfetch(control_connection_t *conn, uint32_t len,
                                  const char *body);
static int handle_control_hspost(control_connection_t *conn, uint32_t len,
                                 const char *body);
static int handle_control_add_onion(control_connection_t *conn, uint32_t len,
                                    const char *body);
static int handle_control_del_onion(control_connection_t *conn, uint32_t len,
                                    const char *body);
static int write_stream_target_to_buf(entry_connection_t *conn, char *buf,
                                      size_t len);
static void orconn_target_get_name(char *buf, size_t len,
                                   or_connection_t *conn);

static int get_cached_network_liveness(void);
static void set_cached_network_liveness(int liveness);

static void flush_queued_events_cb(mainloop_event_t *event, void *arg);

static char * download_status_to_string(const download_status_t *dl);
static void control_get_bytes_rw_last_sec(uint64_t *r, uint64_t *w);

/** Convert a connection_t* to an control_connection_t*; assert if the cast is
 * invalid. */
control_connection_t *
TO_CONTROL_CONN(connection_t *c)
{
  tor_assert(c->magic == CONTROL_CONNECTION_MAGIC);
  return DOWNCAST(control_connection_t, c);
}

/** Given a control event code for a message event, return the corresponding
 * log severity. */
static inline int
event_to_log_severity(int event)
{
  switch (event) {
    case EVENT_DEBUG_MSG: return LOG_DEBUG;
    case EVENT_INFO_MSG: return LOG_INFO;
    case EVENT_NOTICE_MSG: return LOG_NOTICE;
    case EVENT_WARN_MSG: return LOG_WARN;
    case EVENT_ERR_MSG: return LOG_ERR;
    default: return -1;
  }
}

/** Given a log severity, return the corresponding control event code. */
static inline int
log_severity_to_event(int severity)
{
  switch (severity) {
    case LOG_DEBUG: return EVENT_DEBUG_MSG;
    case LOG_INFO: return EVENT_INFO_MSG;
    case LOG_NOTICE: return EVENT_NOTICE_MSG;
    case LOG_WARN: return EVENT_WARN_MSG;
    case LOG_ERR: return EVENT_ERR_MSG;
    default: return -1;
  }
}

/** Helper: clear bandwidth counters of all origin circuits. */
static void
clear_circ_bw_fields(void)
{
  origin_circuit_t *ocirc;
  SMARTLIST_FOREACH_BEGIN(circuit_get_global_list(), circuit_t *, circ) {
    if (!CIRCUIT_IS_ORIGIN(circ))
      continue;
    ocirc = TO_ORIGIN_CIRCUIT(circ);
    ocirc->n_written_circ_bw = ocirc->n_read_circ_bw = 0;
    ocirc->n_overhead_written_circ_bw = ocirc->n_overhead_read_circ_bw = 0;
    ocirc->n_delivered_written_circ_bw = ocirc->n_delivered_read_circ_bw = 0;
  }
  SMARTLIST_FOREACH_END(circ);
}

/** Set <b>global_event_mask*</b> to the bitwise OR of each live control
 * connection's event_mask field. */
void
control_update_global_event_mask(void)
{
  smartlist_t *conns = get_connection_array();
  event_mask_t old_mask, new_mask;
  old_mask = global_event_mask;
  int any_old_per_sec_events = control_any_per_second_event_enabled();

  global_event_mask = 0;
  SMARTLIST_FOREACH(conns, connection_t *, _conn,
  {
    if (_conn->type == CONN_TYPE_CONTROL &&
        STATE_IS_OPEN(_conn->state)) {
      control_connection_t *conn = TO_CONTROL_CONN(_conn);
      global_event_mask |= conn->event_mask;
    }
  });

  new_mask = global_event_mask;

  /* Handle the aftermath.  Set up the log callback to tell us only what
   * we want to hear...*/
  control_adjust_event_log_severity();

  /* Macro: true if ev was false before and is true now. */
#define NEWLY_ENABLED(ev) \
  (! (old_mask & (ev)) && (new_mask & (ev)))

  /* ...then, if we've started logging stream or circ bw, clear the
   * appropriate fields. */
  if (NEWLY_ENABLED(EVENT_STREAM_BANDWIDTH_USED)) {
    SMARTLIST_FOREACH(conns, connection_t *, conn,
    {
      if (conn->type == CONN_TYPE_AP) {
        edge_connection_t *edge_conn = TO_EDGE_CONN(conn);
        edge_conn->n_written = edge_conn->n_read = 0;
      }
    });
  }
  if (NEWLY_ENABLED(EVENT_CIRC_BANDWIDTH_USED)) {
    clear_circ_bw_fields();
  }
  if (NEWLY_ENABLED(EVENT_BANDWIDTH_USED)) {
    uint64_t r, w;
    control_get_bytes_rw_last_sec(&r, &w);
  }
  if (any_old_per_sec_events != control_any_per_second_event_enabled()) {
    reschedule_per_second_timer();
  }

#undef NEWLY_ENABLED
}

/** Adjust the log severities that result in control_event_logmsg being called
 * to match the severity of log messages that any controllers are interested
 * in. */
void
control_adjust_event_log_severity(void)
{
  int i;
  int min_log_event=EVENT_ERR_MSG, max_log_event=EVENT_DEBUG_MSG;

  for (i = EVENT_DEBUG_MSG; i <= EVENT_ERR_MSG; ++i) {
    if (EVENT_IS_INTERESTING(i)) {
      min_log_event = i;
      break;
    }
  }
  for (i = EVENT_ERR_MSG; i >= EVENT_DEBUG_MSG; --i) {
    if (EVENT_IS_INTERESTING(i)) {
      max_log_event = i;
      break;
    }
  }
  if (EVENT_IS_INTERESTING(EVENT_STATUS_GENERAL)) {
    if (min_log_event > EVENT_NOTICE_MSG)
      min_log_event = EVENT_NOTICE_MSG;
    if (max_log_event < EVENT_ERR_MSG)
      max_log_event = EVENT_ERR_MSG;
  }
  if (min_log_event <= max_log_event)
    change_callback_log_severity(event_to_log_severity(min_log_event),
                                 event_to_log_severity(max_log_event),
                                 control_event_logmsg);
  else
    change_callback_log_severity(LOG_ERR, LOG_ERR,
                                 control_event_logmsg);
}

/** Return true iff the event with code <b>c</b> is being sent to any current
 * control connection.  This is useful if the amount of work needed to prepare
 * to call the appropriate control_event_...() function is high.
 */
int
control_event_is_interesting(int event)
{
  return EVENT_IS_INTERESTING(event);
}

/** Return true if any event that needs to fire once a second is enabled. */
int
control_any_per_second_event_enabled(void)
{
  return ANY_EVENT_IS_INTERESTING(
      EVENT_MASK_(EVENT_BANDWIDTH_USED) |
      EVENT_MASK_(EVENT_CELL_STATS) |
      EVENT_MASK_(EVENT_CIRC_BANDWIDTH_USED) |
      EVENT_MASK_(EVENT_CONN_BW) |
      EVENT_MASK_(EVENT_STREAM_BANDWIDTH_USED)
  );
}

/* The value of 'get_bytes_read()' the previous time that
 * control_get_bytes_rw_last_sec() as called. */
static uint64_t stats_prev_n_read = 0;
/* The value of 'get_bytes_written()' the previous time that
 * control_get_bytes_rw_last_sec() as called. */
static uint64_t stats_prev_n_written = 0;

/**
 * Set <b>n_read</b> and <b>n_written</b> to the total number of bytes read
 * and written by Tor since the last call to this function.
 *
 * Call this only from the main thread.
 */
static void
control_get_bytes_rw_last_sec(uint64_t *n_read,
                              uint64_t *n_written)
{
  const uint64_t stats_n_bytes_read = get_bytes_read();
  const uint64_t stats_n_bytes_written = get_bytes_written();

  *n_read = stats_n_bytes_read - stats_prev_n_read;
  *n_written = stats_n_bytes_written - stats_prev_n_written;
  stats_prev_n_read = stats_n_bytes_read;
  stats_prev_n_written = stats_n_bytes_written;
}

/**
 * Run all the controller events (if any) that are scheduled to trigger once
 * per second.
 */
void
control_per_second_events(void)
{
  if (!control_any_per_second_event_enabled())
    return;

  uint64_t bytes_read, bytes_written;
  control_get_bytes_rw_last_sec(&bytes_read, &bytes_written);
  control_event_bandwidth_used((uint32_t)bytes_read,(uint32_t)bytes_written);

  control_event_stream_bandwidth_used();
  control_event_conn_bandwidth_used();
  control_event_circ_bandwidth_used();
  control_event_circuit_cell_stats();
}

/** Append a NUL-terminated string <b>s</b> to the end of
 * <b>conn</b>-\>outbuf.
 */
static inline void
connection_write_str_to_buf(const char *s, control_connection_t *conn)
{
  size_t len = strlen(s);
  connection_buf_add(s, len, TO_CONN(conn));
}

/** Given a <b>len</b>-character string in <b>data</b>, made of lines
 * terminated by CRLF, allocate a new string in *<b>out</b>, and copy the
 * contents of <b>data</b> into *<b>out</b>, adding a period before any period
 * that appears at the start of a line, and adding a period-CRLF line at
 * the end. Replace all LF characters sequences with CRLF.  Return the number
 * of bytes in *<b>out</b>.
 */
STATIC size_t
write_escaped_data(const char *data, size_t len, char **out)
{
  tor_assert(len < SIZE_MAX - 9);
  size_t sz_out = len+8+1;
  char *outp;
  const char *start = data, *end;
  size_t i;
  int start_of_line;
  for (i=0; i < len; ++i) {
    if (data[i] == '\n') {
      sz_out += 2; /* Maybe add a CR; maybe add a dot. */
      if (sz_out >= SIZE_T_CEILING) {
        log_warn(LD_BUG, "Input to write_escaped_data was too long");
        *out = tor_strdup(".\r\n");
        return 3;
      }
    }
  }
  *out = outp = tor_malloc(sz_out);
  end = data+len;
  start_of_line = 1;
  while (data < end) {
    if (*data == '\n') {
      if (data > start && data[-1] != '\r')
        *outp++ = '\r';
      start_of_line = 1;
    } else if (*data == '.') {
      if (start_of_line) {
        start_of_line = 0;
        *outp++ = '.';
      }
    } else {
      start_of_line = 0;
    }
    *outp++ = *data++;
  }
  if (outp < *out+2 || fast_memcmp(outp-2, "\r\n", 2)) {
    *outp++ = '\r';
    *outp++ = '\n';
  }
  *outp++ = '.';
  *outp++ = '\r';
  *outp++ = '\n';
  *outp = '\0'; /* NUL-terminate just in case. */
  tor_assert(outp >= *out);
  tor_assert((size_t)(outp - *out) <= sz_out);
  return outp - *out;
}

/** Given a <b>len</b>-character string in <b>data</b>, made of lines
 * terminated by CRLF, allocate a new string in *<b>out</b>, and copy
 * the contents of <b>data</b> into *<b>out</b>, removing any period
 * that appears at the start of a line, and replacing all CRLF sequences
 * with LF.   Return the number of
 * bytes in *<b>out</b>. */
STATIC size_t
read_escaped_data(const char *data, size_t len, char **out)
{
  char *outp;
  const char *next;
  const char *end;

  *out = outp = tor_malloc(len+1);

  end = data+len;

  while (data < end) {
    /* we're at the start of a line. */
    if (*data == '.')
      ++data;
    next = memchr(data, '\n', end-data);
    if (next) {
      size_t n_to_copy = next-data;
      /* Don't copy a CR that precedes this LF. */
      if (n_to_copy && *(next-1) == '\r')
        --n_to_copy;
      memcpy(outp, data, n_to_copy);
      outp += n_to_copy;
      data = next+1; /* This will point at the start of the next line,
                      * or the end of the string, or a period. */
    } else {
      memcpy(outp, data, end-data);
      outp += (end-data);
      *outp = '\0';
      return outp - *out;
    }
    *outp++ = '\n';
  }

  *outp = '\0';
  return outp - *out;
}

/** If the first <b>in_len_max</b> characters in <b>start</b> contain a
 * double-quoted string with escaped characters, return the length of that
 * string (as encoded, including quotes).  Otherwise return -1. */
static inline int
get_escaped_string_length(const char *start, size_t in_len_max,
                          int *chars_out)
{
  const char *cp, *end;
  int chars = 0;

  if (*start != '\"')
    return -1;

  cp = start+1;
  end = start+in_len_max;

  /* Calculate length. */
  while (1) {
    if (cp >= end) {
      return -1; /* Too long. */
    } else if (*cp == '\\') {
      if (++cp == end)
        return -1; /* Can't escape EOS. */
      ++cp;
      ++chars;
    } else if (*cp == '\"') {
      break;
    } else {
      ++cp;
      ++chars;
    }
  }
  if (chars_out)
    *chars_out = chars;
  return (int)(cp - start+1);
}

/** As decode_escaped_string, but does not decode the string: copies the
 * entire thing, including quotation marks. */
static const char *
extract_escaped_string(const char *start, size_t in_len_max,
                       char **out, size_t *out_len)
{
  int length = get_escaped_string_length(start, in_len_max, NULL);
  if (length<0)
    return NULL;
  *out_len = length;
  *out = tor_strndup(start, *out_len);
  return start+length;
}

/** Given a pointer to a string starting at <b>start</b> containing
 * <b>in_len_max</b> characters, decode a string beginning with one double
 * quote, containing any number of non-quote characters or characters escaped
 * with a backslash, and ending with a final double quote.  Place the resulting
 * string (unquoted, unescaped) into a newly allocated string in *<b>out</b>;
 * store its length in <b>out_len</b>.  On success, return a pointer to the
 * character immediately following the escaped string.  On failure, return
 * NULL. */
static const char *
decode_escaped_string(const char *start, size_t in_len_max,
                   char **out, size_t *out_len)
{
  const char *cp, *end;
  char *outp;
  int len, n_chars = 0;

  len = get_escaped_string_length(start, in_len_max, &n_chars);
  if (len<0)
    return NULL;

  end = start+len-1; /* Index of last quote. */
  tor_assert(*end == '\"');
  outp = *out = tor_malloc(len+1);
  *out_len = n_chars;

  cp = start+1;
  while (cp < end) {
    if (*cp == '\\')
      ++cp;
    *outp++ = *cp++;
  }
  *outp = '\0';
  tor_assert((outp - *out) == (int)*out_len);

  return end+1;
}

/** Create and add a new controller connection on <b>sock</b>.  If
 * <b>CC_LOCAL_FD_IS_OWNER</b> is set in <b>flags</b>, this Tor process should
 * exit when the connection closes.  If <b>CC_LOCAL_FD_IS_AUTHENTICATED</b>
 * is set, then the connection does not need to authenticate.
 */
int
control_connection_add_local_fd(tor_socket_t sock, unsigned flags)
{
  if (BUG(! SOCKET_OK(sock)))
    return -1;
  const int is_owner = !!(flags & CC_LOCAL_FD_IS_OWNER);
  const int is_authenticated = !!(flags & CC_LOCAL_FD_IS_AUTHENTICATED);
  control_connection_t *control_conn = control_connection_new(AF_UNSPEC);
  connection_t *conn = TO_CONN(control_conn);
  conn->s = sock;
  tor_addr_make_unspec(&conn->addr);
  conn->port = 1;
  conn->address = tor_strdup("<local socket>");

  /* We take ownership of this socket so that later, when we close it,
   * we don't freak out. */
  tor_take_socket_ownership(sock);

  if (set_socket_nonblocking(sock) < 0 ||
      connection_add(conn) < 0) {
    connection_free(conn);
    return -1;
  }

  control_conn->is_owning_control_connection = is_owner;

  if (connection_init_accepted_conn(conn, NULL) < 0) {
    connection_mark_for_close(conn);
    return -1;
  }

  if (is_authenticated) {
    conn->state = CONTROL_CONN_STATE_OPEN;
  }

  return 0;
}

/** Acts like sprintf, but writes its formatted string to the end of
 * <b>conn</b>-\>outbuf. */
static void
connection_printf_to_buf(control_connection_t *conn, const char *format, ...)
{
  va_list ap;
  char *buf = NULL;
  int len;

  va_start(ap,format);
  len = tor_vasprintf(&buf, format, ap);
  va_end(ap);

  if (len < 0) {
    log_err(LD_BUG, "Unable to format string for controller.");
    tor_assert(0);
  }

  connection_buf_add(buf, (size_t)len, TO_CONN(conn));

  tor_free(buf);
}

/** Write all of the open control ports to ControlPortWriteToFile */
void
control_ports_write_to_file(void)
{
  smartlist_t *lines;
  char *joined = NULL;
  const or_options_t *options = get_options();

  if (!options->ControlPortWriteToFile)
    return;

  lines = smartlist_new();

  SMARTLIST_FOREACH_BEGIN(get_connection_array(), const connection_t *, conn) {
    if (conn->type != CONN_TYPE_CONTROL_LISTENER || conn->marked_for_close)
      continue;
#ifdef AF_UNIX
    if (conn->socket_family == AF_UNIX) {
      smartlist_add_asprintf(lines, "UNIX_PORT=%s\n", conn->address);
      continue;
    }
#endif /* defined(AF_UNIX) */
    smartlist_add_asprintf(lines, "PORT=%s:%d\n", conn->address, conn->port);
  } SMARTLIST_FOREACH_END(conn);

  joined = smartlist_join_strings(lines, "", 0, NULL);

  if (write_str_to_file(options->ControlPortWriteToFile, joined, 0) < 0) {
    log_warn(LD_CONTROL, "Writing %s failed: %s",
             options->ControlPortWriteToFile, strerror(errno));
  }
#ifndef _WIN32
  if (options->ControlPortFileGroupReadable) {
    if (chmod(options->ControlPortWriteToFile, 0640)) {
      log_warn(LD_FS,"Unable to make %s group-readable.",
               options->ControlPortWriteToFile);
    }
  }
#endif /* !defined(_WIN32) */
  tor_free(joined);
  SMARTLIST_FOREACH(lines, char *, cp, tor_free(cp));
  smartlist_free(lines);
}

/** Send a "DONE" message down the control connection <b>conn</b>. */
static void
send_control_done(control_connection_t *conn)
{
  connection_write_str_to_buf("250 OK\r\n", conn);
}

/** Represents an event that's queued to be sent to one or more
 * controllers. */
typedef struct queued_event_s {
  uint16_t event;
  char *msg;
} queued_event_t;

/** Pointer to int. If this is greater than 0, we don't allow new events to be
 * queued. */
static tor_threadlocal_t block_event_queue_flag;

/** Holds a smartlist of queued_event_t objects that may need to be sent
 * to one or more controllers */
static smartlist_t *queued_control_events = NULL;

/** True if the flush_queued_events_event is pending. */
static int flush_queued_event_pending = 0;

/** Lock to protect the above fields. */
static tor_mutex_t *queued_control_events_lock = NULL;

/** An event that should fire in order to flush the contents of
 * queued_control_events. */
static mainloop_event_t *flush_queued_events_event = NULL;

void
control_initialize_event_queue(void)
{
  if (queued_control_events == NULL) {
    queued_control_events = smartlist_new();
  }

  if (flush_queued_events_event == NULL) {
    struct event_base *b = tor_libevent_get_base();
    if (b) {
      flush_queued_events_event =
        mainloop_event_new(flush_queued_events_cb, NULL);
      tor_assert(flush_queued_events_event);
    }
  }

  if (queued_control_events_lock == NULL) {
    queued_control_events_lock = tor_mutex_new();
    tor_threadlocal_init(&block_event_queue_flag);
  }
}

static int *
get_block_event_queue(void)
{
  int *val = tor_threadlocal_get(&block_event_queue_flag);
  if (PREDICT_UNLIKELY(val == NULL)) {
    val = tor_malloc_zero(sizeof(int));
    tor_threadlocal_set(&block_event_queue_flag, val);
  }
  return val;
}

/** Helper: inserts an event on the list of events queued to be sent to
 * one or more controllers, and schedules the events to be flushed if needed.
 *
 * This function takes ownership of <b>msg</b>, and may free it.
 *
 * We queue these events rather than send them immediately in order to break
 * the dependency in our callgraph from code that generates events for the
 * controller, and the network layer at large.  Otherwise, nearly every
 * interesting part of Tor would potentially call every other interesting part
 * of Tor.
 */
MOCK_IMPL(STATIC void,
queue_control_event_string,(uint16_t event, char *msg))
{
  /* This is redundant with checks done elsewhere, but it's a last-ditch
   * attempt to avoid queueing something we shouldn't have to queue. */
  if (PREDICT_UNLIKELY( ! EVENT_IS_INTERESTING(event) )) {
    tor_free(msg);
    return;
  }

  int *block_event_queue = get_block_event_queue();
  if (*block_event_queue) {
    tor_free(msg);
    return;
  }

  queued_event_t *ev = tor_malloc(sizeof(*ev));
  ev->event = event;
  ev->msg = msg;

  /* No queueing an event while queueing an event */
  ++*block_event_queue;

  tor_mutex_acquire(queued_control_events_lock);
  tor_assert(queued_control_events);
  smartlist_add(queued_control_events, ev);

  int activate_event = 0;
  if (! flush_queued_event_pending && in_main_thread()) {
    activate_event = 1;
    flush_queued_event_pending = 1;
  }

  tor_mutex_release(queued_control_events_lock);

  --*block_event_queue;

  /* We just put an event on the queue; mark the queue to be
   * flushed.  We only do this from the main thread for now; otherwise,
   * we'd need to incur locking overhead in Libevent or use a socket.
   */
  if (activate_event) {
    tor_assert(flush_queued_events_event);
    mainloop_event_activate(flush_queued_events_event);
  }
}

#define queued_event_free(ev) \
  FREE_AND_NULL(queued_event_t, queued_event_free_, (ev))

/** Release all storage held by <b>ev</b>. */
static void
queued_event_free_(queued_event_t *ev)
{
  if (ev == NULL)
    return;

  tor_free(ev->msg);
  tor_free(ev);
}

/** Send every queued event to every controller that's interested in it,
 * and remove the events from the queue.  If <b>force</b> is true,
 * then make all controllers send their data out immediately, since we
 * may be about to shut down. */
static void
queued_events_flush_all(int force)
{
  /* Make sure that we get all the pending log events, if there are any. */
  flush_pending_log_callbacks();

  if (PREDICT_UNLIKELY(queued_control_events == NULL)) {
    return;
  }
  smartlist_t *all_conns = get_connection_array();
  smartlist_t *controllers = smartlist_new();
  smartlist_t *queued_events;

  int *block_event_queue = get_block_event_queue();
  ++*block_event_queue;

  tor_mutex_acquire(queued_control_events_lock);
  /* No queueing an event while flushing events. */
  flush_queued_event_pending = 0;
  queued_events = queued_control_events;
  queued_control_events = smartlist_new();
  tor_mutex_release(queued_control_events_lock);

  /* Gather all the controllers that will care... */
  SMARTLIST_FOREACH_BEGIN(all_conns, connection_t *, conn) {
    if (conn->type == CONN_TYPE_CONTROL &&
        !conn->marked_for_close &&
        conn->state == CONTROL_CONN_STATE_OPEN) {
      control_connection_t *control_conn = TO_CONTROL_CONN(conn);

      smartlist_add(controllers, control_conn);
    }
  } SMARTLIST_FOREACH_END(conn);

  SMARTLIST_FOREACH_BEGIN(queued_events, queued_event_t *, ev) {
    const event_mask_t bit = ((event_mask_t)1) << ev->event;
    const size_t msg_len = strlen(ev->msg);
    SMARTLIST_FOREACH_BEGIN(controllers, control_connection_t *,
                            control_conn) {
      if (control_conn->event_mask & bit) {
        connection_buf_add(ev->msg, msg_len, TO_CONN(control_conn));
      }
    } SMARTLIST_FOREACH_END(control_conn);

    queued_event_free(ev);
  } SMARTLIST_FOREACH_END(ev);

  if (force) {
    SMARTLIST_FOREACH_BEGIN(controllers, control_connection_t *,
                            control_conn) {
      connection_flush(TO_CONN(control_conn));
    } SMARTLIST_FOREACH_END(control_conn);
  }

  smartlist_free(queued_events);
  smartlist_free(controllers);

  --*block_event_queue;
}

/** Libevent callback: Flushes pending events to controllers that are
 * interested in them. */
static void
flush_queued_events_cb(mainloop_event_t *event, void *arg)
{
  (void) event;
  (void) arg;
  queued_events_flush_all(0);
}