/* packet-cql.c

 * Routines for Apache Cassandra CQL dissection

 * Copyright 2015, Aaron Ten Clay <aarontc@aarontc.com>

 *

 * Wireshark - Network traffic analyzer

 * By Gerald Combs <gerald@wireshark.org>

 * Copyright 1998 Gerald Combs

 *

 * SPDX-License-Identifier: GPL-2.0-or-later

 */

/*

 * CQL V3 reference: https://github.com/apache/cassandra/blob/trunk/doc/native_protocol_v3.spec

 * CQL V4 reference: https://github.com/apache/cassandra/blob/trunk/doc/native_protocol_v4.spec

 */

#include "config.h"

#include <epan/conversation.h>

#include <epan/packet.h>

#include "packet-tcp.h"

#include <epan/wmem_scopes.h>

#include <epan/expert.h>

#include <epan/to_str.h>

#include <epan/addr_resolv.h>

#ifdef HAVE_LZ4

#include <lz4.h>

#endif

#ifdef HAVE_SNAPPY

#include <snappy-c.h>

#endif

#define CQL_DEFAULT_PORT 9042 /* Not IANA registered */

/* the code can reasonably attempt to decompress buffer up to 10MB */

#define MAX_UNCOMPRESSED_SIZE (10 * 1024 * 1024)

void proto_reg_handoff_cql(void);

void proto_register_cql(void);

static int proto_cql;

/* CQL header frame fields */

static int hf_cql_version;

static int hf_cql_protocol_version;

static int hf_cql_direction;

/* CQL header frame fields */

static int hf_cql_flags_bitmap;

static int hf_cql_flag_compression;

static int hf_cql_flag_tracing;

static int hf_cql_flag_reserved3;

static int hf_cql_flag_custom_payload;

static int hf_cql_flag_warning;

static int hf_cql_flag_reserved4;

static int hf_cql_stream;

static int hf_cql_opcode;

static int hf_cql_length;

/* CQL data types */

/*

static int hf_cql_int;

static int hf_cql_long;

static int hf_cql_uuid;

static int hf_cql_bytes;

static int hf_cql_inet;

*/

/* Batch flags */

static int hf_cql_batch_flag_serial_consistency;

static int hf_cql_batch_flag_default_timestamp;

static int hf_cql_batch_flag_with_name_for_values;

static int hf_cql_batch_flags_bitmap;

static int ett_cql_batch_flags_bitmap;

static int hf_cql_consistency;

static int hf_cql_string_length;

static int hf_cql_string_map_size;

static int hf_cql_string;

static int hf_cql_auth_token;

static int hf_cql_value_count;

static int hf_cql_short_bytes_length;

static int hf_cql_bytes_length;

static int hf_cql_bytes;

static int hf_cql_bigint;

static int hf_cql_scale;

static int hf_cql_boolean;

static int hf_cql_ascii;

static int hf_cql_double;

static int hf_cql_float;

static int hf_cql_custom;

static int hf_cql_null_value;

static int hf_cql_int;

static int hf_cql_uuid;

static int hf_cql_tracing_uuid;

static int hf_cql_port;

static int hf_cql_timeuuid;

static int hf_cql_varchar;

static int hf_cql_varint_count8;

static int hf_cql_varint_count16;

static int hf_cql_varint_count32;

static int hf_cql_varint_count64;

static int hf_cql_raw_compressed_bytes;

static int hf_cql_paging_state;

static int hf_cql_page_size;

static int hf_cql_timestamp;

static int hf_cql_query_id;

static int hf_cql_event_type;

static int hf_cql_event_schema_change_type;

static int hf_cql_event_schema_change_type_target;

static int hf_cql_event_schema_change_keyspace;

static int hf_cql_event_schema_change_object;

static int hf_cql_result_timestamp;

static int hf_cql_string_list_size;

static int hf_cql_batch_type;

static int hf_cql_batch_query_type;

static int hf_cql_batch_query_size;

static int hf_cql_error_code;

static int hf_cql_result_kind;

static int hf_cql_result_rows_data_type;

static int hf_cql_query_flags_bitmap;

static int hf_cql_query_flags_values;

static int hf_cql_query_flags_skip_metadata;

static int hf_cql_query_flags_page_size;

static int hf_cql_query_flags_paging_state;

static int hf_cql_query_flags_serial_consistency;

static int hf_cql_query_flags_default_timestamp;

static int hf_cql_query_flags_names_for_values;

static int hf_cql_query_flags_reserved3;

static int hf_cql_result_rows_flags_values;

static int hf_cql_result_prepared_flags_values;

static int hf_cql_result_rows_flag_global_tables_spec;

static int hf_cql_result_rows_flag_has_more_pages;

static int hf_cql_result_rows_flag_no_metadata;

static int hf_cql_result_rows_column_count;

static int hf_cql_result_rows_tuple_size;

static int hf_cql_result_prepared_pk_count;

static int hf_cql_string_result_rows_global_table_spec_ksname;

static int hf_cql_string_result_rows_global_table_spec_table_name;

static int hf_cql_string_result_rows_table_name;

static int hf_cql_string_result_rows_keyspace_name;

static int hf_cql_string_result_rows_column_name;

static int hf_cql_result_rows_row_count;

static int hf_cql_string_result_rows_udt_name;

static int hf_cql_string_result_rows_udt_size;

static int hf_cql_string_result_rows_udt_field_name;

static int hf_cql_string_result_rows_list_size;

static int hf_cql_string_result_rows_map_size;

static int hf_cql_string_result_rows_set_size;

static int hf_cql_bytesmap_string;

static int ett_cql_protocol;

static int ett_cql_version;

static int ett_cql_message;

static int ett_cql_result_columns;

