MySQL analyzer

scripts/base/init-default.bro

@@ -46,6 +46,7 @@
 @load base/protocols/http
 @load base/protocols/irc
 @load base/protocols/modbus
+@load base/protocols/mysql
 @load base/protocols/pop3
 @load base/protocols/radius
 @load base/protocols/snmp

scripts/base/protocols/mysql/__load__.bro

@@ -0,0 +1 @@
+@load ./main

scripts/base/protocols/mysql/main.bro

@@ -0,0 +1,109 @@
+##! Implements base functionality for MySQL analysis. Generates the mysql.log file.
+
+module MySQL;
+
+export {
+	redef enum Log::ID += { mysql::LOG };
+
+	type Info: record {
+		## Timestamp for when the event happened.
+		ts:     time    &log;
+		## Unique ID for the connection.
+		uid:    string  &log;
+		## The connection's 4-tuple of endpoint addresses/ports.
+		id:     conn_id &log;
+		## The command that was issued
+		cmd:	string	&log;
+		## The argument issued to the command
+		arg:	string	&log;
+		## The result (error, OK, etc.) from the server
+		result: string &log &optional;
+		## Server message, if any
+		response: string &log &optional;
+	};
+
+	## Event that can be handled to access the MySQL record as it is sent on
+	## to the logging framework.
+	global log_mysql: event(rec: Info);
+}
+
+redef record connection += {
+	mysql: Info &optional;
+};
+
+const ports = { 1434/tcp, 3306/tcp };
+
+const commands: table[count] of string = {
+	[0] = "sleep",
+	[1] = "quit",
+	[2] = "init_db",
+	[3] = "query",
+	[4] = "field_list",
+};
+
+event bro_init() &priority=5
+	{
+	Log::create_stream(mysql::LOG, [$columns=Info, $ev=log_mysql]);
+	Analyzer::register_for_ports(Analyzer::ANALYZER_MYSQL, ports);
+	}
+
+event mysql_handshake_response(c: connection, username: string)
+	{
+	if ( !c?$mysql )
+		{
+		local info: Info;
+		info$ts = network_time();
+		info$uid = c$uid;
+		info$id = c$id;
+		info$cmd = "login";
+		info$arg = username;
+		c$mysql = info;
+		}
+	}
+
+event mysql_command_request(c: connection, command: count, arg: string)
+	{
+	if ( !c?$mysql )
+		{
+		local info: Info;
+		info$ts = network_time();
+		info$uid = c$uid;
+		info$id = c$id;
+		info$cmd = commands[command];
+		info$arg = sub(arg, /\0$/, "");
+		c$mysql = info;
+		}
+	}
+
+event mysql_command_response(c: connection, response: count)
+	{
+	if ( c?$mysql )
+		{
+		c$mysql$result = "ok";
+		c$mysql$response = fmt("Affected rows: %d", response);
+		Log::write(mysql::LOG, c$mysql);
+		delete c$mysql;
+		}
+	}
+
+event mysql_error(c: connection, code: count, msg: string)
+	{
+	if ( c?$mysql )
+		{
+		c$mysql$result = "error";
+		c$mysql$response = msg;
+		Log::write(mysql::LOG, c$mysql);
+		delete c$mysql;
+		}	
+	}
+
+event mysql_ok(c: connection, affected_rows: count)
+	{
+	if ( c?$mysql )
+		{
+		c$mysql$result = "ok";
+		c$mysql$response = fmt("Affected rows: %d", affected_rows);
+		Log::write(mysql::LOG, c$mysql);
+		delete c$mysql;
+		}	
+	}

src/analyzer/protocol/mysql/CMakeLists.txt

@@ -0,0 +1,9 @@
+include(BroPlugin)
+
+include_directories(BEFORE ${CMAKE_CURRENT_SOURCE_DIR} ${CMAKE_CURRENT_BINARY_DIR})
+
+bro_plugin_begin(Bro MySQL)
+	bro_plugin_cc(MySQL.cc Plugin.cc)
+	bro_plugin_bif(events.bif)
+	bro_plugin_pac(mysql.pac mysql-analyzer.pac mysql-protocol.pac)
+bro_plugin_end()

