GH-1506: Fix Broker unserialization of set/table function indices

Zeek function types are serialized as a broker::vector, but the
unserialization logic for Zeek set/table types that use a function for
as an index incorrectly identified it as a composite-index, which also
use broker::vector, rather than a singleton-index, and makes such
unserialization fail.

A general example where this failure can happen in practice is when
trying to unserialize a connection record for which there's a
Conn::RemovalHook, since that's a set[function], and a specific case of
that is use of the Intel Framework in a Zeek cluster.

src/broker/Data.cc

@@ -48,6 +48,22 @@ TEST_CASE("converting Zeek to Broker protocol constants")
 
 namespace zeek::Broker::detail {
 
+// Returns true if the given Zeek type is serialized as a broker::vector
+static bool serialized_as_vector(TypeTag tag)
+	{
+	switch ( tag ) {
+	case TYPE_VECTOR:
+	case TYPE_RECORD:
+	case TYPE_FUNC:
+	case TYPE_PATTERN:
+	case TYPE_OPAQUE:
+		return true;
+	default:
+		return false;
+	}
+	return false;
+	}
+
 static bool data_type_check(const broker::data& d, Type* t);
 
 TransportProto to_zeek_port_proto(broker::port::protocol tp)
@@ -216,11 +232,9 @@ struct val_converter {
 				{
 				if ( expected_index_types.size() == 1 )
 					{
-					auto index_is_vector_or_record =
+					auto disambiguate = serialized_as_vector(expected_index_types[0]->Tag());
-					     expected_index_types[0]->Tag() == TYPE_RECORD ||
-					     expected_index_types[0]->Tag() == TYPE_VECTOR;
 
-					if ( index_is_vector_or_record )
+					if ( disambiguate )
 						{
 						// Disambiguate from composite key w/ multiple vals.
 						composite_key.emplace_back(move(item));
@@ -275,11 +289,9 @@ struct val_converter {
 				{
 				if ( expected_index_types.size() == 1 )
 					{
-					auto index_is_vector_or_record =
+					auto disambiguate = serialized_as_vector(expected_index_types[0]->Tag());
-					     expected_index_types[0]->Tag() == TYPE_RECORD ||
-					     expected_index_types[0]->Tag() == TYPE_VECTOR;
 
-					if ( index_is_vector_or_record )
+					if ( disambiguate )
 						{
 						// Disambiguate from composite key w/ multiple vals.
 						composite_key.emplace_back(move(item.first));
@@ -606,11 +618,9 @@ struct type_checker {
 				{
 				if ( expected_index_types.size() == 1 )
 					{
-					auto index_is_vector_or_record =
+					auto disambiguate = serialized_as_vector(expected_index_types[0]->Tag());
-					     expected_index_types[0]->Tag() == TYPE_RECORD ||
-					     expected_index_types[0]->Tag() == TYPE_VECTOR;
 
-					if ( index_is_vector_or_record )
+					if ( disambiguate )
 						// Disambiguate from composite key w/ multiple vals.
 						indices_to_check.emplace_back(&item);
 					else
@@ -665,11 +675,9 @@ struct type_checker {
 				{
 				if ( expected_index_types.size() == 1 )
 					{
-					auto index_is_vector_or_record =
+					auto disambiguate = serialized_as_vector(expected_index_types[0]->Tag());
-					     expected_index_types[0]->Tag() == TYPE_RECORD ||
-					     expected_index_types[0]->Tag() == TYPE_VECTOR;
 
-					if ( index_is_vector_or_record )
+					if ( disambiguate )
 						// Disambiguate from composite key w/ multiple vals.
 						indices_to_check.emplace_back(&item.first);
 					else

testing/btest/Baseline/broker.remote_event_index_types/recv.recv.out

@@ -0,0 +1,37 @@
+### BTest baseline data generated by btest-diff. Do not edit. Use "btest -U/-u" to update. Requires BTest >= 0.63.
+set1, {
+1
+}
+set2, {
+[2, two] 
+}
+setrecord, {
+[a=97, b=B]
+}
+setfunction, {
+foo
+{ 
+print foo;
+}
+}
+setpattern, {
+/^?(foobar)$?/
+}
+table1, {
+[1] = t1
+}
+table2, {
+[2, two] = t2
+}
+tablerecord, {
+[[a=97, b=B]] = trec
+}
+tablefunction, {
+[foo
+{ 
+print foo;
+}] = tfunc
+}
+tablepattern, {
+[/^?(foobar)$?/] = tpat
+}

testing/btest/Baseline/scripts.policy.frameworks.intel.seen.smtp-cluster/manager-1.intel.log

@@ -0,0 +1,11 @@
+### BTest baseline data generated by btest-diff. Do not edit. Use "btest -U/-u" to update. Requires BTest >= 0.63.
+#separator \x09
+#set_separator	,
+#empty_field	(empty)
+#unset_field	-
+#path	intel
+#open XXXX-XX-XX-XX-XX-XX
+#fields	ts	uid	id.orig_h	id.orig_p	id.resp_h	id.resp_p	seen.indicator	seen.indicator_type	seen.where	seen.node	matched	sources	fuid	file_mime_type	file_desc
+#types	time	string	addr	port	addr	port	string	enum	enum	string	set[enum]	set[string]	string	string	string
+XXXXXXXXXX.XXXXXX	CHhAvVGS1DHFjwGM9	1.1.1.1	1	2.2.2.2	2	example@gmail.com	Intel::EMAIL	Intel::IN_ANYWHERE	worker-1	Intel::EMAIL	source1	-	-	-
+#close XXXX-XX-XX-XX-XX-XX

testing/btest/broker/remote_event_index_types.zeek

@@ -0,0 +1,126 @@
+# @TEST-PORT: BROKER_PORT
+#
+# @TEST-EXEC: btest-bg-run recv "zeek -B broker -b ../recv.zeek >recv.out"
+# @TEST-EXEC: btest-bg-run send "zeek -B broker -b ../send.zeek >send.out"
+#
+# @TEST-EXEC: btest-bg-wait 45
+# @TEST-EXEC: btest-diff recv/recv.out
+
+@TEST-START-FILE common.zeek
+redef exit_only_after_terminate = T;
+
+type MyRecord: record {
+	a: count;
+	b: string;
+};
+
+type MyVector: vector of string;
+
+type MyOpaque: opaque of topk;
+
+type MyFunction: function();
+
+function foo()
+	{ print "foo"; }
+
+type Set1: set[count];
+type Set2: set[count, string];
+type SetVector: set[MyVector];
+type SetRecord: set[MyRecord];
+type SetFunction: set[MyFunction];
+type SetPattern: set[pattern];
+
+type Table1: table[count] of string;
+type Table2: table[count, string] of string;
+type TableVector: table[MyVector] of string;
+type TableRecord: table[MyRecord] of string;
+type TableFunction: table[MyFunction] of string;
+type TablePattern: table[pattern] of string;
+
+global set1: event(x: Set1);
+global set2: event(x: Set2);
+global setvector: event(x: SetVector);
+global setrecord: event(x: SetRecord);
+global setfunction: event(x: SetFunction);
+global setpattern: event(x: SetPattern);
+
+global table1: event(x: Table1);
+global table2: event(x: Table2);
+global tablevector: event(x: TableVector);
+global tablerecord: event(x: TableRecord);
+global tablefunction: event(x: TableFunction);
+global tablepattern: event(x: TablePattern);
+
+global done: event();
+@TEST-END-FILE
+
+@TEST-START-FILE send.zeek
+@load ./common.zeek
+
+event zeek_init()
+	{ Broker::peer("127.0.0.1", to_port(getenv("BROKER_PORT"))); }
+
+event Broker::peer_added(endpoint: Broker::EndpointInfo, msg: string)
+	{
+	Broker::publish("test", set1, Set1(1));
+	Broker::publish("test", set2, Set2([2, "two"]));
+	# Broker::publish("test", setvector, SetVector(MyVector("one", "two")));
+	Broker::publish("test", setrecord, SetRecord(MyRecord($a=97, $b="B")));
+	Broker::publish("test", setfunction, SetFunction(foo));
+	Broker::publish("test", setpattern, SetPattern(/foobar/));
+
+	Broker::publish("test", table1, Table1([1] = "t1"));
+	Broker::publish("test", table2, Table2([2, "two"] = "t2"));
+	# Broker::publish("test", tablevector, TableVector([MyVector("one", "two")] = "tvec"));
+	Broker::publish("test", tablerecord, TableRecord([MyRecord($a=97, $b="B")] = "trec"));
+	Broker::publish("test", tablefunction, TableFunction([foo] = "tfunc"));
+	Broker::publish("test", tablepattern, TablePattern([/foobar/] = "tpat"));
+
+	Broker::publish("test", done);
+	}
+
+event Broker::peer_lost(endpoint: Broker::EndpointInfo, msg: string)
+	{ terminate(); }
+@TEST-END-FILE
+
+@TEST-START-FILE recv.zeek
+@load ./common.zeek
+
+event set1(x: Set1)
+	{ print "set1", x; }
+event set2(x: Set2)
+	{ print "set2", x; }
+event setvector(x: SetVector)
+	{ print "setvector", x; }
+event setrecord(x: SetRecord)
+	{ print "setrecord", x; }
+event setfunction(x: SetFunction)
+	{ print "setfunction", x; }
+event setpattern(x: SetPattern)
+	{ print "setpattern", x; }
+
+event table1(x: Table1)
+	{ print "table1", x; }
+event table2(x: Table2)
+	{ print "table2", x; }
+event tablevector(x: TableVector)
+	{ print "tablevector", x; }
+event tablerecord(x: TableRecord)
+	{ print "tablerecord", x; }
+event tablefunction(x: TableFunction)
+	{ print "tablefunction", x; }
+event tablepattern(x: TablePattern)
+	{ print "tablepattern", x; }
+
+event zeek_init()
+	{
+	Broker::subscribe("test");
+	Broker::listen("127.0.0.1", to_port(getenv("BROKER_PORT")));
+	}
+
+event done()
+	{ terminate(); }
+
+event Broker::peer_lost(endpoint: Broker::EndpointInfo, msg: string)
+	{ terminate(); }
+@TEST-END-FILE

testing/btest/scripts/policy/frameworks/intel/seen/smtp-cluster.zeek

@@ -0,0 +1,98 @@
+# @TEST-PORT: BROKER_PORT1
+# @TEST-PORT: BROKER_PORT2
+
+# @TEST-EXEC: btest-bg-run manager-1 ZEEKPATH=$ZEEKPATH:.. CLUSTER_NODE=manager-1 zeek -b %INPUT
+# @TEST-EXEC: btest-bg-run worker-1  ZEEKPATH=$ZEEKPATH:.. CLUSTER_NODE=worker-1 zeek -b %INPUT
+# @TEST-EXEC: btest-bg-wait 90
+
+# @TEST-EXEC: btest-diff manager-1/intel.log
+
+@TEST-START-FILE intel.dat
+#fields	indicator	indicator_type	meta.source	meta.desc	meta.url
+example@gmail.com	Intel::EMAIL	source1	test entry	http://some-data-distributor.com/100000
+@TEST-END-FILE
+
+@TEST-START-FILE cluster-layout.zeek
+redef Cluster::nodes = {
+	["manager-1"] = [$node_type=Cluster::MANAGER, $ip=127.0.0.1, $p=to_port(getenv("BROKER_PORT1"))],
+	["worker-1"]  = [$node_type=Cluster::WORKER,  $ip=127.0.0.1, $p=to_port(getenv("BROKER_PORT2")), $manager="manager-1"],
+};
+@TEST-END-FILE
+
+@load base/frameworks/cluster
+@load base/frameworks/intel
+@load frameworks/intel/seen
+
+redef Log::default_rotation_interval = 0secs;
+redef Intel::read_files += { "../intel.dat" };
+
+global done_reading = F;
+global connected = F;
+global log_count = 0;
+
+event zeek_init()
+	{
+	Broker::subscribe("test");
+	}
+
+hook my_removal_hook(c: connection)
+	{
+	}
+
+event proceed()
+	{
+	# This is an entirely artificial connection record because reading from
+	# a real pcap tends to make this test timeout on CI under ASan.
+	local c = connection(
+			$id = conn_id($orig_h=1.1.1.1, $orig_p=1/tcp,
+			              $resp_h=2.2.2.2, $resp_p=2/tcp),
+			$orig = endpoint($size=1, $state=4, $flow_label=0),
+			$resp = endpoint($size=1, $state=4, $flow_label=0),
+			$start_time=current_time(),
+			$duration=1sec,
+			$service=set("smtp"),
+			$history="ShAdDa",
+			$uid="CHhAvVGS1DHFjwGM9",
+			$removal_hooks=set(my_removal_hook)
+	);
+
+	local iseen = Intel::Seen(
+			$indicator="example@gmail.com",
+			$indicator_type=Intel::EMAIL,
+			$where=Intel::IN_ANYWHERE,
+			$conn=c
+	);
+	Intel::seen(iseen);
+	}
+
+event Cluster::node_up(name: string, id: string)
+	{
+	if ( Cluster::node != "manager-1" )
+		return;
+
+	connected = T;
+
+	if ( done_reading )
+		Broker::publish("test", proceed);
+	}
+
+event Cluster::node_down(name: string, id: string)
+	{
+	terminate();
+	}
+
+event Input::end_of_data(name: string, source: string)
+	{
+	if ( Cluster::node != "manager-1" )
+		return;
+
+	done_reading = T;
+
+	if ( connected )
+		Broker::publish("test", proceed);
+	}
+
+event Intel::log_intel(rec: Intel::Info)
+	{
+	terminate();
+	}