GH-474: add MQTT::max_payload_size option

This caps size of payload strings within mqtt_publish events and
mqtt_publish.log files.  A new "payload_len" field in the log file
shows the real payload size in cases where it may have been truncated.

scripts/base/init-bare.zeek

@@ -5083,7 +5083,17 @@ export {
 
 		## Payload of the published message.
 		payload : string;
+
+		## The actual length of the payload in the case the *payload*
+		## field's contents were truncated according to
+		## :zeek:see:`MQTT::max_payload_size`.
+		payload_len : count;
 	};
+
+	## The maximum payload size to allocate for the purpose of
+	## payload information in :zeek:see:`mqtt_publish` events (and the
+	## default MQTT logs generated from that).
+	const max_payload_size = 100 &redef;
 }
 
 module Cluster;

scripts/base/protocols/mqtt/main.zeek

@@ -85,6 +85,11 @@ export {
 		## Payload of the message
 		payload:     string  &log;
 
+		## The actual length of the payload in the case the *payload*
+		## field's contents were truncated according to
+		## :zeek:see:`MQTT::max_payload_size`.
+		payload_len: count   &log;
+
 		## Track if the message was acked
 		ack:         bool    &default=F;
 		## Indicates if the server sent the RECEIVED qos message
@@ -190,7 +195,8 @@ event mqtt_publish(c: connection, is_orig: bool, msg_id: count, msg: MQTT::Publi
 	                       $qos=qos_levels[msg$qos],
 	                       $qos_level=msg$qos,
 	                       $topic=msg$topic,
-	                       $payload=msg$payload);
+	                       $payload=msg$payload,
+	                       $payload_len=msg$payload_len);
 	if ( pi$qos_level == 0 )
 		pi$status="ok";
 

src/analyzer/protocol/mqtt/CMakeLists.txt

