add broker output plugin for openflow (at the moment we more or less

just send the flow_mod event along - there still is no feedback) and add
a testcase for it.

Also fix a few other small problems.

scripts/base/frameworks/openflow/plugins/__load__.bro

@@ -1,2 +1,3 @@
 @load ./ryu
 @load ./log
+@load ./broker

scripts/base/frameworks/openflow/plugins/broker.bro

@@ -0,0 +1,64 @@
+@load base/frameworks/openflow
+
+module OpenFlow;
+
+export {
+	redef enum Plugin += {
+		BROKER,
+	};
+
+	## Broker controller constructor.
+	##
+	## host: Controller ip.
+	##
+	## host_port: Controller listen port.
+	##
+	## dpid: OpenFlow switch datapath id.
+	##
+	## Returns: OpenFlow::Controller record
+	global broker_new: function(host: addr, host_port: port, dpid: count): OpenFlow::Controller;
+
+	redef record ControllerState += {
+		## Controller ip.
+		broker_host: addr &optional;
+		## Controller listen port.
+		broker_port: port &optional;
+		## OpenFlow switch datapath id.
+		broker_dpid: count &optional;
+	};
+
+	global broker_flow_mod: event(dpid: count, match: ofp_match, flow_mod: ofp_flow_mod);
+	global broker_flow_clear: event(dpid: count);
+}
+
+function broker_describe(state: ControllerState): string
+	{
+	return fmt("Broker Plugin - %s:%d - DPID: %d", state$broker_host, state$broker_port, state$broker_dpid);
+	}
+
+function broker_flow_mod_fun(state: ControllerState, match: ofp_match, flow_mod: OpenFlow::ofp_flow_mod): bool
+	{
+	event OpenFlow::broker_flow_mod(state$broker_dpid, match, flow_mod);
+
+	return T;
+	}
+
+function broker_flow_clear_fun(state: OpenFlow::ControllerState): bool
+	{
+	event OpenFlow::broker_flow_clear(state$broker_dpid);
+
+	return T;
+	}
+
+# broker controller constructor
+function broker_new(host: addr, host_port: port, dpid: count): OpenFlow::Controller
+	{
+	BrokerComm::enable();
+	BrokerComm::auto_event("bro/event/openflow", broker_flow_mod);
+	BrokerComm::auto_event("bro/event/openflow", broker_flow_clear);
+	BrokerComm::connect(cat(host), host_port, 1sec);
+
+	return [$state=[$broker_host=host, $broker_port=host_port, $broker_dpid=dpid, $_plugin=OpenFlow::BROKER],
+		$flow_mod=broker_flow_mod_fun, $flow_clear=broker_flow_clear_fun, $describe=broker_describe];
+	}
+

scripts/base/frameworks/openflow/plugins/log.bro

@@ -1,16 +1,18 @@
 ##! OpenFlow module that outputs flow-modification commands
 ##! to a Bro log file.
 
-module OpenFlow;
-
 @load base/frameworks/openflow
 @load base/frameworks/logging
 