static int ett_cql_results_no_metadata;

static int ett_cql_result_map;

static int ett_cql_result_set;

static int ett_cql_result_metadata;

static int ett_cql_result_rows;

static int ett_cql_result_metadata_colspec;

static int ett_cql_header_flags_bitmap;

static int ett_cql_query_flags_bitmap;

static int ett_cql_custom_payload;

static int hf_cql_response_in;

static int hf_cql_response_to;

static int hf_cql_response_time;

static int hf_cql_ipv4;

static int hf_cql_ipv6;

/* desegmentation of CQL */

static bool cql_desegment = true;

static expert_field ei_cql_data_not_dissected_yet;

static expert_field ei_cql_unexpected_negative_value;

typedef struct _cql_transaction_type {

  uint32_t req_frame;

  uint32_t rep_frame;

  nstime_t req_time;

} cql_transaction_type;

typedef struct _cql_conversation_info_type {

  wmem_map_t* streams;

} cql_conversation_type;

static const value_string cql_direction_names[] = {

  { 0x0, "Request" },

  { 0x8, "Response" },

  { 0x0, NULL }

};

typedef enum {

  CQL_BATCH_FLAG_SERIAL_CONSISTENCY =   0x10,

  CQL_BATCH_FLAG_DEFAULT_TIMESTAMP =    0x20,

  CQL_BATCH_FLAG_WITH_NAME_FOR_VALUES = 0x40

} cql_batch_flags;

typedef enum {

  CQL_OPCODE_ERROR = 0x00,

  CQL_OPCODE_STARTUP = 0x01,

  CQL_OPCODE_READY = 0x02,

  CQL_OPCODE_AUTHENTICATE = 0x03,

  /* Opcode 0x04 not used in CQL */

  CQL_OPCODE_OPTIONS = 0x05,

  CQL_OPCODE_SUPPORTED = 0x06,

  CQL_OPCODE_QUERY = 0x07,

  CQL_OPCODE_RESULT = 0x08,

  CQL_OPCODE_PREPARE = 0x09,

  CQL_OPCODE_EXECUTE = 0x0A,

  CQL_OPCODE_REGISTER = 0x0B,

  CQL_OPCODE_EVENT = 0x0C,

  CQL_OPCODE_BATCH = 0x0D,

  CQL_OPCODE_AUTH_CHALLENGE = 0x0E,

  CQL_OPCODE_AUTH_RESPONSE = 0x0F,

  CQL_OPCODE_AUTH_SUCCESS = 0x10

} cql_opcode_type;

static const value_string cql_opcode_names[] = {

  { CQL_OPCODE_ERROR, "ERROR" },

  { CQL_OPCODE_STARTUP, "STARTUP" },

  { CQL_OPCODE_READY, "READY" },

  { CQL_OPCODE_AUTHENTICATE, "AUTHENTICATE" },

  { CQL_OPCODE_OPTIONS, "OPTIONS" },

  { CQL_OPCODE_SUPPORTED, "SUPPORTED" },

  { CQL_OPCODE_QUERY, "QUERY" },

  { CQL_OPCODE_RESULT, "RESULT" },

  { CQL_OPCODE_PREPARE, "PREPARE" },

  { CQL_OPCODE_EXECUTE, "EXECUTE" },

  { CQL_OPCODE_REGISTER, "REGISTER" },

  { CQL_OPCODE_EVENT, "EVENT" },

  { CQL_OPCODE_BATCH, "BATCH" },

  { CQL_OPCODE_AUTH_CHALLENGE, "AUTH_CHALLENGE" },

  { CQL_OPCODE_AUTH_RESPONSE, "AUTH_RESPONSE" },

  { CQL_OPCODE_AUTH_SUCCESS, "AUTH_SUCCESS" },

  { 0x00, NULL }

};

typedef enum {

  CQL_HEADER_FLAG_COMPRESSION = 0x01,

  CQL_HEADER_FLAG_TRACING = 0x02,

  CQL_HEADER_FLAG_V3_RESERVED = 0xFC,

  CQL_HEADER_FLAG_CUSTOM_PAYLOAD = 0x04,

  CQL_HEADER_FLAG_WARNING = 0x08,

  CQL_HEADER_FLAG_V4_RESERVED = 0xF0

} cql_flags;

typedef enum {

  CQL_QUERY_FLAG_VALUES = 0x01,

  CQL_QUERY_FLAG_SKIP_METADATA = 0x02,

  CQL_QUERY_FLAG_PAGE_SIZE = 0x04,

  CQL_QUERY_FLAG_PAGING_STATE = 0x08,

  CQL_QUERY_FLAG_SERIAL_CONSISTENCY = 0x10,

  CQL_QUERY_FLAG_DEFAULT_TIMESTAMP = 0x20,

  CQL_QUERY_FLAG_VALUE_NAMES = 0x40,

  CQL_QUERY_FLAG_V3_RESERVED = 0x80

} cql_query_flags;

typedef enum {

  CQL_RESULT_ROWS_FLAG_GLOBAL_TABLES_SPEC = 0x00000001,

  CQL_RESULT_ROWS_FLAG_HAS_MORE_PAGES =     0x00000002,

  CQL_RESULT_ROWS_FLAG_NO_METADATA =        0x00000004

} cql_result_rows_flags;

typedef enum {

  CQL_CONSISTENCY_ANY = 0x0000,

  CQL_CONSISTENCY_ONE = 0x0001,

  CQL_CONSISTENCY_TWO = 0x0002,

  CQL_CONSISTENCY_THREE = 0x003,

  CQL_CONSISTENCY_QUORUM = 0x0004,

  CQL_CONSISTENCY_ALL = 0x0005,

  CQL_CONSISTENCY_LOCAL_QUORUM = 0x0006,

  CQL_CONSISTENCY_EACH_QUORUM = 0x0007,

  CQL_CONSISTENCY_SERIAL = 0x0008,

  CQL_CONSISTENCY_LOCAL_SERIAL = 0x0009,

  CQL_CONSISTENCY_LOCAL_ONE = 0x000A

} cql_consistencies;

