mojo / core / node_channel.cc [blame]

// Copyright 2016 The Chromium Authors
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.

#ifdef UNSAFE_BUFFERS_BUILD
// TODO(crbug.com/351564777): Remove this and convert code to safer constructs.
#pragma allow_unsafe_buffers
#endif

#include "mojo/core/node_channel.h"

#include <cstring>
#include <limits>
#include <sstream>

#include "base/functional/bind.h"
#include "base/location.h"
#include "base/logging.h"
#include "base/memory/ptr_util.h"
#include "base/task/single_thread_task_runner.h"
#include "build/build_config.h"
#include "mojo/core/broker_host.h"
#include "mojo/core/channel.h"
#include "mojo/core/configuration.h"
#include "mojo/core/core.h"
#include "mojo/core/ipcz_driver/envelope.h"
#include "mojo/core/request_context.h"

namespace mojo {
namespace core {

namespace {

// NOTE: Please ONLY append messages to the end of this enum.
enum class MessageType : uint32_t {
  ACCEPT_INVITEE,
  ACCEPT_INVITATION,
  ADD_BROKER_CLIENT,
  BROKER_CLIENT_ADDED,
  ACCEPT_BROKER_CLIENT,
  EVENT_MESSAGE,
  REQUEST_PORT_MERGE,
  REQUEST_INTRODUCTION,
  INTRODUCE,
#if BUILDFLAG(IS_WIN)
  RELAY_EVENT_MESSAGE,
#endif
  BROADCAST_EVENT,
#if BUILDFLAG(IS_WIN)
  EVENT_MESSAGE_FROM_RELAY,
#endif
  ACCEPT_PEER,
  BIND_BROKER_HOST,
};

#pragma pack(push, 1)

struct alignas(8) Header {
  MessageType type;
};

static_assert(sizeof(Header) == kNodeChannelHeaderSize);

static_assert(IsAlignedForChannelMessage(sizeof(Header)),
              "Invalid header size.");

struct alignas(8) AcceptInviteeDataV0 {
  ports::NodeName inviter_name;
  ports::NodeName token;
};

struct alignas(8) AcceptInviteeDataV1 : AcceptInviteeDataV0 {
  uint64_t capabilities = kNodeCapabilityNone;
};

using AcceptInviteeData = AcceptInviteeDataV1;

struct alignas(8) AcceptInvitationDataV0 {
  ports::NodeName token;
  ports::NodeName invitee_name;
};

struct alignas(8) AcceptInvitationDataV1 : AcceptInvitationDataV0 {
  uint64_t capabilities = kNodeCapabilityNone;
};

using AcceptInvitationData = AcceptInvitationDataV1;

struct alignas(8) AcceptPeerDataV0 {
  ports::NodeName token;
  ports::NodeName peer_name;
  ports::PortName port_name;
};

using AcceptPeerData = AcceptPeerDataV0;

// This message may include a process handle on platforms that require it.
struct alignas(8) AddBrokerClientDataV0 {
  ports::NodeName client_name;
#if !BUILDFLAG(IS_WIN)
  uint32_t process_handle;
#endif
};

using AddBrokerClientData = AddBrokerClientDataV0;

#if !BUILDFLAG(IS_WIN)
static_assert(sizeof(base::ProcessHandle) == sizeof(uint32_t),
              "Unexpected pid size");
static_assert(sizeof(AddBrokerClientData) % kChannelMessageAlignment == 0,
              "Invalid AddBrokerClientData size.");
#endif

// This data is followed by a platform channel handle to the broker.
struct alignas(8) BrokerClientAddedDataV0 {
  ports::NodeName client_name;
};

using BrokerClientAddedData = BrokerClientAddedDataV0;

// This data may be followed by a platform channel handle to the broker. If not,
// then the inviter is the broker and its channel should be used as such.
struct alignas(8) AcceptBrokerClientDataV0 {
  ports::NodeName broker_name;
};

struct alignas(8) AcceptBrokerClientDataV1 : AcceptBrokerClientDataV0 {
  uint64_t capabilities = kNodeCapabilityNone;
  uint64_t broker_capabilities = kNodeCapabilityNone;
};

using AcceptBrokerClientData = AcceptBrokerClientDataV1;

// This is followed by arbitrary payload data which is interpreted as a token
// string for port location.
// NOTE: Because this field is variable length it cannot be versioned.
struct alignas(8) RequestPortMergeData {
  ports::PortName connector_port_name;
};

// Used for both REQUEST_INTRODUCTION and INTRODUCE.
//
// For INTRODUCE the message also includes a valid platform handle for a
// channel the receiver may use to communicate with the named node directly,
// or an invalid platform handle if the node is unknown to the sender or
// otherwise cannot be introduced.
struct alignas(8) IntroductionDataV0 {
  ports::NodeName name;
};

struct alignas(8) IntroductionDataV1 : IntroductionDataV0 {
  uint64_t capabilities = kNodeCapabilityNone;
};

using IntroductionData = IntroductionDataV1;

// This message is just a PlatformHandle. The data struct alignas(8) here has
// only a padding field to ensure an aligned, non-zero-length payload.
struct alignas(8) BindBrokerHostDataV0 {};

using BindBrokerHostData = BindBrokerHostDataV0;

#if BUILDFLAG(IS_WIN)
// This struct alignas(8) is followed by the full payload of a message to be
// relayed.
// NOTE: Because this field is variable length it cannot be versioned.
struct alignas(8) RelayEventMessageData {
  ports::NodeName destination;
};

// This struct alignas(8) is followed by the full payload of a relayed
// message.
struct alignas(8) EventMessageFromRelayDataV0 {
  ports::NodeName source;
};

using EventMessageFromRelayData = EventMessageFromRelayDataV0;

#endif

#pragma pack(pop)

Channel::MessagePtr CreateMessage(MessageType type,
                                  size_t payload_size,
                                  size_t num_handles,
                                  void** out_data,
                                  size_t capacity = 0) {
  const size_t total_size = payload_size + sizeof(Header);
  if (capacity == 0)
    capacity = total_size;
  else
    capacity = std::max(total_size, capacity);
  auto message =
      Channel::Message::CreateMessage(capacity, total_size, num_handles);
  Header* header = reinterpret_cast<Header*>(message->mutable_payload());

  // Make sure any header padding gets zeroed.
  memset(header, 0, sizeof(Header));
  header->type = type;

  // The out_data starts beyond the header.
  *out_data = reinterpret_cast<void*>(header + 1);
  return message;
}

template <typename DataType>
Channel::MessagePtr CreateMessage(MessageType type,
                                  size_t payload_size,
                                  size_t num_handles,
                                  DataType** out_data,
                                  size_t capacity = 0) {
  auto msg_ptr = CreateMessage(type, payload_size, num_handles,
                               reinterpret_cast<void**>(out_data), capacity);

  // Since we know the type let's make sure any padding areas are zeroed.
  memset(*out_data, 0, sizeof(DataType));

  return msg_ptr;
}

// This method takes a second template argument which is another datatype
// which represents the smallest size this payload can be to be considered
// valid this MUST be used when there is more than one version of a message to
// specify the oldest version of the message.
template <typename DataType, typename MinSizedDataType>
bool GetMessagePayloadMinimumSized(const void* bytes,
                                   size_t num_bytes,
                                   DataType* out_data) {
  static_assert(sizeof(DataType) > 0, "DataType must have non-zero size.");
  if (num_bytes < sizeof(Header) + sizeof(MinSizedDataType)) {
    return false;
  }

  // Always make sure that the full object is zeored and default constructed
  // as we may not have the complete type. The default construction allows
  // fields to be default initialized to be resilient to older message
  // versions.
  memset(out_data, 0, sizeof(*out_data));
  new (out_data) DataType;

  // Overwrite any fields we received.
  memcpy(out_data, static_cast<const uint8_t*>(bytes) + sizeof(Header),
         std::min(sizeof(DataType), num_bytes - sizeof(Header)));
  return true;
}

template <typename DataType>
bool GetMessagePayload(const void* bytes,
                       size_t num_bytes,
                       DataType* out_data) {
  return GetMessagePayloadMinimumSized<DataType, DataType>(bytes, num_bytes,
                                                           out_data);
}

}  // namespace

// static
scoped_refptr<NodeChannel> NodeChannel::Create(
    Delegate* delegate,
    ConnectionParams connection_params,
    Channel::HandlePolicy channel_handle_policy,
    scoped_refptr<base::SingleThreadTaskRunner> io_task_runner,
    const ProcessErrorCallback& process_error_callback) {
#if BUILDFLAG(IS_NACL)
  LOG(FATAL) << "Multi-process not yet supported on NaCl-SFI";
#else
  return new NodeChannel(delegate, std::move(connection_params),
                         channel_handle_policy, io_task_runner,
                         process_error_callback);
#endif
}

// static
Channel::MessagePtr NodeChannel::CreateEventMessage(size_t capacity,
                                                    size_t payload_size,
                                                    void** payload,
                                                    size_t num_handles) {
  return CreateMessage(MessageType::EVENT_MESSAGE, payload_size, num_handles,
                       payload, capacity);
}

// static
bool NodeChannel::GetEventMessageData(Channel::Message& message,
                                      void** data,
                                      size_t* num_data_bytes) {
  // NOTE: Callers must guarantee that the payload in `message` must be at least
  // large enough to hold a Header.
  if (message.payload_size() < sizeof(Header))
    return false;
  *data = reinterpret_cast<Header*>(message.mutable_payload()) + 1;
  *num_data_bytes = message.payload_size() - sizeof(Header);
  return true;
}

void NodeChannel::Start() {
  base::AutoLock lock(channel_lock_);
  // ShutDown() may have already been called, in which case |channel_| is null.
  if (channel_)
    channel_->Start();
}

void NodeChannel::ShutDown() {
  base::AutoLock lock(channel_lock_);
  if (channel_) {
    channel_->ShutDown();
    channel_ = nullptr;
  }
}

void NodeChannel::LeakHandleOnShutdown() {
  base::AutoLock lock(channel_lock_);
  if (channel_) {
    channel_->LeakHandle();
  }
}

void NodeChannel::NotifyBadMessage(const std::string& error) {
  DCHECK(HasBadMessageHandler());
  process_error_callback_.Run("Received bad user message: " + error);
}

void NodeChannel::SetRemoteProcessHandle(base::Process process_handle) {
  DCHECK(owning_task_runner()->RunsTasksInCurrentSequence());
  {
    base::AutoLock lock(channel_lock_);
    if (channel_)
      channel_->set_remote_process(process_handle.Duplicate());
  }
  base::AutoLock lock(remote_process_handle_lock_);
  DCHECK(!remote_process_handle_.IsValid());
  CHECK_NE(remote_process_handle_.Handle(), base::GetCurrentProcessHandle());
  remote_process_handle_ = std::move(process_handle);
}

bool NodeChannel::HasRemoteProcessHandle() {
  base::AutoLock lock(remote_process_handle_lock_);
  return remote_process_handle_.IsValid();
}

base::Process NodeChannel::CloneRemoteProcessHandle() {
  base::AutoLock lock(remote_process_handle_lock_);
  return remote_process_handle_.Duplicate();
}

void NodeChannel::SetRemoteNodeName(const ports::NodeName& name) {
  DCHECK(owning_task_runner()->RunsTasksInCurrentSequence());
  remote_node_name_ = name;
}

void NodeChannel::AcceptInvitee(const ports::NodeName& inviter_name,
                                const ports::NodeName& token) {
  AcceptInviteeData* data;
  Channel::MessagePtr message = CreateMessage(
      MessageType::ACCEPT_INVITEE, sizeof(AcceptInviteeData), 0, &data);
  data->inviter_name = inviter_name;
  data->token = token;
  data->capabilities = local_capabilities_;
  WriteChannelMessage(std::move(message));
}

void NodeChannel::AcceptInvitation(const ports::NodeName& token,
                                   const ports::NodeName& invitee_name) {
  AcceptInvitationData* data;
  Channel::MessagePtr message = CreateMessage(
      MessageType::ACCEPT_INVITATION, sizeof(AcceptInvitationData), 0, &data);
  data->token = token;
  data->invitee_name = invitee_name;
  data->capabilities = local_capabilities_;
  WriteChannelMessage(std::move(message));
}

void NodeChannel::AcceptPeer(const ports::NodeName& sender_name,
                             const ports::NodeName& token,
                             const ports::PortName& port_name) {
  AcceptPeerData* data;
  Channel::MessagePtr message =
      CreateMessage(MessageType::ACCEPT_PEER, sizeof(AcceptPeerData), 0, &data);
  data->token = token;
  data->peer_name = sender_name;
  data->port_name = port_name;
  WriteChannelMessage(std::move(message));
}

void NodeChannel::AddBrokerClient(const ports::NodeName& client_name,
                                  base::Process process_handle) {
  AddBrokerClientData* data;
  std::vector<PlatformHandle> handles;
#if BUILDFLAG(IS_WIN)
  handles.emplace_back(base::win::ScopedHandle(process_handle.Release()));
#endif
  Channel::MessagePtr message =
      CreateMessage(MessageType::ADD_BROKER_CLIENT, sizeof(AddBrokerClientData),
                    handles.size(), &data);
  message->SetHandles(std::move(handles));
  data->client_name = client_name;
#if !BUILDFLAG(IS_WIN)
  // Older clients treat this as a real process handle, but don't actually need
  // it, so send a valid null handle.
  data->process_handle = base::kNullProcessHandle;
#endif
  WriteChannelMessage(std::move(message));
}

void NodeChannel::BrokerClientAdded(const ports::NodeName& client_name,
                                    PlatformHandle broker_channel) {
  BrokerClientAddedData* data;
  std::vector<PlatformHandle> handles;
  if (broker_channel.is_valid())
    handles.emplace_back(std::move(broker_channel));
  Channel::MessagePtr message =
      CreateMessage(MessageType::BROKER_CLIENT_ADDED,
                    sizeof(BrokerClientAddedData), handles.size(), &data);
  message->SetHandles(std::move(handles));
  data->client_name = client_name;
  WriteChannelMessage(std::move(message));
}

void NodeChannel::AcceptBrokerClient(const ports::NodeName& broker_name,
                                     PlatformHandle broker_channel,
                                     const uint64_t broker_capabilities) {
  AcceptBrokerClientData* data;
  std::vector<PlatformHandle> handles;
  if (broker_channel.is_valid())
    handles.emplace_back(std::move(broker_channel));
  Channel::MessagePtr message =
      CreateMessage(MessageType::ACCEPT_BROKER_CLIENT,
                    sizeof(AcceptBrokerClientData), handles.size(), &data);
  message->SetHandles(std::move(handles));
  data->broker_name = broker_name;
  data->broker_capabilities = broker_capabilities;
  data->capabilities = local_capabilities_;
  WriteChannelMessage(std::move(message));
}

void NodeChannel::RequestPortMerge(const ports::PortName& connector_port_name,
                                   const std::string& token) {
  RequestPortMergeData* data;
  Channel::MessagePtr message =
      CreateMessage(MessageType::REQUEST_PORT_MERGE,
                    sizeof(RequestPortMergeData) + token.size(), 0, &data);
  data->connector_port_name = connector_port_name;
  memcpy(data + 1, token.data(), token.size());
  WriteChannelMessage(std::move(message));
}

void NodeChannel::RequestIntroduction(const ports::NodeName& name) {
  IntroductionData* data;
  Channel::MessagePtr message = CreateMessage(
      MessageType::REQUEST_INTRODUCTION, sizeof(IntroductionData), 0, &data);
  data->name = name;
  WriteChannelMessage(std::move(message));
}

void NodeChannel::Introduce(const ports::NodeName& name,
                            PlatformHandle channel_handle,
                            uint64_t capabilities) {
  IntroductionData* data;
  std::vector<PlatformHandle> handles;
  if (channel_handle.is_valid())
    handles.emplace_back(std::move(channel_handle));
  Channel::MessagePtr message = CreateMessage(
      MessageType::INTRODUCE, sizeof(IntroductionData), handles.size(), &data);
  message->SetHandles(std::move(handles));
  data->name = name;
  // Note that these are not our capabilities, but the capabilities of the peer
  // we're introducing.
  data->capabilities = capabilities;
  WriteChannelMessage(std::move(message));
}

void NodeChannel::SendChannelMessage(Channel::MessagePtr message) {
  WriteChannelMessage(std::move(message));
}

void NodeChannel::Broadcast(Channel::MessagePtr message) {
  DCHECK(!message->has_handles());
  void* data;
  Channel::MessagePtr broadcast_message = CreateMessage(
      MessageType::BROADCAST_EVENT, message->data_num_bytes(), 0, &data);
  memcpy(data, message->data(), message->data_num_bytes());
  WriteChannelMessage(std::move(broadcast_message));
}

void NodeChannel::BindBrokerHost(PlatformHandle broker_host_handle) {
#if !BUILDFLAG(IS_APPLE) && !BUILDFLAG(IS_NACL) && !BUILDFLAG(IS_FUCHSIA)
  DCHECK(broker_host_handle.is_valid());
  BindBrokerHostData* data;
  std::vector<PlatformHandle> handles;
  handles.push_back(std::move(broker_host_handle));
  Channel::MessagePtr message =
      CreateMessage(MessageType::BIND_BROKER_HOST, sizeof(BindBrokerHostData),
                    handles.size(), &data);
  message->SetHandles(std::move(handles));
  WriteChannelMessage(std::move(message));
#endif
}

#if BUILDFLAG(IS_WIN)
void NodeChannel::RelayEventMessage(const ports::NodeName& destination,
                                    Channel::MessagePtr message) {
  DCHECK(message->has_handles());

  // Note that this is only used on Windows, and on Windows all platform
  // handles are included in the message data. We blindly copy all the data
  // here and the relay node (the broker) will duplicate handles as needed.
  size_t num_bytes = sizeof(RelayEventMessageData) + message->data_num_bytes();
  RelayEventMessageData* data;
  Channel::MessagePtr relay_message =
      CreateMessage(MessageType::RELAY_EVENT_MESSAGE, num_bytes, 0, &data);
  data->destination = destination;
  memcpy(data + 1, message->data(), message->data_num_bytes());

  // When the handles are duplicated in the broker, the source handles will
  // be closed. If the broker never receives this message then these handles
  // will leak, but that means something else has probably broken and the
  // sending process won't likely be around much longer.
  //
  // TODO(crbug.com/40563346): We would like to be able to violate the
  // above stated assumption. We should not leak handles in cases where we
  // outlive the broker, as we may continue existing and eventually accept a new
  // broker invitation.
  std::vector<PlatformHandleInTransit> handles = message->TakeHandles();
  for (auto& handle : handles)
    handle.TakeHandle().release();

  WriteChannelMessage(std::move(relay_message));
}

void NodeChannel::EventMessageFromRelay(const ports::NodeName& source,
                                        Channel::MessagePtr message) {
  size_t num_bytes =
      sizeof(EventMessageFromRelayData) + message->payload_size();
  EventMessageFromRelayData* data;
  Channel::MessagePtr relayed_message =
      CreateMessage(MessageType::EVENT_MESSAGE_FROM_RELAY, num_bytes,
                    message->num_handles(), &data);
  data->source = source;
  if (message->payload_size())
    memcpy(data + 1, message->payload(), message->payload_size());
  relayed_message->SetHandles(message->TakeHandles());
  WriteChannelMessage(std::move(relayed_message));
}
#endif  // BUILDFLAG(IS_WIN)

NodeChannel::NodeChannel(
    Delegate* delegate,
    ConnectionParams connection_params,
    Channel::HandlePolicy channel_handle_policy,
    scoped_refptr<base::SingleThreadTaskRunner> io_task_runner,
    const ProcessErrorCallback& process_error_callback)
    : base::RefCountedDeleteOnSequence<NodeChannel>(io_task_runner),
      delegate_(delegate),
      process_error_callback_(process_error_callback)
#if !BUILDFLAG(IS_NACL)
      ,
      channel_(Channel::Create(this,
                               std::move(connection_params),
                               channel_handle_policy,
                               std::move(io_task_runner)))
#endif
{
  InitializeLocalCapabilities();
}

NodeChannel::~NodeChannel() {
  ShutDown();
}

void NodeChannel::CreateAndBindLocalBrokerHost(
    PlatformHandle broker_host_handle) {
#if !BUILDFLAG(IS_APPLE) && !BUILDFLAG(IS_NACL) && !BUILDFLAG(IS_FUCHSIA)
  // Self-owned.
  ConnectionParams connection_params(
      PlatformChannelEndpoint(std::move(broker_host_handle)));
  new BrokerHost(remote_process_handle_.Duplicate(),
                 std::move(connection_params), process_error_callback_);
#endif
}

void NodeChannel::OnChannelMessage(
    const void* payload,
    size_t payload_size,
    std::vector<PlatformHandle> handles,
    scoped_refptr<ipcz_driver::Envelope> envelope) {
  DCHECK(owning_task_runner()->RunsTasksInCurrentSequence());

  RequestContext request_context(RequestContext::Source::SYSTEM);

  if (payload_size <= sizeof(Header)) {
    delegate_->OnChannelError(remote_node_name_, this);
    return;
  }

  const Header* header = static_cast<const Header*>(payload);
  switch (header->type) {
    case MessageType::ACCEPT_INVITEE: {
      AcceptInviteeData data;
      if (GetMessagePayloadMinimumSized<AcceptInviteeData, AcceptInviteeDataV0>(
              payload, payload_size, &data)) {
        // Attach any capabilities that the other side advertised.
        SetRemoteCapabilities(data.capabilities);
        delegate_->OnAcceptInvitee(remote_node_name_, data.inviter_name,
                                   data.token);
        return;
      }
      break;
    }

    case MessageType::ACCEPT_INVITATION: {
      AcceptInvitationData data;
      if (GetMessagePayloadMinimumSized<AcceptInvitationData,
                                        AcceptInvitationDataV0>(
              payload, payload_size, &data)) {
        // Attach any capabilities that the other side advertised.
        SetRemoteCapabilities(data.capabilities);
        delegate_->OnAcceptInvitation(remote_node_name_, data.token,
                                      data.invitee_name);
        return;
      }
      break;
    }

    case MessageType::ADD_BROKER_CLIENT: {
      AddBrokerClientData data;
      if (GetMessagePayload(payload, payload_size, &data)) {
#if BUILDFLAG(IS_WIN)
        if (handles.size() != 1) {
          DLOG(ERROR) << "Dropping invalid AddBrokerClient message.";
          break;
        }
        delegate_->OnAddBrokerClient(remote_node_name_, data.client_name,
                                     handles[0].ReleaseHandle());
#else
        if (!handles.empty()) {
          DLOG(ERROR) << "Dropping invalid AddBrokerClient message.";
          break;
        }
        delegate_->OnAddBrokerClient(remote_node_name_, data.client_name,
                                     base::kNullProcessHandle);
#endif
        return;
      }
      break;
    }

    case MessageType::BROKER_CLIENT_ADDED: {
      BrokerClientAddedData data;
      if (GetMessagePayload(payload, payload_size, &data)) {
        if (handles.size() != 1) {
          DLOG(ERROR) << "Dropping invalid BrokerClientAdded message.";
          break;
        }
        delegate_->OnBrokerClientAdded(remote_node_name_, data.client_name,
                                       std::move(handles[0]));
        return;
      }
      break;
    }

    case MessageType::ACCEPT_BROKER_CLIENT: {
      AcceptBrokerClientData data;
      if (GetMessagePayloadMinimumSized<AcceptBrokerClientData,
                                        AcceptBrokerClientDataV0>(
              payload, payload_size, &data)) {
        PlatformHandle broker_channel;
        if (handles.size() > 1) {
          DLOG(ERROR) << "Dropping invalid AcceptBrokerClient message.";
          break;
        }
        if (handles.size() == 1)
          broker_channel = std::move(handles[0]);

        // Attach any capabilities that the other side advertised.
        SetRemoteCapabilities(data.capabilities);
        delegate_->OnAcceptBrokerClient(remote_node_name_, data.broker_name,
                                        std::move(broker_channel),
                                        data.broker_capabilities);
        return;
      }
      break;
    }

    case MessageType::EVENT_MESSAGE: {
      Channel::MessagePtr message =
          Channel::Message::CreateMessage(payload_size, handles.size());
      message->SetHandles(std::move(handles));
      memcpy(message->mutable_payload(), payload, payload_size);
      delegate_->OnEventMessage(remote_node_name_, std::move(message));
      return;
    }

    case MessageType::REQUEST_PORT_MERGE: {
      RequestPortMergeData data;
      if (GetMessagePayload(payload, payload_size, &data)) {
        // Don't accept an empty token.
        size_t token_size = payload_size - sizeof(data) - sizeof(Header);
        if (token_size == 0)
          break;
        std::string token(reinterpret_cast<const char*>(payload) +
                              sizeof(Header) + sizeof(data),
                          token_size);
        delegate_->OnRequestPortMerge(remote_node_name_,
                                      data.connector_port_name, token);
        return;
      }
      break;
    }

    case MessageType::REQUEST_INTRODUCTION: {
      IntroductionData data;
      if (GetMessagePayloadMinimumSized<IntroductionData, IntroductionDataV0>(
              payload, payload_size, &data)) {
        delegate_->OnRequestIntroduction(remote_node_name_, data.name);
        return;
      }
      break;
    }

    case MessageType::INTRODUCE: {
      IntroductionData data;
      if (GetMessagePayloadMinimumSized<IntroductionData, IntroductionDataV0>(
              payload, payload_size, &data)) {
        if (handles.size() > 1) {
          DLOG(ERROR) << "Dropping invalid introduction message.";
          break;
        }
        PlatformHandle channel_handle;
        if (handles.size() == 1)
          channel_handle = std::move(handles[0]);

        // The node channel for this introduction will be created later, so we
        // can only pass up the capabilities we received from the broker for
        // that remote.
        delegate_->OnIntroduce(remote_node_name_, data.name,
                               std::move(channel_handle), data.capabilities);
        return;
      }
      break;
    }

#if BUILDFLAG(IS_WIN)
    case MessageType::RELAY_EVENT_MESSAGE: {
      base::ProcessHandle from_process;
      {
        base::AutoLock lock(remote_process_handle_lock_);
        // NOTE: It's safe to retain a weak reference to this process handle
        // through the extent of this call because |this| is kept alive and
        // |remote_process_handle_| is never reset once set.
        from_process = remote_process_handle_.Handle();

        // If we don't have a handle to the remote process, we should not be
        // receiving relay requests from them because we're not the broker.
        if (from_process == base::kNullProcessHandle)
          break;
      }
      RelayEventMessageData data;
      if (GetMessagePayload(payload, payload_size, &data)) {
        // Don't try to relay an empty message.
        if (payload_size <= sizeof(Header) + sizeof(data))
          break;

        Channel::HandlePolicy handle_policy;
        {
          base::AutoLock lock(channel_lock_);
          handle_policy = channel_->handle_policy();
        }

        const void* message_start = reinterpret_cast<const uint8_t*>(payload) +
                                    sizeof(Header) + sizeof(data);
        Channel::MessagePtr message = Channel::Message::Deserialize(
            message_start, payload_size - sizeof(Header) - sizeof(data),
            handle_policy, from_process);
        if (!message) {
          DLOG(ERROR) << "Dropping invalid relay message.";
          break;
        }
        delegate_->OnRelayEventMessage(remote_node_name_, from_process,
                                       data.destination, std::move(message));
        return;
      }
      break;
    }
#endif

    case MessageType::BROADCAST_EVENT: {
      if (payload_size <= sizeof(Header))
        break;
      const void* data = static_cast<const void*>(
          reinterpret_cast<const Header*>(payload) + 1);
      Channel::MessagePtr message =
          Channel::Message::Deserialize(data, payload_size - sizeof(Header),
                                        Channel::HandlePolicy::kRejectHandles);
      if (!message) {
        DLOG(ERROR) << "Dropping invalid broadcast message.";
        break;
      }
      delegate_->OnBroadcast(remote_node_name_, std::move(message));
      return;
    }

#if BUILDFLAG(IS_WIN)
    case MessageType::EVENT_MESSAGE_FROM_RELAY: {
      EventMessageFromRelayData data;
      if (GetMessagePayload(payload, payload_size, &data)) {
        if (payload_size < (sizeof(Header) + sizeof(data)))
          break;

        size_t num_bytes = payload_size - sizeof(data) - sizeof(Header);

        Channel::MessagePtr message =
            Channel::Message::CreateMessage(num_bytes, handles.size());
        message->SetHandles(std::move(handles));
        if (num_bytes)
          memcpy(message->mutable_payload(),
                 static_cast<const uint8_t*>(payload) + sizeof(Header) +
                     sizeof(data),
                 num_bytes);
        delegate_->OnEventMessageFromRelay(remote_node_name_, data.source,
                                           std::move(message));
        return;
      }
      break;
    }
#endif  // BUILDFLAG(IS_WIN)

    case MessageType::ACCEPT_PEER: {
      AcceptPeerData data;
      if (GetMessagePayload(payload, payload_size, &data)) {
        delegate_->OnAcceptPeer(remote_node_name_, data.token, data.peer_name,
                                data.port_name);
        return;
      }
      break;
    }

    case MessageType::BIND_BROKER_HOST:
      if (handles.size() == 1) {
        CreateAndBindLocalBrokerHost(std::move(handles[0]));
        return;
      }
      break;

    default:
      // Ignore unrecognized message types, allowing for future extensibility.
      return;
  }

  DLOG(ERROR) << "Received invalid message type: "
              << static_cast<int>(header->type) << " closing channel.";
  if (process_error_callback_)
    process_error_callback_.Run("NodeChannel received a malformed message");
  delegate_->OnChannelError(remote_node_name_, this);
}

void NodeChannel::OnChannelError(Channel::Error error) {
  DCHECK(owning_task_runner()->RunsTasksInCurrentSequence());

  RequestContext request_context(RequestContext::Source::SYSTEM);

  ShutDown();

  if (process_error_callback_ &&
      error == Channel::Error::kReceivedMalformedData) {
    process_error_callback_.Run("Channel received a malformed message");
  }

  // |OnChannelError()| may cause |this| to be destroyed, but still need
  // access to the name after that destruction. So make a copy of
  // |remote_node_name_| so it can be used if |this| becomes destroyed.
  ports::NodeName node_name = remote_node_name_;
  delegate_->OnChannelError(node_name, this);
}

void NodeChannel::WriteChannelMessage(Channel::MessagePtr message) {
  base::AutoLock lock(channel_lock_);
  if (!channel_)
    DLOG(ERROR) << "Dropping message on closed channel.";
  else
    channel_->Write(std::move(message));
}

void NodeChannel::OfferChannelUpgrade() {
#if !BUILDFLAG(IS_NACL)
  base::AutoLock lock(channel_lock_);
  channel_->OfferChannelUpgrade();
#endif
}

uint64_t NodeChannel::RemoteCapabilities() const {
  return remote_capabilities_;
}

bool NodeChannel::HasRemoteCapability(const uint64_t capability) const {
  return (remote_capabilities_ & capability) == capability;
}

void NodeChannel::SetRemoteCapabilities(const uint64_t capabilities) {
  remote_capabilities_ |= capabilities;
}

uint64_t NodeChannel::LocalCapabilities() const {
  return local_capabilities_;
}

bool NodeChannel::HasLocalCapability(const uint64_t capability) const {
  return (local_capabilities_ & capability) == capability;
}

void NodeChannel::SetLocalCapabilities(const uint64_t capabilities) {
  if (GetConfiguration().dont_advertise_capabilities) {
    return;
  }

  local_capabilities_ |= capabilities;
}

void NodeChannel::InitializeLocalCapabilities() {
  if (GetConfiguration().dont_advertise_capabilities) {
    return;
  }

  if (core::Channel::SupportsChannelUpgrade()) {
    SetLocalCapabilities(kNodeCapabilitySupportsUpgrade);
  }
}

}  // namespace core
}  // namespace mojo