blob: d5b37dce38323701cdd8efb88a1a5200352307e8 [file] [log] [blame]
// Copyright 2017 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 "mojo/core/channel.h"
#include "base/bind.h"
#include "base/memory/ptr_util.h"
#include "base/message_loop/message_loop.h"
#include "base/optional.h"
#include "base/process/process_handle.h"
#include "base/run_loop.h"
#include "base/threading/thread.h"
#include "mojo/core/platform_handle_utils.h"
#include "mojo/public/cpp/platform/platform_channel.h"
#include "testing/gmock/include/gmock/gmock.h"
#include "testing/gtest/include/gtest/gtest.h"
namespace mojo {
namespace core {
namespace {
class TestChannel : public Channel {
public:
TestChannel(Channel::Delegate* delegate)
: Channel(delegate, Channel::HandlePolicy::kAcceptHandles) {}
char* GetReadBufferTest(size_t* buffer_capacity) {
return GetReadBuffer(buffer_capacity);
}
bool OnReadCompleteTest(size_t bytes_read, size_t* next_read_size_hint) {
return OnReadComplete(bytes_read, next_read_size_hint);
}
MOCK_METHOD7(GetReadPlatformHandles,
bool(const void* payload,
size_t payload_size,
size_t num_handles,
const void* extra_header,
size_t extra_header_size,
std::vector<PlatformHandle>* handles,
bool* deferred));
MOCK_METHOD0(Start, void());
MOCK_METHOD0(ShutDownImpl, void());
MOCK_METHOD0(LeakHandle, void());
void Write(MessagePtr message) override {}
protected:
~TestChannel() override {}
};
// Not using GMock as I don't think it supports movable types.
class MockChannelDelegate : public Channel::Delegate {
public:
MockChannelDelegate() {}
size_t GetReceivedPayloadSize() const { return payload_size_; }
const void* GetReceivedPayload() const { return payload_.get(); }
protected:
void OnChannelMessage(const void* payload,
size_t payload_size,
std::vector<PlatformHandle> handles) override {
payload_.reset(new char[payload_size]);
memcpy(payload_.get(), payload, payload_size);
payload_size_ = payload_size;
}
// Notify that an error has occured and the Channel will cease operation.
void OnChannelError(Channel::Error error) override {}
private:
size_t payload_size_ = 0;
std::unique_ptr<char[]> payload_;
};
Channel::MessagePtr CreateDefaultMessage(bool legacy_message) {
const size_t payload_size = 100;
Channel::MessagePtr message = std::make_unique<Channel::Message>(
payload_size, 0,
legacy_message ? Channel::Message::MessageType::NORMAL_LEGACY
: Channel::Message::MessageType::NORMAL);
char* payload = static_cast<char*>(message->mutable_payload());
for (size_t i = 0; i < payload_size; i++) {
payload[i] = static_cast<char>(i);
}
return message;
}
void TestMemoryEqual(const void* data1,
size_t data1_size,
const void* data2,
size_t data2_size) {
ASSERT_EQ(data1_size, data2_size);
const unsigned char* data1_char = static_cast<const unsigned char*>(data1);
const unsigned char* data2_char = static_cast<const unsigned char*>(data2);
for (size_t i = 0; i < data1_size; i++) {
// ASSERT so we don't log tons of errors if the data is different.
ASSERT_EQ(data1_char[i], data2_char[i]);
}
}
void TestMessagesAreEqual(Channel::Message* message1,
Channel::Message* message2,
bool legacy_messages) {
// If any of the message is null, this is probably not what you wanted to
// test.
ASSERT_NE(nullptr, message1);
ASSERT_NE(nullptr, message2);
ASSERT_EQ(message1->payload_size(), message2->payload_size());
EXPECT_EQ(message1->has_handles(), message2->has_handles());
TestMemoryEqual(message1->payload(), message1->payload_size(),
message2->payload(), message2->payload_size());
if (legacy_messages)
return;
ASSERT_EQ(message1->extra_header_size(), message2->extra_header_size());
TestMemoryEqual(message1->extra_header(), message1->extra_header_size(),
message2->extra_header(), message2->extra_header_size());
}
TEST(ChannelTest, LegacyMessageDeserialization) {
Channel::MessagePtr message = CreateDefaultMessage(true /* legacy_message */);
Channel::MessagePtr deserialized_message =
Channel::Message::Deserialize(message->data(), message->data_num_bytes());
TestMessagesAreEqual(message.get(), deserialized_message.get(),
true /* legacy_message */);
}
TEST(ChannelTest, NonLegacyMessageDeserialization) {
Channel::MessagePtr message =
CreateDefaultMessage(false /* legacy_message */);
Channel::MessagePtr deserialized_message =
Channel::Message::Deserialize(message->data(), message->data_num_bytes());
TestMessagesAreEqual(message.get(), deserialized_message.get(),
false /* legacy_message */);
}
TEST(ChannelTest, OnReadLegacyMessage) {
size_t buffer_size = 100 * 1024;
Channel::MessagePtr message = CreateDefaultMessage(true /* legacy_message */);
MockChannelDelegate channel_delegate;
scoped_refptr<TestChannel> channel = new TestChannel(&channel_delegate);
char* read_buffer = channel->GetReadBufferTest(&buffer_size);
ASSERT_LT(message->data_num_bytes(),
buffer_size); // Bad test. Increase buffer
// size.
memcpy(read_buffer, message->data(), message->data_num_bytes());
size_t next_read_size_hint = 0;
EXPECT_TRUE(channel->OnReadCompleteTest(message->data_num_bytes(),
&next_read_size_hint));
TestMemoryEqual(message->payload(), message->payload_size(),
channel_delegate.GetReceivedPayload(),
channel_delegate.GetReceivedPayloadSize());
}
TEST(ChannelTest, OnReadNonLegacyMessage) {
size_t buffer_size = 100 * 1024;
Channel::MessagePtr message =
CreateDefaultMessage(false /* legacy_message */);
MockChannelDelegate channel_delegate;
scoped_refptr<TestChannel> channel = new TestChannel(&channel_delegate);
char* read_buffer = channel->GetReadBufferTest(&buffer_size);
ASSERT_LT(message->data_num_bytes(),
buffer_size); // Bad test. Increase buffer
// size.
memcpy(read_buffer, message->data(), message->data_num_bytes());
size_t next_read_size_hint = 0;
EXPECT_TRUE(channel->OnReadCompleteTest(message->data_num_bytes(),
&next_read_size_hint));
TestMemoryEqual(message->payload(), message->payload_size(),
channel_delegate.GetReceivedPayload(),
channel_delegate.GetReceivedPayloadSize());
}
class ChannelTestShutdownAndWriteDelegate : public Channel::Delegate {
public:
ChannelTestShutdownAndWriteDelegate(
PlatformChannelEndpoint endpoint,
scoped_refptr<base::TaskRunner> task_runner,
scoped_refptr<Channel> client_channel,
std::unique_ptr<base::Thread> client_thread,
base::RepeatingClosure quit_closure)
: quit_closure_(std::move(quit_closure)),
client_channel_(std::move(client_channel)),
client_thread_(std::move(client_thread)) {
channel_ = Channel::Create(this, ConnectionParams(std::move(endpoint)),
Channel::HandlePolicy::kAcceptHandles,
std::move(task_runner));
channel_->Start();
}
~ChannelTestShutdownAndWriteDelegate() override { channel_->ShutDown(); }
// Channel::Delegate implementation
void OnChannelMessage(const void* payload,
size_t payload_size,
std::vector<PlatformHandle> handles) override {
++message_count_;
// If |client_channel_| exists then close it and its thread.
if (client_channel_) {
// Write a fresh message, making our channel readable again.
Channel::MessagePtr message = CreateDefaultMessage(false);
client_thread_->task_runner()->PostTask(
FROM_HERE, base::BindOnce(&Channel::Write, client_channel_,
base::Passed(&message)));
// Close the channel and wait for it to shutdown.
client_channel_->ShutDown();
client_channel_ = nullptr;
client_thread_->Stop();
client_thread_ = nullptr;
}
// Write a message to the channel, to verify whether this triggers an
// OnChannelError callback before all messages were read.
Channel::MessagePtr message = CreateDefaultMessage(false);
channel_->Write(std::move(message));
}
void OnChannelError(Channel::Error error) override {
EXPECT_EQ(2, message_count_);
quit_closure_.Run();
}
base::RepeatingClosure quit_closure_;
int message_count_ = 0;
scoped_refptr<Channel> channel_;
scoped_refptr<Channel> client_channel_;
std::unique_ptr<base::Thread> client_thread_;
};
TEST(ChannelTest, PeerShutdownDuringRead) {
base::MessageLoop message_loop(base::MessageLoop::TYPE_IO);
PlatformChannel channel;
// Create a "client" Channel with one end of the pipe, and Start() it.
std::unique_ptr<base::Thread> client_thread =
std::make_unique<base::Thread>("clientio_thread");
client_thread->StartWithOptions(
base::Thread::Options(base::MessageLoop::TYPE_IO, 0));
scoped_refptr<Channel> client_channel = Channel::Create(
nullptr, ConnectionParams(channel.TakeRemoteEndpoint()),
Channel::HandlePolicy::kAcceptHandles, client_thread->task_runner());
client_channel->Start();
// On the "client" IO thread, create and write a message.
Channel::MessagePtr message = CreateDefaultMessage(false);
client_thread->task_runner()->PostTask(
FROM_HERE,
base::BindOnce(&Channel::Write, client_channel, base::Passed(&message)));
// Create a "server" Channel with the other end of the pipe, and process the
// messages from it. The |server_delegate| will ShutDown the client end of
// the pipe after the first message, and quit the RunLoop when OnChannelError
// is received.
base::RunLoop run_loop;
ChannelTestShutdownAndWriteDelegate server_delegate(
channel.TakeLocalEndpoint(), message_loop.task_runner(),
std::move(client_channel), std::move(client_thread),
run_loop.QuitClosure());
run_loop.Run();
}
class RejectHandlesDelegate : public Channel::Delegate {
public:
RejectHandlesDelegate() = default;
size_t num_messages() const { return num_messages_; }
// Channel::Delegate:
void OnChannelMessage(const void* payload,
size_t payload_size,
std::vector<PlatformHandle> handles) override {
++num_messages_;
}
void OnChannelError(Channel::Error error) override {
if (wait_for_error_loop_)
wait_for_error_loop_->Quit();
}
void WaitForError() {
wait_for_error_loop_.emplace();
wait_for_error_loop_->Run();
}
private:
size_t num_messages_ = 0;
base::Optional<base::RunLoop> wait_for_error_loop_;
DISALLOW_COPY_AND_ASSIGN(RejectHandlesDelegate);
};
TEST(ChannelTest, RejectHandles) {
base::MessageLoop message_loop(base::MessageLoop::TYPE_IO);
PlatformChannel platform_channel;
RejectHandlesDelegate receiver_delegate;
scoped_refptr<Channel> receiver = Channel::Create(
&receiver_delegate,
ConnectionParams(platform_channel.TakeLocalEndpoint()),
Channel::HandlePolicy::kRejectHandles, message_loop.task_runner());
receiver->Start();
RejectHandlesDelegate sender_delegate;
scoped_refptr<Channel> sender = Channel::Create(
&sender_delegate, ConnectionParams(platform_channel.TakeRemoteEndpoint()),
Channel::HandlePolicy::kRejectHandles, message_loop.task_runner());
sender->Start();
// Create another platform channel just to stuff one of its endpoint handles
// into a message. Sending this message to the receiver should cause the
// receiver to reject it and close the Channel without ever dispatching the
// message.
PlatformChannel dummy_channel;
std::vector<mojo::PlatformHandle> handles;
handles.push_back(dummy_channel.TakeLocalEndpoint().TakePlatformHandle());
auto message = std::make_unique<Channel::Message>(0 /* payload_size */,
1 /* max_handles */);
message->SetHandles(std::move(handles));
sender->Write(std::move(message));
receiver_delegate.WaitForError();
EXPECT_EQ(0u, receiver_delegate.num_messages());
}
TEST(ChannelTest, DeserializeMessage_BadExtraHeaderSize) {
// Verifies that a message payload is rejected when the extra header chunk
// size not properly aligned.
constexpr uint16_t kBadAlignment = kChannelMessageAlignment + 1;
constexpr uint16_t kTotalHeaderSize =
sizeof(Channel::Message::Header) + kBadAlignment;
constexpr uint32_t kEmptyPayloadSize = 8;
constexpr uint32_t kMessageSize = kTotalHeaderSize + kEmptyPayloadSize;
char message[kMessageSize];
memset(message, 0, kMessageSize);
Channel::Message::Header* header =
reinterpret_cast<Channel::Message::Header*>(&message[0]);
header->num_bytes = kMessageSize;
header->num_header_bytes = kTotalHeaderSize;
header->message_type = Channel::Message::MessageType::NORMAL;
header->num_handles = 0;
EXPECT_EQ(nullptr, Channel::Message::Deserialize(&message[0], kMessageSize,
base::kNullProcessHandle));
}
} // namespace
} // namespace core
} // namespace mojo