static const value_string cql_consistency_names[] = {

  { CQL_CONSISTENCY_ANY, "ANY" },

  { CQL_CONSISTENCY_ONE, "ONE" },

  { CQL_CONSISTENCY_TWO, "TWO" },

  { CQL_CONSISTENCY_THREE, "THREE" },

  { CQL_CONSISTENCY_QUORUM, "QUORUM" },

  { CQL_CONSISTENCY_ALL, "ALL" },

  { CQL_CONSISTENCY_LOCAL_QUORUM, "LOCAL_QUORUM" },

  { CQL_CONSISTENCY_EACH_QUORUM, "EACH_QUORUM" },

  { CQL_CONSISTENCY_SERIAL, "SERIAL" },

  { CQL_CONSISTENCY_LOCAL_SERIAL, "LOCAL_SERIAL" },

  { CQL_CONSISTENCY_LOCAL_ONE, "LOCAL_ONE" },

  { 0x00, NULL }

};

typedef enum {

  CQL_BATCH_TYPE_LOGGED = 0,

  CQL_BATCH_TYPE_UNLOGGED = 1,

  CQL_BATCH_TYPE_COUNTER = 2

} cql_batch_types;

static const value_string cql_batch_type_names[] = {

  { CQL_BATCH_TYPE_LOGGED, "LOGGED" },

  { CQL_BATCH_TYPE_UNLOGGED, "UNLOGGED" },

  { CQL_BATCH_TYPE_COUNTER, "COUNTER" },

  { 0x00, NULL }

};

typedef enum {

  CQL_BATCH_QUERY_TYPE_QUERY = 0,

  CQL_BATCH_QUERY_TYPE_PREPARED = 1

} cql_batch_query_types;

static const value_string cql_batch_query_type_names[] = {

  { CQL_BATCH_QUERY_TYPE_QUERY, "QUERY" },

  { CQL_BATCH_QUERY_TYPE_PREPARED, "PREPARED" },

  { 0x00, NULL }

};

typedef enum {

  CQL_RESULT_KIND_VOID = 0x0001,

  CQL_RESULT_KIND_ROWS = 0x0002,

  CQL_RESULT_KIND_SET_KEYSPACE = 0x0003,

  CQL_RESULT_KIND_PREPARED = 0x0004,

  CQL_RESULT_KIND_SCHEMA_CHANGE = 0x0005

} cql_result_kinds;

static const value_string cql_result_kind_names[] = {

  { CQL_RESULT_KIND_VOID, "VOID" },

  { CQL_RESULT_KIND_ROWS, "Rows" },

  { CQL_RESULT_KIND_SET_KEYSPACE, "Set Keyspace" },

  { CQL_RESULT_KIND_PREPARED, "Prepared" },

  { CQL_RESULT_KIND_SCHEMA_CHANGE, "Schema Change" },

  { 0x00, NULL }

};

typedef enum {

  CQL_RESULT_ROW_TYPE_CUSTOM = 0x0000,

  CQL_RESULT_ROW_TYPE_ASCII = 0x0001,

  CQL_RESULT_ROW_TYPE_BIGINT = 0x0002,

  CQL_RESULT_ROW_TYPE_BLOB = 0x0003,

  CQL_RESULT_ROW_TYPE_BOOLEAN = 0x0004,

  CQL_RESULT_ROW_TYPE_COUNTER = 0x0005,

  CQL_RESULT_ROW_TYPE_DECIMAL = 0x0006,

  CQL_RESULT_ROW_TYPE_DOUBLE = 0x0007,

  CQL_RESULT_ROW_TYPE_FLOAT = 0x0008,

  CQL_RESULT_ROW_TYPE_INT = 0x0009,

  CQL_RESULT_ROW_TYPE_TIMESTAMP = 0x000B,

  CQL_RESULT_ROW_TYPE_UUID = 0x000C,

  CQL_RESULT_ROW_TYPE_VARCHAR = 0x000D,

  CQL_RESULT_ROW_TYPE_VARINT = 0x000E,

  CQL_RESULT_ROW_TYPE_TIMEUUID = 0x000F,

  CQL_RESULT_ROW_TYPE_INET = 0x0010,

  CQL_RESULT_ROW_TYPE_DATE = 0x0011,

  CQL_RESULT_ROW_TYPE_TIME = 0x0012,

  CQL_RESULT_ROW_TYPE_SMALLINT = 0x0013,

  CQL_RESULT_ROW_TYPE_TINYINT = 0x0014,

  CQL_RESULT_ROW_TYPE_LIST = 0x0020,

  CQL_RESULT_ROW_TYPE_MAP = 0x0021,

  CQL_RESULT_ROW_TYPE_SET = 0x0022,

  CQL_RESULT_ROW_TYPE_UDT = 0x0030,

  CQL_RESULT_ROW_TYPE_TUPLE = 0x0031

} cql_result_row_data_types;