src/analyzer/protocol/mysql/MySQL.cc

@@ -0,0 +1,71 @@
+#include "MySQL.h"
+
+#include "analyzer/protocol/tcp/TCP_Reassembler.h"
+
+#include "Reporter.h"
+
+#include "events.bif.h"
+
+using namespace analyzer::MySQL;
+
+MySQL_Analyzer::MySQL_Analyzer(Connection* c)
+
+: tcp::TCP_ApplicationAnalyzer("MySQL", c)
+
+	{
+	interp = new binpac::MySQL::MySQL_Conn(this);
+	
+	had_gap = false;
+	
+	}
+
+MySQL_Analyzer::~MySQL_Analyzer()
+	{
+	delete interp;
+	}
+
+void MySQL_Analyzer::Done()
+	{
+	
+	tcp::TCP_ApplicationAnalyzer::Done();
+
+	interp->FlowEOF(true);
+	interp->FlowEOF(false);
+	
+	}
+
+void MySQL_Analyzer::EndpointEOF(bool is_orig)
+	{
+	tcp::TCP_ApplicationAnalyzer::EndpointEOF(is_orig);
+	interp->FlowEOF(is_orig);
+	}
+
+void MySQL_Analyzer::DeliverStream(int len, const u_char* data, bool orig)
+	{
+	tcp::TCP_ApplicationAnalyzer::DeliverStream(len, data, orig);
+
+	assert(TCP());
+	if ( TCP()->IsPartial() )
+		return;
+
+	if ( had_gap )
+		// If only one side had a content gap, we could still try to
+		// deliver data to the other side if the script layer can handle this.
+		return;
+
+	try
+		{
+		interp->NewData(orig, data, data + len);
+		}
+	catch ( const binpac::Exception& e )
+		{
+		reporter->Weird(e.msg().c_str());
+		}
+	}
+
+void MySQL_Analyzer::Undelivered(uint64 seq, int len, bool orig)
+	{
+	tcp::TCP_ApplicationAnalyzer::Undelivered(seq, len, orig);
+	had_gap = true;
+	interp->NewGap(orig, len);
+	}

src/analyzer/protocol/mysql/MySQL.h

@@ -0,0 +1,48 @@
+#ifndef ANALYZER_PROTOCOL_MYSQL_MYSQL_H
+#define ANALYZER_PROTOCOL_MYSQL_MYSQL_H
+
+#include "events.bif.h"
+
+
+#include "analyzer/protocol/tcp/TCP.h"
+
+#include "mysql_pac.h"
+
+namespace analyzer { namespace MySQL {
+
+class MySQL_Analyzer
+
+: public tcp::TCP_ApplicationAnalyzer {
+
+public:
+	MySQL_Analyzer(Connection* conn);
+	virtual ~MySQL_Analyzer();
+
+	// Overriden from Analyzer.
+	virtual void Done();
+	
+	virtual void DeliverStream(int len, const u_char* data, bool orig);
+	virtual void Undelivered(uint64 seq, int len, bool orig);
+
+	// Overriden from tcp::TCP_ApplicationAnalyzer.
+	virtual void EndpointEOF(bool is_orig);
+	
+
+	static analyzer::Analyzer* InstantiateAnalyzer(Connection* conn)
+		{ return new MySQL_Analyzer(conn); }
+
+	static bool Available()
+		{
+		return ( mysql_command_response || mysql_server_version || mysql_debug || mysql_handshake_response || mysql_login || mysql_command_request );
+		}
+
+protected:
+	binpac::MySQL::MySQL_Conn* interp;
+	
+	bool had_gap;
+	
+};
+
+} } // namespace analyzer::* 
+
+#endif

src/analyzer/protocol/mysql/Plugin.cc

@@ -0,0 +1,9 @@
+#include "plugin/Plugin.h"
+
+#include "MySQL.h"
+
+BRO_PLUGIN_BEGIN(Bro, MySQL)
+	BRO_PLUGIN_DESCRIPTION("MySQL analyzer");
+	BRO_PLUGIN_ANALYZER("MySQL", MySQL::MySQL_Analyzer);
+	BRO_PLUGIN_BIF_FILE(events);
+BRO_PLUGIN_END

