Work around race condition during encryption setup

The sending party may finish setting up encryption and start
sending encrypted packets while the receiving party is still configuring
encryption and expecting unencrypted packets

PiperOrigin-RevId: 549267838

connections/implementation/base_endpoint_channel.cc

@@ -49,7 +49,7 @@ ByteArray IntToBytes(std::int32_t value) {
   int_bytes[0] = static_cast<char>((value >> 24) & 0x0FF);
   int_bytes[1] = static_cast<char>((value >> 16) & 0x0FF);
   int_bytes[2] = static_cast<char>((value >> 8) & 0x0FF);
-  int_bytes[3] = static_cast<char>((value)&0x0FF);
+  int_bytes[3] = static_cast<char>((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<ByteArray> BaseEndpointChannel::TryDecrypt(const ByteArray& data) {
+  MutexLock crypto_lock(&crypto_mutex_);
+  if (!IsEncryptionEnabledLocked()) {
+    return Exception::kFailed;
+  }
+  std::unique_ptr<std::string> decrypted_data =
+      crypto_context_->DecodeMessageFromPeer(data.string_data());
+  if (decrypted_data) {
+    return ExceptionOr<ByteArray>(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<std::string> BaseEndpointChannel::EncodeMessageForTests(
+    absl::string_view data) {
+  MutexLock lock(&crypto_mutex_);
+  DCHECK(IsEncryptionEnabledLocked());
+  return crypto_context_->EncodeMessageToPeer(std::string(data));
+}
+
 }  // namespace connections
 }  // namespace nearby

connections/implementation/base_endpoint_channel.h

@@ -68,6 +68,8 @@ class BaseEndpointChannel : public EndpointChannel {
   int GetMaxTransmitPacketSize() const override;
   void EnableEncryption(std::shared_ptr<EncryptionContext> context) override;
   void DisableEncryption() override;
+  bool IsEncrypted() override;
+  ExceptionOr<ByteArray> 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<std::string> EncodeMessageForTests(absl::string_view data);
 
  private:
   // Used to sanity check that our frame sizes are reasonable.

connections/implementation/base_endpoint_channel_test.cc

@@ -18,7 +18,6 @@
 #include <string>
 #include <utility>
 
-#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<ByteArray> 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<std::string> encrypted_message =
+      channel_a.EncodeMessageForTests(kMessage);
+
+  ExceptionOr<ByteArray> 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<ByteArray> 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"};

connections/implementation/connections_authentication_transport_test.cc

@@ -59,6 +59,9 @@ class MockEndpointChannel : public EndpointChannel {
   MOCK_METHOD(void, EnableEncryption, (std::shared_ptr<EncryptionContext>),
               (override));
   MOCK_METHOD(void, DisableEncryption, (), (override));
+  MOCK_METHOD(bool, IsEncrypted, (), (override));
+  MOCK_METHOD(ExceptionOr<ByteArray>, TryDecrypt, (const ByteArray& data),
+              (override));
   MOCK_METHOD(bool, IsPaused, (), (const override));
   MOCK_METHOD(void, Pause, (), (override));
   MOCK_METHOD(void, Resume, (), (override));

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<EncryptionContext> context) override {}
   void DisableEncryption() override {}
+  bool IsEncrypted() override { return false; }
+  ExceptionOr<ByteArray> TryDecrypt(const ByteArray& data) override {
+    return Exception::kFailed;
+  }
+
   bool IsPaused() const override { return false; }
   void Pause() override {}
   void Resume() override {}

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<ByteArray> TryDecrypt(const ByteArray& data) = 0;
+
   // True if the EndpointChannel is currently pausing all writes.
   virtual bool IsPaused() const = 0;
 

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<OfflineFrame> EndpointManager::TryDecryptFrame(
+    const ByteArray& data, EndpointChannel* endpoint_channel) {
+  auto start_time = SystemClock::ElapsedRealtime();
+  while (true) {
+    ExceptionOr<ByteArray> 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<bool> 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<bool> EndpointManager::HandleData(
       return ExceptionOr<bool>(bytes.exception());
     }
     ExceptionOr<OfflineFrame> 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<OfflineFrame> 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)) {

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<EndpointChannel> 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<EndpointChannel> 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<OfflineFrame> TryDecryptFrame(const ByteArray& data,
+                                            EndpointChannel* endpoint_channel);
   EndpointChannelManager* channel_manager_;
 
   RecursiveMutex frame_processors_lock_;

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<EncryptionContext> context), (override));
   MOCK_METHOD(void, DisableEncryption, (), (override));
   MOCK_METHOD(bool, IsPaused, (), (const override));
+  MOCK_METHOD(bool, IsEncrypted, (), (override));
+  MOCK_METHOD(ExceptionOr<ByteArray>, 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<MockEndpointChannel>();
   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<MockEndpointChannel>();
+  EXPECT_CALL(*endpoint_channel, Read(_))
+      .WillOnce(Return(ExceptionOr<ByteArray>(payload)))
+      .WillRepeatedly(Return(ExceptionOr<ByteArray>(Exception::kIo)));
+  EXPECT_CALL(*endpoint_channel, TryDecrypt(Eq(payload)))
+      .WillRepeatedly(Return(ExceptionOr<ByteArray>(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<MockEndpointChannel>();
+  EXPECT_CALL(*endpoint_channel, Read(_))
+      .WillOnce(Return(ExceptionOr<ByteArray>(payload)))
+      .WillRepeatedly(Return(ExceptionOr<ByteArray>(Exception::kIo)));
+  EXPECT_CALL(*endpoint_channel, TryDecrypt(Eq(payload)))
+      .WillOnce(Return(ExceptionOr<ByteArray>(Exception::kFailed)))
+      .WillRepeatedly(Return(ExceptionOr<ByteArray>(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<MockEndpointChannel>();
+  EXPECT_CALL(*endpoint_channel, Read(_))
+      .WillOnce(Return(ExceptionOr<ByteArray>(payload)))
+      .WillRepeatedly(Return(ExceptionOr<ByteArray>(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<MockEndpointChannel>();
+  auto connect_request = std::make_unique<MockFrameProcessor>();
+  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>{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<ByteArray>(payload)))
+      .WillRepeatedly(Return(ExceptionOr<ByteArray>(Exception::kIo)));
+  EXPECT_CALL(*endpoint_channel, TryDecrypt(Eq(payload)))
+      .WillOnce(Return(ExceptionOr<ByteArray>(Exception::kFailed)))
+      .WillOnce(Return(ExceptionOr<ByteArray>(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)

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<EncryptionContext> context) override {}
   void DisableEncryption() override {}
+  bool IsEncrypted() override { return false; }
+  ExceptionOr<ByteArray> 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; }