static const value_string cql_result_row_type_names[] = {

  { CQL_RESULT_ROW_TYPE_CUSTOM, "CUSTOM" },

  { CQL_RESULT_ROW_TYPE_ASCII, "ASCII" },

  { CQL_RESULT_ROW_TYPE_BIGINT, "BIGINT" },

  { CQL_RESULT_ROW_TYPE_BLOB, "BLOB" },

  { CQL_RESULT_ROW_TYPE_BOOLEAN, "BOOLEAN" },

  { CQL_RESULT_ROW_TYPE_COUNTER, "COUNTER" },

  { CQL_RESULT_ROW_TYPE_DECIMAL, "DECIMAL" },

  { CQL_RESULT_ROW_TYPE_DOUBLE, "DOUBLE" },

  { CQL_RESULT_ROW_TYPE_FLOAT, "FLOAT" },

  { CQL_RESULT_ROW_TYPE_INT, "INT" },

  { CQL_RESULT_ROW_TYPE_TIMESTAMP, "TIMESTAMP" },

  { CQL_RESULT_ROW_TYPE_UUID, "UUID" },

  { CQL_RESULT_ROW_TYPE_VARCHAR, "VARCHAR" },

  { CQL_RESULT_ROW_TYPE_VARINT, "VARINT" },

  { CQL_RESULT_ROW_TYPE_TIMEUUID, "TIMEUUID" },

  { CQL_RESULT_ROW_TYPE_INET, "INET" },

  { CQL_RESULT_ROW_TYPE_DATE, "DATE" },

  { CQL_RESULT_ROW_TYPE_TIME, "TIME" },

  { CQL_RESULT_ROW_TYPE_SMALLINT, "SMALLINT" },

  { CQL_RESULT_ROW_TYPE_TINYINT, "TINYINT" },

  { CQL_RESULT_ROW_TYPE_LIST, "LIST" },

  { CQL_RESULT_ROW_TYPE_MAP, "MAP" },

  { CQL_RESULT_ROW_TYPE_SET, "SET" },

  { CQL_RESULT_ROW_TYPE_UDT, "UDT" },

  { CQL_RESULT_ROW_TYPE_TUPLE, "TUPLE" },

  { 0x0, NULL }

};

/* From https://github.com/apache/cassandra/blob/cbf4dcb3345c7e2f42f6a897c66b6460b7acc2ca/doc/native_protocol_v4.spec#L1046 */

typedef enum {

  CQL_ERROR_SERVER = 0x0000,

  CQL_ERROR_PROTOCOL = 0x000A,

  CQL_ERROR_AUTH = 0x0100,

  CQL_ERROR_UNAVAILABLE = 0x1000,

  CQL_ERROR_OVERLOADED = 0x1001,

  CQL_ERROR_BOOTSTRAPPING = 0x1002,

  CQL_ERROR_TRUNCATE = 0x1003,

  CQL_ERROR_WRITE_TIMEOUT = 0x1100,

  CQL_ERROR_READ_TIMEOUT = 0x1200,

  CQL_ERROR_READ_FAILURE = 0x1300,

  CQL_ERROR_FUNCTION_FAILURE = 0x1400,

  CQL_ERROR_WRITE_FAILURE = 0x1500,

  CQL_ERROR_SYNTAX = 0x2000,

  CQL_ERROR_UNAUTHORIEZED = 0x2100,

  CQL_ERROR_INVALID = 0x2200,

  CQL_ERROR_CONFIG = 0x2300,

  CQL_ERROR_ALREADY_EXISTS = 0x2400,

  CQL_ERROR_UNPREPARED = 0x2500

} cql_error_types;

static const value_string cql_error_names[] = {

  { CQL_ERROR_SERVER, "Server error" },

  { CQL_ERROR_PROTOCOL, "Protocol error" },

  { CQL_ERROR_AUTH, "Authentication error" },

  { CQL_ERROR_UNAVAILABLE, "Unavailable exception" },

  { CQL_ERROR_OVERLOADED, "Overloaded" },

  { CQL_ERROR_BOOTSTRAPPING, "Is_bootstrapping" },

  { CQL_ERROR_TRUNCATE, "Truncate_error" },

  { CQL_ERROR_WRITE_TIMEOUT, "Write_timeout" },

  { CQL_ERROR_READ_TIMEOUT, "Read_timeout" },

  { CQL_ERROR_READ_FAILURE, "Read_failure" },

  { CQL_ERROR_FUNCTION_FAILURE, "Function_failure" },

  { CQL_ERROR_WRITE_FAILURE, "Write_failure" },

  { CQL_ERROR_SYNTAX, "Syntax_error" },

  { CQL_ERROR_UNAUTHORIEZED, "Unauthorized" },

  { CQL_ERROR_INVALID, "Invalid" },

  {CQL_ERROR_CONFIG, "Config_error" },

  { CQL_ERROR_ALREADY_EXISTS, "Already_exists" },

  { CQL_ERROR_UNPREPARED, "Unprepared" },

  { 0x0, NULL}

};

static int

dissect_cql_query_parameters(proto_tree* cql_subtree, tvbuff_t* tvb, int offset, int execute)