src/analyzer/protocol/mysql/events.bif

@@ -0,0 +1,10 @@
+event mysql_command_response%(c: connection, response: count%);
+event mysql_server_version%(c: connection, ver: string%);
+event mysql_debug%(c: connection, ver: count%);
+event mysql_handshake_response%(c: connection, username: string%);
+
+event mysql_login%(c: connection, username: string, success: bool%);
+event mysql_command_request%(c: connection, command: count, arg: string%);
+
+event mysql_error%(c: connection, code: count, msg: string%);
+event mysql_ok%(c: connection, affected_rows: count%);

src/analyzer/protocol/mysql/mysql-analyzer.pac

@@ -0,0 +1,73 @@
+refine flow MySQL_Flow += {
+	function proc_mysql_handshakev10(msg: Handshake_v10): bool
+		%{
+		if ( mysql_server_version ) 
+			BifEvent::generate_mysql_server_version(connection()->bro_analyzer(), connection()->bro_analyzer()->Conn(),
+								bytestring_to_val(${msg.server_version}));
+		connection()->bro_analyzer()->ProtocolConfirmation();
+		return true;
+		%}
+
+	function proc_mysql_handshake_response_packet(msg: Handshake_Response_Packet): bool
+		%{
+		if ( mysql_handshake_response )
+			BifEvent::generate_mysql_handshake_response(connection()->bro_analyzer(), connection()->bro_analyzer()->Conn(),
+								    bytestring_to_val(${msg.username}));
+		return true;
+		%}
+
+	function proc_mysql_command_request_packet(msg: Command_Request_Packet): bool
+		%{
+		if ( mysql_command_request )
+			BifEvent::generate_mysql_command_request(connection()->bro_analyzer(), connection()->bro_analyzer()->Conn(),
+								${msg.command}, bytestring_to_val(${msg.arg}));
+		return true;
+		%}
+
+	function proc_err_packet(msg: ERR_Packet): bool
+		%{
+		if ( mysql_error )
+			BifEvent::generate_mysql_error(connection()->bro_analyzer(), connection()->bro_analyzer()->Conn(),
+						       ${msg.code}, bytestring_to_val(${msg.msg}));
+		return true;
+		%}
+
+	function proc_ok_packet(msg: OK_Packet): bool
+		%{
+		if ( mysql_ok )
+			BifEvent::generate_mysql_ok(connection()->bro_analyzer(), connection()->bro_analyzer()->Conn(), ${msg.rows});
+		return true;
+		%}
+	
+	function proc_resultset(msg: Resultset): bool
+		%{
+		if ( mysql_command_response )
+			BifEvent::generate_mysql_command_response(connection()->bro_analyzer(), connection()->bro_analyzer()->Conn(), ${msg.rows}->size());
+		return true;
+		%}
+
+};
+
+refine typeattr Handshake_v10 += &let {
+	proc = $context.flow.proc_mysql_handshakev10(this);
+};
+
+refine typeattr Handshake_Response_Packet += &let {
+	proc = $context.flow.proc_mysql_handshake_response_packet(this);
+};
+
+refine typeattr Command_Request_Packet += &let {
+	proc = $context.flow.proc_mysql_command_request_packet(this);
+};
+
+refine typeattr ERR_Packet += &let {
+	proc = $context.flow.proc_err_packet(this);
+};
+
+refine typeattr OK_Packet += &let {
+	proc = $context.flow.proc_ok_packet(this);
+};
+
+refine typeattr Resultset += &let {
+	debug = $context.flow.proc_resultset(this);
+};

src/analyzer/protocol/mysql/mysql-protocol.pac

