device/fido: switch to 24-bit routing IDs.

caBLEv2 shard IDs have turned into 24-bit routing IDs.

Also, fix a race caused by the fact that the network service sends
notice of a data frame before writing its contents to the Mojo data
pipe. Thus the data might not be ready to read as the code previously
assumed.

BUG=1002262

Change-Id: I2af2ce1ffa909f09559fc061881ce8bc72fbea55
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/2416562
Commit-Queue: Adam Langley <agl@chromium.org>
Reviewed-by: Martin Kreichgauer <martinkr@google.com>
Cr-Commit-Position: refs/heads/master@{#810053}

device/fido/cable/fido_tunnel_device.cc

@@ -236,12 +236,7 @@ base::WeakPtr<FidoDevice> FidoTunnelDevice::GetWeakPtr() {
   return weak_factory_.GetWeakPtr();
 }
 
-// This is a dummy function to allow things to compile at each step of a
-// multi-CL sequence.
-void FidoTunnelDevice::OnTunnelReady(bool ok,
-                                     base::Optional<uint8_t> routing_id) {}
-
-void FidoTunnelDevice::OnTunnelReady_Future(
+void FidoTunnelDevice::OnTunnelReady(
     bool ok,
     base::Optional<std::array<uint8_t, kRoutingIdSize>> routing_id) {
   DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);

device/fido/cable/fido_tunnel_device.h

@@ -9,7 +9,7 @@
 
 #include "base/callback_forward.h"
 #include "base/sequence_checker.h"
-#include "device/fido/cable/v2_handshake.h"
+#include "device/fido/cable/v2_constants.h"
 #include "device/fido/cable/websocket_adapter.h"
 #include "device/fido/fido_device.h"
 #include "third_party/abseil-cpp/absl/types/variant.h"
@@ -95,10 +95,7 @@ class COMPONENT_EXPORT(DEVICE_FIDO) FidoTunnelDevice : public FidoDevice {
     base::Optional<std::vector<uint8_t>> handshake_message;
   };
 
-  // This is a dummy function to allow things to compile at each step of a
-  // multi-CL sequence.
-  void OnTunnelReady(bool ok, base::Optional<uint8_t> routing_id);
-  void OnTunnelReady_Future(
+  void OnTunnelReady(
       bool ok,
       base::Optional<std::array<uint8_t, kRoutingIdSize>> routing_id);
   void OnTunnelData(base::Optional<base::span<const uint8_t>> data);

device/fido/cable/v2_authenticator.cc

@@ -332,11 +332,7 @@ class TunnelTransport : public Transport {
     kReady,
   };
 
-  // This is a dummy function to allow things to compile at each step of a
-  // multi-CL sequence.
-  void OnTunnelReady(bool ok, base::Optional<uint8_t> routing_id) {}
-
-  void OnTunnelReady_Future(
+  void OnTunnelReady(
       bool ok,
       base::Optional<std::array<uint8_t, device::cablev2::kRoutingIdSize>>
           routing_id) {

device/fido/cable/websocket_adapter.cc

@@ -20,7 +20,9 @@ static constexpr size_t kMaxIncomingMessageSize = 1 << 20;
 WebSocketAdapter::WebSocketAdapter(TunnelReadyCallback on_tunnel_ready,
                                    TunnelDataCallback on_tunnel_data)
     : on_tunnel_ready_(std::move(on_tunnel_ready)),
-      on_tunnel_data_(std::move(on_tunnel_data)) {}
+      on_tunnel_data_(std::move(on_tunnel_data)),
+      read_pipe_watcher_(FROM_HERE, mojo::SimpleWatcher::ArmingPolicy::MANUAL) {
+}
 
 WebSocketAdapter::~WebSocketAdapter() = default;
 
@@ -68,61 +70,72 @@ void WebSocketAdapter::OnConnectionEstablished(
     return;
   }
 
-  base::Optional<uint8_t> shard_id;
+  base::Optional<std::array<uint8_t, kRoutingIdSize>> routing_id;
   for (const auto& header : response->headers) {
     if (base::EqualsCaseInsensitiveASCII(header->name.c_str(),
-                                         kCableShardIdHeader)) {
-      unsigned u;
-      if (!base::StringToUint(header->value, &u) || shard_id > 63) {
-        FIDO_LOG(ERROR) << "Invalid shard ID from tunnel server";
+                                         kCableRoutingIdHeader)) {
+      if (routing_id.has_value() ||
+          !base::HexStringToSpan(header->value, routing_id.emplace())) {
+        FIDO_LOG(ERROR) << "Invalid routing ID from tunnel server: "
+                        << header->value;
         return;
       }
-      shard_id = u;
-      break;
     }
   }
 
   socket_remote_.Bind(std::move(socket));
   read_pipe_ = std::move(readable);
+  read_pipe_watcher_.Watch(
+      read_pipe_.get(), MOJO_HANDLE_SIGNAL_READABLE,
+      MOJO_TRIGGER_CONDITION_SIGNALS_SATISFIED,
+      base::BindRepeating(&WebSocketAdapter::OnDataPipeReady,
+                          base::Unretained(this)));
   write_pipe_ = std::move(writable);
   client_receiver_.Bind(std::move(client_receiver));
+
   socket_remote_->StartReceiving();
 
-  std::move(on_tunnel_ready_).Run(true, shard_id);
+  std::move(on_tunnel_ready_).Run(true, routing_id);
 }
 
 void WebSocketAdapter::OnDataFrame(bool finish,
                                    network::mojom::WebSocketMessageType type,
                                    uint64_t data_len) {
   DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
+  DCHECK_EQ(pending_message_i_, pending_message_.size());
+  DCHECK(!pending_message_finished_);
+
+  if (data_len == 0) {
+    if (finish) {
+      FlushPendingMessage();
+    }
+    return;
+  }
 
   const size_t old_size = pending_message_.size();
   const size_t new_size = old_size + data_len;
   if (type != network::mojom::WebSocketMessageType::BINARY ||
       data_len > std::numeric_limits<uint32_t>::max() || new_size < old_size ||
       new_size > kMaxIncomingMessageSize) {
-    FIDO_LOG(ERROR) << "invalid WebSocket frame";
+    FIDO_LOG(ERROR) << "invalid WebSocket frame (type: "
+                    << static_cast<int>(type) << ", len: " << data_len << ")";
     Close();
     return;
   }
 
-  if (data_len > 0) {
-    pending_message_.resize(new_size);
-    uint32_t data_len_32 = data_len;
-    if (read_pipe_->ReadData(&pending_message_.data()[old_size], &data_len_32,
-                             MOJO_READ_DATA_FLAG_ALL_OR_NONE) !=
-        MOJO_RESULT_OK) {
-      FIDO_LOG(ERROR) << "reading WebSocket frame failed";
-      Close();
-      return;
-    }
-    DCHECK_EQ(static_cast<size_t>(data_len_32), data_len);
-  }
-
-  if (finish) {
-    on_tunnel_data_.Run(pending_message_);
-    pending_message_.resize(0);
-  }
+  // The network process sends the |OnDataFrame| message before writing to
+  // |read_pipe_|. Therefore we cannot depend on the message bytes being
+  // immediately available in |read_pipe_| without a race. Thus
+  // |read_pipe_watcher_| is used to wait for the data if needed.
+
+  pending_message_.resize(new_size);
+  pending_message_finished_ = finish;
+  // Suspend more |OnDataFrame| callbacks until frame's data has been read. The
+  // network service has successfully read |data_len| bytes before calling this
+  // function so there's no I/O errors to worry about while reading; we know
+  // that the bytes are coming.
+  client_receiver_.Pause();
+  OnDataPipeReady(MOJO_RESULT_OK, mojo::HandleSignalsState());
 }
 
 void WebSocketAdapter::OnDropChannel(bool was_clean,
@@ -137,6 +150,41 @@ void WebSocketAdapter::OnClosingHandshake() {
   DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
 }
 
+void WebSocketAdapter::OnDataPipeReady(MojoResult,
+                                       const mojo::HandleSignalsState&) {
+  const size_t todo = pending_message_.size() - pending_message_i_;
+  DCHECK_GT(todo, 0u);
+
+  // Truncation to 32-bits cannot overflow because |pending_message_.size()| is
+  // bound by |kMaxIncomingMessageSize| when it is resized in |OnDataFrame|.
+  uint32_t todo_32 = static_cast<uint32_t>(todo);
+  static_assert(
+      kMaxIncomingMessageSize <= std::numeric_limits<decltype(todo_32)>::max(),
+      "");
+  const MojoResult result =
+      read_pipe_->ReadData(&pending_message_.data()[pending_message_i_],
+                           &todo_32, MOJO_READ_DATA_FLAG_NONE);
+  if (result == MOJO_RESULT_OK) {
+    pending_message_i_ += todo_32;
+    DCHECK_LE(pending_message_i_, pending_message_.size());
+
+    if (pending_message_i_ < pending_message_.size()) {
+      read_pipe_watcher_.Arm();
+    } else {
+      client_receiver_.Resume();
+      if (pending_message_finished_) {
+        FlushPendingMessage();
+      }
+    }
+  } else if (result == MOJO_RESULT_SHOULD_WAIT) {
+    read_pipe_watcher_.Arm();
+  } else {
+    FIDO_LOG(ERROR) << "reading WebSocket frame failed: "
+                    << static_cast<int>(result);
+    Close();
+  }
+}
+
 void WebSocketAdapter::OnMojoPipeDisconnect() {
   DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
 
@@ -160,5 +208,14 @@ void WebSocketAdapter::Close() {
   on_tunnel_data_.Run(base::nullopt);
 }
 
+void WebSocketAdapter::FlushPendingMessage() {
+  std::vector<uint8_t> message;
+  message.swap(pending_message_);
+  pending_message_i_ = 0;
+  pending_message_finished_ = false;
+
+  on_tunnel_data_.Run(message);
+}
+
 }  // namespace cablev2
 }  // namespace device

device/fido/cable/websocket_adapter.h

@@ -12,6 +12,7 @@
 #include "base/containers/span.h"
 #include "base/optional.h"
 #include "base/sequence_checker.h"
+#include "device/fido/cable/v2_handshake.h"
 #include "services/network/public/mojom/network_context.mojom.h"
 
 namespace device {
@@ -24,14 +25,13 @@ class COMPONENT_EXPORT(DEVICE_FIDO) WebSocketAdapter
     : public network::mojom::WebSocketHandshakeClient,
       network::mojom::WebSocketClient {
  public:
-  using TunnelReadyCallback =
-      base::OnceCallback<void(bool, base::Optional<uint8_t>)>;
+  using TunnelReadyCallback = base::OnceCallback<
+      void(bool, base::Optional<std::array<uint8_t, kRoutingIdSize>>)>;
   using TunnelDataCallback =
       base::RepeatingCallback<void(base::Optional<base::span<const uint8_t>>)>;
   WebSocketAdapter(
       // on_tunnel_ready is called once with a boolean that indicates whether
-      // the WebSocket successfully connected and an optional shard ID taken
-      // from the X-caBLE-Shard header in the HTTP response, if any.
+      // the WebSocket successfully connected and an optional routing ID.
       TunnelReadyCallback on_tunnel_ready,
       // on_tunnel_ready is called repeatedly, after successful connection, with
       // the contents of WebSocket messages. Framing is preserved so a single
@@ -72,11 +72,22 @@ class COMPONENT_EXPORT(DEVICE_FIDO) WebSocketAdapter
 
  private:
   void OnMojoPipeDisconnect();
+  void OnDataPipeReady(MojoResult result,
+                       const mojo::HandleSignalsState& state);
   void Close();
+  void FlushPendingMessage();
 
   bool closed_ = false;
+
   // pending_message_ contains a partial message that is being reassembled.
   std::vector<uint8_t> pending_message_;
+  // pending_message_i_ contains the number of valid bytes of
+  // |pending_message_|.
+  size_t pending_message_i_ = 0;
+  // pending_message_finished_ is true if |pending_message_| is the full size of
+  // an application frame and thus should be passed up once filled with bytes.
+  bool pending_message_finished_ = false;
+
   TunnelReadyCallback on_tunnel_ready_;
   const TunnelDataCallback on_tunnel_data_;
   mojo::Receiver<network::mojom::WebSocketHandshakeClient> handshake_receiver_{
@@ -84,6 +95,7 @@ class COMPONENT_EXPORT(DEVICE_FIDO) WebSocketAdapter
   mojo::Receiver<network::mojom::WebSocketClient> client_receiver_{this};
   mojo::Remote<network::mojom::WebSocket> socket_remote_;
   mojo::ScopedDataPipeConsumerHandle read_pipe_;
+  mojo::SimpleWatcher read_pipe_watcher_;
   mojo::ScopedDataPipeProducerHandle write_pipe_;
   SEQUENCE_CHECKER(sequence_checker_);
 };