{

  int32_t bytes_length = 0;

  uint32_t flags = 0;

  uint64_t i = 0;

  uint32_t string_length = 0;

  uint32_t value_count = 0;

  static int * const cql_query_bitmaps[] = {

    &hf_cql_query_flags_values,

    &hf_cql_query_flags_skip_metadata,

    &hf_cql_query_flags_page_size,

    &hf_cql_query_flags_paging_state,

    &hf_cql_query_flags_serial_consistency,

    &hf_cql_query_flags_default_timestamp,

    &hf_cql_query_flags_names_for_values,

    &hf_cql_query_flags_reserved3,

    NULL

  };

  /* consistency */

  proto_tree_add_item(cql_subtree, hf_cql_consistency, tvb, offset, 2, ENC_BIG_ENDIAN);

  offset += 2;

  /* flags */

  proto_tree_add_bitmask(cql_subtree, tvb, offset, hf_cql_query_flags_bitmap, ett_cql_query_flags_bitmap, cql_query_bitmaps, ENC_BIG_ENDIAN);

  flags = tvb_get_uint8(tvb, offset);

  offset += 1;

  if(flags & CQL_QUERY_FLAG_VALUES) {

    proto_tree_add_item_ret_uint(cql_subtree, hf_cql_value_count, tvb, offset, 2, ENC_BIG_ENDIAN, &value_count);

    offset += 2;

    for (i = 0; i < value_count; ++i) {

      if (!execute && flags & CQL_QUERY_FLAG_VALUE_NAMES) {

        proto_tree_add_item_ret_uint(cql_subtree, hf_cql_string_length, tvb, offset, 2, ENC_BIG_ENDIAN, &string_length);

        offset += 2;

        proto_tree_add_item(cql_subtree, hf_cql_string, tvb, offset, string_length, ENC_UTF_8);

        offset += string_length;

      }

      proto_tree_add_item_ret_int(cql_subtree, hf_cql_bytes_length, tvb, offset, 4, ENC_BIG_ENDIAN, &bytes_length);

      offset += 4;

      if (bytes_length > 0) {

        proto_tree_add_item(cql_subtree, hf_cql_bytes, tvb, offset, bytes_length, ENC_NA);

        offset += bytes_length;

      }

    }

  }

  if (flags & CQL_QUERY_FLAG_PAGE_SIZE) {

    proto_tree_add_item(cql_subtree, hf_cql_page_size, tvb, offset, 4, ENC_BIG_ENDIAN);

    offset += 4;

  }

  if (flags & CQL_QUERY_FLAG_PAGING_STATE) {

    proto_tree_add_item_ret_int(cql_subtree, hf_cql_bytes_length, tvb, offset, 4, ENC_BIG_ENDIAN, &bytes_length);

    offset += 4;

    if (bytes_length > 0) {

      proto_tree_add_item(cql_subtree, hf_cql_bytes, tvb, offset, bytes_length, ENC_NA);

      offset += bytes_length;

    }

  }

  if (flags & CQL_QUERY_FLAG_SERIAL_CONSISTENCY) {

    proto_tree_add_item(cql_subtree, hf_cql_consistency, tvb, offset, 2, ENC_BIG_ENDIAN);

    offset += 2;

  }

  if (flags & CQL_QUERY_FLAG_DEFAULT_TIMESTAMP) {

    proto_tree_add_item(cql_subtree, hf_cql_timestamp, tvb, offset, 8, ENC_TIME_USECS|ENC_BIG_ENDIAN);

    offset += 8;

  }

  return offset;

}

static unsigned

get_cql_pdu_len(packet_info* pinfo _U_, tvbuff_t* tvb, int offset, void* data _U_)

{

  /* CQL has 32-bit length at 5th byte in frame. */

  uint32_t length = tvb_get_ntohl(tvb, offset + 5);

  /* Include length of frame header. */

  return length + 9;

}

static cql_transaction_type*

cql_transaction_add_request(cql_conversation_type* conv,

        packet_info* pinfo,

        int32_t stream,

        int fake)

{

  cql_transaction_type* trans;

  wmem_list_t* list;

  list = (wmem_list_t*)wmem_map_lookup(conv->streams, GINT_TO_POINTER(stream));

  if(!list) {

    list = wmem_list_new(wmem_file_scope());

  } else {

    wmem_map_remove(conv->streams, GINT_TO_POINTER(stream));

  }

  trans = wmem_new(wmem_file_scope(), cql_transaction_type);

  if (fake) {

    trans->req_frame = 0;

  } else {

    trans->req_frame = pinfo->fd->num;

  }

  trans->rep_frame = 0;

  trans->req_time = pinfo->abs_ts;

  wmem_list_append(list, (void *)trans);

  wmem_map_insert(conv->streams, GINT_TO_POINTER(stream), (void*)list);

  return trans;

}

static cql_transaction_type*

cql_enrich_transaction_with_response(cql_conversation_type* conv,

          packet_info* pinfo,

          int32_t stream)

{

  cql_transaction_type* trans;

  wmem_list_frame_t* frame;

  wmem_list_t* list;

  list = (wmem_list_t*)wmem_map_lookup(conv->streams, GINT_TO_POINTER(stream));

  if (!list) {

    return NULL;

  }

  frame = wmem_list_tail(list);

  if (!frame) {

    return NULL;

  }

  trans = (cql_transaction_type *)wmem_list_frame_data(frame);

  if (!trans) {

    return NULL;

  }

  trans->rep_frame = pinfo->fd->num;

  return trans;

}

static cql_transaction_type*

cql_transaction_lookup(cql_conversation_type* conv,

      packet_info* pinfo,

      int32_t stream)

{

  wmem_list_frame_t* frame;

  wmem_list_t* list;

  list = (wmem_list_t*)wmem_map_lookup(conv->streams, GINT_TO_POINTER(stream));

  if (!list) {

    return NULL;

  }

  frame = wmem_list_head(list);

  if (!frame) {

    return NULL;

  }

  do {

    cql_transaction_type* trans = NULL;

    trans = (cql_transaction_type *)wmem_list_frame_data(frame);

    if (trans->req_frame == pinfo->fd->num || trans->rep_frame == pinfo->fd->num) {

      return trans;

    }

  } while ((frame = wmem_list_frame_next(frame)));

  return NULL;

}

typedef enum {

  CQL_COMPRESSION_NONE = 0,

  CQL_COMPRESSION_LZ4 = 1,

  CQL_COMPRESSION_SNAPPY = 2,

  CQL_DECOMPRESSION_ATTEMPTED = 3,

} cql_compression_level;

// NOLINTNEXTLINE(misc-no-recursion)

static int parse_option(proto_tree* metadata_subtree, packet_info *pinfo, tvbuff_t* tvb, int offset)