+module OpenFlow;
+
 export {
 	redef enum Plugin += {
-		LOG,
+		OFLOG,
 	};
 
+	redef enum Log::ID += { LOG };
+
 	## Log controller constructor.
 	##
 	## dpid: OpenFlow switch datapath id.
@@ -42,12 +44,12 @@ export {
 
 event bro_init() &priority=5
 	{
-	Log::create_stream(LOG, [$columns=Info, $ev=log_openflow, $path="openflow"]);
+	Log::create_stream(OpenFlow::LOG, [$columns=Info, $ev=log_openflow, $path="openflow"]);
 	}
 
 function log_flow_mod(state: ControllerState, match: ofp_match, flow_mod: OpenFlow::ofp_flow_mod): bool
 	{
-	Log::write(LOG, [$ts=network_time(), $dpid=state$log_dpid, $match=match, $flow_mod=flow_mod]);
+	Log::write(OpenFlow::LOG, [$ts=network_time(), $dpid=state$log_dpid, $match=match, $flow_mod=flow_mod]);
 
 	return T;
 	}

scripts/base/frameworks/openflow/plugins/ryu.bro

@@ -149,7 +149,7 @@ function ryu_flow_mod(state: OpenFlow::ControllerState, match: ofp_match, flow_m
 	return T;
 	}
 
-function ryu_flow_clear(state: OpenFlow::ControllerState): bool	
+function ryu_flow_clear(state: OpenFlow::ControllerState): bool
 	{
 	local url=cat("http://", cat(state$ryu_host), ":", cat(state$ryu_port), RYU_FLOWENTRY_PATH, "clear", "/", state$ryu_dpid);
 

scripts/base/frameworks/pacf/plugins/openflow.bro

@@ -163,7 +163,7 @@ function openflow_rule_to_flow_mod(p: PluginState, r: Rule) : OpenFlow::ofp_flow
 	if ( r?$expire )
 		flow_mod$hard_timeout = double_to_count(interval_to_double(r$expire));
 	if ( c?$table_id )
-		flow_mod$table_id = c$table_id;	
+		flow_mod$table_id = c$table_id;
 
 	if ( r$ty == DROP )
 		{
@@ -181,7 +181,7 @@ function openflow_rule_to_flow_mod(p: PluginState, r: Rule) : OpenFlow::ofp_flow
 		}
 	else
 		{
-		Reporter::error(fmt("Rule type %s not supported for openflow yet", cat(r$ty)));		
+		Reporter::error(fmt("Rule type %s not supported for openflow yet", cat(r$ty)));
 		}
 
 	return openflow_flow_mod_pred(p, r, flow_mod);

testing/btest/Baseline/scripts.base.frameworks.openflow.broker-basic/recv.recv.out

@@ -0,0 +1,5 @@
+BrokerComm::incoming_connection_established
+flow_clear, 42
+got flow_mod, 42, [in_port=<uninitialized>, dl_src=<uninitialized>, dl_dst=<uninitialized>, dl_vlan=<uninitialized>, dl_vlan_pcp=<uninitialized>, dl_type=<uninitialized>, nw_tos=<uninitialized>, nw_proto=<uninitialized>, nw_src=<uninitialized>, nw_dst=<uninitialized>, tp_src=<uninitialized>, tp_dst=<uninitialized>], [cookie=1, table_id=<uninitialized>, command=OpenFlow::OFPFC_ADD, idle_timeout=0, hard_timeout=0, priority=0, out_port=<uninitialized>, out_group=<uninitialized>, flags=0, out_ports=[3, 7]]
+got flow_mod, 42, [in_port=<uninitialized>, dl_src=<uninitialized>, dl_dst=<uninitialized>, dl_vlan=<uninitialized>, dl_vlan_pcp=<uninitialized>, dl_type=2048, nw_tos=<uninitialized>, nw_proto=6, nw_src=10.10.1.4/32, nw_dst=74.53.140.153/32, tp_src=1470/tcp, tp_dst=25/tcp], [cookie=42, table_id=<uninitialized>, command=OpenFlow::OFPFC_ADD, idle_timeout=30, hard_timeout=0, priority=5, out_port=<uninitialized>, out_group=<uninitialized>, flags=0, out_ports=[]]
+got flow_mod, 42, [in_port=<uninitialized>, dl_src=<uninitialized>, dl_dst=<uninitialized>, dl_vlan=<uninitialized>, dl_vlan_pcp=<uninitialized>, dl_type=2048, nw_tos=<uninitialized>, nw_proto=6, nw_src=74.53.140.153/32, nw_dst=10.10.1.4/32, tp_src=25/tcp, tp_dst=25/tcp], [cookie=42, table_id=<uninitialized>, command=OpenFlow::OFPFC_ADD, idle_timeout=30, hard_timeout=0, priority=5, out_port=<uninitialized>, out_group=<uninitialized>, flags=0, out_ports=[]]

testing/btest/Baseline/scripts.base.frameworks.openflow.broker-basic/send.send.out

@@ -0,0 +1,2 @@
+BrokerComm::outgoing_connection_established, 127.0.0.1, 9999/tcp
+connection established

testing/btest/Baseline/scripts.base.frameworks.openflow.log-basic/openflow.log

@@ -3,10 +3,10 @@
 #empty_field	(empty)
 #unset_field	-
 #path	openflow
-#open	2015-04-14-21-58-29
+#open	2015-04-20-20-22-44
-#fields	ts	dpid	match.in_port	match.dl_src	match.dl_dst	match.dl_vlan	match.dl_vlan_pcp	match.dl_type	match.nw_tos	match.nw_proto	match.nw_src	match.nw_dst	match.tp_src	match.tp_dst	flow_mod.cookie	flow_mod.table_id	flow_mod.command	flow_mod.idle_timeout	flow_mod.hard_timeout	flow_mod.priority	flow_mod.out_group	flow_mod.flags	flow_mod.out_ports
+#fields	ts	dpid	match.in_port	match.dl_src	match.dl_dst	match.dl_vlan	match.dl_vlan_pcp	match.dl_type	match.nw_tos	match.nw_proto	match.nw_src	match.nw_dst	match.tp_src	match.tp_dst	flow_mod.cookie	flow_mod.table_id	flow_mod.command	flow_mod.idle_timeout	flow_mod.hard_timeout	flow_mod.priority	flow_mod.out_port	flow_mod.out_group	flow_mod.flags	flow_mod.out_ports
-#types	time	count	count	string	string	count	count	count	count	count	subnet	subnet	port	port	count	count	enum	count	count	count	count	count	vector[count]
+#types	time	count	count	string	string	count	count	count	count	count	subnet	subnet	port	port	count	count	enum	count	count	count	count	count	count	vector[count]
-0.000000	42	-	-	-	-	-	-	-	-	-	-	-	-	1	-	OpenFlow::OFPFC_ADD	0	0	0	-	0	3,7
+0.000000	42	-	-	-	-	-	-	-	-	-	-	-	-	1	-	OpenFlow::OFPFC_ADD	0	0	0	-	-	0	3,7
-1254722767.875996	42	-	-	-	-	-	2048	-	6	10.10.1.4/32	74.53.140.153/32	1470	25	42	-	OpenFlow::OFPFC_ADD	30	0	5	-	0	(empty)
+1254722767.875996	42	-	-	-	-	-	2048	-	6	10.10.1.4/32	74.53.140.153/32	1470	25	42	-	OpenFlow::OFPFC_ADD	30	0	5	-	-	0	(empty)
-1254722767.875996	42	-	-	-	-	-	2048	-	6	74.53.140.153/32	10.10.1.4/32	25	25	42	-	OpenFlow::OFPFC_ADD	30	0	5	-	0	(empty)
+1254722767.875996	42	-	-	-	-	-	2048	-	6	74.53.140.153/32	10.10.1.4/32	25	25	42	-	OpenFlow::OFPFC_ADD	30	0	5	-	-	0	(empty)
-#close	2015-04-14-21-58-29
+#close	2015-04-20-20-22-44

testing/btest/scripts/base/frameworks/openflow/broker-basic.bro

@@ -0,0 +1,106 @@
+# @TEST-SERIALIZE: brokercomm
+# @TEST-REQUIRES: grep -q ENABLE_BROKER $BUILD/CMakeCache.txt
+# @TEST-EXEC: btest-bg-run recv "bro -b ../recv.bro broker_port=$BROKER_PORT >recv.out"
+# @TEST-EXEC: btest-bg-run send "bro -b -r $TRACES/smtp.trace --pseudo-realtime ../send.bro broker_port=$BROKER_PORT >send.out"
+
+# @TEST-EXEC: btest-bg-wait 20
+# @TEST-EXEC: btest-diff recv/recv.out
+# @TEST-EXEC: btest-diff send/send.out
+
+@TEST-START-FILE send.bro
+
+@load base/protocols/conn
+@load base/frameworks/openflow
+
+const broker_port: port &redef;
+redef exit_only_after_terminate = T;
+
+global of_controller: OpenFlow::Controller;
+
+event bro_init()
+	{
+	suspend_processing();
+	of_controller = OpenFlow::broker_new(127.0.0.1, broker_port, 42);
+	}
+
+event BrokerComm::outgoing_connection_established(peer_address: string,
+                                            peer_port: port,
+                                            peer_name: string)
+	{
+	print "BrokerComm::outgoing_connection_established", peer_address, peer_port;
+	continue_processing();
+	OpenFlow::flow_clear(of_controller);
+	OpenFlow::flow_mod(of_controller, [], [$cookie=1, $command=OpenFlow::OFPFC_ADD, $out_ports=vector(3, 7)]);
+	}
+
+event BrokerComm::outgoing_connection_broken(peer_address: string,
+                                       peer_port: port)
+	{
+	terminate();
+	}
+
+event connection_established(c: connection)
+	{
+	print "connection established";
+	local match = OpenFlow::match_conn(c$id);
+	local match_rev = OpenFlow::match_conn(c$id, T);
+
+	local flow_mod: OpenFlow::ofp_flow_mod = [
+		$cookie=42,
+		$command=OpenFlow::OFPFC_ADD,
+		$idle_timeout=30,
+		$priority=5
+	];
+
+	OpenFlow::flow_mod(of_controller, match, flow_mod);
+	OpenFlow::flow_mod(of_controller, match_rev, flow_mod);
+	}
+
+
+@TEST-END-FILE
+
+@TEST-START-FILE recv.bro
+
+@load base/frameworks/openflow
+
+const broker_port: port &redef;
+redef exit_only_after_terminate = T;
+
+global msg_count: count = 0;
+
+event bro_init()
+	{
+	BrokerComm::enable();
+	BrokerComm::subscribe_to_events("bro/event/openflow");
+	BrokerComm::subscribe_to_events("bro/event/test_event");
+	BrokerComm::listen(broker_port, "127.0.0.1");
+	}
+
+event BrokerComm::incoming_connection_established(peer_name: string)
+	{
+	print "BrokerComm::incoming_connection_established";
+	}
+
+function got_message()
+	{
+	++msg_count;
+
+	if ( msg_count >= 4 )
+		terminate();
+	}
+
+event OpenFlow::broker_flow_mod(dpid: count, match: OpenFlow::ofp_match, flow_mod: OpenFlow::ofp_flow_mod)
+	{
+	print "got flow_mod", dpid, match, flow_mod;
+	got_message();
+	}
+
+event OpenFlow::broker_flow_clear(dpid: count)
+	{
+	print "flow_clear", dpid;
+	got_message();
+	}
+
+
+@TEST-END-FILE
+