tools/gnmsg/client_message_decoder.py - geode-native - Git at Google

 #!/usr/local/bin/python3

 # contributor license agreements.  See the NOTICE file distributed with
 # this work for additional information regarding copyright ownership.
 # The ASF licenses this file to You 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.
 import re
 import struct

 from dateutil import parser

 from client_messages import parse_client_message
 from decoder_base import DecoderBase
 from message_types import message_types
 from numeric_conversion import to_hex_digit


 class ClientMessageDecoder(DecoderBase):
     def __init__(self, output_queue):
         super(ClientMessageDecoder, self).__init__(output_queue)
         self.STATE_NEUTRAL_ = 0
         self.STATE_FOUND_SECURITY_FOOTER_ = 1
         self.send_trace_parts_retriever_ = None
         self.send_trace_parser_ = None
         self.connection_states_ = {}
         self.nc_version_ = None
         self.send_trace_parts_retriever_ = None
         self.get_send_trace_parts_functions = {
             "0.0.42": self.get_send_trace_parts_base,
             "10.0.3": self.get_send_trace_parts_base,
             "10.1.1": self.get_send_trace_parts_base,
             "10.1.2": self.get_send_trace_parts_base,
             "10.1.3": self.get_send_trace_parts_base,
             "9.1.1": self.get_send_trace_parts_v911,
         }
         self.send_trace_parsers = {
             "0.0.42": self.parse_request_fields_base,
             "10.0.3": self.parse_request_fields_base,
             "10.1.1": self.parse_request_fields_base,
             "10.1.2": self.parse_request_fields_base,
             "10.1.3": self.parse_request_fields_base,
             "9.1.1": self.parse_request_fields_v911,
         }
         #
         # Native client code believes this is the list of messages that require a security footer.
         # We will use this list to verify and report if a message is sent that needs one but doesn't
         # have it, since this has been the source of at least one difficult-to-diagnose bug in the
         # past.  To see the decision-making code that filters on this message list, look at
         # ThinClientBaseDM::beforeSendingRequest and TcrMessage::isUserInitiativeOps in geode-native
         # C++ code base.
         self.message_requires_security_part = [
             "ADD_PDX_ENUM",
             "ADD_PDX_TYPE",
             "CLIENT_READY",
             "CLOSE_CONNECTION",
             "COMMIT",
             "GETCQSTATS_MSG_TYPE",
             "GET_CLIENT_PARTITION_ATTRIBUTES",
             "GET_CLIENT_PR_METADATA",
             "GET_ENTRY",
             "GET_FUNCTION_ATTRIBUTES",
             "GET_PDX_ENUM_BY_ID",
             "GET_PDX_ID_FOR_ENUM",
             "GET_PDX_ID_FOR_TYPE",
             "GET_PDX_TYPE_BY_ID",
             "INVALID",
             "MAKE_PRIMARY",
             "MONITORCQ_MSG_TYPE",
             "PERIODIC_ACK",
             "PING",
             "REQUEST_EVENT_VALUE",
             "ROLLBACK"
             "SIZE",
             "TX_FAILOVER",
             "TX_SYNCHRONIZATION",
             "USER_CREDENTIAL_MESSAGE",
         ]
     def search_for_version(self, line):
         if self.nc_version_ == None:
             expression = re.compile(r"Product version:.*Native (\d+)\.(\d+)\.(\d+)-")
             match = expression.search(line)
             if match:
                 major = match.group(1)
                 minor = match.group(2)
                 patch = match.group(3)
                 self.nc_version_ = major + "." + minor + "." + patch
                 self.send_trace_parts_retriever_ = self.get_send_trace_parts_functions[
                     self.nc_version_
                 ]
                 self.send_trace_parser_ = self.send_trace_parsers[self.nc_version_]

     def get_send_trace_parts_v911(self, line, parts):
         result = False
         expression = re.compile(
             r"(\d\d:\d\d:\d\d\.\d+).*TcrConnection::send:\s*\[([\d|a-f|A-F|x|X]+).*sending request to endpoint.*bytes:\s*([\d| ]+)"
         )
         match = expression.search(line)
         if match:
             parts.append(dateutil.parser.parse(match.group(1)))
             parts.append(match.group(2))
             parts.append(match.group(3))
             result = True

         return result

     def get_send_trace_parts_base(self, line, parts):
         result = False
         expression = re.compile(
             r"(\d\d:\d\d:\d\d\.\d+).*TcrConnection::send:\s*\[([\d|a-f|A-F|x|X]+).*sending request to endpoint.*bytes:\s*([\d|a-f|A-F]+)"
         )
         match = expression.search(line)
         if match:
             parts.append(parser.parse(match.group(1)))
             parts.append(match.group(2))
             parts.append(match.group(3))
             result = True

         return result

     def get_send_trace_parts(self, line, parts):
         if self.send_trace_parts_retriever_ is not None:
             return self.send_trace_parts_retriever_(line, parts)

     def get_add_security_trace_parts(self, line, parts):
         result = False
         expression = re.compile(
             r"(\d\d:\d\d:\d\d\.\d+).*TcrMessage::addSecurityPart\s*\[(0x[\d|a-f|A-F]*).*length\s*=\s*(\d+)\s*,\s*encrypted\s+ID\s*=\s*([\d|a-f|A-F]+)"
         )
         match = expression.search(line)
         if match:
             parts.append(parser.parse(match.group(1)))
             parts.append(match.group(2))
             parts.append(match.group(3))
             parts.append(match.group(4))
             result = True

         return result

     def decimal_string_to_hex_string(self, byte):
         high_nibble = int(int(byte) / 16)
         low_nibble = int(byte) % 16
         return to_hex_digit[high_nibble] + to_hex_digit[low_nibble]

     def format_bytes_as_hex_v911(self, message_bytes):
         byte_list = message_bytes.split(" ")
         hex_string = ""
         for byte in byte_list:
             if byte:
                 hex_string += self.decimal_string_to_hex_string(byte)
         return hex_string

     def parse_request_fields_v911(self, message_bytes):
         hex_message_bytes = self.format_bytes_as_hex_v911(message_bytes)
         message_type = message_types[int(hex_message_bytes[0:8], 16)]
         message_length = int(hex_message_bytes[8:16], 16)
         message_number_of_parts = int(hex_message_bytes[16:24], 16)
         message_transaction_id = struct.unpack(
             ">i", bytes.fromhex(hex_message_bytes[24:32])
         )[0]
         message_security_flag = (int(hex_message_bytes[32:34], 16) & 0x02) >> 1
         return (
             message_type,
             message_length,
             message_number_of_parts,
             message_transaction_id,
             message_security_flag,
         )

     def parse_request_fields_base(self, message_bytes):
         message_type = message_types[int(message_bytes[0:8], 16)]
         message_length = int(message_bytes[8:16], 16)
         message_number_of_parts = int(message_bytes[16:24], 16)
         message_transaction_id = struct.unpack(
             ">i", bytes.fromhex(message_bytes[24:32])
         )[0]
         message_security_flag = (int(message_bytes[32:34], 16) & 0x02) >> 1
         return (
             message_type,
             message_length,
             message_number_of_parts,
             message_transaction_id,
             message_security_flag,
         )

     def parse_request_fields(self, message_bytes):
         if self.send_trace_parser_ is not None:
             return self.send_trace_parser_(message_bytes)

     def request_requires_security_footer(self, message_type):
         return message_type in self.message_requires_security_part

     def process_line(self, line):
         connection = None
         is_send_trace = False
         is_add_security_trace = False
         send_trace = {}

         self.search_for_version(line)

         parts = []
         if self.get_send_trace_parts(line, parts):
             send_trace["Timestamp"], send_trace["Connection"], message_bytes = parts
             is_send_trace = True
         elif self.get_add_security_trace_parts(line, parts):
             timestamp, connection, security_footer_length, message_bytes = parts
             is_add_security_trace = True
         else:
             return

         if connection not in self.connection_states_:
             self.connection_states_[connection] = self.STATE_NEUTRAL_

         if self.connection_states_[connection] == self.STATE_NEUTRAL_:
             if is_add_security_trace:
                 self.connection_states_[connection] = self.STATE_FOUND_SECURITY_FOOTER_
             elif is_send_trace:
                 send_trace["Direction"] = "--->"
                 (
                     send_trace["Type"],
                     send_trace["Length"],
                     send_trace["Parts"],
                     send_trace["TransactionId"],
                     send_trace["SecurityFlag"],
                 ) = self.parse_request_fields(message_bytes)
                 if (send_trace["SecurityFlag"] == 1) and (
                     self.request_requires_security_footer(str(send_trace["Type"]))
                 ):
                     print(
                         "ERROR: Security flag is set, but no footer was added for this message!"
                     )

                 parse_client_message(send_trace, message_bytes)
                 self.output_queue_.put({"message": send_trace})
         elif self.connection_states_[connection] == self.STATE_FOUND_SECURITY_FOOTER_:
             if is_send_trace:
                 send_trace["Direction"] = "--->"
                 (
                     send_trace["Type"],
                     send_trace["Length"],
                     send_trace["Parts"],
                     send_trace["TransactionId"],
                     send_trace["SecurityFlag"],
                 ) = self.parse_request_fields(message_bytes)
                 self.output_queue_.put({"message": send_trace})
	#!/usr/local/bin/python3

	# contributor license agreements. See the NOTICE file distributed with
	# this work for additional information regarding copyright ownership.
	# The ASF licenses this file to You 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.
	import re
	import struct

	from dateutil import parser

	from client_messages import parse_client_message
	from decoder_base import DecoderBase
	from message_types import message_types
	from numeric_conversion import to_hex_digit


	class ClientMessageDecoder(DecoderBase):
	def __init__(self, output_queue):
	super(ClientMessageDecoder, self).__init__(output_queue)
	self.STATE_NEUTRAL_ = 0
	self.STATE_FOUND_SECURITY_FOOTER_ = 1
	self.send_trace_parts_retriever_ = None
	self.send_trace_parser_ = None
	self.connection_states_ = {}
	self.nc_version_ = None
	self.send_trace_parts_retriever_ = None
	self.get_send_trace_parts_functions = {
	"0.0.42": self.get_send_trace_parts_base,
	"10.0.3": self.get_send_trace_parts_base,
	"10.1.1": self.get_send_trace_parts_base,
	"10.1.2": self.get_send_trace_parts_base,
	"10.1.3": self.get_send_trace_parts_base,
	"9.1.1": self.get_send_trace_parts_v911,
	}
	self.send_trace_parsers = {
	"0.0.42": self.parse_request_fields_base,
	"10.0.3": self.parse_request_fields_base,
	"10.1.1": self.parse_request_fields_base,
	"10.1.2": self.parse_request_fields_base,
	"10.1.3": self.parse_request_fields_base,
	"9.1.1": self.parse_request_fields_v911,
	}
	#
	# Native client code believes this is the list of messages that require a security footer.
	# We will use this list to verify and report if a message is sent that needs one but doesn't
	# have it, since this has been the source of at least one difficult-to-diagnose bug in the
	# past. To see the decision-making code that filters on this message list, look at
	# ThinClientBaseDM::beforeSendingRequest and TcrMessage::isUserInitiativeOps in geode-native
	# C++ code base.
	self.message_requires_security_part = [
	"ADD_PDX_ENUM",
	"ADD_PDX_TYPE",
	"CLIENT_READY",
	"CLOSE_CONNECTION",
	"COMMIT",
	"GETCQSTATS_MSG_TYPE",
	"GET_CLIENT_PARTITION_ATTRIBUTES",
	"GET_CLIENT_PR_METADATA",
	"GET_ENTRY",
	"GET_FUNCTION_ATTRIBUTES",
	"GET_PDX_ENUM_BY_ID",
	"GET_PDX_ID_FOR_ENUM",
	"GET_PDX_ID_FOR_TYPE",
	"GET_PDX_TYPE_BY_ID",
	"INVALID",
	"MAKE_PRIMARY",
	"MONITORCQ_MSG_TYPE",
	"PERIODIC_ACK",
	"PING",
	"REQUEST_EVENT_VALUE",
	"ROLLBACK"
	"SIZE",
	"TX_FAILOVER",
	"TX_SYNCHRONIZATION",
	"USER_CREDENTIAL_MESSAGE",
	]
	def search_for_version(self, line):
	if self.nc_version_ == None:
	expression = re.compile(r"Product version:.*Native (\d+)\.(\d+)\.(\d+)-")
	match = expression.search(line)
	if match:
	major = match.group(1)
	minor = match.group(2)
	patch = match.group(3)
	self.nc_version_ = major + "." + minor + "." + patch
	self.send_trace_parts_retriever_ = self.get_send_trace_parts_functions[
	self.nc_version_
	]
	self.send_trace_parser_ = self.send_trace_parsers[self.nc_version_]

	def get_send_trace_parts_v911(self, line, parts):
	result = False
	expression = re.compile(
	r"(\d\d:\d\d:\d\d\.\d+).TcrConnection::send:\s\[([\d\|a-f\|A-F\|x\|X]+).sending request to endpoint.bytes:\s*([\d\| ]+)"
	)
	match = expression.search(line)
	if match:
	parts.append(dateutil.parser.parse(match.group(1)))
	parts.append(match.group(2))
	parts.append(match.group(3))
	result = True

	return result

	def get_send_trace_parts_base(self, line, parts):
	result = False
	expression = re.compile(
	r"(\d\d:\d\d:\d\d\.\d+).TcrConnection::send:\s\[([\d\|a-f\|A-F\|x\|X]+).sending request to endpoint.bytes:\s*([\d\|a-f\|A-F]+)"
	)
	match = expression.search(line)
	if match:
	parts.append(parser.parse(match.group(1)))
	parts.append(match.group(2))
	parts.append(match.group(3))
	result = True

	return result

	def get_send_trace_parts(self, line, parts):
	if self.send_trace_parts_retriever_ is not None:
	return self.send_trace_parts_retriever_(line, parts)

	def get_add_security_trace_parts(self, line, parts):
	result = False
	expression = re.compile(
	r"(\d\d:\d\d:\d\d\.\d+).TcrMessage::addSecurityPart\s\[(0x[\d\|a-f\|A-F]).length\s=\s(\d+)\s,\sencrypted\s+ID\s=\s([\d\|a-f\|A-F]+)"
	)
	match = expression.search(line)
	if match:
	parts.append(parser.parse(match.group(1)))
	parts.append(match.group(2))
	parts.append(match.group(3))
	parts.append(match.group(4))
	result = True

	return result

	def decimal_string_to_hex_string(self, byte):
	high_nibble = int(int(byte) / 16)
	low_nibble = int(byte) % 16
	return to_hex_digit[high_nibble] + to_hex_digit[low_nibble]

	def format_bytes_as_hex_v911(self, message_bytes):
	byte_list = message_bytes.split(" ")
	hex_string = ""
	for byte in byte_list:
	if byte:
	hex_string += self.decimal_string_to_hex_string(byte)
	return hex_string

	def parse_request_fields_v911(self, message_bytes):
	hex_message_bytes = self.format_bytes_as_hex_v911(message_bytes)
	message_type = message_types[int(hex_message_bytes[0:8], 16)]
	message_length = int(hex_message_bytes[8:16], 16)
	message_number_of_parts = int(hex_message_bytes[16:24], 16)
	message_transaction_id = struct.unpack(
	">i", bytes.fromhex(hex_message_bytes[24:32])
	)[0]
	message_security_flag = (int(hex_message_bytes[32:34], 16) & 0x02) >> 1
	return (
	message_type,
	message_length,
	message_number_of_parts,
	message_transaction_id,
	message_security_flag,
	)

	def parse_request_fields_base(self, message_bytes):
	message_type = message_types[int(message_bytes[0:8], 16)]
	message_length = int(message_bytes[8:16], 16)
	message_number_of_parts = int(message_bytes[16:24], 16)
	message_transaction_id = struct.unpack(
	">i", bytes.fromhex(message_bytes[24:32])
	)[0]
	message_security_flag = (int(message_bytes[32:34], 16) & 0x02) >> 1
	return (
	message_type,
	message_length,
	message_number_of_parts,
	message_transaction_id,
	message_security_flag,
	)

	def parse_request_fields(self, message_bytes):
	if self.send_trace_parser_ is not None:
	return self.send_trace_parser_(message_bytes)

	def request_requires_security_footer(self, message_type):
	return message_type in self.message_requires_security_part

	def process_line(self, line):
	connection = None
	is_send_trace = False
	is_add_security_trace = False
	send_trace = {}

	self.search_for_version(line)

	parts = []
	if self.get_send_trace_parts(line, parts):
	send_trace["Timestamp"], send_trace["Connection"], message_bytes = parts
	is_send_trace = True
	elif self.get_add_security_trace_parts(line, parts):
	timestamp, connection, security_footer_length, message_bytes = parts
	is_add_security_trace = True
	else:
	return

	if connection not in self.connection_states_:
	self.connection_states_[connection] = self.STATE_NEUTRAL_

	if self.connection_states_[connection] == self.STATE_NEUTRAL_:
	if is_add_security_trace:
	self.connection_states_[connection] = self.STATE_FOUND_SECURITY_FOOTER_
	elif is_send_trace:
	send_trace["Direction"] = "--->"
	(
	send_trace["Type"],
	send_trace["Length"],
	send_trace["Parts"],
	send_trace["TransactionId"],
	send_trace["SecurityFlag"],
	) = self.parse_request_fields(message_bytes)
	if (send_trace["SecurityFlag"] == 1) and (
	self.request_requires_security_footer(str(send_trace["Type"]))
	):
	print(
	"ERROR: Security flag is set, but no footer was added for this message!"
	)

	parse_client_message(send_trace, message_bytes)
	self.output_queue_.put({"message": send_trace})
	elif self.connection_states_[connection] == self.STATE_FOUND_SECURITY_FOOTER_:
	if is_send_trace:
	send_trace["Direction"] = "--->"
	(
	send_trace["Type"],
	send_trace["Length"],
	send_trace["Parts"],
	send_trace["TransactionId"],
	send_trace["SecurityFlag"],
	) = self.parse_request_fields(message_bytes)
	self.output_queue_.put({"message": send_trace})