{

  uint32_t data_type = 0;

  uint32_t string_length = 0;

  uint32_t tuple_size = 0;

  uint32_t udt_size = 0;

  uint32_t i = 0;

  proto_tree_add_item_ret_uint(metadata_subtree, hf_cql_result_rows_data_type, tvb, offset, 2, ENC_BIG_ENDIAN, &data_type);

  offset += 2;

  increment_dissection_depth(pinfo);

  switch (data_type) {

    case CQL_RESULT_ROW_TYPE_LIST:

      offset = parse_option(metadata_subtree, pinfo, tvb, offset);

      break;

    case CQL_RESULT_ROW_TYPE_MAP:

      offset = parse_option(metadata_subtree, pinfo, tvb, offset);

      offset = parse_option(metadata_subtree, pinfo, tvb, offset);

      break;

    case CQL_RESULT_ROW_TYPE_SET:

      offset = parse_option(metadata_subtree, pinfo, tvb, offset);

      break;

    case CQL_RESULT_ROW_TYPE_UDT:

      /* keyspace */

      proto_tree_add_item_ret_uint(metadata_subtree, hf_cql_string_length, tvb, offset, 2, ENC_BIG_ENDIAN, &string_length);

      offset += 2;

      proto_tree_add_item(metadata_subtree, hf_cql_string_result_rows_keyspace_name, tvb, offset, string_length, ENC_UTF_8);

      offset += string_length;

      /* UDT name */

      proto_tree_add_item_ret_uint(metadata_subtree, hf_cql_string_length, tvb, offset, 2, ENC_BIG_ENDIAN, &string_length);

      offset += 2;

      proto_tree_add_item(metadata_subtree, hf_cql_string_result_rows_udt_name, tvb, offset, string_length, ENC_UTF_8);

      offset += string_length;

      /* UDT size */

      proto_tree_add_item_ret_uint(metadata_subtree, hf_cql_string_result_rows_udt_size, tvb, offset, 2, ENC_BIG_ENDIAN, &udt_size);

      offset += 2;

      for (i = 0; i < udt_size; i++) {

        /* UDT field name */

        proto_tree_add_item_ret_uint(metadata_subtree, hf_cql_string_length, tvb, offset, 2, ENC_BIG_ENDIAN, &string_length);

        offset += 2;

        proto_tree_add_item(metadata_subtree, hf_cql_string_result_rows_udt_field_name, tvb, offset, string_length, ENC_UTF_8);

        offset += string_length;

        /* UDT field option */

        offset = parse_option(metadata_subtree, pinfo, tvb, offset);

      }

      break;

    case CQL_RESULT_ROW_TYPE_TUPLE:

      proto_tree_add_item_ret_uint(metadata_subtree, hf_cql_result_rows_tuple_size, tvb, offset, 2, ENC_BIG_ENDIAN, &tuple_size);

      offset += 2;

      for (i = 0; i < tuple_size; i++) {

        offset = parse_option(metadata_subtree, pinfo, tvb, offset);

      }

      break;

    default:

      break;

  }

  decrement_dissection_depth(pinfo);

  return offset;

}

static void add_varint_item(proto_tree *tree, tvbuff_t *tvb, const int offset, int length)

{

  switch (length)

  {

  case 1:

    proto_tree_add_item(tree, hf_cql_varint_count8,  tvb, offset, 1, ENC_BIG_ENDIAN);

    break;

  case 2:

    proto_tree_add_item(tree, hf_cql_varint_count16, tvb, offset, 2, ENC_BIG_ENDIAN);

    break;

  case 3:

    proto_tree_add_item(tree, hf_cql_varint_count32, tvb, offset, 3, ENC_BIG_ENDIAN);

    break;

  case 4:

    proto_tree_add_item(tree, hf_cql_varint_count32, tvb, offset, 4, ENC_BIG_ENDIAN);

    break;

  case 5:

    proto_tree_add_item(tree, hf_cql_varint_count64, tvb, offset, 5, ENC_BIG_ENDIAN);

    break;

  case 6:

    proto_tree_add_item(tree, hf_cql_varint_count64, tvb, offset, 6, ENC_BIG_ENDIAN);

    break;

  case 7:

    proto_tree_add_item(tree, hf_cql_varint_count64, tvb, offset, 7, ENC_BIG_ENDIAN);

    break;

  case 8:

    proto_tree_add_item(tree, hf_cql_varint_count64, tvb, offset, 8, ENC_BIG_ENDIAN);

    break;

  }

}

static void add_cql_uuid(proto_tree* tree, int hf_uuid, tvbuff_t* tvb, int offset)

{

  e_guid_t guid;

  int i;

  guid.data1 = tvb_get_letohl(tvb, offset+12);

  guid.data2 = tvb_get_letohl(tvb, offset+10);

  guid.data3 = tvb_get_letohl(tvb, offset+8);

  for (i = 0; i < 8; i++)

  {

    guid.data4[i] = tvb_get_uint8(tvb, offset+(7-i));

  }

  proto_tree_add_guid(tree, hf_uuid, tvb, offset, 16, &guid);

}

// NOLINTNEXTLINE(misc-no-recursion)

static int parse_value(proto_tree* columns_subtree, packet_info *pinfo, tvbuff_t* tvb, int* offset_metadata, int offset)

