Reland binder transport fuzzers (#28258)

* Revert "Temporarily remove binder fuzzers it fails to build with old llvm (#27599)"

This reverts commit 6b922f871f.

* Migrate to protobuf based structured fuzzing

* Fix crash happening due to recent change

test/core/transport/binder/end2end/fuzzers/BUILD

@@ -0,0 +1,95 @@
+# Copyright 2021 gRPC authors.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+load("//bazel:grpc_build_system.bzl", "grpc_cc_library", "grpc_package", "grpc_proto_library")
+load("//test/core/util:grpc_fuzzer.bzl", "grpc_proto_fuzzer")
+
+grpc_package(
+    name = "test/core/transport/binder/end2end/fuzzers",
+    features = [
+        "layering_check",
+    ],
+)
+
+licenses(["notice"])
+
+# Protobuf messages for generating inputs. We manually define proto
+# library rule here so the same proto file can be shared between multiple
+# grpc_proto_fuzzer targets
+grpc_proto_library(
+    name = "binder_transport_fuzzer_proto",
+    srcs = ["binder_transport_fuzzer.proto"],
+)
+
+grpc_cc_library(
+    name = "fuzzer_utils",
+    srcs = ["fuzzer_utils.cc"],
+    language = "c++",
+    public_hdrs = ["fuzzer_utils.h"],
+    deps = [
+        "binder_transport_fuzzer_proto",
+        "//:gpr",
+        "//:gpr_base",
+        "//:grpc++",
+        "//:grpc++_base",
+        "//:grpc_base",
+        "//test/core/util:grpc_test_util",
+    ],
+)
+
+grpc_proto_fuzzer(
+    name = "binder_transport_client_fuzzer",
+    srcs = [
+        "client_fuzzer.cc",
+    ],
+    corpus = "binder_transport_client_fuzzer_corpus",
+    proto = "client.proto",
+    tags = [
+        "no_mac",
+        "no_windows",
+    ],
+    deps = [
+        "binder_transport_fuzzer_proto",
+        ":fuzzer_utils",
+        "//:gpr",
+        "//:gpr_base",
+        "//:grpc++",
+        "//:grpc++_base",
+        "//:grpc_base",
+        "//test/core/util:grpc_test_util",
+    ],
+)
+
+grpc_proto_fuzzer(
+    name = "binder_transport_server_fuzzer",
+    srcs = [
+        "server_fuzzer.cc",
+    ],
+    corpus = "binder_transport_server_fuzzer_corpus",
+    proto = "server.proto",
+    tags = [
+        "no_mac",
+        "no_windows",
+    ],
+    deps = [
+        "binder_transport_fuzzer_proto",
+        ":fuzzer_utils",
+        "//:gpr",
+        "//:gpr_base",
+        "//:grpc++",
+        "//:grpc++_base",
+        "//:grpc_base",
+        "//test/core/util:grpc_test_util",
+    ],
+)

test/core/transport/binder/end2end/fuzzers/binder_transport_fuzzer.proto

@@ -0,0 +1,81 @@
+// Copyright 2021 gRPC authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+syntax = "proto3";
+
+package binder_transport_fuzzer;
+
+message Binder {}
+
+message Value {
+  oneof data_type {
+    int32 i32 = 1;
+    int64 i64 = 2;
+    bytes byte_array = 3;
+    // Strings in Parcel could also contain non UTF-8 data so we use bytes
+    // to represent string here
+    bytes str = 4;
+    Binder binder = 5;
+  }
+}
+
+message Parcel {
+  repeated Value values = 1;
+
+  // Simulates the return value of AParcel_getDataSize
+  // (The value generated by protobuf libprotobuf-mutator might not always make sense
+  // but the transport implementation should handle that)
+  int32 data_size = 2;
+}
+
+enum TransactionCode {
+  INVALID = 0;
+  SETUP_TRANSPORT = 1;
+  SHUTDOWN_TRANSPORT = 2;
+  ACKNOWLEDGE_BYTES = 3;
+  PING = 4;
+  PING_RESPONSE = 5;
+}
+
+message Transaction {
+  TransactionCode code = 1;
+  int32 uid = 2;
+  Parcel parcel = 3;
+}
+
+// Special parcel that used for setting up transport.
+// TODO(mingcl): Consider also fuzzing the setup transport code path
+message SetupTransportParcel {
+  int32 version = 1;
+
+  // Simulates the return value of AParcel_getDataSize
+  // (The value generated by protobuf libprotobuf-mutator might not always make sense
+  // but the transport implementation should handle that)
+  int32 data_size = 2;
+}
+
+message SetupTransportTransaction {
+  int32 uid = 1;
+  SetupTransportParcel parcel = 2;
+}
+
+message IncomingParcels {
+  SetupTransportTransaction setup_transport_transaction = 1;
+  repeated Transaction transactions = 2;
+}
+
+message Input {
+  IncomingParcels incoming_parcels = 1;
+}
+

test/core/transport/binder/end2end/fuzzers/client.proto

@@ -0,0 +1,17 @@
+// Copyright 2021 gRPC authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+syntax = "proto3";
+
+package binder_transport_fuzzer;

test/core/transport/binder/end2end/fuzzers/client_fuzzer.cc

@@ -0,0 +1,155 @@
+// Copyright 2021 gRPC authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include <thread>
+#include <utility>
+
+#include "absl/memory/memory.h"
+
+#include <grpc/grpc.h>
+#include <grpcpp/security/binder_security_policy.h>
+
+#include "src/core/ext/transport/binder/transport/binder_transport.h"
+#include "src/core/lib/config/core_configuration.h"
+#include "src/core/lib/iomgr/executor.h"
+#include "src/core/lib/surface/channel.h"
+#include "src/libfuzzer/libfuzzer_macro.h"
+#include "test/core/transport/binder/end2end/fuzzers/binder_transport_fuzzer.pb.h"
+#include "test/core/transport/binder/end2end/fuzzers/fuzzer_utils.h"
+
+bool squelch = true;
+bool leak_check = true;
+
+static void* tag(intptr_t t) { return reinterpret_cast<void*>(t); }
+
+static void dont_log(gpr_log_func_args*) {}
+
+DEFINE_PROTO_FUZZER(const binder_transport_fuzzer::Input& input) {
+  grpc_test_only_set_slice_hash_seed(0);
+  if (squelch) gpr_set_log_function(dont_log);
+  grpc_init();
+  {
+    // Copied and modified from grpc/test/core/end2end/fuzzers/client_fuzzer.cc
+    grpc_core::ExecCtx exec_ctx;
+    grpc_core::Executor::SetThreadingAll(false);
+
+    grpc_completion_queue* cq = grpc_completion_queue_create_for_next(nullptr);
+    grpc_transport* client_transport = grpc_create_binder_transport_client(
+        absl::make_unique<grpc_binder::fuzzing::BinderForFuzzing>(
+            input.incoming_parcels()),
+        std::make_shared<
+            grpc::experimental::binder::UntrustedSecurityPolicy>());
+    grpc_arg authority_arg = grpc_channel_arg_string_create(
+        const_cast<char*>(GRPC_ARG_DEFAULT_AUTHORITY),
+        const_cast<char*>("test-authority"));
+    grpc_channel_args* args =
+        grpc_channel_args_copy_and_add(nullptr, &authority_arg, 1);
+    const grpc_channel_args* channel_args = grpc_core::CoreConfiguration::Get()
+                                                .channel_args_preconditioning()
+                                                .PreconditionChannelArgs(args);
+    grpc_channel* channel = grpc_channel_create("test-target", channel_args,
+                                                GRPC_CLIENT_DIRECT_CHANNEL,
+                                                client_transport, nullptr);
+    grpc_channel_args_destroy(channel_args);
+    grpc_channel_args_destroy(args);
+    grpc_slice host = grpc_slice_from_static_string("localhost");
+    grpc_call* call = grpc_channel_create_call(
+        channel, nullptr, 0, cq, grpc_slice_from_static_string("/foo"), &host,
+        gpr_inf_future(GPR_CLOCK_REALTIME), nullptr);
+    grpc_metadata_array initial_metadata_recv;
+    grpc_metadata_array_init(&initial_metadata_recv);
+    grpc_byte_buffer* response_payload_recv = nullptr;
+    grpc_metadata_array trailing_metadata_recv;
+    grpc_metadata_array_init(&trailing_metadata_recv);
+    grpc_status_code status;
+    grpc_slice details = grpc_empty_slice();
+
+    grpc_op ops[6];
+    memset(ops, 0, sizeof(ops));
+    grpc_op* op = ops;
+    op->op = GRPC_OP_SEND_INITIAL_METADATA;
+    op->data.send_initial_metadata.count = 0;
+    op->flags = 0;
+    op->reserved = nullptr;
+    op++;
+    op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
+    op->flags = 0;
+    op->reserved = nullptr;
+    op++;
+    op->op = GRPC_OP_RECV_INITIAL_METADATA;
+    op->data.recv_initial_metadata.recv_initial_metadata =
+        &initial_metadata_recv;
+    op->flags = 0;
+    op->reserved = nullptr;
+    op++;
+    op->op = GRPC_OP_RECV_MESSAGE;
+    op->data.recv_message.recv_message = &response_payload_recv;
+    op->flags = 0;
+    op->reserved = nullptr;
+    op++;
+    op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
+    op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
+    op->data.recv_status_on_client.status = &status;
+    op->data.recv_status_on_client.status_details = &details;
+    op->flags = 0;
+    op->reserved = nullptr;
+    op++;
+    grpc_call_error error = grpc_call_start_batch(
+        call, ops, static_cast<size_t>(op - ops), tag(1), nullptr);
+    int requested_calls = 1;
+    GPR_ASSERT(GRPC_CALL_OK == error);
+    grpc_event ev;
+    while (true) {
+      grpc_core::ExecCtx::Get()->Flush();
+      ev = grpc_completion_queue_next(cq, gpr_inf_past(GPR_CLOCK_REALTIME),
+                                      nullptr);
+      switch (ev.type) {
+        case GRPC_QUEUE_TIMEOUT:
+          goto done;
+        case GRPC_QUEUE_SHUTDOWN:
+          break;
+        case GRPC_OP_COMPLETE:
+          requested_calls--;
+          break;
+      }
+    }
+
+  done:
+    if (requested_calls) {
+      grpc_call_cancel(call, nullptr);
+    }
+    grpc_binder::fuzzing::JoinFuzzingThread();
+    for (int i = 0; i < requested_calls; i++) {
+      ev = grpc_completion_queue_next(cq, gpr_inf_past(GPR_CLOCK_REALTIME),
+                                      nullptr);
+      GPR_ASSERT(ev.type == GRPC_OP_COMPLETE);
+    }
+    grpc_completion_queue_shutdown(cq);
+    for (int i = 0; i < requested_calls; i++) {
+      ev = grpc_completion_queue_next(cq, gpr_inf_past(GPR_CLOCK_REALTIME),
+                                      nullptr);
+      GPR_ASSERT(ev.type == GRPC_QUEUE_SHUTDOWN);
+    }
+    grpc_call_unref(call);
+    grpc_completion_queue_destroy(cq);
+    grpc_metadata_array_destroy(&initial_metadata_recv);
+    grpc_metadata_array_destroy(&trailing_metadata_recv);
+    grpc_slice_unref(details);
+    grpc_channel_destroy(channel);
+    if (response_payload_recv != nullptr) {
+      grpc_byte_buffer_destroy(response_payload_recv);
+    }
+  }
+  grpc_shutdown();
+}

test/core/transport/binder/end2end/fuzzers/fuzzer_utils.cc

@@ -0,0 +1,154 @@
+// Copyright 2021 gRPC authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include "test/core/transport/binder/end2end/fuzzers/fuzzer_utils.h"
+
+namespace grpc_binder {
+namespace fuzzing {
+
+namespace {
+
+std::thread* g_fuzzing_thread = nullptr;
+
+template <typename... Args>
+void CreateFuzzingThread(Args&&... args) {
+  GPR_ASSERT(g_fuzzing_thread == nullptr);
+  g_fuzzing_thread = new std::thread(std::forward<Args>(args)...);
+}
+
+}  // namespace
+
+void JoinFuzzingThread() {
+  if (g_fuzzing_thread) {
+    g_fuzzing_thread->join();
+    delete g_fuzzing_thread;
+    g_fuzzing_thread = nullptr;
+  }
+}
+
+int32_t ReadableParcelForFuzzing::GetDataSize() const {
+  return parcel_data_size_;
+}
+
+absl::Status ReadableParcelForFuzzing::ReadInt32(int32_t* data) {
+  if (consumed_data_size_ >= kParcelDataSizeLimit) {
+    return absl::InternalError("Parcel size limit exceeds");
+  }
+  if (values_.empty() || !values_.front().has_i32()) {
+    return absl::InternalError("error");
+  }
+  *data = values_.front().i32();
+  values_.pop();
+  consumed_data_size_ += sizeof(int32_t);
+  return absl::OkStatus();
+}
+
+absl::Status ReadableParcelForFuzzing::ReadInt64(int64_t* data) {
+  if (consumed_data_size_ >= kParcelDataSizeLimit) {
+    return absl::InternalError("Parcel size limit exceeds");
+  }
+  if (values_.empty() || !values_.front().has_i64()) {
+    return absl::InternalError("error");
+  }
+  *data = values_.front().i64();
+  values_.pop();
+  consumed_data_size_ += sizeof(int64_t);
+  return absl::OkStatus();
+}
+
+absl::Status ReadableParcelForFuzzing::ReadBinder(
+    std::unique_ptr<Binder>* binder) {
+  if (consumed_data_size_ >= kParcelDataSizeLimit) {
+    return absl::InternalError("Parcel size limit exceeds");
+  }
+  if (values_.empty() || !values_.front().has_binder()) {
+    return absl::InternalError("error");
+  }
+  *binder = absl::make_unique<BinderForFuzzing>();
+  values_.pop();
+  consumed_data_size_ += sizeof(void*);
+  return absl::OkStatus();
+}
+
+absl::Status ReadableParcelForFuzzing::ReadByteArray(std::string* data) {
+  if (consumed_data_size_ >= kParcelDataSizeLimit) {
+    return absl::InternalError("Parcel size limit exceeds");
+  }
+  if (values_.empty() || !values_.front().has_byte_array()) {
+    return absl::InternalError("error");
+  }
+  *data = values_.front().byte_array();
+  values_.pop();
+  consumed_data_size_ += data->size();
+  return absl::OkStatus();
+}
+
+absl::Status ReadableParcelForFuzzing::ReadString(std::string* data) {
+  if (consumed_data_size_ >= kParcelDataSizeLimit) {
+    return absl::InternalError("Parcel size limit exceeds");
+  }
+  if (values_.empty() || !values_.front().has_str()) {
+    return absl::InternalError("error");
+  }
+  *data = values_.front().str();
+  values_.pop();
+  consumed_data_size_ += data->size();
+  return absl::OkStatus();
+}
+
+void FuzzingLoop(
+    binder_transport_fuzzer::IncomingParcels incoming_parcels,
+    grpc_core::RefCountedPtr<grpc_binder::WireReader> wire_reader_ref,
+    grpc_binder::TransactionReceiver::OnTransactCb callback) {
+  {
+    // Send SETUP_TRANSPORT request.
+    std::unique_ptr<grpc_binder::ReadableParcel> parcel =
+        absl::make_unique<ReadableParcelForFuzzing>(
+            incoming_parcels.setup_transport_transaction().parcel());
+    callback(static_cast<transaction_code_t>(
+                 grpc_binder::BinderTransportTxCode::SETUP_TRANSPORT),
+             parcel.get(),
+             /*uid=*/incoming_parcels.setup_transport_transaction().uid())
+        .IgnoreError();
+  }
+  for (const auto& tx_iter : incoming_parcels.transactions()) {
+    transaction_code_t tx_code = tx_iter.code();
+    std::unique_ptr<grpc_binder::ReadableParcel> parcel =
+        absl::make_unique<ReadableParcelForFuzzing>(tx_iter.parcel());
+    callback(tx_code, parcel.get(),
+             /*uid=*/tx_iter.uid())
+        .IgnoreError();
+  }
+  wire_reader_ref = nullptr;
+}
+
+TranasctionReceiverForFuzzing::TranasctionReceiverForFuzzing(
+    binder_transport_fuzzer::IncomingParcels incoming_parcels,
+    grpc_core::RefCountedPtr<WireReader> wire_reader_ref,
+    TransactionReceiver::OnTransactCb cb) {
+  gpr_log(GPR_INFO, "Construct TranasctionReceiverForFuzzing");
+  CreateFuzzingThread(FuzzingLoop, std::move(incoming_parcels),
+                      std::move(wire_reader_ref), std::move(cb));
+}
+
+std::unique_ptr<TransactionReceiver> BinderForFuzzing::ConstructTxReceiver(
+    grpc_core::RefCountedPtr<WireReader> wire_reader_ref,
+    TransactionReceiver::OnTransactCb cb) const {
+  auto tx_receiver = absl::make_unique<TranasctionReceiverForFuzzing>(
+      incoming_parcels_, wire_reader_ref, cb);
+  return tx_receiver;
+}
+
+}  // namespace fuzzing
+}  // namespace grpc_binder

test/core/transport/binder/end2end/fuzzers/fuzzer_utils.h

@@ -0,0 +1,154 @@
+// Copyright 2021 gRPC authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#ifndef GRPC_TEST_CORE_TRANSPORT_BINDER_END2END_FUZZERS_FUZZER_UTILS_H
+#define GRPC_TEST_CORE_TRANSPORT_BINDER_END2END_FUZZERS_FUZZER_UTILS_H
+
+#include <memory>
+#include <queue>
+#include <string>
+#include <thread>
+#include <vector>
+
+#include "absl/memory/memory.h"
+#include "absl/status/status.h"
+
+#include <grpc/support/log.h>
+
+#include "src/core/ext/transport/binder/wire_format/binder.h"
+#include "src/core/ext/transport/binder/wire_format/wire_reader.h"
+#include "test/core/transport/binder/end2end/fuzzers/binder_transport_fuzzer.pb.h"
+
+namespace grpc_binder {
+namespace fuzzing {
+
+// A WritableParcel implementation that simply does nothing. Don't use
+// MockWritableParcel here since capturing calls is expensive.
+class NoOpWritableParcel : public WritableParcel {
+ public:
+  int32_t GetDataSize() const override { return 0; }
+  absl::Status WriteInt32(int32_t /*data*/) override {
+    return absl::OkStatus();
+  }
+  absl::Status WriteInt64(int64_t /*data*/) override {
+    return absl::OkStatus();
+  }
+  absl::Status WriteBinder(HasRawBinder* /*binder*/) override {
+    return absl::OkStatus();
+  }
+  absl::Status WriteString(absl::string_view /*s*/) override {
+    return absl::OkStatus();
+  }
+  absl::Status WriteByteArray(const int8_t* /*buffer*/,
+                              int32_t /*length*/) override {
+    return absl::OkStatus();
+  }
+};
+
+// Binder implementation used in fuzzing.
+//
+// Most of its the functionalities are no-op, except ConstructTxReceiver now
+// returns a TranasctionReceiverForFuzzing.
+class BinderForFuzzing : public Binder {
+ public:
+  BinderForFuzzing() : input_(absl::make_unique<NoOpWritableParcel>()) {}
+
+  explicit BinderForFuzzing(const binder_transport_fuzzer::IncomingParcels& p)
+      : incoming_parcels_(p), input_(absl::make_unique<NoOpWritableParcel>()) {}
+
+  void Initialize() override {}
+  absl::Status PrepareTransaction() override { return absl::OkStatus(); }
+
+  absl::Status Transact(BinderTransportTxCode /*tx_code*/) override {
+    return absl::OkStatus();
+  }
+
+  std::unique_ptr<TransactionReceiver> ConstructTxReceiver(
+      grpc_core::RefCountedPtr<WireReader> wire_reader_ref,
+      TransactionReceiver::OnTransactCb cb) const override;
+
+  WritableParcel* GetWritableParcel() const override { return input_.get(); }
+  void* GetRawBinder() override { return nullptr; }
+
+ private:
+  binder_transport_fuzzer::IncomingParcels incoming_parcels_;
+  std::unique_ptr<WritableParcel> input_;
+};
+
+// ReadableParcel implementation used in fuzzing.
+//
+// It consumes a Parcel generated by mutator, and returns the data in the Parcel
+// upon user's requests.
+class ReadableParcelForFuzzing : public ReadableParcel {
+ public:
+  explicit ReadableParcelForFuzzing(const binder_transport_fuzzer::Parcel& p)
+      : parcel_data_size_(p.data_size()), consumed_data_size_(0) {
+    for (const auto& v : p.values()) {
+      values_.push(v);
+    }
+  }
+
+  // Construct from SetupTransportParcel, which have fixed types of data in it.
+  explicit ReadableParcelForFuzzing(
+      const binder_transport_fuzzer::SetupTransportParcel& p)
+      : parcel_data_size_(p.data_size()), consumed_data_size_(0) {
+    // Creates value for protocol version and put it into the queue
+    binder_transport_fuzzer::Value version_value;
+    version_value.set_i32(p.version());
+    values_.push(version_value);
+
+    // Creates a binder value and put it into the queue
+    binder_transport_fuzzer::Value binder_value;
+    binder_value.mutable_binder();  // sets one-of field
+    values_.push(binder_value);
+  }
+
+  int32_t GetDataSize() const override;
+  absl::Status ReadInt32(int32_t* data) override;
+  absl::Status ReadInt64(int64_t* data) override;
+  absl::Status ReadBinder(std::unique_ptr<Binder>* binder) override;
+  absl::Status ReadByteArray(std::string* data) override;
+  absl::Status ReadString(std::string* data) override;
+
+ private:
+  // Stores data/objects in binder in their order. Since we don't support random
+  // access using a std::queue is enough here.
+  std::queue<binder_transport_fuzzer::Value> values_;
+
+  const int32_t parcel_data_size_;
+
+  static constexpr size_t kParcelDataSizeLimit = 1024 * 1024;
+  size_t consumed_data_size_;
+};
+
+void JoinFuzzingThread();
+
+// TransactionReceiver implementation used in fuzzing.
+//
+// When constructed, start sending fuzzed requests to the client. When all the
+// bytes are consumed, the reference to WireReader will be released.
+class TranasctionReceiverForFuzzing : public TransactionReceiver {
+ public:
+  TranasctionReceiverForFuzzing(
+      binder_transport_fuzzer::IncomingParcels incoming_parcels,
+      grpc_core::RefCountedPtr<WireReader> wire_reader_ref,
+      TransactionReceiver::OnTransactCb cb);
+
+  void* GetRawBinder() override { return nullptr; }
+};
+
+}  // namespace fuzzing
+}  // namespace grpc_binder
+
+#endif  // GRPC_TEST_CORE_TRANSPORT_BINDER_END2END_FUZZERS_FUZZER_UTILS_H

test/core/transport/binder/end2end/fuzzers/server.proto

@@ -0,0 +1,17 @@
+// Copyright 2021 gRPC authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+syntax = "proto3";
+
+package binder_transport_fuzzer;

test/core/transport/binder/end2end/fuzzers/server_fuzzer.cc

@@ -0,0 +1,131 @@
+// Copyright 2021 gRPC authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include <grpc/grpc.h>
+
+#include "src/core/ext/transport/binder/transport/binder_transport.h"
+#include "src/core/lib/config/core_configuration.h"
+#include "src/core/lib/iomgr/executor.h"
+#include "src/core/lib/slice/slice_internal.h"
+#include "src/core/lib/surface/server.h"
+#include "src/libfuzzer/libfuzzer_macro.h"
+#include "test/core/transport/binder/end2end/fuzzers/binder_transport_fuzzer.pb.h"
+#include "test/core/transport/binder/end2end/fuzzers/fuzzer_utils.h"
+
+bool squelch = true;
+bool leak_check = true;
+
+static void* tag(intptr_t t) { return reinterpret_cast<void*>(t); }
+
+static void dont_log(gpr_log_func_args* /*args*/) {}
+
+DEFINE_PROTO_FUZZER(const binder_transport_fuzzer::Input& input) {
+  grpc_test_only_set_slice_hash_seed(0);
+  if (squelch) gpr_set_log_function(dont_log);
+  grpc_init();
+  {
+    // Copied and modified from grpc/test/core/end2end/fuzzers/server_fuzzer.cc
+    grpc_core::ExecCtx exec_ctx;
+    grpc_core::Executor::SetThreadingAll(false);
+
+    grpc_server* server = grpc_server_create(nullptr, nullptr);
+    grpc_completion_queue* cq = grpc_completion_queue_create_for_next(nullptr);
+    grpc_server_register_completion_queue(server, cq, nullptr);
+    // TODO(ctiller): add more registered methods (one for POST, one for PUT)
+    grpc_server_register_method(server, "/reg", nullptr, {}, 0);
+    grpc_server_start(server);
+    grpc_transport* server_transport = grpc_create_binder_transport_server(
+        absl::make_unique<grpc_binder::fuzzing::BinderForFuzzing>(
+            input.incoming_parcels()),
+        std::make_shared<
+            grpc::experimental::binder::UntrustedSecurityPolicy>());
+    const grpc_channel_args* channel_args =
+        grpc_core::CoreConfiguration::Get()
+            .channel_args_preconditioning()
+            .PreconditionChannelArgs(nullptr);
+    grpc_core::Server::FromC(server)->SetupTransport(server_transport, nullptr,
+                                                     channel_args, nullptr);
+    grpc_channel_args_destroy(channel_args);
+    grpc_call* call1 = nullptr;
+    grpc_call_details call_details1;
+    grpc_metadata_array request_metadata1;
+    grpc_call_details_init(&call_details1);
+    grpc_metadata_array_init(&request_metadata1);
+    int requested_calls = 0;
+
+    GPR_ASSERT(GRPC_CALL_OK ==
+               grpc_server_request_call(server, &call1, &call_details1,
+                                        &request_metadata1, cq, cq, tag(1)));
+    requested_calls++;
+
+    grpc_event ev;
+    while (true) {
+      grpc_core::ExecCtx::Get()->Flush();
+      ev = grpc_completion_queue_next(cq, gpr_inf_past(GPR_CLOCK_REALTIME),
+                                      nullptr);
+      switch (ev.type) {
+        case GRPC_QUEUE_TIMEOUT:
+          goto done;
+        case GRPC_QUEUE_SHUTDOWN:
+          break;
+        case GRPC_OP_COMPLETE:
+          if (ev.tag == tag(1)) {
+            requested_calls--;
+            // TODO(ctiller): keep reading that call!
+          }
+          break;
+      }
+    }
+
+  done:
+    grpc_binder::fuzzing::JoinFuzzingThread();
+    if (call1 != nullptr) grpc_call_unref(call1);
+    grpc_call_details_destroy(&call_details1);
+    grpc_metadata_array_destroy(&request_metadata1);
+    grpc_server_shutdown_and_notify(server, cq, tag(0xdead));
+    grpc_server_cancel_all_calls(server);
+    grpc_millis deadline = grpc_core::ExecCtx::Get()->Now() + 5000;
+    for (int i = 0; i <= requested_calls; i++) {
+      // A single grpc_completion_queue_next might not be sufficient for getting
+      // the tag from shutdown, because we might potentially get blocked by
+      // an operation happening on the timer thread.
+      // For example, the deadline timer might expire, leading to the timer
+      // thread trying to cancel the RPC and thereby acquiring a few references
+      // to the call. This will prevent the shutdown to complete till the timer
+      // thread releases those references.
+      // As a solution, we are going to keep performing a cq_next for a
+      // liberal period of 5 seconds for the timer thread to complete its work.
+      do {
+        ev = grpc_completion_queue_next(cq, gpr_inf_past(GPR_CLOCK_REALTIME),
+                                        nullptr);
+        grpc_core::ExecCtx::Get()->InvalidateNow();
+      } while (ev.type != GRPC_OP_COMPLETE &&
+               grpc_core::ExecCtx::Get()->Now() < deadline);
+      GPR_ASSERT(ev.type == GRPC_OP_COMPLETE);
+    }
+    grpc_completion_queue_shutdown(cq);
+    for (int i = 0; i <= requested_calls; i++) {
+      do {
+        ev = grpc_completion_queue_next(cq, gpr_inf_past(GPR_CLOCK_REALTIME),
+                                        nullptr);
+        grpc_core::ExecCtx::Get()->InvalidateNow();
+      } while (ev.type != GRPC_QUEUE_SHUTDOWN &&
+               grpc_core::ExecCtx::Get()->Now() < deadline);
+      GPR_ASSERT(ev.type == GRPC_QUEUE_SHUTDOWN);
+    }
+    grpc_server_destroy(server);
+    grpc_completion_queue_destroy(cq);
+  }
+  grpc_shutdown();
+}