dissectors: add ScyllaDB dissector

This commit adds a basic dissector for ScyllaDB RPC protocol.
ScyllaDB (www.scylladb.com) is a No-SQL database serving multiple
client protocols (e.g. CQL). The newly introduced dissector
provides a way to inspect Scylla's internal protocol, used by
the nodes to communicate with each other - share data, gossip
the cluster state, update the schemas, etc.
This dissector implements only a shallow dissection of most packets,
i.e. recognizing the packet type. Two requests with deeper dissection
are MUTATION and READ_DATA, used by I/O operations in the database.

Bug: 16471
Change-Id: Ibba8262bd4e5a637b24b3e7846c42c6534ef811b
Signed-off-by: Piotr Sarna <sarna@scylladb.com>
Reviewed-on: https://code.wireshark.org/review/36633
Petri-Dish: Martin Mathieson <martin.r.mathieson@googlemail.com>
Tested-by: Petri Dish Buildbot
Reviewed-by: Martin Mathieson <martin.r.mathieson@googlemail.com>

2 files changed, 505 insertions, 0 deletions

diff --git a/epan/dissectors/CMakeLists.txt b/epan/dissectors/CMakeLists.txt
index b656aa57da..fb46521a39 100644
--- a/epan/dissectors/CMakeLists.txt
+++ b/epan/dissectors/CMakeLists.txt
@@ -1694,6 +1694,7 @@ set(DISSECTOR_SRC
 	${CMAKE_CURRENT_SOURCE_DIR}/packet-scsi.c
 	${CMAKE_CURRENT_SOURCE_DIR}/packet-scte35.c
 	${CMAKE_CURRENT_SOURCE_DIR}/packet-sctp.c
+	${CMAKE_CURRENT_SOURCE_DIR}/packet-scylla.c
 	${CMAKE_CURRENT_SOURCE_DIR}/packet-sdh.c
 	${CMAKE_CURRENT_SOURCE_DIR}/packet-sdlc.c
 	${CMAKE_CURRENT_SOURCE_DIR}/packet-sdp.c
diff --git a/epan/dissectors/packet-scylla.c b/epan/dissectors/packet-scylla.c
new file mode 100644
index 0000000000..71d9974fe1
--- /dev/null
+++ b/epan/dissectors/packet-scylla.c
@@ -0,0 +1,504 @@
+/* packet-scylla.c
+ * Routines for Scylla RPC dissection
+ * Copyright 2020 ScyllaDB, Piotr Sarna <sarna@scylladb.com>
+ *
+ * Wireshark - Network traffic analyzer
+ * By Gerald Combs <gerald@wireshark.org>
+ * Copyright 1998 Gerald Combs
+ *
+ * SPDX-License-Identifier: GPL-2.0-or-later
+ */
+
+/*
+ * ScyllaDB RPC protocol is used for inter-node communication
+ * in the ScyllaDB database - reading/sending data, exchanging
+ * cluster information through gossip, updating schemas, etc.
+ *
+ * Protocol references:
+ * https://github.com/scylladb/seastar/blob/master/doc/rpc.md
+ * https://github.com/scylladb/scylla/blob/master/message/messaging_service.hh
+ *
+ */
+
+#include <config.h>
+
+#include <epan/packet.h>
+#include <epan/prefs.h>
+#include <epan/dissectors/packet-tcp.h>
+
+void proto_reg_handoff_scylla(void);
+void proto_register_scylla(void);
+
+#define SCYLLA_PORT 0 /* Not IANA registered, 7000 is the expected value */
+
+#define SCYLLA_HEADER_SIZE 28
+#define SCYLLA_HEADER_VERB_OFFSET 8
+#define SCYLLA_HEADER_MSG_ID_OFFSET 16
+#define SCYLLA_HEADER_LEN_OFFSET 24
+
+#define SCYLLA_RESPONSE_SIZE 12
+#define SCYLLA_RESPONSE_MSG_ID_OFFSET 0
+#define SCYLLA_RESPONSE_LEN_OFFSET 8
+
+#define SCYLLA_NEGOTIATION_SIZE 12
+#define SCYLLA_NEGOTIATION_LEN_OFFSET 8
+
+static int proto_scylla = -1;
+
+static int hf_scylla_request = -1;
+static int hf_scylla_timeout = -1;
+static int hf_scylla_verb = -1;
+static int hf_scylla_msg_id = -1;
+static int hf_scylla_len = -1;
+static int hf_scylla_response = -1;
+static int hf_scylla_response_size = -1;
+static int hf_scylla_response_request_frame = -1;
+static int hf_scylla_negotiation_size = -1;
+static int hf_scylla_payload = -1; // TODO: dissect everything, so that generic "payload" is not needed
+
+// Mutation
+static int hf_scylla_mut_size1 = -1;
+static int hf_scylla_mut_size2 = -1;
+static int hf_scylla_mut_table_id = -1;
+static int hf_scylla_mut_schema_id = -1;
+static int hf_scylla_mut_len_pkeys = -1;
+static int hf_scylla_mut_num_pkeys = -1;
+static int hf_scylla_mut_len_pkey = -1;
+static int hf_scylla_mut_pkey = -1;
+
+// Read data
+static int hf_scylla_read_data_timeout = -1;
+static int hf_scylla_read_data_table_id = -1;
+static int hf_scylla_read_data_schema_version = -1;
+
+static gint ett_scylla = -1;
+static gint ett_scylla_header = -1;
+static gint ett_scylla_response = -1;
+static gint ett_scylla_negotiation = -1;
+static gint ett_scylla_mut = -1;
+static gint ett_scylla_mut_pkey = -1;
+static gint ett_scylla_read_data = -1;
+
+static gboolean scylla_desegment = TRUE;
+
+enum scylla_packets {
+    CLIENT_ID = 0,
+    MUTATION = 1,
+    MUTATION_DONE = 2,
+    READ_DATA = 3,
+    READ_MUTATION_DATA = 4,
+    READ_DIGEST = 5,
+    // Used by gossip
+    GOSSIP_DIGEST_SYN = 6,
+    GOSSIP_DIGEST_ACK = 7,
+    GOSSIP_DIGEST_ACK2 = 8,
+    GOSSIP_ECHO = 9,
+    GOSSIP_SHUTDOWN = 10,
+    // end of gossip verb
+    DEFINITIONS_UPDATE = 11,
+    TRUNCATE = 12,
+    REPLICATION_FINISHED = 13,
+    MIGRATION_REQUEST = 14,
+    // Used by streaming
+    PREPARE_MESSAGE = 15,
+    PREPARE_DONE_MESSAGE = 16,
+    STREAM_MUTATION = 17,
+    STREAM_MUTATION_DONE = 18,
+    COMPLETE_MESSAGE = 19,
+    // end of streaming verbs
+    REPAIR_CHECKSUM_RANGE = 20,
+    GET_SCHEMA_VERSION = 21,
+    SCHEMA_CHECK = 22,
+    COUNTER_MUTATION = 23,
+    MUTATION_FAILED = 24,
+    STREAM_MUTATION_FRAGMENTS = 25,
+    REPAIR_ROW_LEVEL_START = 26,
+    REPAIR_ROW_LEVEL_STOP = 27,
+    REPAIR_GET_FULL_ROW_HASHES = 28,
+    REPAIR_GET_COMBINED_ROW_HASH = 29,
+    REPAIR_GET_SYNC_BOUNDARY = 30,
+    REPAIR_GET_ROW_DIFF = 31,
+    REPAIR_PUT_ROW_DIFF = 32,
+    REPAIR_GET_ESTIMATED_PARTITIONS = 33,
+    REPAIR_SET_ESTIMATED_PARTITIONS = 34,
+    REPAIR_GET_DIFF_ALGORITHMS = 35,
+    REPAIR_GET_ROW_DIFF_WITH_RPC_STREAM = 36,
+    REPAIR_PUT_ROW_DIFF_WITH_RPC_STREAM = 37,
+    REPAIR_GET_FULL_ROW_HASHES_WITH_RPC_STREAM = 38,
+    PAXOS_PREPARE = 39,
+    PAXOS_ACCEPT = 40,
+    PAXOS_LEARN = 41,
+    HINT_MUTATION = 42,
+    LAST = 43,
+};
+
+static const val64_string packettypenames[] = {
+    {CLIENT_ID,                                  "CLIENT_ID"},
+    {MUTATION,                                   "MUTATION"},
+    {MUTATION_DONE,                              "MUTATION_DONE"},
+    {READ_DATA,                                  "READ_DATA"},
+    {READ_MUTATION_DATA,                         "READ_MUTATION_DATA"},
+    {READ_DIGEST,                                "READ_DIGEST"},
+    {GOSSIP_DIGEST_SYN,                          "GOSSIP_DIGEST_SYN"},
+    {GOSSIP_DIGEST_ACK,                          "GOSSIP_DIGEST_ACK"},
+    {GOSSIP_DIGEST_ACK2,                         "GOSSIP_DIGEST_ACK2"},
+    {GOSSIP_ECHO,                                "GOSSIP_ECHO"},
+    {GOSSIP_SHUTDOWN,                            "GOSSIP_SHUTDOWN"},
+    {DEFINITIONS_UPDATE,                         "DEFINITIONS_UPDATE"},
+    {TRUNCATE,                                   "TRUNCATE"},
+    {REPLICATION_FINISHED,                       "REPLICATION_FINISHED"},
+    {MIGRATION_REQUEST,                          "MIGRATION_REQUEST"},
+    {PREPARE_MESSAGE,                            "PREPARE_MESSAGE"},
+    {PREPARE_DONE_MESSAGE,                       "PREPARE_DONE_MESSAGE"},
+    {STREAM_MUTATION,                            "STREAM_MUTATION"},
+    {STREAM_MUTATION_DONE,                       "STREAM_MUTATION_DONE"},
+    {COMPLETE_MESSAGE,                           "COMPLETE_MESSAGE"},
+    {REPAIR_CHECKSUM_RANGE,                      "REPAIR_CHECKSUM_RANGE"},
+    {GET_SCHEMA_VERSION,                         "GET_SCHEMA_VERSION"},
+    {SCHEMA_CHECK,                               "SCHEMA_CHECK"},
+    {COUNTER_MUTATION,                           "COUNTER_MUTATION"},
+    {MUTATION_FAILED,                            "MUTATION_FAILED"},
+    {STREAM_MUTATION_FRAGMENTS,                  "STREAM_MUTATION_FRAGMENTS"},
+    {REPAIR_ROW_LEVEL_START,                     "REPAIR_ROW_LEVEL_START"},
+    {REPAIR_ROW_LEVEL_STOP,                      "REPAIR_ROW_LEVEL_STOP"},
+    {REPAIR_GET_FULL_ROW_HASHES,                 "REPAIR_GET_FULL_ROW_HASHES"},
+    {REPAIR_GET_COMBINED_ROW_HASH,               "REPAIR_GET_COMBINED_ROW_HASH"},
+    {REPAIR_GET_SYNC_BOUNDARY,                   "REPAIR_GET_SYNC_BOUNDARY"},
+    {REPAIR_GET_ROW_DIFF,                        "REPAIR_GET_ROW_DIFF"},
+    {REPAIR_PUT_ROW_DIFF,                        "REPAIR_PUT_ROW_DIFF"},
+    {REPAIR_GET_ESTIMATED_PARTITIONS,            "REPAIR_GET_ESTIMATED_PARTITIONS"},
+    {REPAIR_SET_ESTIMATED_PARTITIONS,            "REPAIR_SET_ESTIMATED_PARTITIONS"},
+    {REPAIR_GET_DIFF_ALGORITHMS,                 "REPAIR_GET_DIFF_ALGORITHMS"},
+    {REPAIR_GET_ROW_DIFF_WITH_RPC_STREAM,        "REPAIR_GET_ROW_DIFF_WITH_RPC_STREAM"},
+    {REPAIR_PUT_ROW_DIFF_WITH_RPC_STREAM,        "REPAIR_PUT_ROW_DIFF_WITH_RPC_STREAM"},
+    {REPAIR_GET_FULL_ROW_HASHES_WITH_RPC_STREAM, "REPAIR_GET_FULL_ROW_HASHES_WITH_RPC_STREAM"},
+    {PAXOS_PREPARE,                              "PAXOS_PREPARE"},
+    {PAXOS_ACCEPT,                               "PAXOS_ACCEPT"},
+    {PAXOS_LEARN,                                "PAXOS_LEARN"},
+    {HINT_MUTATION,                              "HINT_MUTATION"},
+    {0, NULL}
+};
+
+static gboolean
+looks_like_rpc_negotiation(tvbuff_t *tvb, const gint offset) {
+    return tvb_memeql(tvb, offset, (const guint8 *)"SSTARRPC", 8) == 0;
+}
+
+static gboolean
+looks_like_response(guint64 verb_type, guint32 len) {
+    return verb_type > LAST || len > 64*1024*1024;
+}
+
+typedef struct {
+    guint64 verb_type;
+    guint32 request_frame_num;
+    guint32 response_frame_num;
+} request_response_t;
+
+static guint
+get_scylla_pdu_len(packet_info *pinfo _U_, tvbuff_t *tvb, int offset, void *data _U_)
+{
+    guint64 verb_type = tvb_get_letoh64(tvb, offset + SCYLLA_HEADER_VERB_OFFSET);
+    guint32 plen = tvb_get_letohl(tvb, offset + SCYLLA_HEADER_LEN_OFFSET);
+    if (looks_like_rpc_negotiation(tvb, offset)) {
+        return tvb_get_letohl(tvb, offset + SCYLLA_NEGOTIATION_LEN_OFFSET) + SCYLLA_NEGOTIATION_SIZE;
+    } else if (looks_like_response(verb_type, plen)) {
+        return tvb_get_letohl(tvb, offset + SCYLLA_RESPONSE_LEN_OFFSET) + SCYLLA_RESPONSE_SIZE;
+    }
+    return plen + SCYLLA_HEADER_SIZE;
+}
+
+static int
+dissect_scylla_negotiation_pdu(tvbuff_t *tvb, packet_info *pinfo, proto_tree *scylla_tree)
+{
+    gint offset = 0;
+
+    proto_tree *scylla_negotiation_tree = proto_tree_add_subtree(scylla_tree, tvb, offset,
+            SCYLLA_NEGOTIATION_SIZE, ett_scylla_negotiation, NULL, "Protocol negotiation");
+    guint32 len = tvb_get_letohl(tvb, offset + SCYLLA_NEGOTIATION_LEN_OFFSET);
+    gint negotiation_offset = 8;
+    proto_tree_add_item(scylla_negotiation_tree, hf_scylla_negotiation_size, tvb, offset + negotiation_offset, 4, ENC_LITTLE_ENDIAN);
+    negotiation_offset += 4;
+    proto_tree_add_item(scylla_negotiation_tree, hf_scylla_payload, tvb, offset + negotiation_offset, len - negotiation_offset, ENC_NA);
+
+    col_set_str(pinfo->cinfo, COL_PROTOCOL, "Scylla");
+    col_set_str(pinfo->cinfo, COL_INFO, "Protocol negotiation");
+    return tvb_reported_length(tvb);
+}
+
+static int
+dissect_scylla_response_pdu(tvbuff_t *tvb, packet_info *pinfo, proto_tree *scylla_tree, request_response_t *req_resp)
+{
+    gint offset = 0;
+
+    /* Add response subtree */
+    proto_item *response_ti = proto_tree_add_string_format(scylla_tree, hf_scylla_response,
+                                                           tvb, offset, SCYLLA_RESPONSE_SIZE,
+                                                           "", "Response");
+    proto_tree *scylla_response_tree = proto_item_add_subtree(response_ti, ett_scylla_response);
+
+    gint resp_offset = 0;
+    guint32 len = tvb_get_letohl(tvb, offset + SCYLLA_RESPONSE_LEN_OFFSET);
+
+    guint64 msg_id;
+    proto_tree_add_item_ret_uint64(scylla_response_tree, hf_scylla_msg_id, tvb, offset + resp_offset, 8, ENC_LITTLE_ENDIAN, &msg_id);
+    resp_offset += 8;
+    proto_tree_add_item(scylla_response_tree, hf_scylla_response_size, tvb, offset + resp_offset, 4, ENC_LITTLE_ENDIAN);
+    resp_offset += 4;
+    proto_tree_add_item(scylla_response_tree, hf_scylla_payload, tvb, offset + resp_offset, len, ENC_NA);
+
+    col_set_str(pinfo->cinfo, COL_PROTOCOL, "Scylla");
+    if (req_resp) {
+        /* Fill in the response frame */
+        req_resp->response_frame_num = pinfo->num;
+
+        proto_item *verb_item = proto_tree_add_uint64(scylla_response_tree, hf_scylla_verb, tvb, offset + len, 8, req_resp->verb_type);
+        proto_item_set_generated(verb_item);
+        proto_item *req = proto_tree_add_uint(scylla_tree, hf_scylla_response_request_frame, tvb, 0, 0, req_resp->request_frame_num);
+        proto_item_set_generated(req);
+
+        proto_item_append_text(response_ti, " (msg_id=%" G_GINT64_MODIFIER "u, %s)",
+                               msg_id, val64_to_str(req_resp->verb_type, packettypenames, "Unknown (0x%02x)"));
+
+        col_clear(pinfo->cinfo, COL_INFO);
+        col_add_fstr(pinfo->cinfo, COL_INFO, "Response for %s",
+            val64_to_str(req_resp->verb_type, packettypenames, "Unknown (0x%02x)"));
+    } else {
+        col_set_str(pinfo->cinfo, COL_INFO, "Response for unknown packet");
+    }
+    return tvb_reported_length(tvb);
+}
+
+static int
+dissect_scylla_msg_pdu(tvbuff_t *tvb, packet_info *pinfo, proto_tree *scylla_tree, proto_item *ti, guint64 verb_type, guint32 len)
+{
+    gint offset = 0;
+
+    /* Add request subtree */
+    proto_item *request_ti = proto_tree_add_string_format(scylla_tree, hf_scylla_request,
+                                                          tvb, offset, SCYLLA_RESPONSE_SIZE,
+                                                          "", "Header for %s",
+                                                          val64_to_str(verb_type, packettypenames, "Unknown (0x%02x)"));
+    proto_tree *scylla_header_tree = proto_item_add_subtree(request_ti, ett_scylla_response);
+
+    proto_tree_add_item(scylla_header_tree, hf_scylla_timeout, tvb, offset, 8, ENC_LITTLE_ENDIAN);
+    offset += 8;
+    proto_item_append_text(ti, ", Type %s", val64_to_str(verb_type, packettypenames, "Unknown (0x%02x)"));
+    proto_tree_add_item(scylla_header_tree, hf_scylla_verb, tvb, offset, 8, ENC_LITTLE_ENDIAN);
+    offset += 8;
+    guint64 msg_id;
+    proto_tree_add_item_ret_uint64(scylla_header_tree, hf_scylla_msg_id, tvb, offset, 8, ENC_LITTLE_ENDIAN, &msg_id);
+    offset += 8;
+    proto_tree_add_item(scylla_header_tree, hf_scylla_len, tvb, offset, 4, ENC_LITTLE_ENDIAN);
+    offset += 4;
+
+    proto_item_append_text(request_ti, " (msg_id=%" G_GINT64_MODIFIER "u)", msg_id);
+
+    switch (verb_type) {
+    case MUTATION: {
+        guint32 mut_len = len - SCYLLA_HEADER_SIZE;
+        proto_tree* scylla_mut_tree = proto_tree_add_subtree(scylla_tree, tvb, offset,
+                mut_len, ett_scylla_mut, NULL, "Mutation");
+        gint mut_offset = 0;
+        guint32 len_keys;
+        guint32 num_keys;
+        proto_tree_add_item(scylla_mut_tree, hf_scylla_mut_size1, tvb, offset + mut_offset, 4, ENC_LITTLE_ENDIAN);
+        mut_offset += 4;
+        proto_tree_add_item(scylla_mut_tree, hf_scylla_mut_size2, tvb, offset + mut_offset, 4, ENC_LITTLE_ENDIAN);
+        mut_offset += 4;
+        proto_tree_add_item(scylla_mut_tree, hf_scylla_mut_table_id, tvb, offset + mut_offset, 16, ENC_NA);
+        mut_offset += 16;
+        proto_tree_add_item(scylla_mut_tree, hf_scylla_mut_schema_id, tvb, offset + mut_offset, 16, ENC_NA);
+        mut_offset += 16;
+        proto_tree_add_item_ret_uint(scylla_mut_tree, hf_scylla_mut_len_pkeys, tvb, offset + mut_offset, 4, ENC_LITTLE_ENDIAN, &len_keys);
+        mut_offset += 4;
+        proto_tree* scylla_mut_pkey_tree = proto_tree_add_subtree(scylla_mut_tree, tvb, offset,
+                mut_len - mut_offset, ett_scylla_mut_pkey, NULL, "Partition key");
+        proto_tree_add_item_ret_uint(scylla_mut_pkey_tree, hf_scylla_mut_num_pkeys, tvb, offset + mut_offset, 4, ENC_LITTLE_ENDIAN, &num_keys);
+        mut_offset += 4;
+        guint i;
+        for (i = 0; i < num_keys; ++i) {
+            guint32 len_pkey = tvb_get_letohl(tvb, offset + mut_offset);
+            proto_tree_add_item(scylla_mut_pkey_tree, hf_scylla_mut_len_pkey, tvb, offset + mut_offset, 4, ENC_LITTLE_ENDIAN);
+            mut_offset += 4;
+            proto_tree_add_item(scylla_mut_pkey_tree, hf_scylla_mut_pkey, tvb, offset + mut_offset, len_pkey, ENC_NA);
+            mut_offset += len_pkey;
+        }
+        // TODO: dissect further
+        proto_tree_add_item(scylla_mut_tree, hf_scylla_payload, tvb, offset + mut_offset, len - mut_offset, ENC_NA);
+        }
+        break;
+    case READ_DATA: {
+        guint32 rd_len = len - SCYLLA_HEADER_SIZE;
+        proto_tree* scylla_read_tree = proto_tree_add_subtree(scylla_tree, tvb, offset,
+                rd_len, ett_scylla_read_data, NULL, "Read data");
+        gint rd_offset = 0;
+
+        proto_tree_add_item(scylla_read_tree, hf_scylla_read_data_timeout, tvb, offset + rd_offset, 4, ENC_LITTLE_ENDIAN);
+        rd_offset += 4;
+        proto_tree_add_item(scylla_read_tree, hf_scylla_read_data_table_id, tvb, offset + rd_offset, 16, ENC_NA);
+        rd_offset += 16;
+        proto_tree_add_item(scylla_read_tree, hf_scylla_read_data_schema_version, tvb, offset + rd_offset, 16, ENC_NA);
+        rd_offset += 16;
+        //TODO: dissect further
+        proto_tree_add_item(scylla_read_tree, hf_scylla_payload, tvb, offset + rd_offset, len - rd_offset, ENC_NA);
+        }
+        break;
+    default:
+        // Generic payload. TODO: dissect
+        proto_tree_add_item(scylla_tree, hf_scylla_payload, tvb, offset, len, ENC_NA);
+        break;
+    }
+    col_set_str(pinfo->cinfo, COL_PROTOCOL, "Scylla");
+    col_clear(pinfo->cinfo, COL_INFO);
+    col_add_fstr(pinfo->cinfo, COL_INFO, "Request %s",
+             val64_to_str(verb_type, packettypenames, "Unknown (0x%02x)"));
+    return tvb_reported_length(tvb);
+}
+
+static gboolean
+response_expected(guint64 verb_type)
+{
+    switch (verb_type) {
+    case GOSSIP_DIGEST_SYN:
+    case GOSSIP_DIGEST_ACK:
+    case GOSSIP_DIGEST_ACK2:
+    case GOSSIP_SHUTDOWN:
+    case MUTATION:
+    case MUTATION_DONE:
+    case MUTATION_FAILED:
+    case HINT_MUTATION:
+    case PAXOS_LEARN:
+        return FALSE;
+    default:
+        return TRUE;
+    }
+}
+
+
+static int
+dissect_scylla_pdu(tvbuff_t *tvb, packet_info *pinfo, proto_tree *tree, void* data _U_)
+{
+    gint offset = 0;
+    conversation_t *conversation;
+    wmem_map_t *conv_map;
+
+    proto_item *ti = proto_tree_add_item(tree, proto_scylla, tvb, 0, -1, ENC_NA);
+    proto_tree *scylla_tree = proto_item_add_subtree(ti, ett_scylla);
+
+    guint64 verb_type = tvb_get_letoh64(tvb, offset + SCYLLA_HEADER_VERB_OFFSET);
+    guint32 len = tvb_get_letohl(tvb, offset + SCYLLA_HEADER_LEN_OFFSET);
+
+    conversation = find_or_create_conversation(pinfo);
+    conv_map  = (wmem_map_t *)conversation_get_proto_data(conversation, proto_scylla);
+    if (conv_map == NULL) {
+        conv_map = wmem_map_new(wmem_file_scope(), wmem_int64_hash, g_int64_equal);
+        conversation_add_proto_data(conversation, proto_scylla, conv_map);
+    }
+
+    if (looks_like_rpc_negotiation(tvb, offset)) {
+        return dissect_scylla_negotiation_pdu(tvb, pinfo, scylla_tree);
+    } else if (looks_like_response(verb_type, len)) {
+        void *req_resp;
+        guint64 msg_id;
+        msg_id = tvb_get_letoh64(tvb, offset + SCYLLA_RESPONSE_MSG_ID_OFFSET);
+        req_resp = wmem_map_lookup(conv_map, &msg_id);
+        return dissect_scylla_response_pdu(tvb, pinfo, scylla_tree, (request_response_t *)req_resp);
+    }
+
+    guint64 msg_id = tvb_get_letoh64(tvb, offset + SCYLLA_HEADER_MSG_ID_OFFSET);
+
+    if (!PINFO_FD_VISITED(pinfo) && response_expected(verb_type)) {
+        guint64 *key = (guint64 *)wmem_alloc(wmem_file_scope(), sizeof(guint64));
+        request_response_t *val = (request_response_t *)wmem_alloc(wmem_file_scope(), sizeof(request_response_t));
+        *key = msg_id;
+        val->verb_type = verb_type;
+        val->request_frame_num = pinfo->num;
+        wmem_map_insert(conv_map, key, val);
+    }
+
+    return dissect_scylla_msg_pdu(tvb, pinfo, scylla_tree, ti, verb_type, len);
+}
+
+static int
+dissect_scylla(tvbuff_t *tvb, packet_info *pinfo, proto_tree *tree, void* data)
+{
+    tcp_dissect_pdus(tvb, pinfo, tree, scylla_desegment, SCYLLA_HEADER_SIZE,
+        get_scylla_pdu_len, dissect_scylla_pdu, data);
+    return tvb_reported_length(tvb);
+}
+
+void
+proto_register_scylla(void)
+{
+    static hf_register_info hf[] = {
+        // RPC header
+        { &hf_scylla_request, { "request", "scylla.request", FT_STRING, BASE_NONE, NULL, 0x0, NULL, HFILL } },
+        { &hf_scylla_timeout, { "RPC timeout", "scylla.timeout", FT_UINT64, BASE_DEC, NULL, 0x0, NULL, HFILL } },
+        { &hf_scylla_verb, { "verb", "scylla.verb", FT_UINT64, BASE_DEC|BASE_VAL64_STRING, VALS64(packettypenames), 0x0, NULL, HFILL } },
+        { &hf_scylla_msg_id, { "msg id", "scylla.msg_id", FT_UINT64, BASE_DEC, NULL, 0x0, NULL, HFILL } },
+        { &hf_scylla_len, { "packet length", "scylla.len", FT_UINT32, BASE_DEC, NULL, 0x0, NULL, HFILL } },
+        { &hf_scylla_payload, { "payload", "scylla.payload", FT_BYTES, BASE_NONE, NULL, 0x0, NULL, HFILL } },
+        { &hf_scylla_response, { "response", "scylla.response", FT_STRING, BASE_NONE, NULL, 0x0, NULL, HFILL } },
+        { &hf_scylla_response_size, { "response size", "scylla.response.size", FT_UINT32, BASE_DEC, NULL, 0x0, NULL, HFILL } },
+        { &hf_scylla_response_request_frame, { "Request frame", "scylla.response.request", FT_FRAMENUM, BASE_NONE, FRAMENUM_TYPE(FT_FRAMENUM_REQUEST), 0x0, NULL, HFILL } },
+        { &hf_scylla_negotiation_size, { "negotiation size", "scylla.negotiation.size", FT_UINT32, BASE_DEC, NULL, 0x0, NULL, HFILL } },
+        // mutation verb
+        { &hf_scylla_mut_size1, { "mutation size 1", "scylla.mut.size1", FT_UINT32, BASE_DEC, NULL, 0x0, NULL, HFILL } },
+        { &hf_scylla_mut_size2, { "mutation size 2", "scylla.mut.size2", FT_UINT32, BASE_DEC, NULL, 0x0, NULL, HFILL } },
+        { &hf_scylla_mut_table_id, { "mutation table id", "scylla.mut.table_id", FT_BYTES, BASE_NONE, NULL, 0x0, NULL, HFILL } },
+        { &hf_scylla_mut_schema_id, { "mutation schema id", "scylla.mut.schema_id", FT_BYTES, BASE_NONE, NULL, 0x0, NULL, HFILL } },
+        { &hf_scylla_mut_len_pkeys, { "size of partition keys payload", "scylla.mut.len_pkeys", FT_UINT32, BASE_DEC, NULL, 0x0, NULL, HFILL } },
+        { &hf_scylla_mut_num_pkeys, { "number of partition keys", "scylla.mut.num_pkeys", FT_UINT32, BASE_DEC, NULL, 0x0, NULL, HFILL } },
+        { &hf_scylla_mut_len_pkey, { "length of a partition key", "scylla.mut.len_pkey", FT_UINT32, BASE_DEC, NULL, 0x0, NULL, HFILL } },
+        { &hf_scylla_mut_pkey, { "partition key", "scylla.mut.pkey", FT_BYTES, BASE_NONE, NULL, 0x0, NULL, HFILL } },
+        // read_data verb
+        { &hf_scylla_read_data_timeout, { "timeout", "scylla.read_data.timeout", FT_UINT32, BASE_DEC, NULL, 0x0, NULL, HFILL } },
+        { &hf_scylla_read_data_table_id, { "table ID", "scylla.read_data.table_id", FT_BYTES, BASE_NONE, NULL, 0x0, NULL, HFILL } },
+        { &hf_scylla_read_data_schema_version, { "Schema version", "scylla.read_data.schema_version", FT_BYTES, BASE_NONE, NULL, 0x0, NULL, HFILL } },
+
+    };
+
+    /* Setup protocol subtree array */
+    static gint *ett[] = {
+        &ett_scylla,
+        &ett_scylla_header,
+        &ett_scylla_response,
+        &ett_scylla_negotiation,
+        &ett_scylla_mut,
+        &ett_scylla_mut_pkey,
+        &ett_scylla_read_data,
+    };
+
+    proto_scylla = proto_register_protocol("Scylla RPC protocol", "Scylla", "scylla");
+    module_t* scylla_module = prefs_register_protocol(proto_scylla, NULL);
+    prefs_register_bool_preference(scylla_module, "desegment",
+        "Desegment all Scylla messages spanning multiple TCP segments",
+        "Whether Scylla dissector should desegment all messages spanning multiple TCP segments",
+        &scylla_desegment);
+
+    proto_register_field_array(proto_scylla, hf, array_length(hf));
+    proto_register_subtree_array(ett, array_length(ett));
+}
+
+void
+proto_reg_handoff_scylla(void)
+{
+    static dissector_handle_t scylla_handle;
+
+    scylla_handle = create_dissector_handle(dissect_scylla, proto_scylla);
+    dissector_add_uint_with_preference("tcp.port", SCYLLA_PORT, scylla_handle);
+}
+
+/*
+ * Editor modelines  -  https://www.wireshark.org/tools/modelines.html
+ *
+ * Local variables:
+ * c-basic-offset: 4
+ * tab-width: 8
+ * indent-tabs-mode: nil
+ * End:
+ *
+ * vi: set shiftwidth=4 tabstop=8 expandtab:
+ * :indentSize=4:tabSize=8:noTabs=true:
+ */