diff --git a/connections/implementation/base_endpoint_channel.cc b/connections/implementation/base_endpoint_channel.cc index b2b2f96e..d9b98916 100644 --- a/connections/implementation/base_endpoint_channel.cc +++ b/connections/implementation/base_endpoint_channel.cc @@ -49,7 +49,7 @@ ByteArray IntToBytes(std::int32_t value) { int_bytes[0] = static_cast((value >> 24) & 0x0FF); int_bytes[1] = static_cast((value >> 16) & 0x0FF); int_bytes[2] = static_cast((value >> 8) & 0x0FF); - int_bytes[3] = static_cast((value)&0x0FF); + int_bytes[3] = static_cast((value) & 0x0FF); return ByteArray(int_bytes, sizeof(int_bytes)); } @@ -348,6 +348,24 @@ void BaseEndpointChannel::DisableEncryption() { crypto_context_.reset(); } +bool BaseEndpointChannel::IsEncrypted() { + MutexLock crypto_lock(&crypto_mutex_); + return IsEncryptionEnabledLocked(); +} + +ExceptionOr BaseEndpointChannel::TryDecrypt(const ByteArray& data) { + MutexLock crypto_lock(&crypto_mutex_); + if (!IsEncryptionEnabledLocked()) { + return Exception::kFailed; + } + std::unique_ptr decrypted_data = + crypto_context_->DecodeMessageFromPeer(data.string_data()); + if (decrypted_data) { + return ExceptionOr(ByteArray(std::move(*decrypted_data))); + } + return Exception::kExecution; +} + bool BaseEndpointChannel::IsPaused() const { MutexLock lock(&is_paused_mutex_); return is_paused_; @@ -415,5 +433,12 @@ void BaseEndpointChannel::UnblockPausedWriter() { is_paused_cond_.Notify(); } +std::unique_ptr BaseEndpointChannel::EncodeMessageForTests( + absl::string_view data) { + MutexLock lock(&crypto_mutex_); + DCHECK(IsEncryptionEnabledLocked()); + return crypto_context_->EncodeMessageToPeer(std::string(data)); +} + } // namespace connections } // namespace nearby diff --git a/connections/implementation/base_endpoint_channel.h b/connections/implementation/base_endpoint_channel.h index 18df11e2..e3d77f65 100644 --- a/connections/implementation/base_endpoint_channel.h +++ b/connections/implementation/base_endpoint_channel.h @@ -68,6 +68,8 @@ class BaseEndpointChannel : public EndpointChannel { int GetMaxTransmitPacketSize() const override; void EnableEncryption(std::shared_ptr context) override; void DisableEncryption() override; + bool IsEncrypted() override; + ExceptionOr TryDecrypt(const ByteArray& data) override; bool IsPaused() const ABSL_LOCKS_EXCLUDED(is_paused_mutex_) override; void Pause() ABSL_LOCKS_EXCLUDED(is_paused_mutex_) override; void Resume() ABSL_LOCKS_EXCLUDED(is_paused_mutex_) override; @@ -80,6 +82,8 @@ class BaseEndpointChannel : public EndpointChannel { protected: virtual void CloseImpl() = 0; + // For tests only. + std::unique_ptr EncodeMessageForTests(absl::string_view data); private: // Used to sanity check that our frame sizes are reasonable. diff --git a/connections/implementation/base_endpoint_channel_test.cc b/connections/implementation/base_endpoint_channel_test.cc index da2b05ce..03ca6898 100644 --- a/connections/implementation/base_endpoint_channel_test.cc +++ b/connections/implementation/base_endpoint_channel_test.cc @@ -18,7 +18,6 @@ #include #include -#include "securegcm/d2d_connection_context_v1.h" #include "securegcm/ukey2_handshake.h" #include "gmock/gmock.h" #include "protobuf-matchers/protocol-buffer-matchers.h" @@ -28,12 +27,12 @@ #include "connections/implementation/encryption_runner.h" #include "connections/implementation/offline_frames.h" #include "internal/platform/byte_array.h" +#include "internal/platform/count_down_latch.h" #include "internal/platform/exception.h" #include "internal/platform/input_stream.h" -#include "internal/platform/output_stream.h" -#include "internal/platform/count_down_latch.h" #include "internal/platform/logging.h" #include "internal/platform/multi_thread_executor.h" +#include "internal/platform/output_stream.h" #include "internal/platform/pipe.h" #include "internal/platform/single_thread_executor.h" #include "proto/connections_enums.pb.h" @@ -51,6 +50,8 @@ class TestEndpointChannel : public BaseEndpointChannel { explicit TestEndpointChannel(InputStream* input, OutputStream* output) : BaseEndpointChannel("service_id", "channel", input, output) {} + using BaseEndpointChannel::EncodeMessageForTests; + MOCK_METHOD(Medium, GetMedium, (), (const override)); MOCK_METHOD(void, CloseImpl, (), (override)); }; @@ -178,6 +179,60 @@ TEST(BaseEndpointChannelTest, ReadWrite) { EXPECT_EQ(rx_message, tx_message); } +TEST(BaseEndpointChannelTest, ChannelUnencryptedByDefault) { + Pipe pipe; + TestEndpointChannel channel(&pipe.GetInputStream(), &pipe.GetOutputStream()); + + ExceptionOr result = channel.TryDecrypt(ByteArray("message")); + + EXPECT_FALSE(channel.IsEncrypted()); + EXPECT_FALSE(result.ok()); + EXPECT_EQ(result.exception(), Exception::kFailed); +} + +TEST(BaseEndpointChannelTest, TryDecrypt) { + absl::string_view kMessage = "message"; + Pipe pipe_a; // channel_a writes to pipe_a, reads from pipe_b. + Pipe pipe_b; // channel_b writes to pipe_b, reads from pipe_a. + TestEndpointChannel channel_a(&pipe_b.GetInputStream(), + &pipe_a.GetOutputStream()); + TestEndpointChannel channel_b(&pipe_a.GetInputStream(), + &pipe_b.GetOutputStream()); + auto [context_a, context_b] = DoDhKeyExchange(&channel_a, &channel_b); + ASSERT_NE(context_a, nullptr); + ASSERT_NE(context_b, nullptr); + channel_a.EnableEncryption(context_a); + channel_b.EnableEncryption(context_b); + std::unique_ptr encrypted_message = + channel_a.EncodeMessageForTests(kMessage); + + ExceptionOr decrypted_message = + channel_b.TryDecrypt(ByteArray(*encrypted_message)); + + EXPECT_TRUE(channel_b.IsEncrypted()); + EXPECT_TRUE(decrypted_message.ok()); + EXPECT_EQ(decrypted_message.result().AsStringView(), kMessage); +} + +TEST(BaseEndpointChannelTest, TryDecryptFailsWhenDecryptionFails) { + Pipe pipe_a; // channel_a writes to pipe_a, reads from pipe_b. + Pipe pipe_b; // channel_b writes to pipe_b, reads from pipe_a. + TestEndpointChannel channel_a(&pipe_b.GetInputStream(), + &pipe_a.GetOutputStream()); + TestEndpointChannel channel_b(&pipe_a.GetInputStream(), + &pipe_b.GetOutputStream()); + auto [context_a, context_b] = DoDhKeyExchange(&channel_a, &channel_b); + ASSERT_NE(context_a, nullptr); + channel_a.EnableEncryption(context_a); + + ExceptionOr result = + channel_a.TryDecrypt(ByteArray("invalid message")); + + EXPECT_TRUE(channel_a.IsEncrypted()); + EXPECT_FALSE(result.ok()); + EXPECT_EQ(result.exception(), Exception::kExecution); +} + TEST(BaseEndpointChannelTest, NotEncryptedReadWriteCanBeIntercepted) { // Not encrypted IO; MITM scenario. @@ -267,6 +322,8 @@ TEST(BaseEndpointChannelTest, EncryptedReadWriteCanNotBeIntercepted) { EXPECT_EQ(channel_a.GetType(), "ENCRYPTED_BLUETOOTH"); EXPECT_EQ(channel_b.GetType(), "ENCRYPTED_BLUETOOTH"); + EXPECT_TRUE(channel_a.IsEncrypted()); + EXPECT_TRUE(channel_b.IsEncrypted()); // Start data transfer ByteArray tx_message{"data message"}; diff --git a/connections/implementation/connections_authentication_transport_test.cc b/connections/implementation/connections_authentication_transport_test.cc index 492430e0..ef72c23e 100644 --- a/connections/implementation/connections_authentication_transport_test.cc +++ b/connections/implementation/connections_authentication_transport_test.cc @@ -59,6 +59,9 @@ class MockEndpointChannel : public EndpointChannel { MOCK_METHOD(void, EnableEncryption, (std::shared_ptr), (override)); MOCK_METHOD(void, DisableEncryption, (), (override)); + MOCK_METHOD(bool, IsEncrypted, (), (override)); + MOCK_METHOD(ExceptionOr, TryDecrypt, (const ByteArray& data), + (override)); MOCK_METHOD(bool, IsPaused, (), (const override)); MOCK_METHOD(void, Pause, (), (override)); MOCK_METHOD(void, Resume, (), (override)); diff --git a/connections/implementation/encryption_runner_test.cc b/connections/implementation/encryption_runner_test.cc index 270b8273..0bc91960 100644 --- a/connections/implementation/encryption_runner_test.cc +++ b/connections/implementation/encryption_runner_test.cc @@ -82,6 +82,11 @@ class FakeEndpointChannel : public EndpointChannel { int GetMaxTransmitPacketSize() const override { return 512; } void EnableEncryption(std::shared_ptr context) override {} void DisableEncryption() override {} + bool IsEncrypted() override { return false; } + ExceptionOr TryDecrypt(const ByteArray& data) override { + return Exception::kFailed; + } + bool IsPaused() const override { return false; } void Pause() override {} void Resume() override {} diff --git a/connections/implementation/endpoint_channel.h b/connections/implementation/endpoint_channel.h index 05b093bb..f0fc3e67 100644 --- a/connections/implementation/endpoint_channel.h +++ b/connections/implementation/endpoint_channel.h @@ -91,6 +91,14 @@ class EndpointChannel { // Disables encryption on the EndpointChannel. virtual void DisableEncryption() = 0; + // Returns true if EndpointChannel is encrypted. + virtual bool IsEncrypted() = 0; + + // Decrypts `data` if encryption is enabled. + // Returns `kExecution` exception if encryption is enabled but decryption + // failed. Returns `kFailed` exception if encryption is not enabled. + virtual ExceptionOr TryDecrypt(const ByteArray& data) = 0; + // True if the EndpointChannel is currently pausing all writes. virtual bool IsPaused() const = 0; diff --git a/connections/implementation/endpoint_manager.cc b/connections/implementation/endpoint_manager.cc index 5ad88e8d..57f75942 100644 --- a/connections/implementation/endpoint_manager.cc +++ b/connections/implementation/endpoint_manager.cc @@ -35,13 +35,21 @@ namespace nearby { namespace connections { +namespace { using ::location::nearby::connections::OfflineFrame; using ::location::nearby::connections::V1Frame; using ::nearby::analytics::PacketMetaData; -using ::nearby::connections::PayloadDirection; -constexpr absl::Duration EndpointManager::kProcessEndpointDisconnectionTimeout; -constexpr absl::Time EndpointManager::kInvalidTimestamp; +// We set this to 11s to provide sufficient time for an in-progress WebRTC +// bandwidth upgrade to resolve. This is chosen to be slightly longer than the +// 10s timeout in WebRtc::AttemptToConnect(). +constexpr absl::Duration kProcessEndpointDisconnectionTimeout = + absl::Seconds(11); +constexpr absl::Time kInvalidTimestamp = absl::InfinitePast(); +// The maximum time we will wait for the encryption setup during negotiating a +// connection. +constexpr absl::Duration kDecryptRetryTimeout = absl::Seconds(3); +} // namespace class EndpointManager::LockedFrameProcessor { public: @@ -168,9 +176,33 @@ void EndpointManager::EndpointChannelLoopRunnable( << "; endpoint_id=" << endpoint_id; } +ExceptionOr EndpointManager::TryDecryptFrame( + const ByteArray& data, EndpointChannel* endpoint_channel) { + auto start_time = SystemClock::ElapsedRealtime(); + while (true) { + ExceptionOr decrypted = endpoint_channel->TryDecrypt(data); + if (decrypted.ok()) { + NEARBY_LOGS(VERBOSE) << "Message decrypted after " + << SystemClock::ElapsedRealtime() - start_time; + return parser::FromBytes(decrypted.result()); + } + if (decrypted.exception() == Exception::kExecution) { + return decrypted.exception(); + } + auto elapsed = SystemClock::ElapsedRealtime() - start_time; + if (elapsed > kDecryptRetryTimeout) { + NEARBY_LOGS(WARNING) << "Can't decrypt the mesage. Timeout after " + << elapsed; + return Exception::kTimeout; + } + SystemClock::Sleep(absl::Milliseconds(1)); + } +} + ExceptionOr EndpointManager::HandleData( const std::string& endpoint_id, ClientProxy* client, EndpointChannel* endpoint_channel) { + bool try_decrypting = !endpoint_channel->IsEncrypted(); // Read as much as we can from the healthy EndpointChannel - when it is no // longer in good shape (i.e. our read from it throws an Exception), our // super class will loop back around and try our luck in case there's been @@ -185,6 +217,23 @@ ExceptionOr EndpointManager::HandleData( return ExceptionOr(bytes.exception()); } ExceptionOr wrapped_frame = parser::FromBytes(bytes.result()); + if (!wrapped_frame.ok() && try_decrypting) { + // Workaround for a race condition where the remote party has sent an + // encrypted message but our end was still configured as unencrypted when + // the message was received. The workaround is to wait until the + // encryption set-up has completed on another thread. We run this + // workaround if: + // - the connection was unencrypted when we started reading from the + // channel + // - the received frame looks wrong (corrupted) + // - it's the first invalid frame. + try_decrypting = false; + ExceptionOr decrypted = + TryDecryptFrame(bytes.result(), endpoint_channel); + if (decrypted.ok()) { + wrapped_frame = std::move(decrypted); + } + } if (!wrapped_frame.ok()) { if (wrapped_frame.GetException().Raised( Exception::kInvalidProtocolBuffer)) { diff --git a/connections/implementation/endpoint_manager.h b/connections/implementation/endpoint_manager.h index 922090e9..c60e1484 100644 --- a/connections/implementation/endpoint_manager.h +++ b/connections/implementation/endpoint_manager.h @@ -61,6 +61,8 @@ namespace connections { class EndpointManager { public: + using OfflineFrame = ::location::nearby::connections::OfflineFrame; + class FrameProcessor { public: virtual ~FrameProcessor() = default; @@ -107,12 +109,12 @@ class EndpointManager { // Invoked from the different PcpHandler implementations (of which there can // be only one at a time). // Blocks until registration is complete. - void RegisterEndpoint( - ClientProxy* client, const std::string& endpoint_id, - const ConnectionResponseInfo& info, - const ConnectionOptions& connection_options, - std::unique_ptr channel, - const ConnectionListener& listener, const std::string& connection_token); + void RegisterEndpoint(ClientProxy* client, const std::string& endpoint_id, + const ConnectionResponseInfo& info, + const ConnectionOptions& connection_options, + std::unique_ptr channel, + const ConnectionListener& listener, + const std::string& connection_token); // Called when a client explicitly asks to disconnect from this endpoint. In // this case, we do not notify the client of onDisconnected(). void UnregisterEndpoint(ClientProxy* client, const std::string& endpoint_id); @@ -250,13 +252,6 @@ class EndpointManager { static void WaitForLatch(const std::string& method_name, CountDownLatch* latch, std::int32_t timeout_millis); - // We set this to 11s to provide sufficient time for an in-progress WebRTC - // bandwidth upgrade to resolve. This is chosen to be slightly longer than the - // 10s timeout in WebRtc::AttemptToConnect(). - static constexpr absl::Duration kProcessEndpointDisconnectionTimeout = - absl::Milliseconds(11000); - static constexpr absl::Time kInvalidTimestamp = absl::InfinitePast(); - // It should be noted that this method may be called multiple times (because // invoking this method closes the endpoint channel, which causes the // dedicated reader and KeepAlive threads to terminate, which in turn leads to @@ -283,6 +278,8 @@ class EndpointManager { // Executes all jobs sequentially, on a serial_executor_. void RunOnEndpointManagerThread(const std::string& name, Runnable runnable); + ExceptionOr TryDecryptFrame(const ByteArray& data, + EndpointChannel* endpoint_channel); EndpointChannelManager* channel_manager_; RecursiveMutex frame_processors_lock_; diff --git a/connections/implementation/endpoint_manager_test.cc b/connections/implementation/endpoint_manager_test.cc index 58c83162..60d13ffa 100644 --- a/connections/implementation/endpoint_manager_test.cc +++ b/connections/implementation/endpoint_manager_test.cc @@ -47,6 +47,7 @@ using ::location::nearby::connections::V1Frame; using ::location::nearby::proto::connections::DisconnectionReason; using ::location::nearby::proto::connections::Medium; using ::testing::_; +using ::testing::Eq; using ::testing::MockFunction; using ::testing::Return; using ::testing::StrictMock; @@ -77,6 +78,9 @@ class MockEndpointChannel : public EndpointChannel { (std::shared_ptr context), (override)); MOCK_METHOD(void, DisableEncryption, (), (override)); MOCK_METHOD(bool, IsPaused, (), (const override)); + MOCK_METHOD(bool, IsEncrypted, (), (override)); + MOCK_METHOD(ExceptionOr, TryDecrypt, (const ByteArray& data), + (override)); MOCK_METHOD(void, Pause, (), (override)); MOCK_METHOD(void, Resume, (), (override)); MOCK_METHOD(absl::Time, GetLastReadTimestamp, (), (const override)); @@ -299,7 +303,7 @@ TEST_F(EndpointManagerTest, SendControlMessageWorks) { NEARBY_LOG(INFO, "Will call destructors now"); } -TEST_F(EndpointManagerTest, SingleReadOnInvalidPayload) { +TEST_F(EndpointManagerTest, SingleReadOnReadError) { auto endpoint_channel = std::make_unique(); EXPECT_CALL(*endpoint_channel, Read(_)) .WillOnce( @@ -310,6 +314,102 @@ TEST_F(EndpointManagerTest, SingleReadOnInvalidPayload) { RegisterEndpoint(std::move(endpoint_channel)); } +TEST_F(EndpointManagerTest, ReadInvalidUnencryptedPayloadIgnoresFrame) { + // 1. EndpointChannel is unencrypted. + // 2. EndpointManager receives an invalid unencrypted frame. + // 3. EndpointManager calls EndpointChannel::TryDecrypt(), which keeps failing + // because the channel is not encrypted. + // 4. Invalid frame is ignored. No bad side effects. + CountDownLatch latch(1); + const ByteArray payload("not a valid frame"); + auto endpoint_channel = std::make_unique(); + EXPECT_CALL(*endpoint_channel, Read(_)) + .WillOnce(Return(ExceptionOr(payload))) + .WillRepeatedly(Return(ExceptionOr(Exception::kIo))); + EXPECT_CALL(*endpoint_channel, TryDecrypt(Eq(payload))) + .WillRepeatedly(Return(ExceptionOr(Exception::kFailed))); + EXPECT_CALL(*endpoint_channel, Write(_)) + .WillRepeatedly(Return(Exception{Exception::kSuccess})); + EXPECT_CALL(*endpoint_channel, Close(_)) + .WillOnce([&](DisconnectionReason reason) { latch.CountDown(); }); + RegisterEndpoint(std::move(endpoint_channel), false); + latch.Await(); + em_.UnregisterEndpoint(client_.get(), endpoint_id_); +} + +TEST_F(EndpointManagerTest, ReadInvalidEncryptedPayloadIgnoresFrame) { + // 1. EndpointChannel is unencrypted. + // 2. EndpointManager receives an invalid encrypted frame. + // 3. EndpointManager calls EndpointChannel::TryDecrypt(), decryption fails + // too. + // 4. Invalid frame is ignored. No bad side effects. + const ByteArray payload("not a valid frame"); + auto endpoint_channel = std::make_unique(); + EXPECT_CALL(*endpoint_channel, Read(_)) + .WillOnce(Return(ExceptionOr(payload))) + .WillRepeatedly(Return(ExceptionOr(Exception::kIo))); + EXPECT_CALL(*endpoint_channel, TryDecrypt(Eq(payload))) + .WillOnce(Return(ExceptionOr(Exception::kFailed))) + .WillRepeatedly(Return(ExceptionOr(Exception::kExecution))); + EXPECT_CALL(*endpoint_channel, Write(_)) + .WillRepeatedly(Return(Exception{Exception::kSuccess})); + RegisterEndpoint(std::move(endpoint_channel)); +} + +TEST_F(EndpointManagerTest, ReadInvalidPayloadFromEncryptedChannel) { + // 1. EndpointChannel is encrypted. + // 2. EndpointManager receives an invalid encrypted frame. + // 3. No calls to TryDecrypt. + const ByteArray payload("not a valid frame"); + auto endpoint_channel = std::make_unique(); + EXPECT_CALL(*endpoint_channel, Read(_)) + .WillOnce(Return(ExceptionOr(payload))) + .WillRepeatedly(Return(ExceptionOr(Exception::kIo))); + EXPECT_CALL(*endpoint_channel, IsEncrypted()).WillRepeatedly(Return(true)); + EXPECT_CALL(*endpoint_channel, TryDecrypt(Eq(payload))).Times(0); + EXPECT_CALL(*endpoint_channel, Write(_)) + .WillRepeatedly(Return(Exception{Exception::kSuccess})); + RegisterEndpoint(std::move(endpoint_channel)); +} + +TEST_F(EndpointManagerTest, TryDecrypt) { + // 1. EndpointChannel is unencrypted. + // 2. EndpointManager receives a valid encrypted frame alas it's interpreted + // as unencrypted at first. + // 3. EndpointManager calls EndpointChannel::TryDecrypt(), decryption works. + // 4. Frame is processed. + const ByteArray payload("valid encrypted frame"); + auto endpoint_channel = std::make_unique(); + auto connect_request = std::make_unique(); + ByteArray endpoint_info{"endpoint_name"}; + ConnectionInfo connection_info{ + "endpoint_id", + endpoint_info, + 1234 /*nonce*/, + false /*supports_5_ghz*/, + "" /*bssid*/, + 2412 /*ap_frequency*/, + "8xqT" /*ip_address in 4 bytes format*/, + std::vector{Medium::BLE} /*supported_mediums*/, + 0 /*keep_alive_interval_millis*/, + 0 /*keep_alive_timeout_millis*/}; + ByteArray decrypted_data = parser::ForConnectionRequest(connection_info); + EXPECT_CALL(*connect_request, OnIncomingFrame); + EXPECT_CALL(*connect_request, OnEndpointDisconnect); + EXPECT_CALL(*endpoint_channel, Read(_)) + .WillOnce(Return(ExceptionOr(payload))) + .WillRepeatedly(Return(ExceptionOr(Exception::kIo))); + EXPECT_CALL(*endpoint_channel, TryDecrypt(Eq(payload))) + .WillOnce(Return(ExceptionOr(Exception::kFailed))) + .WillOnce(Return(ExceptionOr(decrypted_data))); + EXPECT_CALL(*endpoint_channel, Write(_)) + .WillRepeatedly(Return(Exception{Exception::kSuccess})); + em_.RegisterFrameProcessor(V1Frame::CONNECTION_REQUEST, + connect_request.get()); + processors_.emplace_back(std::move(connect_request)); + RegisterEndpoint(std::move(endpoint_channel)); +} + // Regression test for b/278729669. // // During the destruction of NearbyConnections, Core (which owns ClientProxy) diff --git a/connections/implementation/fake_endpoint_channel.h b/connections/implementation/fake_endpoint_channel.h index db12c2c1..24744c76 100644 --- a/connections/implementation/fake_endpoint_channel.h +++ b/connections/implementation/fake_endpoint_channel.h @@ -77,6 +77,10 @@ class FakeEndpointChannel : public EndpointChannel { int GetMaxTransmitPacketSize() const override { return 512; } void EnableEncryption(std::shared_ptr context) override {} void DisableEncryption() override {} + bool IsEncrypted() override { return false; } + ExceptionOr TryDecrypt(const ByteArray& data) override { + return Exception::kFailed; + } bool IsPaused() const override { return is_paused_; } void Pause() override { is_paused_ = true; } void Resume() override { is_paused_ = false; }