@@ -5,7 +5,7 @@ include_directories(BEFORE ${CMAKE_CURRENT_SOURCE_DIR} ${CMAKE_CURRENT_BINARY_DI
 
 zeek_plugin_begin(Zeek MQTT)
 zeek_plugin_cc(MQTT.cc Plugin.cc)
-zeek_plugin_bif(types.bif events.bif)
+zeek_plugin_bif(types.bif consts.bif events.bif)
 zeek_plugin_pac(mqtt.pac
                 mqtt-protocol.pac
                 commands/connect.pac

src/analyzer/protocol/mqtt/commands/publish.pac

@@ -29,9 +29,17 @@ refine flow MQTT_Flow += {
 			m->Assign(2, val_mgr->GetBool(${msg.retain}));
 			m->Assign(3, new StringVal(${msg.topic.str}.length(),
 			                           reinterpret_cast<const char*>(${msg.topic.str}.begin())));
-			m->Assign(4, new StringVal(${msg.payload}.length(),
+
+			auto len = ${msg.payload}.length();
+
+			if ( len > static_cast<int>(BifConst::MQTT::max_payload_size) )
+				len = BifConst::MQTT::max_payload_size;
+
+			m->Assign(4, new StringVal(len,
 			                           reinterpret_cast<const char*>(${msg.payload}.begin())));
 
+			m->Assign(5, val_mgr->GetCount(${msg.payload}.length()));
+
 			BifEvent::generate_mqtt_publish(connection()->bro_analyzer(),
 			                                connection()->bro_analyzer()->Conn(),
 			                                ${pdu.is_orig},

src/analyzer/protocol/mqtt/consts.bif

@@ -0,0 +1 @@
+const MQTT::max_payload_size: count;

src/analyzer/protocol/mqtt/mqtt.pac

@@ -6,6 +6,7 @@
 %extern{
 	#include "events.bif.h"
 	#include "types.bif.h"
+	#include "consts.bif.h"
 %}
 
 analyzer MQTT withcontext {

testing/btest/Baseline/coverage.bare-load-baseline/canonified_loaded_scripts.log

@@ -91,6 +91,7 @@ scripts/base/init-frameworks-and-bifs.zeek
     build/scripts/base/bif/plugins/Zeek_MIME.events.bif.zeek
     build/scripts/base/bif/plugins/Zeek_Modbus.events.bif.zeek
     build/scripts/base/bif/plugins/Zeek_MQTT.types.bif.zeek
+    build/scripts/base/bif/plugins/Zeek_MQTT.consts.bif.zeek
     build/scripts/base/bif/plugins/Zeek_MQTT.events.bif.zeek
     build/scripts/base/bif/plugins/Zeek_MySQL.events.bif.zeek
     build/scripts/base/bif/plugins/Zeek_NCP.events.bif.zeek

testing/btest/Baseline/coverage.default-load-baseline/canonified_loaded_scripts.log

@@ -91,6 +91,7 @@ scripts/base/init-frameworks-and-bifs.zeek
     build/scripts/base/bif/plugins/Zeek_MIME.events.bif.zeek
     build/scripts/base/bif/plugins/Zeek_Modbus.events.bif.zeek
     build/scripts/base/bif/plugins/Zeek_MQTT.types.bif.zeek
+    build/scripts/base/bif/plugins/Zeek_MQTT.consts.bif.zeek
     build/scripts/base/bif/plugins/Zeek_MQTT.events.bif.zeek
     build/scripts/base/bif/plugins/Zeek_MySQL.events.bif.zeek
     build/scripts/base/bif/plugins/Zeek_NCP.events.bif.zeek

testing/btest/Baseline/plugins.hooks/output

@@ -283,7 +283,7 @@
 0.000000   MetaHookPost  CallFunction(Log::__create_stream, <frame>, (Weird::LOG, [columns=Weird::Info, ev=Weird::log_weird, path=weird])) -> <no result>
 0.000000   MetaHookPost  CallFunction(Log::__create_stream, <frame>, (X509::LOG, [columns=X509::Info, ev=X509::log_x509, path=x509])) -> <no result>
 0.000000   MetaHookPost  CallFunction(Log::__create_stream, <frame>, (mysql::LOG, [columns=MySQL::Info, ev=MySQL::log_mysql, path=mysql])) -> <no result>
-0.000000   MetaHookPost  CallFunction(Log::__write, <frame>, (PacketFilter::LOG, [ts=1564691423.873691, node=zeek, filter=ip or not ip, init=T, success=T])) -> <no result>
+0.000000   MetaHookPost  CallFunction(Log::__write, <frame>, (PacketFilter::LOG, [ts=1564780864.738576, node=zeek, filter=ip or not ip, init=T, success=T])) -> <no result>
 0.000000   MetaHookPost  CallFunction(Log::add_default_filter, <frame>, (Broker::LOG)) -> <no result>
 0.000000   MetaHookPost  CallFunction(Log::add_default_filter, <frame>, (Cluster::LOG)) -> <no result>
 0.000000   MetaHookPost  CallFunction(Log::add_default_filter, <frame>, (Config::LOG)) -> <no result>
@@ -476,7 +476,7 @@
 0.000000   MetaHookPost  CallFunction(Log::create_stream, <frame>, (Weird::LOG, [columns=Weird::Info, ev=Weird::log_weird, path=weird])) -> <no result>
 0.000000   MetaHookPost  CallFunction(Log::create_stream, <frame>, (X509::LOG, [columns=X509::Info, ev=X509::log_x509, path=x509])) -> <no result>
 0.000000   MetaHookPost  CallFunction(Log::create_stream, <frame>, (mysql::LOG, [columns=MySQL::Info, ev=MySQL::log_mysql, path=mysql])) -> <no result>
-0.000000   MetaHookPost  CallFunction(Log::write, <frame>, (PacketFilter::LOG, [ts=1564691423.873691, node=zeek, filter=ip or not ip, init=T, success=T])) -> <no result>
+0.000000   MetaHookPost  CallFunction(Log::write, <frame>, (PacketFilter::LOG, [ts=1564780864.738576, node=zeek, filter=ip or not ip, init=T, success=T])) -> <no result>
 0.000000   MetaHookPost  CallFunction(NetControl::check_plugins, <frame>, ()) -> <no result>
 0.000000   MetaHookPost  CallFunction(NetControl::init, <null>, ()) -> <no result>
 0.000000   MetaHookPost  CallFunction(Notice::want_pp, <frame>, ()) -> <no result>
@@ -627,6 +627,7 @@
 0.000000   MetaHookPost  LoadFile(0, .<...>/Zeek_Login.events.bif.zeek) -> -1
 0.000000   MetaHookPost  LoadFile(0, .<...>/Zeek_Login.functions.bif.zeek) -> -1
 0.000000   MetaHookPost  LoadFile(0, .<...>/Zeek_MIME.events.bif.zeek) -> -1
+0.000000   MetaHookPost  LoadFile(0, .<...>/Zeek_MQTT.consts.bif.zeek) -> -1
 0.000000   MetaHookPost  LoadFile(0, .<...>/Zeek_MQTT.events.bif.zeek) -> -1
 0.000000   MetaHookPost  LoadFile(0, .<...>/Zeek_MQTT.types.bif.zeek) -> -1
 0.000000   MetaHookPost  LoadFile(0, .<...>/Zeek_Modbus.events.bif.zeek) -> -1
@@ -1198,7 +1199,7 @@
 0.000000   MetaHookPre   CallFunction(Log::__create_stream, <frame>, (Weird::LOG, [columns=Weird::Info, ev=Weird::log_weird, path=weird]))
 0.000000   MetaHookPre   CallFunction(Log::__create_stream, <frame>, (X509::LOG, [columns=X509::Info, ev=X509::log_x509, path=x509]))
 0.000000   MetaHookPre   CallFunction(Log::__create_stream, <frame>, (mysql::LOG, [columns=MySQL::Info, ev=MySQL::log_mysql, path=mysql]))
-0.000000   MetaHookPre   CallFunction(Log::__write, <frame>, (PacketFilter::LOG, [ts=1564691423.873691, node=zeek, filter=ip or not ip, init=T, success=T]))
+0.000000   MetaHookPre   CallFunction(Log::__write, <frame>, (PacketFilter::LOG, [ts=1564780864.738576, node=zeek, filter=ip or not ip, init=T, success=T]))
 0.000000   MetaHookPre   CallFunction(Log::add_default_filter, <frame>, (Broker::LOG))
 0.000000   MetaHookPre   CallFunction(Log::add_default_filter, <frame>, (Cluster::LOG))
 0.000000   MetaHookPre   CallFunction(Log::add_default_filter, <frame>, (Config::LOG))
@@ -1391,7 +1392,7 @@
 0.000000   MetaHookPre   CallFunction(Log::create_stream, <frame>, (Weird::LOG, [columns=Weird::Info, ev=Weird::log_weird, path=weird]))
 0.000000   MetaHookPre   CallFunction(Log::create_stream, <frame>, (X509::LOG, [columns=X509::Info, ev=X509::log_x509, path=x509]))
 0.000000   MetaHookPre   CallFunction(Log::create_stream, <frame>, (mysql::LOG, [columns=MySQL::Info, ev=MySQL::log_mysql, path=mysql]))
-0.000000   MetaHookPre   CallFunction(Log::write, <frame>, (PacketFilter::LOG, [ts=1564691423.873691, node=zeek, filter=ip or not ip, init=T, success=T]))
+0.000000   MetaHookPre   CallFunction(Log::write, <frame>, (PacketFilter::LOG, [ts=1564780864.738576, node=zeek, filter=ip or not ip, init=T, success=T]))
 0.000000   MetaHookPre   CallFunction(NetControl::check_plugins, <frame>, ())
 0.000000   MetaHookPre   CallFunction(NetControl::init, <null>, ())
 0.000000   MetaHookPre   CallFunction(Notice::want_pp, <frame>, ())
@@ -1542,6 +1543,7 @@
 0.000000   MetaHookPre   LoadFile(0, .<...>/Zeek_Login.events.bif.zeek)
 0.000000   MetaHookPre   LoadFile(0, .<...>/Zeek_Login.functions.bif.zeek)
 0.000000   MetaHookPre   LoadFile(0, .<...>/Zeek_MIME.events.bif.zeek)
+0.000000   MetaHookPre   LoadFile(0, .<...>/Zeek_MQTT.consts.bif.zeek)
 0.000000   MetaHookPre   LoadFile(0, .<...>/Zeek_MQTT.events.bif.zeek)
 0.000000   MetaHookPre   LoadFile(0, .<...>/Zeek_MQTT.types.bif.zeek)
 0.000000   MetaHookPre   LoadFile(0, .<...>/Zeek_Modbus.events.bif.zeek)
@@ -2112,7 +2114,7 @@
 0.000000 | HookCallFunction Log::__create_stream(Weird::LOG, [columns=Weird::Info, ev=Weird::log_weird, path=weird])
 0.000000 | HookCallFunction Log::__create_stream(X509::LOG, [columns=X509::Info, ev=X509::log_x509, path=x509])
 0.000000 | HookCallFunction Log::__create_stream(mysql::LOG, [columns=MySQL::Info, ev=MySQL::log_mysql, path=mysql])
-0.000000 | HookCallFunction Log::__write(PacketFilter::LOG, [ts=1564691423.873691, node=zeek, filter=ip or not ip, init=T, success=T])
+0.000000 | HookCallFunction Log::__write(PacketFilter::LOG, [ts=1564780864.738576, node=zeek, filter=ip or not ip, init=T, success=T])
 0.000000 | HookCallFunction Log::add_default_filter(Broker::LOG)
 0.000000 | HookCallFunction Log::add_default_filter(Cluster::LOG)
 0.000000 | HookCallFunction Log::add_default_filter(Config::LOG)
@@ -2305,7 +2307,7 @@
 0.000000 | HookCallFunction Log::create_stream(Weird::LOG, [columns=Weird::Info, ev=Weird::log_weird, path=weird])
 0.000000 | HookCallFunction Log::create_stream(X509::LOG, [columns=X509::Info, ev=X509::log_x509, path=x509])
 0.000000 | HookCallFunction Log::create_stream(mysql::LOG, [columns=MySQL::Info, ev=MySQL::log_mysql, path=mysql])
-0.000000 | HookCallFunction Log::write(PacketFilter::LOG, [ts=1564691423.873691, node=zeek, filter=ip or not ip, init=T, success=T])
+0.000000 | HookCallFunction Log::write(PacketFilter::LOG, [ts=1564780864.738576, node=zeek, filter=ip or not ip, init=T, success=T])
 0.000000 | HookCallFunction NetControl::check_plugins()
 0.000000 | HookCallFunction NetControl::init()
 0.000000 | HookCallFunction Notice::want_pp()
@@ -2456,6 +2458,7 @@
 0.000000 | HookLoadFile  .<...>/Zeek_Login.events.bif.zeek
 0.000000 | HookLoadFile  .<...>/Zeek_Login.functions.bif.zeek
 0.000000 | HookLoadFile  .<...>/Zeek_MIME.events.bif.zeek
+0.000000 | HookLoadFile  .<...>/Zeek_MQTT.consts.bif.zeek
 0.000000 | HookLoadFile  .<...>/Zeek_MQTT.events.bif.zeek
 0.000000 | HookLoadFile  .<...>/Zeek_MQTT.types.bif.zeek
 0.000000 | HookLoadFile  .<...>/Zeek_Modbus.events.bif.zeek
@@ -2738,7 +2741,7 @@
 0.000000 | HookLoadFile  base<...>/xmpp
 0.000000 | HookLoadFile  base<...>/zeek.bif.zeek
 0.000000 | HookLogInit   packet_filter 1/1 {ts (time), node (string), filter (string), init (bool), success (bool)}
-0.000000 | HookLogWrite  packet_filter [ts=1564691423.873691, node=zeek, filter=ip or not ip, init=T, success=T]
+0.000000 | HookLogWrite  packet_filter [ts=1564780864.738576, node=zeek, filter=ip or not ip, init=T, success=T]
 0.000000 | HookQueueEvent NetControl::init()
 0.000000 | HookQueueEvent filter_change_tracking()
 0.000000 | HookQueueEvent zeek_init()

testing/btest/Baseline/scripts.base.protocols.mqtt.mqtt-payload-cap/mqtt_publish.log

@@ -0,0 +1,12 @@
+#separator \x09
+#set_separator	,
+#empty_field	(empty)
+#unset_field	-
+#path	mqtt_publish
+#open	2019-08-02-21-26-17
+#fields	ts	uid	id.orig_h	id.orig_p	id.resp_h	id.resp_p	from_client	retain	qos	status	topic	payload	payload_len
+#types	time	string	addr	port	addr	port	bool	bool	string	string	string	string	count
+1461170591.219981	CHhAvVGS1DHFjwGM9	10.0.1.4	49327	198.41.30.241	1883	F	T	at most once	ok	SampleTopic	Hello fr	35
+1461170596.653674	ClEkJM2Vm5giqnMf4h	10.0.1.4	49330	198.41.30.241	1883	T	F	at most once	ok	SampleTopic	Hello MQ	10
+1461170596.891281	CHhAvVGS1DHFjwGM9	10.0.1.4	49327	198.41.30.241	1883	F	F	at most once	ok	SampleTopic	Hello MQ	10
+#close	2019-08-02-21-26-17

testing/btest/Baseline/scripts.base.protocols.mqtt.mqtt/mqtt_publish.log

@@ -3,10 +3,10 @@
 #empty_field	(empty)
 #unset_field	-
 #path	mqtt_publish
-#open	2019-07-29-16-44-12
+#open	2019-08-02-21-23-10
-#fields	ts	uid	id.orig_h	id.orig_p	id.resp_h	id.resp_p	from_client	retain	qos	status	topic	payload
+#fields	ts	uid	id.orig_h	id.orig_p	id.resp_h	id.resp_p	from_client	retain	qos	status	topic	payload	payload_len
-#types	time	string	addr	port	addr	port	bool	bool	string	string	string	string
+#types	time	string	addr	port	addr	port	bool	bool	string	string	string	string	count
-1461170591.219981	CHhAvVGS1DHFjwGM9	10.0.1.4	49327	198.41.30.241	1883	F	T	at most once	ok	SampleTopic	Hello from the Paho blocking client
+1461170591.219981	CHhAvVGS1DHFjwGM9	10.0.1.4	49327	198.41.30.241	1883	F	T	at most once	ok	SampleTopic	Hello from the Paho blocking client	35
-1461170596.653674	ClEkJM2Vm5giqnMf4h	10.0.1.4	49330	198.41.30.241	1883	T	F	at most once	ok	SampleTopic	Hello MQTT
+1461170596.653674	ClEkJM2Vm5giqnMf4h	10.0.1.4	49330	198.41.30.241	1883	T	F	at most once	ok	SampleTopic	Hello MQTT	10
-1461170596.891281	CHhAvVGS1DHFjwGM9	10.0.1.4	49327	198.41.30.241	1883	F	F	at most once	ok	SampleTopic	Hello MQTT
+1461170596.891281	CHhAvVGS1DHFjwGM9	10.0.1.4	49327	198.41.30.241	1883	F	F	at most once	ok	SampleTopic	Hello MQTT	10
-#close	2019-07-29-16-44-12
+#close	2019-08-02-21-23-10

testing/btest/scripts/base/protocols/mqtt/mqtt-payload-cap.test

@@ -0,0 +1,6 @@
+# @TEST-EXEC: zeek -b -r $TRACES/mqtt.pcap %INPUT >output
+# @TEST-EXEC: btest-diff mqtt_publish.log
+
+redef MQTT::max_payload_size = 8;
+
+@load base/protocols/mqtt