Web2socket proxy.

net/web2socket_proxy/web2socket_conn.cc

Index: net/web2socket_proxy/web2socket_conn.cc
diff --git a/net/web2socket_proxy/web2socket_conn.cc b/net/web2socket_proxy/web2socket_conn.cc
new file mode 100644
index 0000000000000000000000000000000000000000..b0c86b39de9aa5080b70841b7425a0b4457b173b
--- /dev/null
+++ b/net/web2socket_proxy/web2socket_conn.cc
@@ -0,0 +1,779 @@
+// Copyright (c) 2010 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "web2socket_conn.h"
+
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+
+#include <algorithm>
+#include <limits>
+#include <vector>
+
+#include <arpa/inet.h>
+#include <errno.h>
+#include <netinet/in.h>
+#include <signal.h>
+#include <sys/types.h>
+#include <sys/wait.h>
+
+#include "base/base64.h"
+#include "base/logging.h"
+#include "third_party/libevent/evdns.h"
+
+static int CountSpaces(const std::string& s) {
+  int rv = 0;
+  for (size_t i = 0; i < s.size(); ++i) {
+    rv += (isascii(s[i]) && isspace(s[i]));

[tyoshino (SeeGerritForStatus), 2010/12/03 07:53:33]

If you don't have to keep this code independent from Chromium tree, how about
using functions defined in base/string_util.h that are locale independent
instead of C library?

[Denis Lagno, 2010/12/03 16:28:50]

actually we don't need to be locale-independent here since protocol strictly
specifies that there is sequence of octets in UTF-8.

And actually protocol (the version used ATM) specifies that we should count
spaces as 0x20 octets and not any whitespace as isspace does.  Since it is UTF-8
so just counting 0x20 octets does the job.

[tyoshino (SeeGerritForStatus), 2010/12/06 10:02:51]

e.g. isdigit may recognize some characters other than '0' - '9' on system using
some code page. It's safe to use base/string_util.h to make the implementation
accept only input that follows the spec strictly.
Yes, that's best.

+  }
+  return rv;
+}
+
+static std::string FetchLowerCasedASCIISnippet(uint8_t* begin, uint8_t* end) {
+  std::string rv;
+  for (; begin < end; ++begin) {
+    if (!isascii(*begin)) {
+      return rv;
+    }
+    rv += tolower(*begin);

[tyoshino (SeeGerritForStatus), 2010/12/03 07:53:33]

ditto. ToLowerASCII

[Denis Lagno, 2010/12/03 16:28:50]

Done.

+  }
+  return rv;
+}
+
+// Returns true on success.
+static bool FetchDecimalDigits(const std::string& s, uint32_t* result) {
+  *result = 0;
+  bool got_something = false;
+  for (size_t i = 0; i < s.size(); ++i) {
+    if (isascii(s[i]) && isdigit(s[i])) {
+      got_something = true;
+      if (*result > std::numeric_limits<uint32_t>::max() / 10) {
+        return false;
+      }
+      *result *= 10;
+      int digit = s[i] - '0';
+      if (*result > std::numeric_limits<uint32_t>::max() - digit) {
+        return false;
+      }
+      *result += digit;
+    }
+  }
+  return got_something;
+}
+
+// Returns true on success.
+static bool FetchNamePort(uint8_t* begin, uint8_t* end,
+                          std::string* name, uint32_t* port) {
+  *name = std::string();
+  *port = 0;
+  if (begin == end) {
+    return false;
+  }
+  for (uint8_t* pos = end; --pos != begin;) {
+    if (!isascii(*pos) || !isdigit(*pos)) {

[tyoshino (SeeGerritForStatus), 2010/12/03 07:53:33]

How about isascii(*pos) && isdigit(*pos)?
But if so, *pos will be never ':' below.

[Denis Lagno, 2010/12/03 16:28:50]

actually !x || !y is !(x && y), not just (x && y).
Anyway I reorganized/simplified code here utilizing IsAsciiDigit from
base/string_util.h

[tyoshino (SeeGerritForStatus), 2010/12/06 10:02:51]

Oops, I put wrong comment. You're right. Anyway, thanks for reorganizing.

+      if (*pos == ':') {
+        name->assign(begin, pos);
+        if (end - pos <= 1 ||
+            !FetchDecimalDigits(std::string(pos + 1, end), port)) {
+          *port = 0;
+        }
+      }
+      break;
+    }
+  }
+  if (*port <= 0 || *port >= (2 << 16)) {

[tyoshino (SeeGerritForStatus), 2010/12/03 07:53:33]

Up to your design, but how about handling erroneous port number (>= 2^16) as
failure.

[Denis Lagno, 2010/12/03 16:28:50]

Done.

+    name->assign(begin, end);
+    *port = 22;
+  }
+  return !name->empty();
+}
+
+Conn::Conn(Web2SocketServ* master)
+    : master_(master),
+      phase_(PHASE_WAIT_HANDSHAKE),
+      destresolution_ipv4_failure_(false),
+      destresolution_ipv6_failure_(false) {
+  while (token_map_.find(last_token_) != token_map_.end()) {
+    token_ = last_token_ =
+        reinterpret_cast<Token>(reinterpret_cast<size_t>(last_token_) + 1);
+  }
+  token_map_[token_] = this;
+  // Schedule timeout for initial phase of connection.
+  evtimer_set(&destconnect_timeout_event_,
+              &OnDestConnectTimeout, token_);
+  event_base_set(master_->evbase_,
+                 &destconnect_timeout_event_);
+  struct timeval tv;
+  tv.tv_sec = 30;
+  tv.tv_usec = 0;
+  evtimer_add(&destconnect_timeout_event_, &tv);
+}
+
+Conn::~Conn() {
+  phase_ = PHASE_DEFUNCT;
+  event_del(&destconnect_timeout_event_);
+  if (token_map_[token_] == this) {
+    token_map_.erase(token_);
+  }
+}
+
+Conn* Conn::Get(Token token) {
+  TokenMap::iterator it = token_map_.find(token);
+  if (it == token_map_.end()) {
+    return NULL;
+  }
+  Conn* cs = it->second;
+  if (cs == NULL ||
+      cs->token_ != token ||
+      cs->master_ == NULL ||
+      cs->phase_ < 0 ||
+      cs->phase_ > PHASE_SHUT ||
+      !cs->master_->IsConnSane(cs)) {
+    return NULL;
+  }
+  return cs;
+}
+
+void Conn::Shut() {
+  if (phase_ >= PHASE_SHUT) {
+    return;
+  }
+  master_->MarkConnImportance(this, false);
+  if (primchan_.sock_ >= 0 && primchan_.bev_ != NULL) {
+    static const uint8_t closing_handshake[9] = { 0 };
+    bufferevent_write(primchan_.bev_,
+                      closing_handshake, sizeof(closing_handshake));
+  }
+  phase_ = PHASE_SHUT;
+}
+
+// For our purposes equivalent to strlen(reinterpret_cast<const char*>(X)).
+#define RAWSIZEOF(X) (sizeof(X) - 1)
+
+Conn::Status Conn::ConsumeHeader(struct evbuffer* evb) {
+  uint8_t* buf = EVBUFFER_DATA(evb);
+  size_t buf_size = EVBUFFER_LENGTH(evb);
+
+  static const uint8_t kCRLF[] = "\r\n";
+  static const uint8_t kCRLFCRLF[] = "\r\n\r\n";
+  static const uint8_t kGetMagic[] = "GET /echo ";
+  static const uint8_t kDelimiter[] = ": ";
+
+  if (buf_size <= 0) {
+    return STATUS_INCOMPLETE;
+  }
+  if (!buf) {
+    return STATUS_ABORT;
+  }
+  if (!std::equal(buf, buf + std::min(buf_size, RAWSIZEOF(kGetMagic)),
+                  kGetMagic)) {
+    // Data head does not match what is expected.
+    return STATUS_ABORT;
+  }
+
+  if (buf_size >= Web2SocketServ::kReadBufferLimit) {
+    return STATUS_ABORT;
+  }
+  uint8_t* buf_end = buf + buf_size;
+  uint8_t* term_pos = std::search(buf, buf_end, kCRLFCRLF,
+                                  kCRLFCRLF + RAWSIZEOF(kCRLFCRLF));
+  if (term_pos == buf_end) {
+    return STATUS_INCOMPLETE;
+  }
+  uint8_t key3[8]; // Notation (key3) matches websocket RFC.

[tyoshino (SeeGerritForStatus), 2010/12/03 07:53:33]

Two space before inline comment

[Denis Lagno, 2010/12/03 16:28:50]

Done.

+  if (buf_end - term_pos <
+      static_cast<ptrdiff_t>(RAWSIZEOF(kCRLFCRLF) + sizeof(key3))) {
+    return STATUS_INCOMPLETE;
+  }
+  term_pos += RAWSIZEOF(kCRLFCRLF) + sizeof(key3);
+  memcpy(key3, term_pos - sizeof(key3), sizeof(key3));

[tyoshino (SeeGerritForStatus), 2010/12/03 07:53:33]

how about swapping these lines to remove "- sizeof(key3)"

[Denis Lagno, 2010/12/03 16:28:50]

Done.

+  // First line is "GET /echo" line, so we skip it.
+  uint8_t* pos = std::search(buf, term_pos, kCRLF, kCRLF + RAWSIZEOF(kCRLF));
+  if (pos == term_pos) {
+    return STATUS_ABORT;
+  }
+  for (;;) {
+    pos += RAWSIZEOF(kCRLF);
+    if (term_pos - pos <
+        static_cast<ptrdiff_t>(sizeof(key3) + RAWSIZEOF(kCRLF))) {
+      return STATUS_ABORT;
+    }
+    if (term_pos - pos ==
+        static_cast<ptrdiff_t>(sizeof(key3) + RAWSIZEOF(kCRLF))) {
+      break;
+    }
+    uint8_t* npos = std::search(pos, term_pos, kDelimiter,
+                                kDelimiter + RAWSIZEOF(kDelimiter));
+    if (npos == term_pos) {
+      return STATUS_ABORT;
+    }
+    std::string key = FetchLowerCasedASCIISnippet(pos, npos);
+    pos = std::search(npos += RAWSIZEOF(kDelimiter), term_pos,
+                      kCRLF, kCRLF + RAWSIZEOF(kCRLF));
+    if (pos == term_pos) {
+      return STATUS_ABORT;
+    }
+    if (!key.empty()) {
+      header_fields_[key] = FetchLowerCasedASCIISnippet(npos, pos);
+    }
+  }
+
+  // Values of Upgrade and Connection fields are hardcoded in the protocol.
+  if (header_fields_["upgrade"] != "websocket" ||
+      header_fields_["connection"] != "upgrade") {
+    return STATUS_ABORT;
+  }
+
+  static const std::string kAllowedHost = "127.0.0.1:";
+  if (header_fields_["host"].substr(0, kAllowedHost.size()) !=
+      kAllowedHost) {
+    return STATUS_ABORT;
+  }
+
+  if (!master_->origin_.empty() &&
+      master_->origin_ != "any") {
+    if (header_fields_["origin"] != master_->origin_) {
+      return STATUS_ABORT;
+    }
+  }
+
+  static const std::string kSecKey1 = "sec-websocket-key1";
+  static const std::string kSecKey2 = "sec-websocket-key2";
+  uint32_t key_number1, key_number2;
+  if (!FetchDecimalDigits(header_fields_[kSecKey1],
+                          &key_number1) ||
+      !FetchDecimalDigits(header_fields_[kSecKey2],
+                          &key_number2)) {
+    return STATUS_ABORT;
+  }
+
+  // We limit incoming header size so following numbers shall not be too high.
+  int spaces1 = CountSpaces(header_fields_[kSecKey1]);
+  int spaces2 = CountSpaces(header_fields_[kSecKey2]);
+  if (spaces1 == 0 ||
+      spaces2 == 0 ||
+      key_number1 % spaces1 != 0 ||
+      key_number2 % spaces2 != 0) {
+    return STATUS_ABORT;
+  }
+
+  uint8_t challenge[4 + 4 + sizeof(key3)];
+  uint32_t part1 = htonl(key_number1 / spaces1);
+  uint32_t part2 = htonl(key_number2 / spaces2);
+  memcpy(challenge, &part1, 4);
+  memcpy(challenge + 4, &part2, 4);
+  memcpy(challenge + sizeof(challenge) - sizeof(key3), key3, sizeof(key3));
+  MD5Sum(challenge, sizeof(challenge), &handshake_response_);
+
+  evbuffer_drain(evb, term_pos - buf);
+  return STATUS_OK;
+}
+
+bool Conn::EmitHandshake(struct bufferevent* bev) {
+  std::vector<std::string> boilerplate;
+  boilerplate.push_back("HTTP/1.1 101 WebSocket Protocol Handshake");
+  boilerplate.push_back("Upgrade: WebSocket");
+  boilerplate.push_back("Connection: Upgrade");
+
+  {
+    // Take care of Location field.
+    char buf[128];
+    snprintf(buf, sizeof(buf),

[tyoshino (SeeGerritForStatus), 2010/12/03 07:53:33]

check return value?

[Denis Lagno, 2010/12/03 16:28:50]

Done.

+             "Sec-WebSocket-Location: ws://%s/echo",
+             header_fields_["host"].c_str());
+    boilerplate.push_back(buf);
+  }
+  {
+    // Take care of Origin field.
+    if (header_fields_.find("origin") != header_fields_.end()) {
+      char buf[128];
+      snprintf(buf, sizeof(buf),
+               "Sec-WebSocket-Origin: %s", header_fields_["origin"].c_str());
+      boilerplate.push_back(buf);
+    }
+  }
+
+  boilerplate.push_back("");
+  static const uint8_t kCRLF[] = "\r\n";
+  for (size_t i = 0; i < boilerplate.size(); ++i) {
+    if (bufferevent_write(bev, boilerplate[i].c_str(),
+                          boilerplate[i].size()) ||
+        bufferevent_write(bev, kCRLF, RAWSIZEOF(kCRLF))) {
+      return false;
+    }
+  }
+  return !bufferevent_write(bev, &handshake_response_,
+                            sizeof(handshake_response_));
+}
+
+Conn::Status Conn::ConsumeDestframe(struct evbuffer* evb) {
+  uint8_t* buf = EVBUFFER_DATA(evb);
+  size_t buf_size = EVBUFFER_LENGTH(evb);
+
+  if (buf_size < 1) {
+    return STATUS_INCOMPLETE;
+  }
+  if (buf[0] != 0) {
+    return STATUS_ABORT;
+  }
+  if (buf_size < 1 + 1) {
+    return STATUS_INCOMPLETE;
+  }
+  uint8_t* buf_end = buf + buf_size;
+  uint8_t* term_pos = std::find(buf + 1, buf_end, 0xff);
+  if (term_pos == buf_end) {
+    if (buf_size >= Web2SocketServ::kReadBufferLimit) {
+      // So big and still worth nothing.
+      return STATUS_ABORT;
+    }
+    return STATUS_INCOMPLETE;
+  }
+  if (!FetchNamePort(buf + 1, term_pos, &destname_, &destport_)) {
+    return STATUS_ABORT;
+  }
+  evbuffer_drain(evb, term_pos - buf + 1);
+  return STATUS_OK;
+}
+
+Conn::Status Conn::ConsumeFrameHeader(struct evbuffer* evb) {
+  uint8_t* buf = EVBUFFER_DATA(evb);
+  size_t buf_size = EVBUFFER_LENGTH(evb);
+
+  if (buf_size < 1) {
+    return STATUS_INCOMPLETE;
+  }
+  if (buf[0] != 0) {
+    return STATUS_ABORT;
+  }
+  evbuffer_drain(evb, 1);
+  return STATUS_OK;
+}
+
+Conn::Status Conn::ProcessFrameData(struct evbuffer* evb) {
+  uint8_t* buf = EVBUFFER_DATA(evb);
+  size_t buf_size = EVBUFFER_LENGTH(evb);
+
+  if (buf_size < 1) {

[tyoshino (SeeGerritForStatus), 2010/12/03 07:53:33]

We can remove if-clause?

[Denis Lagno, 2010/12/03 16:28:50]

It shouldn't happen in practice, but just a safety check in case libevent calls
us back with empty buffer (kind of spurious wakeup).

+    return STATUS_INCOMPLETE;
+  }
+  uint8_t* buf_end = buf + buf_size;
+  uint8_t* term_pos = std::find(buf, buf_end, 0xff);
+  bool term_detected = (term_pos != buf_end);
+  if (term_detected) {
+    buf_size = term_pos - buf;
+  }
+  switch (phase_) {
+    case PHASE_INSIDE_FRAME_BASE64: {
+      if (term_detected && buf_size % 4) {
+        // base64 is encoded in chunks of 4 bytes.
+        return STATUS_ABORT;
+      }
+      if (buf_size < 4) {
+        DCHECK(!term_detected);
+        return STATUS_INCOMPLETE;
+      }
+      size_t bytes_to_process_atm = (buf_size / 4) * 4;
+      std::string out_bytes;
+      base::Base64Decode(std::string(buf, buf + bytes_to_process_atm),
+                         &out_bytes);
+      evbuffer_drain(evb, bytes_to_process_atm);
+      DCHECK(destchan_.bev_ != NULL);
+      if (bufferevent_write(destchan_.bev_,
+                            out_bytes.c_str(), out_bytes.size())) {
+        return STATUS_ABORT;
+      }
+      break;
+    }
+    case PHASE_INSIDE_FRAME_SKIP: {
+      evbuffer_drain(evb, buf_size);
+      break;
+    }
+    default: {
+      return STATUS_ABORT;
+    }
+  }
+  if (term_detected) {
+    evbuffer_drain(evb, 1);
+    return STATUS_OK;
+  }
+  return STATUS_INCOMPLETE;
+}
+
+bool Conn::TryConnectDest(const struct sockaddr* addr,
+                          socklen_t addrlen) {
+  if (destchan_.sock_ >= 0 ||
+      destchan_.bev_ != NULL) {
+    return false;
+  }
+  destchan_.sock_ = socket(addr->sa_family, SOCK_STREAM, 0);
+  if (destchan_.sock_ < 0) {
+    return false;
+  }
+  if (!Web2SocketServ::SetNonBlock(destchan_.sock_)) {
+    return false;
+  }
+  if (connect(destchan_.sock_, addr, addrlen)) {
+    if (errno != EINPROGRESS) {
+      return false;
+    }
+  }
+  destchan_.bev_ = bufferevent_new(destchan_.sock_,
+                                   &OnDestchanRead,
+                                   &OnDestchanWrite,
+                                   &OnDestchanError,
+                                   token_);
+  if (destchan_.bev_ == NULL) {
+    return false;
+  }
+  if (bufferevent_base_set(master_->evbase_, destchan_.bev_)) {
+    return false;
+  }
+  bufferevent_setwatermark(destchan_.bev_, EV_READ,
+                           0, Web2SocketServ::kReadBufferLimit);
+  if (bufferevent_enable(destchan_.bev_, EV_READ | EV_WRITE)) {
+    return false;
+  }
+  return true;
+}
+
+void Conn::OnPrimchanRead(struct bufferevent* bev, Token token) {
+  Conn* cs = Conn::Get(token);
+  if (bev == NULL ||
+      cs == NULL ||
+      bev != cs->primchan_.bev_) {
+    // Sanity check failed.
+    return;
+  }
+  if (EVBUFFER_LENGTH(EVBUFFER_INPUT(bev)) <= 0) {
+    return;
+  }
+  cs->master_->MarkConnImportance(cs, true);
+  for (;;) {
+    switch (cs->phase_) {
+      case PHASE_WAIT_HANDSHAKE: {
+        switch (cs->ConsumeHeader(EVBUFFER_INPUT(bev))) {
+          case STATUS_OK: {
+            break;
+          }
+          case STATUS_INCOMPLETE: {
+            return;
+          }
+          case STATUS_ABORT:
+          default: {
+            cs->master_->ZapConn(cs);
+            return;
+          }
+        }
+        // Header consumed OK.  Do respond.
+        if (!cs->EmitHandshake(bev)) {
+          cs->master_->ZapConn(cs);
+          return;
+        }
+        cs->phase_ = PHASE_WAIT_DESTFRAME;
+        return;
+      }
+      case PHASE_WAIT_DESTFRAME: {
+        switch (cs->ConsumeDestframe(EVBUFFER_INPUT(bev))) {
+          case STATUS_OK: {
+            {
+              struct sockaddr_in sa;
+              memset(&sa, 0, sizeof(sa));
+              sa.sin_port = htons(cs->destport_);
+              if (inet_pton(sa.sin_family = AF_INET,
+                            cs->destname_.c_str(),
+                            &sa.sin_addr) == 1) {
+                // valid IPv4 address supplied.
+                if (cs->TryConnectDest((struct sockaddr*)&sa, sizeof(sa))) {
+                  cs->phase_ = PHASE_WAIT_DESTCONNECT;
+                  return;
+                }
+              }
+            }
+            {
+              if (cs->destname_.size() >= 2 &&
+                  cs->destname_[0] == '[' &&
+                  cs->destname_[cs->destname_.size() - 1] == ']') {
+                // Literal IPv6 address in brackets.
+                cs->destname_ =
+                    cs->destname_.substr(1, cs->destname_.size() - 2);
+              }
+              struct sockaddr_in6 sa;
+              memset(&sa, 0, sizeof(sa));
+              sa.sin6_port = htons(cs->destport_);
+              if (inet_pton(sa.sin6_family = AF_INET6,
+                            cs->destname_.c_str(),
+                            &sa.sin6_addr) == 1) {
+                // valid IPv6 address supplied.
+                if (cs->TryConnectDest((struct sockaddr*)&sa, sizeof(sa))) {
+                  cs->phase_ = PHASE_WAIT_DESTCONNECT;
+                  return;
+                }
+              }
+            }
+            // Try to asynchronously perform DNS resolution.
+            evdns_resolve_ipv4(cs->destname_.c_str(), 0,
+                               &OnDestResolutionIPv4, token);
+            evdns_resolve_ipv6(cs->destname_.c_str(), 0,
+                               &OnDestResolutionIPv6, token);
+            cs->phase_ = PHASE_WAIT_DESTCONNECT;
+            return;
+          }
+          case STATUS_INCOMPLETE: {
+            return;
+          }
+          case STATUS_ABORT:
+          default: {
+            cs->Shut();
+            return;
+          }
+        }
+      }
+      case PHASE_WAIT_DESTCONNECT: {
+        if (EVBUFFER_LENGTH(EVBUFFER_INPUT(bev)) >=
+            Web2SocketServ::kReadBufferLimit) {
+          cs->Shut();
+        }
+        return;
+      }
+      case PHASE_OUTSIDE_FRAME: {
+        switch (cs->ConsumeFrameHeader(EVBUFFER_INPUT(bev))) {
+          case STATUS_OK: {
+            cs->phase_ = PHASE_INSIDE_FRAME_BASE64;
+            // Process remaining data if any.
+            break;
+          }
+          case STATUS_SKIP: {
+            cs->phase_ = PHASE_INSIDE_FRAME_SKIP;
+            // Process remaining data if any.
+            break;
+          }
+          case STATUS_INCOMPLETE: {
+            return;
+          }
+          case STATUS_ABORT:
+          default: {
+            cs->Shut();
+            return;
+          }
+        }
+        break;
+      }
+      case PHASE_INSIDE_FRAME_BASE64:
+      case PHASE_INSIDE_FRAME_SKIP: {
+        switch (cs->ProcessFrameData(EVBUFFER_INPUT(bev))) {
+          case STATUS_OK: {
+            cs->phase_ = PHASE_OUTSIDE_FRAME;
+            // Handle remaining data if any.
+            break;
+          }
+          case STATUS_INCOMPLETE: {
+            return;
+          }
+          case STATUS_ABORT:
+          default: {
+            cs->Shut();
+            return;
+          }
+        }
+        break;
+      }
+      case PHASE_SHUT: {
+        evbuffer_drain(EVBUFFER_INPUT(bev),
+                       EVBUFFER_LENGTH(EVBUFFER_INPUT(bev)));
+        return;
+      }
+      case PHASE_DEFUNCT:
+      default: {
+        // Must not reach here.
+        cs->master_->ZapConn(cs);
+        return;
+      }
+    }
+  }
+}
+
+void Conn::OnPrimchanWrite(struct bufferevent* bev, Token token) {
+  Conn* cs = Conn::Get(token);
+  if (bev == NULL ||
+      cs == NULL ||
+      bev != cs->primchan_.bev_) {
+    // Sanity check failed.
+    return;
+  }
+  if (cs->phase_ >= PHASE_SHUT) {
+    cs->master_->ZapConn(cs);
+    return;
+  }
+  if (cs->phase_ > PHASE_WAIT_DESTCONNECT) {
+    OnDestchanRead(cs->destchan_.bev_, token);
+  }
+}
+
+void Conn::OnPrimchanError(struct bufferevent* bev,
+                                short what, Token token) {
+  Conn* cs = Conn::Get(token);
+  if (bev == NULL ||
+      cs == NULL ||
+      bev != cs->primchan_.bev_) {
+    // Sanity check failed.
+    return;
+  }
+  if (cs->phase_ >= PHASE_SHUT) {
+    cs->master_->ZapConn(cs);
+  } else {
+    cs->Shut();
+  }
+}
+
+void Conn::OnDestResolutionIPv4(int result, char type,
+                                int count, int ttl,
+                                void* addr_list, Token token) {
+  Conn* cs = Conn::Get(token);
+  if (cs == NULL) {
+    // Sanity check failed.
+    return;
+  }
+  if (cs->phase_ != PHASE_WAIT_DESTCONNECT) {
+    return;
+  }
+  if (result == DNS_ERR_NONE &&
+      count >= 1 &&
+      addr_list != NULL &&
+      type == DNS_IPv4_A) {
+    for (int i = 0; i < count; ++i) {
+      struct sockaddr_in sa;
+      memset(&sa, 0, sizeof(sa));
+      sa.sin_family = AF_INET;
+      sa.sin_port = htons(cs->destport_);
+      DCHECK(sizeof(sa.sin_addr) == sizeof(struct in_addr));
+      memcpy(&sa.sin_addr,
+             static_cast<struct in_addr*>(addr_list) + i,
+             sizeof(sa.sin_addr));
+      if (cs->TryConnectDest((struct sockaddr*)&sa, sizeof(sa))) {
+        return;
+      }
+    }
+  }
+  cs->destresolution_ipv4_failure_ = true;
+  if (cs->destresolution_ipv4_failure_ && cs->destresolution_ipv6_failure_) {
+    cs->Shut();
+  }
+}
+
+void Conn::OnDestResolutionIPv6(int result, char type,
+                                int count, int ttl,
+                                void* addr_list, Token token) {
+  Conn* cs = Conn::Get(token);
+  if (cs == NULL) {
+    // Sanity check failed.
+    return;
+  }
+  if (cs->phase_ != PHASE_WAIT_DESTCONNECT) {
+    return;
+  }
+  if (result == DNS_ERR_NONE &&
+      count >= 1 &&
+      addr_list != NULL &&
+      type == DNS_IPv6_AAAA) {
+    for (int i = 0; i < count; ++i) {
+      struct sockaddr_in6 sa;
+      memset(&sa, 0, sizeof(sa));
+      sa.sin6_family = AF_INET6;
+      sa.sin6_port = htons(cs->destport_);
+      DCHECK(sizeof(sa.sin6_addr) == sizeof(struct in6_addr));
+      memcpy(&sa.sin6_addr,
+             static_cast<struct in6_addr*>(addr_list) + i,
+             sizeof(sa.sin6_addr));
+      if (cs->TryConnectDest((struct sockaddr*)&sa, sizeof(sa))) {
+        return;
+      }
+    }
+  }
+  cs->destresolution_ipv6_failure_ = true;
+  if (cs->destresolution_ipv4_failure_ && cs->destresolution_ipv6_failure_) {
+    cs->Shut();
+  }
+}
+
+void Conn::OnDestConnectTimeout(int, short, Token token) {
+  Conn* cs = Conn::Get(token);
+  if (cs == NULL) {
+    // Sanity check failed.
+  }
+  if (cs->phase_ > PHASE_WAIT_DESTCONNECT) {
+    return;
+  }
+  cs->Shut();
+}
+
+void Conn::OnDestchanRead(struct bufferevent* bev, Token token) {
+  Conn* cs = Conn::Get(token);
+  if (bev == NULL ||
+      cs == NULL ||
+      bev != cs->destchan_.bev_) {
+    // Sanity check failed.
+    return;
+  }
+  if (EVBUFFER_LENGTH(EVBUFFER_INPUT(bev)) <= 0) {
+    return;
+  }
+  if (cs->primchan_.bev_ == NULL) {
+    cs->master_->ZapConn(cs);
+    return;
+  }
+  cs->master_->MarkConnImportance(cs, true);
+  std::string out_bytes;
+  base::Base64Encode(
+      std::string(
+          static_cast<const char*>(static_cast<void*>(
+              EVBUFFER_DATA(EVBUFFER_INPUT(bev)))),
+          EVBUFFER_LENGTH(EVBUFFER_INPUT(bev))),
+      &out_bytes);
+  evbuffer_drain(EVBUFFER_INPUT(bev), EVBUFFER_LENGTH(EVBUFFER_INPUT(bev)));
+  static const uint8_t frame_header[] = { 0x00 };
+  static const uint8_t frame_terminator[] = { 0xff };
+  if (bufferevent_write(cs->primchan_.bev_,
+                        frame_header,  sizeof(frame_header)) ||
+      bufferevent_write(cs->primchan_.bev_,
+                        out_bytes.c_str(), out_bytes.size()) ||
+      bufferevent_write(cs->primchan_.bev_,
+                        frame_terminator, sizeof(frame_terminator))) {
+    cs->Shut();
+  }
+}
+
+
+void Conn::OnDestchanWrite(struct bufferevent* bev, Token token) {
+  Conn* cs = Conn::Get(token);
+  if (bev == NULL ||
+      cs == NULL ||
+      bev != cs->destchan_.bev_) {
+    // Sanity check failed.
+    return;
+  }
+  if (cs->phase_ == PHASE_WAIT_DESTCONNECT) {
+    cs->phase_ = PHASE_OUTSIDE_FRAME;
+  }
+  OnPrimchanRead(cs->primchan_.bev_, token);
+}
+
+void Conn::OnDestchanError(struct bufferevent* bev,
+                           short what, Token token) {
+  Conn* cs = Conn::Get(token);
+  if (bev == NULL ||
+      cs == NULL ||
+      bev != cs->destchan_.bev_) {
+    // Sanity check failed.
+    return;
+  }
+  if (cs->phase_ >= PHASE_SHUT) {
+    cs->master_->ZapConn(cs);
+  } else {
+    cs->Shut();
+  }
+}
+
+Conn::Token Conn::last_token_ = 0;
+Conn::TokenMap Conn::token_map_;