@@ -0,0 +1,348 @@
+###
+#
+# All information is from the MySQL internals documentation at:
+# <http://dev.mysql.com/doc/internals/en/connection-phase.html>
+#
+###
+
+type uint24le = record {
+	byte3 : uint8;
+	byte2 : uint8;
+	byte1 : uint8;
+};
+
+type LengthEncodedInteger = record {
+	i1: uint8;
+	val: case i1 of {
+		0xfb    -> i0: empty;
+		0xfc    -> i2: uint16;
+		0xfd    -> i3: uint24le;
+		0xfe    -> i4: uint64;
+		0xff    -> err_packet: empty;
+		default -> one: empty;
+	};
+};
+
+type LengthEncodedString = record {
+	len: LengthEncodedInteger;
+	val: bytestring &length=to_int()(len);
+};
+
+%header{
+	class to_int 
+		{
+		public:
+		int operator()(uint24le * num) const
+			{
+			return (num->byte1() << 16) | (num->byte2() << 8) | num->byte3();
+			}
+		int operator()(LengthEncodedInteger* lei) const
+			{
+			if ( lei->i1() < 0xfb )
+				return lei->i1();
+			else if ( lei->i1() == 0xfc )
+				return lei->i2();
+			else if ( lei->i1() == 0xfd )
+				return to_int()(lei->i3());
+			else if ( lei->i1() == 0xfe )
+				return lei->i4();
+			else 
+				return 0;
+			}
+
+		};
+%}
+
+extern type to_int;
+
+enum command_consts {
+	COM_SLEEP               = 0x00,
+	COM_QUIT                = 0x01,
+	COM_INIT_DB             = 0x02,
+	COM_QUERY               = 0x03,
+	COM_FIELD_LIST          = 0x04,
+	COM_CREATE_DB           = 0x05,
+	COM_DROP_DB             = 0x06,
+	COM_REFRESH             = 0x07,
+	COM_SHUTDOWN            = 0x08,
+	COM_STATISTICS          = 0x09,
+	COM_PROCESS_INFO        = 0x0a,
+	COM_CONNECT             = 0x0b,
+	COM_PROCESS_KILL        = 0x0c,
+	COM_DEBUG               = 0x0d,
+	COM_PING                = 0x0e,
+	COM_TIME                = 0x0f,
+	COM_DELAYED_INSERT      = 0x10,
+	COM_CHANGE_USER         = 0x11,
+	COM_BINLOG_DUMP         = 0x12,
+	COM_TABLE_DUMP          = 0x13,
+	COM_CONNECT_OUT         = 0x14,
+	COM_REGISTER_SLAVE      = 0x15,
+	COM_STMT_PREPARE        = 0x16,
+	COM_STMT_EXECUTE        = 0x17,
+	COM_STMT_SEND_LONG_DATA = 0x18,
+	COM_STMT_CLOSE          = 0x19,
+	COM_STMT_RESET          = 0x1a,
+	COM_SET_OPTION          = 0x1b,
+	COM_STMT_FETCH          = 0x1c,
+	COM_DAEMON              = 0x1d,
+	COM_BINLOG_DUMP_GTID    = 0x1e
+};
+
+enum state {
+	CONNECTION_PHASE = 0,
+	COMMAND_PHASE    = 1,
+};
+
+enum Expected {
+	NO_EXPECTATION,
+	EXPECT_STATUS,
+	EXPECT_COLUMN_DEFINITION,
+	EXPECT_COLUMN_COUNT,
+	EXPECT_EOF1,
+	EXPECT_EOF2,
+	EXPECT_RESULTSET,
+};
+
+type NUL_String = RE/[^\0]*/;
+
+type Header = record {
+	le_len: uint24le;
+	seq_id: uint8;
+} &let {
+	len: uint32 = to_int()(le_len) + 4;	
+} &length=4;
+
+type MySQL_PDU(is_orig: bool) = record {
+	hdr: Header;
+#	todo: bytestring &length=56;
+	msg:    case is_orig of {
+		false -> server_msg: Server_Message(hdr.seq_id);
+		true  -> client_msg: Client_Message(state);
+	} &requires(state);
+
+	# In case there is trash left over from not parsing something completely.
+	#blah: bytestring &restofdata;
+} &let {
+	state = $context.connection.get_state();
+} &length=hdr.len &byteorder=bigendian;
+
+type Client_Message(state: int) = case state of {
+	CONNECTION_PHASE -> connection_phase: Handshake_Response_Packet;
+	COMMAND_PHASE    -> command_phase:    Command_Request_Packet;
+};
+
+type Server_Message(seq_id: uint8) = case seq_id of {
+	0       -> initial_handshake: Initial_Handshake_Packet;
+	default -> command_response:  Command_Response;
+};
+
+type Initial_Handshake_Packet = record {
+	protocol_version: uint8;
+	pkt: case protocol_version of {
+		10      -> handshake10 : Handshake_v10;
+		9       -> handshake9  : Handshake_v9;
+		default -> error       : ERR_Packet;
+	};
+};
+
+type Handshake_v10 = record {
+	server_version          : NUL_String;
+	connection_id           : uint32;
+	auth_plugin_data_part_1 : bytestring &length=8;
+	filler_1                : uint8;
+	capability_flag_1       : uint16;
+	character_set           : uint8;
+	status_flags            : uint16;
+	capability_flags_2      : uint16;
+	auth_plugin_data_len    : uint8;
+	auth_plugin_name        : NUL_String;
+};
+
+type Handshake_v9 = record {
+	todo: bytestring &restofdata;
+};
+
+type Handshake_Response_Packet = record {
+	cap_flags    : uint32;
+	max_pkt_size : uint32;
+	char_set     : uint8;
+	pad          : padding[23];
+	username     : NUL_String;
+	password     : bytestring &restofdata;
+} &byteorder=bigendian;
+
+type Command_Request_Packet = record {
+	command: uint8;
+	arg:     bytestring &restofdata;
+} &let {
+	update_expectation: bool = $context.connection.set_next_expected(EXPECT_COLUMN_COUNT);
+};
+
+type Command_Response = case $context.connection.get_expectation() of {
+	EXPECT_COLUMN_COUNT		-> col_count	: ColumnCount;
+	EXPECT_COLUMN_DEFINITION	-> col_defs	: ColumnDefinitions;
+	EXPECT_RESULTSET		-> resultset    : Resultset;
+#	EXPECT_RESULTSETROW		-> resultsetrow : ResultsetRow;
+	EXPECT_STATUS			-> status       : Command_Response_Status;
+	EXPECT_EOF1			-> eof1		: EOF1;
+	EXPECT_EOF2			-> eof2		: EOF2;
+	default				-> unknown      : empty;
+};
+
+type Command_Response_Status = record {
+	pkt_type: uint8;
+	response: case pkt_type of {
+		0x00    -> data_ok:   OK_Packet;
+		0xfe    -> data_eof:  EOF_Packet;
+		0xff    -> data_err:  ERR_Packet;
+		default -> unknown:   empty;
+	};
+};
+
+type ColumnCount = record {
+	le_column_count	: LengthEncodedInteger;
+} &let {
+	col_num: uint32 = to_int()(le_column_count);
+	update_col_num: bool = $context.connection.set_col_count(col_num);
+	update_expectation: bool = $context.connection.set_next_expected(EXPECT_COLUMN_DEFINITION);
+};
+
+type ColumnDefinitions = record {
+#	defs: ColumnDefinition41[$context.connection.get_col_count()];
+	defs: ColumnDefinition41[1];
+} &let {
+	update_expectation: bool = $context.connection.set_next_expected(EXPECT_EOF1);
+};
+
+type EOF1 = record {
+	eof	: EOF_Packet;
+} &let {
+	update_expectation: bool = $context.connection.set_next_expected(EXPECT_RESULTSET);
+};
+
+type EOF2 = record {
+	eof	: EOF_Packet;
+} &let {
+	update_expectation: bool = $context.connection.set_next_expected(NO_EXPECTATION);
+};
+
+type Resultset = record {
+	rows		: ResultsetRow[] &until($input.length()==0);
+} &let {
+	update_expectation: bool = $context.connection.set_next_expected(EXPECT_EOF2);
+};
+
+type ResultsetRow = record {
+	fields: LengthEncodedString[$context.connection.get_col_count()];
+};
+
+type ColumnDefinition41 = record {
+	catalog:   LengthEncodedString;
+#	todo: bytestring &length=2;
+	schema:    LengthEncodedString;
+	table:     LengthEncodedString;
+	org_table: LengthEncodedString;
+	name:      LengthEncodedString;
+	org_name:  LengthEncodedString;
+	next_len:  LengthEncodedInteger;
+	char_set:  uint16;
+	col_len:   uint32;
+	type:      uint8;
+	flags:     uint16;
+	decimals:  uint8;
+	filler:    padding[2];
+	#if command was COM_FIELD_LIST {
+	#  lenenc_int     length of default-values
+	#  string[$len]   default values
+	#}
+};
+
+type ColumnDefinition320 = record {
+	table:   LengthEncodedString;
+	name:    LengthEncodedString;
+	length_of_col_len: LengthEncodedInteger;
+	col_len: uint24le;
+	type_len: LengthEncodedInteger;
+	type:     uint8;
+	#if capabilities & CLIENT_LONG_FLAG {
+	#lenenc_int     [03] length of flags+decimals fields
+	#2              flags
+	#1              decimals
+	#  } else {
+	#1              [02] length of flags+decimals fields
+	#1              flags
+	#1              decimals
+	#  }
+	#  if command was COM_FIELD_LIST {
+	#lenenc_int     length of default-values
+	#string[$len]   default values
+	#  }
+};
+
+type OK_Packet = record {
+	le_rows: LengthEncodedInteger;
+	todo: bytestring &restofdata;
+} &let {
+	rows: uint32 = to_int()(le_rows);
+	update_state: bool = $context.connection.update_state(COMMAND_PHASE);
+};
+
+type ERR_Packet = record {
+	code: uint16;
+	state: bytestring &length=6;
+	msg: bytestring &restofdata;
+};
+
+type EOF_Packet = record {
+	warnings: uint16;
+	status  : uint16;
+};
+
+
+refine connection MySQL_Conn += {
+	%member{
+		int state_;
+		Expected expected_;
+		uint32 col_count_;
+	%}
+
+	%init{
+		state_ = CONNECTION_PHASE;
+		expected_ = EXPECT_STATUS;
+		col_count_ = 0;
+	%}
+
+	function get_state(): int
+		%{ 
+		return state_; 
+		%}
+
+	function update_state(s: state): bool
+		%{
+		state_ = s;
+		return true;
+		%}
+
+	function get_expectation(): Expected
+		%{
+		return expected_;
+		%}
+
+	function set_next_expected(e: Expected): bool
+		%{
+		expected_ = e;
+		return true;
+		%}
+
+	function get_col_count(): uint32
+		%{
+		return col_count_;
+		%}
+
+	function set_col_count(i: uint32): bool
+		%{
+		col_count_ = i;
+		return true;
+		%}
+};

