inspector: Fix crash for WS connection

Attaching WS session will now include a roundtrip onto the main thread
to make sure there is no other session (e.g. JS bindings)

This change also required refactoring WS socket implementation to better
support scenarios like this.

Fixes: https://github.com/nodejs/node/issues/16852
PR-URL: https://github.com/nodejs/node/pull/17085
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Timothy Gu <timothygu99@gmail.com>

1 files changed, 141 insertions, 165 deletions

diff --git a/src/inspector_socket_server.cc b/src/inspector_socket_server.cc
index 958c41a654..bf114251cb 100644
--- a/src/inspector_socket_server.cc
+++ b/src/inspector_socket_server.cc
@@ -33,7 +33,6 @@ std::string FormatWsAddress(const std::string& host, int port,
   return url.str();
 }
 
-
 namespace {
 
 static const uint8_t PROTOCOL_JSON[] = {
@@ -114,8 +113,8 @@ void SendHttpResponse(InspectorSocket* socket, const std::string& response) {
                          "\r\n";
   char header[sizeof(HEADERS) + 20];
   int header_len = snprintf(header, sizeof(header), HEADERS, response.size());
-  inspector_write(socket, header, header_len);
-  inspector_write(socket, response.data(), response.size());
+  socket->Write(header, header_len);
+  socket->Write(response.data(), response.size());
 }
 
 void SendVersionResponse(InspectorSocket* socket) {
@@ -145,28 +144,6 @@ void SendProtocolJson(InspectorSocket* socket) {
   CHECK_EQ(Z_OK, inflateEnd(&strm));
   SendHttpResponse(socket, data);
 }
-
-int GetSocketHost(uv_tcp_t* socket, std::string* out_host) {
-  char ip[INET6_ADDRSTRLEN];
-  sockaddr_storage addr;
-  int len = sizeof(addr);
-  int err = uv_tcp_getsockname(socket,
-                               reinterpret_cast<struct sockaddr*>(&addr),
-                               &len);
-  if (err != 0)
-    return err;
-  if (addr.ss_family == AF_INET6) {
-    const sockaddr_in6* v6 = reinterpret_cast<const sockaddr_in6*>(&addr);
-    err = uv_ip6_name(v6, ip, sizeof(ip));
-  } else {
-    const sockaddr_in* v4 = reinterpret_cast<const sockaddr_in*>(&addr);
-    err = uv_ip4_name(v4, ip, sizeof(ip));
-  }
-  if (err != 0)
-    return err;
-  *out_host = ip;
-  return err;
-}
 }  // namespace
 
 
@@ -209,46 +186,58 @@ class Closer {
 
 class SocketSession {
  public:
-  static int Accept(InspectorSocketServer* server, int server_port,
-                    uv_stream_t* server_socket);
+  SocketSession(InspectorSocketServer* server, int id, int server_port);
+  void Close() {
+    ws_socket_.reset();
+  }
   void Send(const std::string& message);
-  void Close();
-
+  void Own(InspectorSocket::Pointer ws_socket) {
+    ws_socket_ = std::move(ws_socket);
+  }
   int id() const { return id_; }
-  bool IsForTarget(const std::string& target_id) const {
-    return target_id_ == target_id;
+  int server_port() {
+    return server_port_;
   }
-  static int ServerPortForClient(InspectorSocket* client) {
-    return From(client)->server_port_;
+  InspectorSocket* ws_socket() {
+    return ws_socket_.get();
   }
-
- private:
-  SocketSession(InspectorSocketServer* server, int server_port);
-  static SocketSession* From(InspectorSocket* socket) {
-    return node::ContainerOf(&SocketSession::socket_, socket);
+  void set_ws_key(const std::string& ws_key) {
+    ws_key_ = ws_key;
+  }
+  void Accept() {
+    ws_socket_->AcceptUpgrade(ws_key_);
+  }
+  void Decline() {
+    ws_socket_->CancelHandshake();
   }
 
-  enum class State { kHttp, kWebSocket, kClosing, kEOF, kDeclined };
-  static bool HandshakeCallback(InspectorSocket* socket,
-                                enum inspector_handshake_event state,
-                                const std::string& path);
-  static void ReadCallback(uv_stream_t* stream, ssize_t read,
-                           const uv_buf_t* buf);
-  static void CloseCallback(InspectorSocket* socket, int code);
+  class Delegate : public InspectorSocket::Delegate {
+   public:
+    Delegate(InspectorSocketServer* server, int session_id)
+             : server_(server), session_id_(session_id) { }
+    ~Delegate() {
+      server_->SessionTerminated(session_id_);
+    }
+    void OnHttpGet(const std::string& path) override;
+    void OnSocketUpgrade(const std::string& path,
+                         const std::string& ws_key) override;
+    void OnWsFrame(const std::vector<char>& data) override;
+
+   private:
+    SocketSession* Session() {
+      return server_->Session(session_id_);
+    }
 
-  void FrontendConnected();
-  void SetDeclined() { state_ = State::kDeclined; }
-  void SetTargetId(const std::string& target_id) {
-    CHECK(target_id_.empty());
-    target_id_ = target_id;
-  }
+    InspectorSocketServer* server_;
+    int session_id_;
+  };
 
+ private:
   const int id_;
-  InspectorSocket socket_;
+  InspectorSocket::Pointer ws_socket_;
   InspectorSocketServer* server_;
-  std::string target_id_;
-  State state_;
   const int server_port_;
+  std::string ws_key_;
 };
 
 class ServerSocket {
@@ -269,7 +258,6 @@ class ServerSocket {
     return node::ContainerOf(&ServerSocket::tcp_socket_,
                              reinterpret_cast<uv_tcp_t*>(socket));
   }
-
   static void SocketConnectedCallback(uv_stream_t* tcp_socket, int status);
   static void SocketClosedCallback(uv_handle_t* tcp_socket);
   static void FreeOnCloseCallback(uv_handle_t* tcp_socket_) {
@@ -296,41 +284,57 @@ InspectorSocketServer::InspectorSocketServer(SocketServerDelegate* delegate,
   state_ = ServerState::kNew;
 }
 
-bool InspectorSocketServer::SessionStarted(SocketSession* session,
-                                           const std::string& id) {
-  if (TargetExists(id) && delegate_->StartSession(session->id(), id)) {
-    connected_sessions_[session->id()] = session;
-    return true;
-  } else {
-    return false;
+InspectorSocketServer::~InspectorSocketServer() = default;
+
+SocketSession* InspectorSocketServer::Session(int session_id) {
+  auto it = connected_sessions_.find(session_id);
+  return it == connected_sessions_.end() ? nullptr : it->second.second.get();
+}
+
+void InspectorSocketServer::SessionStarted(int session_id,
+                                           const std::string& id,
+                                           const std::string& ws_key) {
+  SocketSession* session = Session(session_id);
+  if (!TargetExists(id)) {
+    Session(session_id)->Decline();
+    return;
   }
+  connected_sessions_[session_id].first = id;
+  session->set_ws_key(ws_key);
+  delegate_->StartSession(session_id, id);
 }
 
-void InspectorSocketServer::SessionTerminated(SocketSession* session) {
-  int id = session->id();
-  if (connected_sessions_.erase(id) != 0) {
-    delegate_->EndSession(id);
-    if (connected_sessions_.empty()) {
-      if (state_ == ServerState::kRunning && !server_sockets_.empty()) {
-        PrintDebuggerReadyMessage(host_, server_sockets_[0]->port(),
-                                  delegate_->GetTargetIds(), out_);
-      }
-      if (state_ == ServerState::kStopped) {
-        delegate_->ServerDone();
-      }
+void InspectorSocketServer::SessionTerminated(int session_id) {
+  if (Session(session_id) == nullptr) {
+    return;
+  }
+  bool was_attached = connected_sessions_[session_id].first != "";
+  if (was_attached) {
+    delegate_->EndSession(session_id);
+  }
+  connected_sessions_.erase(session_id);
+  if (connected_sessions_.empty()) {
+    if (was_attached && state_ == ServerState::kRunning
+        && !server_sockets_.empty()) {
+      PrintDebuggerReadyMessage(host_, server_sockets_[0]->port(),
+                                delegate_->GetTargetIds(), out_);
+    }
+    if (state_ == ServerState::kStopped) {
+      delegate_->ServerDone();
     }
   }
-  delete session;
 }
 
-bool InspectorSocketServer::HandleGetRequest(InspectorSocket* socket,
+bool InspectorSocketServer::HandleGetRequest(int session_id,
                                              const std::string& path) {
+  SocketSession* session = Session(session_id);
+  InspectorSocket* socket = session->ws_socket();
   const char* command = MatchPathSegment(path.c_str(), "/json");
   if (command == nullptr)
     return false;
 
   if (MatchPathSegment(command, "list") || command[0] == '\0') {
-    SendListResponse(socket);
+    SendListResponse(socket, session);
     return true;
   } else if (MatchPathSegment(command, "protocol")) {
     SendProtocolJson(socket);
@@ -348,7 +352,8 @@ bool InspectorSocketServer::HandleGetRequest(InspectorSocket* socket,
   return false;
 }
 
-void InspectorSocketServer::SendListResponse(InspectorSocket* socket) {
+void InspectorSocketServer::SendListResponse(InspectorSocket* socket,
+                                             SocketSession* session) {
   std::vector<std::map<std::string, std::string>> response;
   for (const std::string& id : delegate_->GetTargetIds()) {
     response.push_back(std::map<std::string, std::string>());
@@ -366,15 +371,14 @@ void InspectorSocketServer::SendListResponse(InspectorSocket* socket) {
 
     bool connected = false;
     for (const auto& session : connected_sessions_) {
-      if (session.second->IsForTarget(id)) {
+      if (session.second.first == id) {
         connected = true;
         break;
       }
     }
     if (!connected) {
-      std::string host;
-      int port = SocketSession::ServerPortForClient(socket);
-      GetSocketHost(&socket->tcp, &host);
+      std::string host = socket->GetHost();
+      int port = session->server_port();
       std::ostringstream frontend_url;
       frontend_url << "chrome-devtools://devtools/bundled";
       frontend_url << "/inspector.html?experiments=true&v8only=true&ws=";
@@ -444,9 +448,8 @@ void InspectorSocketServer::Stop(ServerCallback cb) {
 }
 
 void InspectorSocketServer::TerminateConnections() {
-  for (const auto& session : connected_sessions_) {
-    session.second->Close();
-  }
+  for (const auto& key_value : connected_sessions_)
+    key_value.second.second->Close();
 }
 
 bool InspectorSocketServer::TargetExists(const std::string& id) {
@@ -455,13 +458,6 @@ bool InspectorSocketServer::TargetExists(const std::string& id) {
   return found != target_ids.end();
 }
 
-void InspectorSocketServer::Send(int session_id, const std::string& message) {
-  auto session_iterator = connected_sessions_.find(session_id);
-  if (session_iterator != connected_sessions_.end()) {
-    session_iterator->second->Send(message);
-  }
-}
-
 void InspectorSocketServer::ServerSocketListening(ServerSocket* server_socket) {
   server_sockets_.push_back(server_socket);
 }
@@ -491,92 +487,73 @@ int InspectorSocketServer::Port() const {
   return port_;
 }
 
-// InspectorSession tracking
-SocketSession::SocketSession(InspectorSocketServer* server, int server_port)
-                             : id_(server->GenerateSessionId()),
-                               server_(server),
-                               state_(State::kHttp),
-                               server_port_(server_port) { }
+void InspectorSocketServer::Accept(int server_port,
+                                   uv_stream_t* server_socket) {
+  std::unique_ptr<SocketSession> session(
+      new SocketSession(this, next_session_id_++, server_port));
+
+  InspectorSocket::DelegatePointer delegate =
+      InspectorSocket::DelegatePointer(
+          new SocketSession::Delegate(this, session->id()));
 
-void SocketSession::Close() {
-  CHECK_NE(state_, State::kClosing);
-  state_ = State::kClosing;
-  inspector_close(&socket_, CloseCallback);
+  InspectorSocket::Pointer inspector =
+      InspectorSocket::Accept(server_socket, std::move(delegate));
+  if (inspector) {
+    session->Own(std::move(inspector));
+    connected_sessions_[session->id()].second = std::move(session);
+  }
 }
 
-// static
-int SocketSession::Accept(InspectorSocketServer* server, int server_port,
-                          uv_stream_t* server_socket) {
-  // Memory is freed when the socket closes.
-  SocketSession* session = new SocketSession(server, server_port);
-  int err = inspector_accept(server_socket, &session->socket_,
-                             HandshakeCallback);
-  if (err != 0) {
-    delete session;
+void InspectorSocketServer::AcceptSession(int session_id) {
+  SocketSession* session = Session(session_id);
+  if (session == nullptr) {
+    delegate_->EndSession(session_id);
+  } else {
+    session->Accept();
   }
-  return err;
 }
 
-// static
-bool SocketSession::HandshakeCallback(InspectorSocket* socket,
-                                      inspector_handshake_event event,
-                                      const std::string& path) {
-  SocketSession* session = SocketSession::From(socket);
-  InspectorSocketServer* server = session->server_;
-  const std::string& id = path.empty() ? path : path.substr(1);
-  switch (event) {
-  case kInspectorHandshakeHttpGet:
-    return server->HandleGetRequest(socket, path);
-  case kInspectorHandshakeUpgrading:
-    if (server->SessionStarted(session, id)) {
-      session->SetTargetId(id);
-      return true;
-    } else {
-      session->SetDeclined();
-      return false;
-    }
-  case kInspectorHandshakeUpgraded:
-    session->FrontendConnected();
-    return true;
-  case kInspectorHandshakeFailed:
-    server->SessionTerminated(session);
-    return false;
-  default:
-    UNREACHABLE();
-    return false;
+void InspectorSocketServer::DeclineSession(int session_id) {
+  auto it = connected_sessions_.find(session_id);
+  if (it != connected_sessions_.end()) {
+    it->second.first.clear();
+    it->second.second->Decline();
   }
 }
 
-// static
-void SocketSession::CloseCallback(InspectorSocket* socket, int code) {
-  SocketSession* session = SocketSession::From(socket);
-  CHECK_EQ(State::kClosing, session->state_);
-  session->server_->SessionTerminated(session);
+void InspectorSocketServer::Send(int session_id, const std::string& message) {
+  SocketSession* session = Session(session_id);
+  if (session != nullptr) {
+    session->Send(message);
+  }
+}
+
+// InspectorSession tracking
+SocketSession::SocketSession(InspectorSocketServer* server, int id,
+                             int server_port)
+                             : id_(id),
+                               server_(server),
+                               server_port_(server_port) { }
+
+
+void SocketSession::Send(const std::string& message) {
+  ws_socket_->Write(message.data(), message.length());
 }
 
-void SocketSession::FrontendConnected() {
-  CHECK_EQ(State::kHttp, state_);
-  state_ = State::kWebSocket;
-  inspector_read_start(&socket_, OnBufferAlloc, ReadCallback);
+void SocketSession::Delegate::OnHttpGet(const std::string& path) {
+  if (!server_->HandleGetRequest(session_id_, path))
+    Session()->ws_socket()->CancelHandshake();
 }
 
-// static
-void SocketSession::ReadCallback(uv_stream_t* stream, ssize_t read,
-                                 const uv_buf_t* buf) {
-  InspectorSocket* socket = inspector_from_stream(stream);
-  SocketSession* session = SocketSession::From(socket);
-  if (read > 0) {
-    session->server_->MessageReceived(session->id_,
-                                      std::string(buf->base, read));
-  } else {
-    session->Close();
-  }
-  if (buf != nullptr && buf->base != nullptr)
-    delete[] buf->base;
+void SocketSession::Delegate::OnSocketUpgrade(const std::string& path,
+                                              const std::string& ws_key) {
+  std::string id = path.empty() ? path : path.substr(1);
+  server_->SessionStarted(session_id_, id, ws_key);
 }
 
-void SocketSession::Send(const std::string& message) {
-  inspector_write(&socket_, message.data(), message.length());
+void SocketSession::Delegate::OnWsFrame(const std::vector<char>& data) {
+  server_->MessageReceived(session_id_,
+                           std::string(data.data(), data.size()));
 }
 
 // ServerSocket implementation
@@ -624,8 +601,7 @@ void ServerSocket::SocketConnectedCallback(uv_stream_t* tcp_socket,
   if (status == 0) {
     ServerSocket* server_socket = ServerSocket::FromTcpSocket(tcp_socket);
     // Memory is freed when the socket closes.
-    SocketSession::Accept(server_socket->server_, server_socket->port_,
-                          tcp_socket);
+    server_socket->server_->Accept(server_socket->port_, tcp_socket);
   }
 }