{

  uint32_t data_type = 0;

  uint32_t string_length = 0;

  int32_t bytes_length = 0;

  uint32_t tuple_size = 0;

  int32_t list_size = 0;

  int32_t map_size = 0;

  int32_t set_size = 0;

  uint32_t udt_size = 0;

  proto_item *item;

  proto_item *sub_item;

  uint32_t i = 0;

  int32_t j = 0;

  int offset_metadata_backup = 0;

  uint32_t addr4;

  ws_in6_addr addr6;

  uint32_t port_number;

  proto_tree* map_subtree;

  proto_tree* set_subtree;

  proto_tree_add_item_ret_int(columns_subtree, hf_cql_bytes_length, tvb, offset, 4, ENC_BIG_ENDIAN, &bytes_length);

  offset += 4;

  item = proto_tree_add_item_ret_uint(columns_subtree, hf_cql_result_rows_data_type, tvb, *offset_metadata, 2, ENC_BIG_ENDIAN, &data_type);

  proto_item_set_hidden(item);

  *offset_metadata += 2;

  if (bytes_length == -1) { // value is NULL, but need to skip metadata offsets

    proto_tree_add_item(columns_subtree, hf_cql_null_value, tvb, offset, 0, ENC_NA);

    if (data_type == CQL_RESULT_ROW_TYPE_MAP) {

      *offset_metadata += 4; /* skip the type fields of *both* key and value in the map in the metadata */

    } else if (data_type == CQL_RESULT_ROW_TYPE_SET) {

      *offset_metadata += 2; /* skip the type field of the elements in the set in the metadata */

    }

    return offset;

  }

  increment_dissection_depth(pinfo);

  switch (data_type) {

    case CQL_RESULT_ROW_TYPE_CUSTOM:

      proto_tree_add_item(columns_subtree, hf_cql_custom, tvb, offset, bytes_length, ENC_UTF_8);

      offset += bytes_length;

      break;

    case CQL_RESULT_ROW_TYPE_ASCII:

      proto_tree_add_item(columns_subtree, hf_cql_ascii, tvb, offset, bytes_length, ENC_ASCII);

      offset += bytes_length;

      break;

    case CQL_RESULT_ROW_TYPE_BIGINT:

      proto_tree_add_item(columns_subtree, hf_cql_bigint, tvb, offset, 8, ENC_BIG_ENDIAN);

      offset += 8;

      break;

    case CQL_RESULT_ROW_TYPE_BLOB:

      proto_tree_add_item(columns_subtree, hf_cql_bytes, tvb, offset, bytes_length, ENC_NA);

      offset += bytes_length;

      break;

    case CQL_RESULT_ROW_TYPE_BOOLEAN:

      proto_tree_add_boolean(columns_subtree, hf_cql_boolean, tvb, offset, 1, true);

      offset += 1;

      break;

    case CQL_RESULT_ROW_TYPE_COUNTER:

      break;

    case CQL_RESULT_ROW_TYPE_DECIMAL:

      proto_tree_add_item(columns_subtree, hf_cql_scale, tvb, offset, 4, ENC_BIG_ENDIAN);

      offset += 4;

      add_varint_item(columns_subtree, tvb, offset, bytes_length - 4);

      offset += bytes_length - 4;

      break;

    case CQL_RESULT_ROW_TYPE_DOUBLE:

      proto_tree_add_item(columns_subtree, hf_cql_double, tvb, offset, 8, ENC_BIG_ENDIAN);

      offset += 8;

      break;

    case CQL_RESULT_ROW_TYPE_FLOAT:

      proto_tree_add_item(columns_subtree, hf_cql_float, tvb, offset, 4, ENC_BIG_ENDIAN);

      offset += 4;

      break;

    case CQL_RESULT_ROW_TYPE_INT:

      proto_tree_add_item(columns_subtree, hf_cql_int, tvb, offset, 4, ENC_BIG_ENDIAN);

      offset += 4;

      break;

    case CQL_RESULT_ROW_TYPE_TIMESTAMP:

      proto_tree_add_item(columns_subtree, hf_cql_result_timestamp, tvb, offset, 8, ENC_BIG_ENDIAN);

      offset += 8;

      break;

    case CQL_RESULT_ROW_TYPE_UUID:

      add_cql_uuid(columns_subtree, hf_cql_uuid, tvb, offset);

      offset += 16;

      break;

    case CQL_RESULT_ROW_TYPE_VARCHAR:

      proto_tree_add_item(columns_subtree, hf_cql_varchar, tvb, offset, bytes_length, ENC_ASCII);

      offset += bytes_length;

      break;

    case CQL_RESULT_ROW_TYPE_VARINT:

      add_varint_item(columns_subtree, tvb, offset, bytes_length);

      offset += bytes_length;

      break;

    case CQL_RESULT_ROW_TYPE_TIMEUUID:

      add_cql_uuid(columns_subtree, hf_cql_timeuuid, tvb, offset);

      offset += 16;

      break;

    case CQL_RESULT_ROW_TYPE_INET:

      switch (bytes_length) {

        case 4:

        case 8:

          addr4 = tvb_get_ipv4(tvb, offset);

          proto_tree_add_ipv4_format_value(columns_subtree, hf_cql_ipv4, tvb, offset, 4, addr4, "%s", get_hostname(addr4));

          offset += 4;

        break;

        case 16:

        case 20:

          tvb_get_ipv6(tvb, offset, &addr6);

          proto_tree_add_ipv6_format_value(columns_subtree, hf_cql_ipv6, tvb, offset, 16, &addr6, "%s", get_hostname6(&addr6));

          offset += 16;

          break;

        default:

          break;

      }

      /* port number is optional */

      if (bytes_length == 16 || bytes_length == 20) {

        sub_item = proto_tree_add_item_ret_uint(columns_subtree, hf_cql_port, tvb, offset, 4, ENC_BIG_ENDIAN, &port_number);

        proto_item_append_text(sub_item, " (%u)", port_number);

        offset += 4;

      }

      break;

    case CQL_RESULT_ROW_TYPE_DATE:

      break;

    case CQL_RESULT_ROW_TYPE_TIME:

      break;

    case CQL_RESULT_ROW_TYPE_SMALLINT:

      break;

    case CQL_RESULT_ROW_TYPE_TINYINT:

      break;

    case CQL_RESULT_ROW_TYPE_LIST:

      item = proto_tree_add_item_ret_int(columns_subtree, hf_cql_string_result_rows_list_size, tvb, offset, 4, ENC_BIG_ENDIAN, &list_size);

      if (list_size < 0) {

        expert_add_info(pinfo, item, &ei_cql_unexpected_negative_value);

        decrement_dissection_depth(pinfo);

        return tvb_reported_length(tvb);

      }

      offset += 4;

      offset_metadata_backup = *offset_metadata;

      for (j = 0; j < list_size; j++) {

        *offset_metadata = offset_metadata_backup;

        offset = parse_value(columns_subtree, pinfo, tvb, offset_metadata, offset);

      }

      break;

    case CQL_RESULT_ROW_TYPE_MAP:

      map_subtree = proto_tree_add_subtree(columns_subtree, tvb, offset, 0, ett_cql_result_map, NULL, "Map");

      item = proto_tree_add_item_ret_int(map_subtree, hf_cql_string_result_rows_map_size, tvb, offset, 4, ENC_BIG_ENDIAN, &map_size);

      offset += 4;

      proto_item_append_text(map_subtree, " with %" PRId32 " element(s)", map_size);

      if (map_size < 0) {

        expert_add_info(pinfo, item, &ei_cql_unexpected_negative_value);

        decrement_dissection_depth(pinfo);

        return tvb_reported_length(tvb);

      } else if (map_size == 0) {

        *offset_metadata += 4; /* skip the type fields of *both* key and value in the map in the metadata */

      } else {

        offset_metadata_backup = *offset_metadata;

        for (j = 0; j < map_size; j++) {

          *offset_metadata = offset_metadata_backup;

          offset = parse_value(map_subtree, pinfo, tvb, offset_metadata, offset);

          offset = parse_value(map_subtree, pinfo, tvb, offset_metadata, offset);

        }

      }

      break;

    case CQL_RESULT_ROW_TYPE_SET:

      set_subtree = proto_tree_add_subtree(columns_subtree, tvb, offset, 0, ett_cql_result_set, NULL, "Set");

      item = proto_tree_add_item_ret_int(set_subtree, hf_cql_string_result_rows_set_size, tvb, offset, 4, ENC_BIG_ENDIAN, &set_size);

      offset += 4;

      if (set_size < 0) {

        expert_add_info(pinfo, item, &ei_cql_unexpected_negative_value);

        decrement_dissection_depth(pinfo);

        return tvb_reported_length(tvb);

      } else if (set_size == 0) {

        *offset_metadata += 2; /* skip the type field of the elements in the set in the metadata */

      } else {

        offset_metadata_backup = *offset_metadata;

        for (j = 0; j < set_size; j++) {

          *offset_metadata = offset_metadata_backup;

          offset = parse_value(set_subtree, pinfo, tvb, offset_metadata, offset);

        }

      }

      break;

    case CQL_RESULT_ROW_TYPE_UDT:

      /* keyspace */

      item = proto_tree_add_item_ret_uint(columns_subtree, hf_cql_string_length, tvb, *offset_metadata, 2, ENC_BIG_ENDIAN, &string_length);

      proto_item_set_hidden(item);

      *offset_metadata += 2;

      item = proto_tree_add_item(columns_subtree, hf_cql_string_result_rows_keyspace_name, tvb, *offset_metadata, string_length, ENC_UTF_8);

      proto_item_set_hidden(item);

      *offset_metadata += string_length;

      /* UDT name */

      item = proto_tree_add_item_ret_uint(columns_subtree, hf_cql_string_length, tvb, *offset_metadata, 2, ENC_BIG_ENDIAN, &string_length);

      proto_item_set_hidden(item);

      *offset_metadata += 2;

      item = proto_tree_add_item(columns_subtree, hf_cql_string_result_rows_udt_name, tvb, *offset_metadata, string_length, ENC_UTF_8);

      proto_item_set_hidden(item);

      *offset_metadata += string_length;

      /* UDT size */

      item = proto_tree_add_item_ret_uint(columns_subtree, hf_cql_string_result_rows_udt_size, tvb, *offset_metadata, 2, ENC_BIG_ENDIAN, &udt_size);

      proto_item_set_hidden(item);

      *offset_metadata += 2;

      for (i = 0; i < udt_size; i++) {

        /* UDT field name */

        item = proto_tree_add_item_ret_uint(columns_subtree, hf_cql_string_length, tvb, *offset_metadata, 2, ENC_BIG_ENDIAN, &string_length);

        proto_item_set_hidden(item);

        *offset_metadata += 2;

        item = proto_tree_add_item(columns_subtree, hf_cql_string_result_rows_udt_field_name, tvb, *offset_metadata, string_length, ENC_UTF_8);

        proto_item_set_hidden(item);

        *offset_metadata += string_length;

        /* UDT field option */

        offset = parse_value(columns_subtree, pinfo, tvb, offset_metadata, offset);

      }

      break;

    case CQL_RESULT_ROW_TYPE_TUPLE:

      item = proto_tree_add_item_ret_uint(columns_subtree, hf_cql_result_rows_tuple_size, tvb, *offset_metadata, 2, ENC_BIG_ENDIAN, &tuple_size);

      proto_item_set_hidden(item);

      *offset_metadata += 2;

      for (i = 0; i < tuple_size; i++) {

        offset = parse_value(columns_subtree, pinfo, tvb, offset_metadata, offset);

      }

      break;

    default:

      break;

  }

  decrement_dissection_depth(pinfo);

  return offset;

}