src/analyzer/protocol/mysql/mysql.pac

@@ -0,0 +1,35 @@
+# Analyzer for MySQL
+#  - mysql-protocol.pac: describes the MySQL protocol messages
+#  - mysql-analyzer.pac: describes the MySQL analyzer code
+
+%include binpac.pac
+%include bro.pac
+
+%extern{
+	#include "events.bif.h"
+%}
+
+analyzer MySQL withcontext {
+	connection: MySQL_Conn;
+	flow:       MySQL_Flow;
+};
+
+# Our connection consists of two flows, one in each direction.
+connection MySQL_Conn(bro_analyzer: BroAnalyzer) {
+	upflow   = MySQL_Flow(true);
+	downflow = MySQL_Flow(false);
+};
+
+%include mysql-protocol.pac
+
+# Now we define the flow:
+flow MySQL_Flow(is_orig: bool) {
+	# There are two options here: flowunit or datagram.
+	# flowunit = MySQL_PDU(is_orig) withcontext(connection, this);
+	flowunit = MySQL_PDU(is_orig) withcontext(connection, this);
+	# Using flowunit will cause the anlayzer to buffer incremental input.
+	# This is needed for &oneline and &length. If you don't need this, you'll
+	# get better performance with datagram.
+};
+
+%include mysql-analyzer.pac